Реализация RAG (Retrieval-Augmented Generation) для AI-бота на сайте

Реализация RAG (Retrieval-Augmented Generation) для AI-бота на сайте

RAG — архитектурный паттерн, при котором LLM не генерирует ответ из обучающих данных, а получает релевантные фрагменты из вашей базы знаний и формирует ответ на их основе. Результат: бот знает ваши продукты, политики и документацию, не галлюцинирует данные компании, ответы поддаются аудиту.

Компоненты RAG-системы

Knowledge base — источник данных: документация, FAQ, статьи базы знаний, страницы сайта, PDF-файлы, тикеты поддержки.

Ingestion pipeline — процесс загрузки, разбивки на чанки и индексации документов.

Vector store — база данных, хранящая эмбеддинги и обеспечивающая семантический поиск.

Retrieval — по запросу пользователя находим топ-N релевантных чанков.

Generation — отправляем найденные чанки + вопрос в LLM, получаем ответ.

Ingestion Pipeline

Разбивка документов на чанки — критичный этап. Слишком маленькие чанки теряют контекст, слишком большие — снижают точность поиска. Оптимально: 500–1000 токенов с перекрытием 100–200 токенов.

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import (
    WebBaseLoader, PyPDFLoader, UnstructuredMarkdownLoader
)

def load_and_chunk_documents(sources: list[dict]) -> list:
    documents = []

    for source in sources:
        if source["type"] == "url":
            loader = WebBaseLoader(source["path"])
        elif source["type"] == "pdf":
            loader = PyPDFLoader(source["path"])
        elif source["type"] == "markdown":
            loader = UnstructuredMarkdownLoader(source["path"])

        docs = loader.load()
        documents.extend(docs)

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=800,
        chunk_overlap=150,
        separators=["\n\n", "\n", ". ", " ", ""]
    )

    return splitter.split_documents(documents)

Эмбеддинги и векторное хранилище

Модели эмбеддингов:

• text-embedding-3-small (OpenAI) — 1536 измерений, $0.02 за 1M токенов, отличное соотношение цены и качества
• text-embedding-3-large — 3072 измерений, лучше для сложных запросов
• multilingual-e5-large (локально, Hugging Face) — бесплатно, хорошо для русского языка

Векторные хранилища:

Для сайта со средней нагрузкой и базой до 100 000 документов — pgvector или Qdrant. Не нужно поднимать отдельный сервис.

import psycopg2
from pgvector.psycopg2 import register_vector
import numpy as np

def store_embeddings(chunks: list, embeddings: list[list[float]]):
    conn = psycopg2.connect(DATABASE_URL)
    register_vector(conn)
    cur = conn.cursor()

    cur.execute("""
        CREATE TABLE IF NOT EXISTS documents (
            id SERIAL PRIMARY KEY,
            content TEXT,
            embedding vector(1536),
            metadata JSONB,
            source_url TEXT,
            created_at TIMESTAMP DEFAULT NOW()
        )
    """)
    cur.execute("CREATE INDEX IF NOT EXISTS documents_embedding_idx ON documents USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100)")

    for chunk, embedding in zip(chunks, embeddings):
        cur.execute(
            "INSERT INTO documents (content, embedding, metadata, source_url) VALUES (%s, %s, %s, %s)",
            (chunk.page_content, np.array(embedding), json.dumps(chunk.metadata), chunk.metadata.get("source", ""))
        )

    conn.commit()

Retrieval: семантический поиск

def retrieve_relevant_chunks(query: str, top_k: int = 5, threshold: float = 0.75) -> list[dict]:
    query_embedding = get_embedding(query)

    conn = psycopg2.connect(DATABASE_URL)
    register_vector(conn)
    cur = conn.cursor()

    cur.execute("""
        SELECT content, source_url, metadata,
               1 - (embedding <=> %s::vector) AS similarity
        FROM documents
        WHERE 1 - (embedding <=> %s::vector) > %s
        ORDER BY embedding <=> %s::vector
        LIMIT %s
    """, (query_embedding, query_embedding, threshold, query_embedding, top_k))

    results = cur.fetchall()
    return [
        {"content": r[0], "source": r[1], "metadata": r[2], "similarity": float(r[3])}
        for r in results
    ]

Гибридный поиск

Семантический поиск иногда промахивается по точным запросам — артикулам, именам, специфическим терминам. Гибридный поиск комбинирует векторный и полнотекстовый (BM25):

def hybrid_search(query: str, top_k: int = 5) -> list[dict]:
    # Семантический поиск
    semantic_results = retrieve_relevant_chunks(query, top_k=top_k * 2)

    # Полнотекстовый поиск через tsvector
    cur.execute("""
        SELECT content, source_url, ts_rank(to_tsvector('russian', content), query) AS rank
        FROM documents, to_tsquery('russian', %s) query
        WHERE to_tsvector('russian', content) @@ query
        ORDER BY rank DESC LIMIT %s
    """, (prepare_ts_query(query), top_k * 2))
    keyword_results = [{"content": r[0], "source": r[1], "rank": r[2]} for r in cur.fetchall()]

    # Reciprocal Rank Fusion
    return reciprocal_rank_fusion(semantic_results, keyword_results, top_k)

Generation: формирование ответа

from openai import OpenAI

client = OpenAI()

SYSTEM_PROMPT = """Ты помощник службы поддержки компании.
Отвечай ТОЛЬКО на основе предоставленного контекста.
Если ответа нет в контексте — честно скажи об этом.
Не придумывай информацию. Указывай источник из контекста."""

def generate_answer(query: str, context_chunks: list[dict]) -> dict:
    context = "\n\n".join([
        f"[Источник: {c['source']}]\n{c['content']}"
        for c in context_chunks
    ])

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Контекст:\n{context}\n\nВопрос: {query}"}
        ],
        temperature=0.1,
        max_tokens=800
    )

    sources = list({c["source"] for c in context_chunks if c.get("source")})

    return {
        "answer": response.choices[0].message.content,
        "sources": sources,
        "chunks_used": len(context_chunks)
    }

Re-ranking

Векторный поиск возвращает кандидатов по косинусному сходству, но не всегда самый семантически близкий чанк — самый полезный. Cross-encoder re-ranking переоценивает кандидатов с учётом вопроса:

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def rerank_chunks(query: str, chunks: list[dict], top_k: int = 3) -> list[dict]:
    pairs = [(query, c["content"]) for c in chunks]
    scores = reranker.predict(pairs)

    ranked = sorted(zip(chunks, scores), key=lambda x: x[1], reverse=True)
    return [chunk for chunk, _ in ranked[:top_k]]

Обновление индекса

При изменении контента на сайте нужно пересчитать эмбеддинги. Стратегии:

Полная переиндексация — раз в сутки, при объёме до 50 000 документов занимает 15–30 минут.

Инкрементальная — при изменении страницы удаляем старые чанки по source_url, добавляем новые. Подходит для CMS с webhook на публикацию.

Мягкое удаление — помечаем устаревшие чанки флагом, не удаляем немедленно. Позволяет откатиться при ошибке.

Оценка качества

Метрики RAG-системы:

• Faithfulness — не противоречит ли ответ контексту
• Answer Relevance — отвечает ли на вопрос
• Context Recall — все ли нужные факты найдены
• Context Precision — нет ли лишнего в контексте

Инструменты оценки: RAGAS (open-source), LangSmith (платный SaaS), DeepEval.

Сроки реализации

Минимально рабочий RAG-бот с одним источником данных — 2 недели. Продуктовая система с несколькими источниками, гибридным поиском и мониторингом — 4–5 недель.