RAG-пайплайны от А до Я: архитектура, продвинутые техники и практическая реализация на Python

Введение: почему RAG стал стандартом де-факто в 2025–2026 годах

Retrieval-Augmented Generation (RAG) — пожалуй, самый обсуждаемый паттерн в мире LLM-приложений. И это неспроста. Большие языковые модели впечатляют своими знаниями, но у них есть три фундаментальных ограничения: данные обучения устаревают, модель склонна «галлюцинировать» факты, и — что особенно болезненно для бизнеса — она понятия не имеет о ваших корпоративных данных. RAG решает все три проблемы разом, подставляя релевантный контекст из внешних источников прямо в промпт.

Но вот в чём подвох.

Между демонстрационным RAG-приложением, которое можно собрать за 20 минут, и production-ready системой, обрабатывающей тысячи запросов в день, лежит настоящая пропасть. Naive RAG — просто разбить документ на чанки, положить в векторную базу и отправить результат поиска в LLM — работает для прототипов. Но стоит выйти в реальный мир, и вы столкнётесь с проблемами качества ответов, скорости, масштабирования и стоимости. Знаю по собственному опыту: первый «боевой» RAG-пайплайн, который я запускал, красиво работал на демо-данных и ужасно — на реальных документах клиента.

В этом руководстве мы пройдём весь путь: от базовой архитектуры до Advanced RAG, Agentic RAG и GraphRAG. Каждую концепцию подкрепим практическим кодом на Python. Если вы уже знакомы с Function Calling и Tool Use в LLM-агентах, считайте эту статью логическим продолжением — RAG-пайплайн часто выступает именно как инструмент, который агент вызывает для получения информации.

Архитектура RAG: три поколения

Naive RAG: основа, которую нужно перерасти

Классический RAG-пайплайн состоит из трёх этапов: индексация (Indexing), извлечение (Retrieval) и генерация (Generation). Звучит просто, и так оно и есть — в этом одновременно и плюс, и минус.

На этапе индексации документы разбиваются на чанки фиксированного размера (например, 512 токенов), для каждого чанка генерируется векторное представление (embedding), и результат сохраняется в векторной базе данных. Затем пользовательский запрос тоже превращается в вектор, выполняется поиск ближайших соседей (approximate nearest neighbor, ANN), найденные чанки подставляются в промпт, и LLM генерирует финальный ответ.

Честно говоря, для демо этого достаточно. Но в продакшене проблемы посыпятся одна за другой:

• Потеря контекста при чанкинге — разрезая документ на куски фиксированного размера, вы неизбежно рвёте логические связи
• Semantic gap — вопрос пользователя и релевантный чанк могут быть сформулированы совершенно по-разному
• Нет верификации — модель генерирует ответ на основе первых K результатов, даже не проверяя их релевантность
• Одноходовость — один запрос, один поиск, один ответ. Без возможности уточнить или переспросить

Advanced RAG: оптимизация на каждом этапе

Advanced RAG добавляет оптимизации до извлечения (pre-retrieval), во время извлечения и после (post-retrieval). Это не одна конкретная техника, а целый набор улучшений, которые можно комбинировать между собой:

• Pre-retrieval: трансформация запроса, HyDE (Hypothetical Document Embeddings), расширение запроса, классификация намерения
• Retrieval: гибридный поиск (BM25 + векторный), re-ranking, multi-query retrieval
• Post-retrieval: сжатие контекста, фильтрация, суммаризация извлечённых документов

Agentic RAG: интеллектуальная оркестрация

А вот это уже по-настоящему интересно. Agentic RAG — следующий эволюционный шаг, который к 2026 году стал рекомендуемым подходом для серьёзных AI-приложений. Вместо линейного пайплайна мы получаем автономного агента, способного:

• Декомпозировать сложный вопрос на подзадачи
• Выбирать стратегию извлечения для каждого подзапроса
• Оценивать качество полученных документов и при необходимости повторять поиск
• Проверять сгенерированный ответ на галлюцинации
• Использовать несколько источников данных — векторная база, граф знаний, SQL, внешние API

По сути, Agentic RAG объединяет паттерны Function Calling (мы разбирали их в предыдущей статье) с продвинутым извлечением информации. Агент сам «решает», когда и как искать, а не просто механически выполняет поиск по шаблону.

Фундамент: векторные базы данных и эмбеддинги

Выбор модели эмбеддингов

Качество RAG начинается с качества эмбеддингов — тут без вариантов. В 2026 году ландшафт моделей эмбеддингов существенно расширился, так что выбирать есть из чего. Вот актуальные рекомендации:

• OpenAI text-embedding-3-large — 3072 измерения, отличный баланс качества и стоимости. Классная фишка — поддержка сокращения размерности (можно задать dimensions=1536 или даже dimensions=256 для экономии)
• Cohere embed-v4 — нативная поддержка мультимодальности и сжатия. Отлично работает с многоязычным контентом
• Открытые модели (BGE, E5, GTE) — можно запускать локально, не отправляя данные на внешние серверы. Критично для корпоративных сценариев с чувствительными данными

Практический совет: всегда тестируйте модель эмбеддингов на вашем конкретном домене. Модель, лидирующая в бенчмарках MTEB, запросто может уступать на ваших данных какой-нибудь более нишевой модели. Я видел это не раз.

Выбор векторной базы данных

Здесь выбор зависит от масштаба и требований:

• Chroma — лёгкая, in-process, идеальна для прототипов и малых приложений
• Qdrant — написана на Rust, высокопроизводительная, с отличной поддержкой фильтрации по метаданным
• Weaviate — встроенный гибридный поиск, GraphQL API
• Pinecone — полностью управляемый сервис, минимум DevOps
• pgvector — расширение для PostgreSQL. Если вы уже используете PostgreSQL и не хотите вводить отдельный сервис, это отличный вариант

Для production-систем в 2026 году Qdrant и pgvector — наиболее популярный выбор. Qdrant берут, когда нужна максимальная производительность векторного поиска. pgvector — когда важнее единая база данных и транзакционность.

Практическая реализация базового RAG

Давайте начнём с минимально работающего примера, чтобы увидеть полный цикл. Используем OpenAI для эмбеддингов и генерации, а ChromaDB — как векторное хранилище:

import chromadb
from openai import OpenAI
from typing import list

client = OpenAI()
chroma_client = chromadb.PersistentClient(path="./chroma_db")

# Создаём или получаем коллекцию
collection = chroma_client.get_or_create_collection(
    name="knowledge_base",
    metadata={"hnsw:space": "cosine"}
)


def get_embedding(text: str) -> list[float]:
    """Получение эмбеддинга через OpenAI API."""
    response = client.embeddings.create(
        model="text-embedding-3-large",
        input=text,
        dimensions=1536  # Сокращаем для экономии
    )
    return response.data[0].embedding


def index_documents(documents: list[dict]):
    """Индексация документов в ChromaDB."""
    for i, doc in enumerate(documents):
        embedding = get_embedding(doc["content"])
        collection.add(
            ids=[f"doc_{i}"],
            embeddings=[embedding],
            documents=[doc["content"]],
            metadatas=[{
                "source": doc.get("source", ""),
                "title": doc.get("title", ""),
                "category": doc.get("category", "")
            }]
        )


def retrieve(query: str, n_results: int = 5) -> list[str]:
    """Поиск релевантных документов."""
    query_embedding = get_embedding(query)
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=n_results
    )
    return results["documents"][0]


def generate_answer(query: str, context_docs: list[str]) -> str:
    """Генерация ответа с контекстом."""
    context = "\n\n---\n\n".join(context_docs)

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "Вы — экспертный ассистент. Отвечайте на вопросы "
                    "ТОЛЬКО на основе предоставленного контекста. "
                    "Если контекст не содержит ответа, так и скажите."
                )
            },
            {
                "role": "user",
                "content": f"Контекст:\n{context}\n\nВопрос: {query}"
            }
        ],
        temperature=0.1
    )
    return response.choices[0].message.content


# Пример использования
query = "Как настроить мониторинг RAG-пайплайна?"
docs = retrieve(query)
answer = generate_answer(query, docs)
print(answer)

Это работает, но это именно Naive RAG. Теперь давайте превратим его в нечто более серьёзное.

Стратегии чанкинга: от наивного к семантическому

Почему размер чанка — это не просто число

Чанкинг — одна из самых недооценённых частей RAG-пайплайна. И я говорю это без преувеличения. Неправильный чанкинг может убить качество даже при идеальных эмбеддингах и самой мощной LLM.

Ключевой компромисс тут такой: маленькие чанки (100–200 токенов) дают более точное попадание в поиске, но теряют контекст. Большие чанки (1000–2000 токенов) сохраняют контекст, но снижают точность поиска и рискуют «разбавить» релевантную информацию нерелевантной. Золотая середина? Зависит от вашего домена (да, я знаю, скучный ответ, но это правда).

Семантический чанкинг

Вместо разрезания по фиксированному количеству токенов мы определяем границы чанков по смысловым переходам. Идея на удивление проста: вычисляем эмбеддинги для последовательных предложений и ищем точки, где косинусное сходство резко падает — это и есть смена темы:

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity


def semantic_chunking(
    sentences: list[str],
    threshold: float = 0.75,
    min_chunk_size: int = 2,
    max_chunk_size: int = 15
) -> list[str]:
    """Семантический чанкинг на основе сходства эмбеддингов."""
    # Получаем эмбеддинги для каждого предложения
    embeddings = [get_embedding(s) for s in sentences]

    chunks = []
    current_chunk = [sentences[0]]

    for i in range(1, len(sentences)):
        similarity = cosine_similarity(
            [embeddings[i-1]], [embeddings[i]]
        )[0][0]

        if (similarity < threshold
                and len(current_chunk) >= min_chunk_size):
            chunks.append(" ".join(current_chunk))
            current_chunk = [sentences[i]]
        elif len(current_chunk) >= max_chunk_size:
            chunks.append(" ".join(current_chunk))
            current_chunk = [sentences[i]]
        else:
            current_chunk.append(sentences[i])

    if current_chunk:
        chunks.append(" ".join(current_chunk))

    return chunks

Контекстные заголовки (Contextual Headers)

Мощная и, честно говоря, недооценённая техника. Суть простая: перед каждым чанком добавляется иерархический контекст — название документа, заголовок раздела, подзаголовок. Это даёт эмбеддингу куда больше информации о том, о чём конкретно этот чанк:

def add_contextual_header(
    chunk: str,
    doc_title: str,
    section: str,
    subsection: str = ""
) -> str:
    """Добавление контекстного заголовка к чанку."""
    header_parts = [f"Документ: {doc_title}", f"Раздел: {section}"]
    if subsection:
        header_parts.append(f"Подраздел: {subsection}")

    header = " | ".join(header_parts)
    return f"{header}\n\n{chunk}"


# Пример
enriched_chunk = add_contextual_header(
    chunk="Для настройки алертов используйте Prometheus Alertmanager...",
    doc_title="Руководство по мониторингу",
    section="Алертинг",
    subsection="Настройка правил"
)
# Результат:
# "Документ: Руководство по мониторингу | Раздел: Алертинг |
#  Подраздел: Настройка правил
#
#  Для настройки алертов используйте Prometheus Alertmanager..."

Parent-Child чанкинг

Ещё одна продвинутая стратегия, которая мне особенно нравится. Идея: для поиска используются маленькие (child) чанки — они дают точное попадание. Но в контекст LLM подставляются большие (parent) чанки, которые сохраняют полный контекст. Лучшее из обоих миров:

from dataclasses import dataclass


@dataclass
class ChunkPair:
    child: str  # Маленький чанк для поиска
    parent: str  # Большой чанк для контекста
    parent_id: str


def create_parent_child_chunks(
    text: str,
    parent_size: int = 2000,
    child_size: int = 400,
    overlap: int = 50
) -> list[ChunkPair]:
    """Создание пар parent-child чанков."""
    words = text.split()
    pairs = []
    parent_id = 0

    for i in range(0, len(words), parent_size):
        parent_chunk = " ".join(words[i:i + parent_size])
        parent_words = parent_chunk.split()

        for j in range(0, len(parent_words), child_size - overlap):
            child_chunk = " ".join(
                parent_words[j:j + child_size]
            )
            if len(child_chunk.split()) > 20:
                pairs.append(ChunkPair(
                    child=child_chunk,
                    parent=parent_chunk,
                    parent_id=f"parent_{parent_id}"
                ))

        parent_id += 1

    return pairs

Гибридный поиск: объединяем лучшее из двух миров

Зачем нужен гибридный поиск

Чисто векторный поиск плохо справляется с точными совпадениями: артикулами, кодами продуктов, именами собственными, аббревиатурами. Классический BM25 (keyword search) прекрасно работает с такими запросами, но не понимает семантику.

Гибридный поиск объединяет оба подхода через Reciprocal Rank Fusion (RRF). И знаете что? В 2026 году гибридный поиск стал фактически стандартной рекомендацией. Если вы его не используете — скорее всего, вы теряете качество.

Реализация гибридного поиска

from rank_bm25 import BM25Okapi
import numpy as np


class HybridRetriever:
    """Гибридный поиск: BM25 + векторный с RRF."""

    def __init__(self, documents: list[dict], collection):
        self.documents = documents
        self.collection = collection

        # Подготовка BM25 индекса
        tokenized = [
            doc["content"].lower().split()
            for doc in documents
        ]
        self.bm25 = BM25Okapi(tokenized)

    def reciprocal_rank_fusion(
        self,
        rankings: list[list[int]],
        k: int = 60
    ) -> list[tuple[int, float]]:
        """Объединение ранжирований через RRF."""
        scores = {}
        for ranking in rankings:
            for rank, doc_id in enumerate(ranking):
                if doc_id not in scores:
                    scores[doc_id] = 0
                scores[doc_id] += 1 / (k + rank + 1)

        return sorted(
            scores.items(), key=lambda x: x[1], reverse=True
        )

    def search(
        self,
        query: str,
        top_k: int = 5,
        vector_weight: float = 0.6,
        bm25_weight: float = 0.4
    ) -> list[dict]:
        """Гибридный поиск с настраиваемыми весами."""
        # BM25 поиск
        bm25_scores = self.bm25.get_scores(
            query.lower().split()
        )
        bm25_ranking = np.argsort(bm25_scores)[::-1][:top_k * 2]

        # Векторный поиск
        query_embedding = get_embedding(query)
        vector_results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k * 2
        )
        vector_ranking = [
            int(id.split("_")[1])
            for id in vector_results["ids"][0]
        ]

        # RRF слияние
        fused = self.reciprocal_rank_fusion(
            [list(bm25_ranking), vector_ranking]
        )

        # Возвращаем top_k результатов
        results = []
        for doc_id, score in fused[:top_k]:
            results.append({
                "content": self.documents[doc_id]["content"],
                "score": score,
                "metadata": self.documents[doc_id].get(
                    "metadata", {}
                )
            })

        return results

Re-ranking: второй этап фильтрации

Зачем нужен re-ranker

Первый этап поиска (BM25, векторный или гибридный) — это, по сути, грубая фильтрация. Мы берём, скажем, топ-20 кандидатов из миллионов документов. Но ранжирование среди этих 20 может быть неточным. Вот тут на сцену выходит re-ranker — вторая модель, которая внимательнее анализирует пару «запрос + документ» и выдаёт более точную оценку релевантности.

Ключевое отличие: bi-encoder (эмбеддинг-модель) кодирует запрос и документ отдельно. А cross-encoder (re-ranker) обрабатывает их совместно, что позволяет уловить тонкие связи между ними. Разница в качестве бывает существенной.

Реализация с Cohere Rerank

import cohere


def rerank_results(
    query: str,
    documents: list[str],
    top_n: int = 5
) -> list[dict]:
    """Re-ranking через Cohere API."""
    co = cohere.Client()

    response = co.rerank(
        model="rerank-v3.5",
        query=query,
        documents=documents,
        top_n=top_n,
        return_documents=True
    )

    return [
        {
            "text": r.document.text,
            "relevance_score": r.relevance_score,
            "index": r.index
        }
        for r in response.results
    ]


# Интеграция в пайплайн
def rag_with_reranking(query: str) -> str:
    """RAG-пайплайн с re-ranking."""
    # Шаг 1: грубый поиск (больше кандидатов)
    candidates = retrieve(query, n_results=20)

    # Шаг 2: re-ranking (оставляем лучшие)
    reranked = rerank_results(query, candidates, top_n=5)

    # Шаг 3: генерация с отфильтрованным контекстом
    context_docs = [r["text"] for r in reranked]
    return generate_answer(query, context_docs)

Трансформация запросов: HyDE, Multi-Query и Step-Back

HyDE — Hypothetical Document Embeddings

HyDE — элегантная техника для преодоления семантического разрыва между вопросом и документом. Идея такая: мы просим LLM сгенерировать гипотетический ответ на вопрос, а затем ищем документы, похожие на этот гипотетический ответ (а не на исходный вопрос). Звучит контринтуитивно? Возможно. Но работает на удивление хорошо:

def hyde_retrieval(query: str, n_results: int = 5) -> list[str]:
    """Поиск с использованием HyDE."""
    # Генерируем гипотетический ответ
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "Напишите подробный параграф, отвечающий "
                    "на вопрос пользователя. Отвечайте так, "
                    "как будто это фрагмент документации."
                )
            },
            {"role": "user", "content": query}
        ],
        temperature=0.7
    )
    hypothetical_doc = response.choices[0].message.content

    # Ищем документы, похожие на гипотетический ответ
    hyp_embedding = get_embedding(hypothetical_doc)
    results = collection.query(
        query_embeddings=[hyp_embedding],
        n_results=n_results
    )
    return results["documents"][0]

Multi-Query Retrieval

Вместо одного запроса генерируем несколько вариаций, выполняем поиск по каждой и объединяем результаты. Это особенно полезно, когда запрос пользователя можно интерпретировать по-разному:

def multi_query_retrieval(
    query: str,
    n_variants: int = 3,
    n_results: int = 5
) -> list[str]:
    """Мульти-запросный поиск."""
    # Генерируем вариации запроса
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    f"Сгенерируйте {n_variants} различных "
                    "формулировки данного вопроса. Каждая "
                    "формулировка должна подходить к теме с "
                    "разных сторон. Верните только список "
                    "вопросов, по одному на строку."
                )
            },
            {"role": "user", "content": query}
        ]
    )

    variants = response.choices[0].message.content.strip().split("\n")
    all_queries = [query] + variants

    # Собираем результаты со всех вариаций
    all_docs = set()
    for q in all_queries:
        docs = retrieve(q, n_results=n_results)
        all_docs.update(docs)

    return list(all_docs)[:n_results * 2]

GraphRAG: когда нужны связи между сущностями

Когда векторного поиска недостаточно

Векторный поиск отлично работает для вопросов типа «Что такое X?» или «Как сделать Y?». Но попробуйте задать что-нибудь вроде: «Какие компании из портфеля фонда Z работают в области генеративного ИИ и были основаны после 2020 года?». Тут нужны структурированные связи между сущностями: фонды → портфельные компании → области деятельности → даты основания. Обычный векторный поиск с этим просто не справится.

GraphRAG строит граф знаний (knowledge graph) поверх вашего корпуса и использует его для извлечения информации. По данным на 2026 год, GraphRAG обходит чисто векторные подходы для задач, требующих сложного рассуждения, — обеспечивая более высокую точность при меньших затратах.

Архитектура GraphRAG с Neo4j

Типичный GraphRAG-пайплайн работает в четыре шага:

1. Извлечение сущностей и связей — LLM анализирует документы и выделяет сущности (люди, компании, технологии) и их взаимосвязи
2. Построение графа — всё это сохраняется в графовой базе данных (Neo4j)
3. Запрос — пользовательский вопрос преобразуется в Cypher-запрос или комбинируется с векторным поиском
4. Генерация — LLM формирует ответ на основе данных из графа

from neo4j import GraphDatabase
from neo4j_graphrag.retrievers import HybridCypherRetriever
from neo4j_graphrag.llm import OpenAILLM
from neo4j_graphrag.embeddings import OpenAIEmbeddings
from neo4j_graphrag.generation import GraphRAG


# Подключение к Neo4j
driver = GraphDatabase.driver(
    "bolt://localhost:7687",
    auth=("neo4j", "password")
)

# Настройка компонентов
llm = OpenAILLM(model_name="gpt-4o")
embedder = OpenAIEmbeddings(model="text-embedding-3-large")

# Гибридный ретривер: Cypher + векторный поиск
retriever = HybridCypherRetriever(
    driver=driver,
    vector_index_name="document_embeddings",
    fulltext_index_name="document_fulltext",
    embedder=embedder,
    # Cypher для обогащения результатов связями из графа
    retrieval_query="""
    MATCH (node)-[r]->(related)
    RETURN node.text AS text,
           score,
           collect(type(r) + ": " + related.name) AS relations
    """,
    top_k=10
)

# Создаём GraphRAG пайплайн
rag = GraphRAG(
    retriever=retriever,
    llm=llm
)

# Запрос
result = rag.search(
    query_text="Какие технологии связаны с компанией X?",
    retriever_config={"top_k": 5}
)
print(result.answer)

Извлечение сущностей для построения графа

Ключевой этап — автоматическое извлечение сущностей и связей из документов. Тут нам снова поможет LLM:

import json


def extract_entities_and_relations(text: str) -> dict:
    """Извлечение сущностей и связей из текста."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": """Извлеките сущности и связи
из текста. Верните JSON с двумя массивами:
- entities: [{name, type, description}]
- relations: [{source, target, type, description}]

Типы сущностей: Person, Company, Technology,
Product, Concept
Типы связей: WORKS_AT, FOUNDED, USES, PART_OF,
RELATED_TO, COMPETES_WITH"""
            },
            {"role": "user", "content": text}
        ],
        response_format={"type": "json_object"},
        temperature=0
    )
    return json.loads(response.choices[0].message.content)


def populate_knowledge_graph(
    driver,
    entities: list[dict],
    relations: list[dict]
):
    """Заполнение графа знаний в Neo4j."""
    with driver.session() as session:
        # Создаём сущности
        for entity in entities:
            session.run(
                """
                MERGE (e:Entity {name: $name})
                SET e.type = $type,
                    e.description = $description
                """,
                name=entity["name"],
                type=entity["type"],
                description=entity.get("description", "")
            )

        # Создаём связи
        for rel in relations:
            session.run(
                """
                MATCH (a:Entity {name: $source})
                MATCH (b:Entity {name: $target})
                MERGE (a)-[r:RELATED {type: $type}]->(b)
                SET r.description = $description
                """,
                source=rel["source"],
                target=rel["target"],
                type=rel["type"],
                description=rel.get("description", "")
            )

Agentic RAG на LangGraph: полная реализация

Архитектура агентного RAG

Итак, пришло время собрать всё воедино. Объединим все описанные техники в интеллектуального агента, используя LangGraph — библиотеку для построения stateful-графов с LLM. Наш агент будет:

1. Классифицировать входящий запрос
2. Решать, нужно ли извлечение (или модель может ответить сама)
3. Выполнять гибридный поиск
4. Оценивать релевантность найденных документов
5. При необходимости переформулировать запрос и повторить поиск
6. Генерировать ответ и проверять его на галлюцинации

Выглядит как много шагов, но каждый из них критически важен для качества:

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator


class RAGState(TypedDict):
    """Состояние агента."""
    question: str
    documents: list[str]
    generation: str
    search_count: int
    is_relevant: bool
    needs_web_search: bool


def route_question(state: RAGState) -> str:
    """Маршрутизация: нужно ли извлечение?"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "Определите, требует ли вопрос поиска "
                    "в базе знаний. Ответьте 'retrieve' "
                    "или 'direct'."
                )
            },
            {"role": "user", "content": state["question"]}
        ],
        temperature=0
    )
    decision = response.choices[0].message.content.strip()
    return "retrieve" if "retrieve" in decision.lower() else "direct"


def retrieve_documents(state: RAGState) -> RAGState:
    """Извлечение документов с гибридным поиском."""
    docs = hybrid_retriever.search(
        state["question"], top_k=10
    )
    # Re-ranking
    reranked = rerank_results(
        state["question"],
        [d["content"] for d in docs],
        top_n=5
    )
    state["documents"] = [r["text"] for r in reranked]
    state["search_count"] = state.get("search_count", 0) + 1
    return state


def grade_documents(state: RAGState) -> RAGState:
    """Оценка релевантности найденных документов."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "Оцените, содержат ли документы "
                    "информацию для ответа на вопрос. "
                    "Ответьте 'relevant' или 'not_relevant'."
                )
            },
            {
                "role": "user",
                "content": (
                    f"Вопрос: {state['question']}\n\n"
                    f"Документы: {state['documents']}"
                )
            }
        ],
        temperature=0
    )
    decision = response.choices[0].message.content.strip()
    state["is_relevant"] = "relevant" in decision.lower()
    return state


def transform_query(state: RAGState) -> RAGState:
    """Переформулирование запроса."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "Переформулируйте вопрос для более "
                    "точного поиска. Верните только "
                    "переформулированный вопрос."
                )
            },
            {"role": "user", "content": state["question"]}
        ]
    )
    state["question"] = (
        response.choices[0].message.content.strip()
    )
    return state


def generate(state: RAGState) -> RAGState:
    """Генерация финального ответа."""
    state["generation"] = generate_answer(
        state["question"],
        state["documents"]
    )
    return state


def check_hallucination(state: RAGState) -> str:
    """Проверка ответа на галлюцинации."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "Проверьте, основан ли ответ на "
                    "предоставленных документах. Ответьте "
                    "'grounded' или 'hallucination'."
                )
            },
            {
                "role": "user",
                "content": (
                    f"Документы: {state['documents']}\n\n"
                    f"Ответ: {state['generation']}"
                )
            }
        ],
        temperature=0
    )
    decision = response.choices[0].message.content.strip()
    if "grounded" in decision.lower():
        return "accept"
    return "retry"


def decide_after_grading(state: RAGState) -> str:
    """Решение после оценки документов."""
    if state["is_relevant"]:
        return "generate"
    if state.get("search_count", 0) >= 3:
        return "generate"  # Максимум попыток
    return "transform"


# Построение графа
workflow = StateGraph(RAGState)

# Добавляем узлы
workflow.add_node("retrieve", retrieve_documents)
workflow.add_node("grade", grade_documents)
workflow.add_node("transform", transform_query)
workflow.add_node("generate", generate)

# Определяем рёбра
workflow.set_conditional_entry_point(
    route_question,
    {"retrieve": "retrieve", "direct": "generate"}
)
workflow.add_edge("retrieve", "grade")
workflow.add_conditional_edges(
    "grade",
    decide_after_grading,
    {"generate": "generate", "transform": "transform"}
)
workflow.add_edge("transform", "retrieve")
workflow.add_conditional_edges(
    "generate",
    check_hallucination,
    {"accept": END, "retry": "transform"}
)

# Компилируем
app = workflow.compile()

# Запуск
result = app.invoke({
    "question": "Как настроить мониторинг RAG?",
    "documents": [],
    "generation": "",
    "search_count": 0,
    "is_relevant": False,
    "needs_web_search": False
})

Corrective RAG (CRAG): самокорректирующийся пайплайн

Идея CRAG

Corrective RAG добавляет петлю обратной связи перед генерацией. Система оценивает, достаточно ли хорош набор извлечённых документов. Если нет — запускается дополнительный поиск, применяются более строгие фильтры или задействуются альтернативные источники (например, веб-поиск).

CRAG выделяет три категории оценки документов:

• Correct — хотя бы один документ содержит релевантную информацию → извлекаем ключевые фрагменты и генерируем ответ
• Incorrect — ни один документ не релевантен → переключаемся на веб-поиск
• Ambiguous — не уверены → комбинируем извлечённые документы с результатами веб-поиска на всякий случай

def corrective_rag_pipeline(query: str) -> str:
    """CRAG: самокорректирующийся RAG."""
    # Шаг 1: Первичное извлечение
    docs = retrieve(query, n_results=10)

    # Шаг 2: Оценка каждого документа
    scored_docs = []
    for doc in docs:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "Оцените релевантность документа "
                        "для вопроса по шкале 1-5. Верните "
                        "только число."
                    )
                },
                {
                    "role": "user",
                    "content": (
                        f"Вопрос: {query}\n"
                        f"Документ: {doc}"
                    )
                }
            ],
            temperature=0
        )
        score = int(
            response.choices[0].message.content.strip()
        )
        if score >= 3:
            scored_docs.append(doc)

    # Шаг 3: Решение о коррекции
    if len(scored_docs) == 0:
        # Incorrect: переключаемся на веб-поиск
        web_results = web_search_fallback(query)
        return generate_answer(query, web_results)
    elif len(scored_docs) < 2:
        # Ambiguous: дополняем из альтернативных источников
        web_results = web_search_fallback(query)
        all_docs = scored_docs + web_results[:3]
        return generate_answer(query, all_docs)
    else:
        # Correct: генерируем на основе найденного
        return generate_answer(query, scored_docs[:5])

Оценка качества RAG-пайплайна

Метрики, которые действительно имеют значение

Невозможно улучшить то, что нельзя измерить — банальность, но от этого не менее верная. Для RAG-пайплайнов существует набор специализированных метрик:

• Retrieval Precision — какая доля извлечённых документов действительно релевантна
• Retrieval Recall — какая доля всех релевантных документов была найдена
• Faithfulness — насколько ответ основан на извлечённых документах (а не на «внутренних знаниях» модели)
• Answer Relevance — насколько ответ соответствует вопросу
• Context Precision — содержится ли ответ в документах с высоким рангом

Автоматическая оценка с RAGAS

RAGAS (Retrieval Augmented Generation Assessment) — пожалуй, самый популярный фреймворк для оценки RAG в 2026 году. Он использует LLM для автоматической оценки, что избавляет от необходимости ручной разметки:

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall
)
from datasets import Dataset


def evaluate_rag(test_questions: list[dict]) -> dict:
    """Оценка RAG-пайплайна с помощью RAGAS."""
    results = {
        "question": [],
        "answer": [],
        "contexts": [],
        "ground_truth": []
    }

    for item in test_questions:
        docs = retrieve(item["question"], n_results=5)
        answer = generate_answer(item["question"], docs)

        results["question"].append(item["question"])
        results["answer"].append(answer)
        results["contexts"].append(docs)
        results["ground_truth"].append(
            item["expected_answer"]
        )

    dataset = Dataset.from_dict(results)

    scores = evaluate(
        dataset,
        metrics=[
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall
        ]
    )
    return scores

Оптимизация для production

Кэширование

Многоуровневое кэширование — одна из тех вещей, которые существенно снижают и латентность, и стоимость. Три уровня, на которые стоит обратить внимание:

• Кэш эмбеддингов — не пересчитывайте эмбеддинги для одинаковых запросов (казалось бы, очевидно, но многие это упускают)
• Семантический кэш — если новый запрос семантически похож на уже кэшированный, можно вернуть кэшированный ответ
• Кэш ответов LLM — для точных совпадений запросов + контекста

import hashlib
from functools import lru_cache


class SemanticCache:
    """Семантический кэш для RAG-ответов."""

    def __init__(self, similarity_threshold: float = 0.95):
        self.threshold = similarity_threshold
        self.cache_collection = (
            chroma_client.get_or_create_collection("cache")
        )

    def get(self, query: str) -> str | None:
        """Поиск в кэше."""
        query_emb = get_embedding(query)
        results = self.cache_collection.query(
            query_embeddings=[query_emb],
            n_results=1
        )

        if (results["distances"]
                and results["distances"][0]
                and results["distances"][0][0] < (
                    1 - self.threshold)):
            return results["metadatas"][0][0]["answer"]
        return None

    def set(self, query: str, answer: str):
        """Сохранение в кэш."""
        query_emb = get_embedding(query)
        cache_id = hashlib.md5(
            query.encode()
        ).hexdigest()

        self.cache_collection.add(
            ids=[cache_id],
            embeddings=[query_emb],
            documents=[query],
            metadatas=[{"answer": answer}]
        )

Guardrails: входной и выходной контроль

В production-системе без защиты от некорректных запросов не обойтись. Вот минимальный набор, который стоит реализовать:

• Input guardrails: фильтрация промпт-инъекций, определение токсичности, проверка на off-topic запросы
• Output guardrails: проверка на галлюцинации, фильтрация PII (персональных данных), контроль тональности
• Access control: синхронизация прав доступа к документам с вашей IAM-системой

Мониторинг и observability

Для production RAG мониторинг каждого этапа пайплайна — это не «было бы неплохо», а необходимость. Вот что стоит отслеживать:

• Латентность каждого этапа (embedding, retrieval, re-ranking, generation)
• Качество извлечения (средний relevance score, процент пустых результатов)
• Стоимость (токены LLM, количество API-вызовов — эти цифры могут неприятно удивить)
• Обратная связь пользователей (thumbs up/down, re-ask rate)

Инструменты вроде LangSmith, LangWatch и Phoenix от Arize предоставляют готовые дашборды для трассировки RAG-пайплайнов. Настоятельно рекомендую подключить хотя бы один из них с самого начала.

Практические рекомендации: чеклист для production RAG

Давайте подведём итог. Вот чеклист, который пригодится вам при запуске RAG-системы в production:

1. Чанкинг: используйте семантический чанкинг с контекстными заголовками. Оптимальный размер чанка — 400–800 токенов с перекрытием 10–15%
2. Поиск: гибридный (BM25 + векторный) — стандарт в 2026 году. Добавьте re-ranking для сложных запросов
3. Эмбеддинги: тестируйте несколько моделей на вашем домене. Рассмотрите fine-tuning эмбеддинг-модели на ваших данных
4. Трансформация запросов: реализуйте HyDE или multi-query для сложных запросов
5. Оценка: настройте автоматическую оценку через RAGAS. Создайте тестовый набор хотя бы из 50–100 пар «вопрос-ответ»
6. Кэширование: семантический кэш значительно снижает стоимость и латентность
7. Guardrails: входной и выходной контроль обязательны для enterprise-систем
8. Мониторинг: трассируйте каждый этап пайплайна, собирайте обратную связь от пользователей
9. GraphRAG: для доменов со сложными связями между сущностями рассмотрите комбинацию векторного и графового поиска
10. Agentic RAG: для сложных запросов используйте агентный подход с самокоррекцией

Заключение

RAG за последние два года прошёл впечатляющий эволюционный путь — от простого «найди похожие чанки и подставь в промпт» до сложных мультимодальных систем с самокоррекцией, графами знаний и агентной оркестрацией.

Но вот что важно: не пытайтесь реализовать всё сразу. Начните с Naive RAG, измерьте качество, определите слабые места и точечно добавляйте Advanced-техники. По моему опыту, гибридный поиск и re-ranking обычно дают наибольший прирост качества при минимальных усилиях. Agentic RAG и GraphRAG — это уже для случаев, когда базовых оптимизаций действительно недостаточно.

И напоследок: лучший RAG-пайплайн — тот, который вы можете измерить, мониторить и итеративно улучшать. Инвестируйте в инфраструктуру оценки не меньше, чем в сам пайплайн, и ваша система будет стабильно расти в качестве. Удачи с вашими пайплайнами!