Agentic RAG з LangGraph: будуємо самокоригувальну RAG-систему на Python

Вступ: чому базовий RAG більше не працює

Якщо ви хоч раз будували RAG-систему (Retrieval-Augmented Generation), то напевно знаєте цей сценарій: користувач ставить запитання, система шукає документи у векторній базі, знайдені фрагменти летять до LLM, і модель генерує відповідь. Просто, елегантно, і... часто недостатньо.

Проблема в тому, що наївний RAG — це одноразова операція. Один запит, один пошук, одна відповідь — і ніякої рефлексії. Система не перевіряє, чи знайдені документи дійсно релевантні. Не переформульовує запит, коли результати так собі. Не вміє поєднати інформацію з кількох джерел. І, чесно кажучи, не має жодного механізму самоконтролю — як студент, який здав курсову не перечитавши.

У 2026 році для продакшн-систем це вже не варіант. За даними порівняльних досліджень, Agentic RAG покращує точність відповідей на 14% порівняно зі стандартним RAG (з 73.1% до 86.9%), а корпоративні системи з самокоригувальними циклами досягають 94% точності оцінки галюцинацій. Різниця відчутна.

У цій статті ми розберемо архітектуру Agentic RAG, напишемо самокоригувальну систему на Python з LangGraph і навчимося оцінювати її якість за допомогою RAGAS. Весь код робочий — можете копіювати прямо у свій проєкт.

Наївний RAG проти Agentic RAG: у чому різниця

Перш ніж писати код, давайте чітко розмежуємо два підходи. Наївний RAG — це конвеєр (pipeline): лінійна послідовність кроків без зворотного зв'язку. Agentic RAG — це цикл (loop): LLM виступає двигуном міркувань, який сам вирішує, як шукати інформацію.

Порівняльна таблиця

Характеристика	Наївний RAG	Agentic RAG
Пошук	Одноразовий, один прохід	Ітеративний, багатопрохідний
Міркування	Відсутнє	Планування та рефлексія
Джерела знань	Одна векторна БД	Кілька джерел + інструменти/API
Багатокрокові запити	Не підтримує	Повна підтримка
Самокорекція	Відсутня	Вбудований цикл перевірки
Латентність	Низька	Вища (компенсується якістю)
Найкращий для	Прості фактологічні запитання	Складні аналітичні завдання

Ключова проблема наївного RAG — відсутність механізму виявлення контекстних прогалин. Якщо відповідь вимагає інформації з документа A і документа B, наївний RAG просто не впорається. Agentic RAG натомість може отримати документ A, зрозуміти, що потребує ще й B, зробити додатковий пошук і зібрати все в одну відповідь.

Архітектура самокоригувальної RAG-системи

Самокоригувальна RAG-система стоїть на трьох патернах, які чудово доповнюють один одного:

• Corrective RAG (CRAG) — перевіряє якість знайдених документів до генерації. Якщо документи нерелевантні, система переформульовує запит або іде шукати в інших джерелах.
• Self-RAG — додає рефлексію після генерації: система оцінює, чи підтримана відповідь контекстом, і чи дійсно вона відповідає на запитання.
• Adaptive RAG — маршрутизує запити за складністю: прості питання обробляються LLM напряму, складні — проходять через повний RAG-цикл.

Архітектура виглядає як граф станів (і це не випадково — ми будуватимемо саме граф у LangGraph):

┌─────────────┐
│   Маршрутизатор   │ ← Вирішує: RAG чи пряма відповідь?
└───────┬─────┘
        │
┌───────▼─────┐
│   Ретрівер       │ ← Пошук у векторній БД
└───────┬─────┘
        │
┌───────▼─────┐     ┌──────────────┐
│   Оцінювач       │────►│ Переформулювання │
│   документів     │ ні  │ запиту            │
└───────┬─────┘     └───────┬──────┘
        │ так                │
┌───────▼─────┐     ┌───────▼──────┐
│   Генератор      │     │ Веб-пошук       │
└───────┬─────┘     │ (фолбек)         │
        │           └──────────────┘
┌───────▼─────┐
│   Перевірка      │ ← Галюцинації? Релевантність?
│   відповіді      │
└───────┬─────┘
        │
   Фінальна відповідь

Підготовка середовища

Для реалізації знадобиться кілька бібліотек. LangGraph — серце нашого графа, ChromaDB — векторна база, а LangChain дає зручні абстракції для LLM та ембедінгів.

# Встановлення залежностей
pip install langgraph langchain langchain-openai chromadb langchain-community langchain-text-splitters ragas

Переконайтеся, що змінні середовища для API-ключів на місці:

export OPENAI_API_KEY="sk-your-key-here"
# Або для Anthropic:
export ANTHROPIC_API_KEY="your-key-here"

Крок 1: Створення векторного сховища

Почнемо з бази знань. Розіб'ємо документи на чанки, згенеруємо ембедінги й збережемо все в ChromaDB:

from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# Завантажуємо документи (ваші джерела знань)
urls = [
    "https://docs.example.com/guide-1",
    "https://docs.example.com/guide-2",
]
docs = []
for url in urls:
    docs.extend(WebBaseLoader(url).load())

# Розбиваємо на чанки з перекриттям
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
)
splits = text_splitter.split_documents(docs)

# Створюємо векторне сховище
vectorstore = Chroma.from_documents(
    documents=splits,
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
    collection_name="knowledge_base",
)

# Створюємо ретрівер
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 5},
)

Зверніть увагу на chunk_overlap=200 — перекриття між чанками потрібне, щоб важлива інформація не загубилась на межі розділення. Здавалось би, дрібниця, але ця оптимізація реально впливає на якість пошуку.

Крок 2: Визначення стану графа

Стан — це, по суті, «пам'ять» нашого агента. Він тримає всю інформацію, яка передається між вузлами графа:

from typing import List, TypedDict
from langchain_core.documents import Document


class GraphState(TypedDict):
    """Стан графа для Agentic RAG."""
    question: str           # Початкове запитання користувача
    generation: str         # Згенерована відповідь
    documents: List[Document]  # Знайдені документи
    retry_count: int        # Лічильник спроб (захист від нескінченних циклів)
    web_search_needed: bool # Чи потрібен веб-пошук

Поле retry_count тут критично важливе. Без нього система може зациклитись — переформулювати запит, не знайти нічого, переформулювати знову, і так до безкінечності (а разом з тим — спалити ваш API-бюджет). Ми обмежимо кількість спроб трьома.

Крок 3: Вузол маршрутизації запитів

Не кожен запит потребує RAG. Прості запитання типу «Яка зараз дата?» краще обробити LLM напряму, а складні доменні запитання — направити через повний цикл:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field

llm = ChatOpenAI(model="gpt-4o", temperature=0)


class RouteDecision(BaseModel):
    """Рішення маршрутизатора."""
    route: str = Field(
        description="Маршрут: 'vectorstore' для пошуку в базі знань, 'direct' для прямої відповіді"
    )


def route_question(state: GraphState) -> str:
    """Маршрутизація запиту: RAG або пряма відповідь LLM."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """Ти — маршрутизатор запитів. Визнач, чи потребує запитання
пошуку в базі знань (vectorstore), чи LLM може відповісти напряму (direct).

Використовуй vectorstore для:
- Запитань про специфічну документацію чи дані
- Технічних деталей, які потребують точних фактів
- Будь-чого, де галюцинація неприпустима

Використовуй direct для:
- Загальних знань та визначень
- Простих обчислень чи логіки
- Світської бесіди"""),
        ("human", "{question}"),
    ])

    structured_llm = llm.with_structured_output(RouteDecision)
    result = structured_llm.invoke(
        prompt.format_messages(question=state["question"])
    )
    return result.route

Крок 4: Оцінювач релевантності документів

А ось і серце Corrective RAG — вузол, що перевіряє кожен знайдений документ на релевантність. Нерелевантні документи відсіюються ще до генерації (навіщо засмічувати контекст?):

class RelevanceGrade(BaseModel):
    """Оцінка релевантності документа."""
    is_relevant: bool = Field(
        description="True, якщо документ релевантний запитанню"
    )


def grade_documents(state: GraphState) -> GraphState:
    """Оцінюємо релевантність кожного знайденого документа."""
    question = state["question"]
    documents = state["documents"]

    prompt = ChatPromptTemplate.from_messages([
        ("system", """Ти — оцінювач релевантності. Визнач, чи містить документ
інформацію, корисну для відповіді на запитання.
Не потрібна точна відповідність — достатньо, щоб документ був тематично пов'язаний."""),
        ("human", "Запитання: {question}\n\nДокумент: {document}"),
    ])

    structured_llm = llm.with_structured_output(RelevanceGrade)

    relevant_docs = []
    for doc in documents:
        grade = structured_llm.invoke(
            prompt.format_messages(
                question=question,
                document=doc.page_content,
            )
        )
        if grade.is_relevant:
            relevant_docs.append(doc)

    # Якщо менше половини документів релевантні — потрібен веб-пошук
    web_search_needed = len(relevant_docs) < len(documents) / 2

    return {
        **state,
        "documents": relevant_docs,
        "web_search_needed": web_search_needed,
    }

Логіка тут проста: якщо менше половини документів пройшли перевірку — значить, база знань не дуже допомагає, і система позначає потребу у веб-пошуку. Це набагато розумніше, ніж генерувати відповідь з поганим контекстом і сподіватись на краще.

Крок 5: Переформулювання запиту

Коли документи нерелевантні, проблема часто не в базі знань, а в самому запиті. Я помічав це неодноразово: варто трохи змінити формулювання — і результати пошуку стають кардинально іншими:

def rewrite_query(state: GraphState) -> GraphState:
    """Переформулювання запиту для покращення результатів пошуку."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """Ти — спеціаліст з оптимізації пошукових запитів.
Переформулюй запитання так, щоб воно краще відповідало семантичному пошуку.
Зроби запит більш конкретним, додай ключові терміни домену.
Поверни лише переформульований запит, без пояснень."""),
        ("human", "Початкове запитання: {question}"),
    ])

    result = llm.invoke(
        prompt.format_messages(question=state["question"])
    )

    return {
        **state,
        "question": result.content,
        "retry_count": state.get("retry_count", 0) + 1,
    }

Крок 6: Генерація відповіді

Генератор формує відповідь на основі відфільтрованих документів. Нічого зайвого — тільки факти з контексту:

def generate_answer(state: GraphState) -> GraphState:
    """Генерація відповіді на основі релевантних документів."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """Ти — корисний ШІ-асистент. Відповідай на запитання,
використовуючи ЛИШЕ надані документи. Якщо в документах недостатньо
інформації, чесно зазнач це. Не вигадуй факти.

Документи:
{context}"""),
        ("human", "{question}"),
    ])

    context = "\n\n---\n\n".join(
        doc.page_content for doc in state["documents"]
    )

    result = llm.invoke(
        prompt.format_messages(context=context, question=state["question"])
    )

    return {**state, "generation": result.content}

Крок 7: Перевірка галюцинацій та релевантності

Тут вступає в гру Self-RAG компонент. Система оцінює власну відповідь за двома критеріями: чи підтримана вона документами (faithfulness) і чи реально відповідає на запитання (relevance). По суті, це вбудований «внутрішній критик»:

class QualityCheck(BaseModel):
    """Результат перевірки якості відповіді."""
    is_grounded: bool = Field(
        description="True, якщо відповідь підтримана наданими документами"
    )
    answers_question: bool = Field(
        description="True, якщо відповідь адресує початкове запитання"
    )


def check_answer_quality(state: GraphState) -> str:
    """Перевірка якості згенерованої відповіді."""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """Оціни якість відповіді за двома критеріями:
1. Заземленість (grounding): чи базується відповідь на наданих документах?
   Відповідь НЕ повинна містити фактів, яких немає в документах.
2. Релевантність: чи відповідає відповідь на поставлене запитання?

Документи: {context}"""),
        ("human", "Запитання: {question}\n\nВідповідь: {generation}"),
    ])

    context = "\n\n".join(
        doc.page_content for doc in state["documents"]
    )

    structured_llm = llm.with_structured_output(QualityCheck)
    result = structured_llm.invoke(
        prompt.format_messages(
            context=context,
            question=state["question"],
            generation=state["generation"],
        )
    )

    if not result.is_grounded:
        return "hallucination"
    elif not result.answers_question:
        return "not_relevant"
    else:
        return "good"

Крок 8: Збирання графа в LangGraph

Ось найцікавіша частина — збираємо всі компоненти в єдиний граф. Тут визначається вся логіка переходів між вузлами:

from langgraph.graph import StateGraph, END

def build_agentic_rag_graph():
    """Побудова графа Agentic RAG."""
    workflow = StateGraph(GraphState)

    # Додаємо вузли
    workflow.add_node("retrieve", retrieve_documents)
    workflow.add_node("grade_documents", grade_documents)
    workflow.add_node("rewrite_query", rewrite_query)
    workflow.add_node("generate", generate_answer)

    # Точка входу
    workflow.set_entry_point("retrieve")

    # Після ретрівалу — завжди оцінюємо документи
    workflow.add_edge("retrieve", "grade_documents")

    # Умовні переходи після оцінки документів
    workflow.add_conditional_edges(
        "grade_documents",
        decide_after_grading,
        {
            "generate": "generate",
            "rewrite": "rewrite_query",
        },
    )

    # Після переформулювання — знову пошук
    workflow.add_edge("rewrite_query", "retrieve")

    # Умовні переходи після генерації
    workflow.add_conditional_edges(
        "generate",
        check_answer_quality,
        {
            "good": END,
            "hallucination": "generate",      # Перегенерувати
            "not_relevant": "rewrite_query",   # Переформулювати запит
        },
    )

    return workflow.compile()


def decide_after_grading(state: GraphState) -> str:
    """Рішення після оцінки документів."""
    if not state["documents"]:
        # Жодного релевантного документа
        if state.get("retry_count", 0) >= 3:
            return "generate"  # Відповідаємо з тим, що є
        return "rewrite"
    return "generate"


def retrieve_documents(state: GraphState) -> GraphState:
    """Пошук документів у векторній базі."""
    documents = retriever.invoke(state["question"])
    return {**state, "documents": documents}


# Компіляція та запуск
app = build_agentic_rag_graph()

# Виконання запиту
result = app.invoke({
    "question": "Як налаштувати моніторинг для RAG-системи?",
    "generation": "",
    "documents": [],
    "retry_count": 0,
    "web_search_needed": False,
})

print(result["generation"])

Зверніть увагу на захисний механізм: retry_count >= 3 запобігає нескінченним циклам. Після трьох невдалих спроб система генерує найкращу можливу відповідь з наявним контекстом — або чесно каже, що не знає. Це важливо, бо нескінченний цикл — це не лише баг, це ще й рахунок за API.

Додавання веб-пошуку як фолбеку

Коли векторна база не рятує, веб-пошук стає запасним варіантом. Додати його нескладно:

from langchain_community.tools import TavilySearchResults

web_search_tool = TavilySearchResults(max_results=3)


def web_search_fallback(state: GraphState) -> GraphState:
    """Веб-пошук як фолбек, коли база знань не допомагає."""
    results = web_search_tool.invoke(state["question"])

    web_docs = [
        Document(
            page_content=r["content"],
            metadata={"source": r["url"]},
        )
        for r in results
    ]

    # Додаємо веб-результати до існуючих документів
    all_docs = state.get("documents", []) + web_docs
    return {**state, "documents": all_docs, "web_search_needed": False}

Для інтеграції в граф додайте вузол і умовне ребро після grade_documents — якщо web_search_needed дорівнює True, направляйте на web_search_fallback.

Оптимізація для продакшну

Працюючий прототип — це тільки початок. Ось кілька оптимізацій, без яких у продакшні буде боляче.

Дешевші моделі для допоміжних вузлів

Маршрутизатор та оцінювач документів не потребують найпотужнішої моделі — там задачі досить прості. Використовуйте GPT-4o-mini або Claude Haiku для цих вузлів, а серйозну модель залиште для генерації:

# Швидка та дешева модель для оцінювання
grading_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Потужна модель для генерації
generation_llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

Гібридний пошук: вектори + ключові слова

Чистий семантичний пошук іноді пропускає документи з точними термінами — номерами версій, кодами помилок, специфічним жаргоном. Гібридний пошук бере найкраще з обох підходів:

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# BM25 для ключових слів
bm25_retriever = BM25Retriever.from_documents(splits)
bm25_retriever.k = 5

# Ансамбль: 60% семантика + 40% ключові слова
hybrid_retriever = EnsembleRetriever(
    retrievers=[retriever, bm25_retriever],
    weights=[0.6, 0.4],
)

Реранкінг результатів

Початковий пошук оптимізований для recall — отримати якнайбільше потенційно релевантних документів. Але ранжування цих кандидатів часто далеке від ідеалу. Реранкер виводить найрелевантніші результати на перші позиції:

from langchain.retrievers.document_compressors import CrossEncoderReranker
from langchain.retrievers import ContextualCompressionRetriever
from langchain_community.cross_encoders import HuggingFaceCrossEncoder

# Модель для реранкінгу
reranker_model = HuggingFaceCrossEncoder(model_name="BAAI/bge-reranker-v2-m3")
compressor = CrossEncoderReranker(model=reranker_model, top_n=3)

# Обгортка з реранкінгом
reranking_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=hybrid_retriever,
)

Оцінка якості з RAGAS

Побудувати систему — це половина справи. Друга половина — довести, що вона працює (і що працює краще за попередню версію). RAGAS (Retrieval Augmented Generation Assessment) — це стандарт де-факто для оцінки RAG-систем у 2026 році. Фреймворк вимірює якість за чотирма метриками:

• Faithfulness (Вірність) — чи підтримана відповідь контекстом? Виявляє галюцинації.
• Answer Relevancy (Релевантність відповіді) — чи відповідає відповідь на запитання?
• Context Precision (Точність контексту) — чи справді знайдені документи містять потрібне?
• Context Recall (Повнота контексту) — чи знайдено всю необхідну інформацію?

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

# Підготовка тестового набору
eval_data = {
    "question": [
        "Як налаштувати моніторинг RAG-системи?",
        "Які метрики використовувати для оцінки RAG?",
    ],
    "answer": [
        result_1["generation"],
        result_2["generation"],
    ],
    "contexts": [
        [doc.page_content for doc in result_1["documents"]],
        [doc.page_content for doc in result_2["documents"]],
    ],
    "ground_truth": [
        "Для моніторингу RAG використовують метрики латентності, точності та повноти...",
        "Основні метрики: faithfulness, answer relevancy, context precision, context recall...",
    ],
}

dataset = Dataset.from_dict(eval_data)

# Запуск оцінки
results = evaluate(
    dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
    ],
)

print(results)
# {'faithfulness': 0.92, 'answer_relevancy': 0.89,
#  'context_precision': 0.85, 'context_recall': 0.88}

Рекомендовані порогові значення для продакшну: Faithfulness > 0.85, Answer Relevancy > 0.80, Context Precision > 0.75. Якщо метрики нижчі — починайте з оптимізації чанкінгу та якості ембедінгів, це зазвичай дає найбільший ефект.

Поширені помилки та як їх уникнути

1. Нескінченні цикли

Без обмеження кількості ітерацій агент може зациклитись. Завжди додавайте retry_count і жорстке обмеження (3-5 спроб). Бонус: це ще й контролює витрати на API.

2. Занадто строгий оцінювач

Якщо оцінювач відсіює забагато документів, система буде вічно переформулювати запити. Налаштуйте промпт так, щоб він пропускав тематично пов'язані документи — не вимагайте ідеального збігу.

3. Відсутність логування

У продакшні логування — це не опція, а необхідність. Який маршрут обрано, скільки документів пройшли оцінку, скільки спроб знадобилось — все це потрібно бачити. Використовуйте LangSmith або OpenTelemetry для трейсингу всього графа.

4. Ігнорування латентності

Кожен додатковий LLM-виклик — це плюс 1-3 секунди. Для інтерактивних систем це критично. Використовуйте стрімінг і повідомляйте користувача про прогрес: «Шукаю додаткову інформацію...» — це набагато краще, ніж мовчазне очікування.

Коли варто використовувати Agentic RAG, а коли — ні

Agentic RAG — потужний інструмент, але не срібна куля. Ось просте правило:

Використовуйте Agentic RAG, коли:

• Запитання вимагають багатокрокового міркування (порівняння даних з різних документів)
• Точність критична, а галюцинації неприпустимі (медичні, юридичні, фінансові системи)
• Користувачі задають складні, неструктуровані запитання
• Потрібна інтеграція з кількома джерелами даних

Залишайтесь на наївному RAG, коли:

• Запитання прості та фактологічні (FAQ-бот)
• Латентність критично важлива (менше 1 секунди)
• Бюджет на API-виклики обмежений
• Документи добре структуровані і близькі до типових запитів

Часті запитання (FAQ)

Яка різниця між Agentic RAG та звичайним RAG?

Звичайний (наївний) RAG — це лінійний конвеєр: один пошук, одна генерація, без перевірки. Agentic RAG додає цикл самокорекції — система оцінює документи, переформульовує запити за потреби, перевіряє відповідь на галюцинації та може звернутися до інших джерел. На практиці це підвищує точність на 14% і більше.

Чи збільшує Agentic RAG вартість API-викликів?

Так, кожен додатковий вузол (оцінювач, перевірка якості) — це виклик LLM. Але вартість можна тримати під контролем: дешевші моделі на допоміжних вузлах, повноцінна модель — тільки для генерації. Зазвичай додаткові витрати складають 30-50%, що цілком виправдовується якістю відповідей.

Які фреймворки найкращі для Agentic RAG у 2026 році?

LangGraph — найпопулярніший вибір завдяки гнучкій графовій архітектурі. LlamaIndex пропонує вбудовані абстракції для агентів та query engines. Для мультиагентних систем варто подивитись на CrewAI або Microsoft Agent Framework. Вибір залежить від складності задачі та екосистеми, яку ви вже використовуєте.

Як оцінити якість Agentic RAG-системи?

Використовуйте RAGAS з чотирма метриками: Faithfulness, Answer Relevancy, Context Precision та Context Recall. Для продакшну орієнтуйтеся на Faithfulness > 0.85, Answer Relevancy > 0.80. Додатково можна підключити LLM-as-a-Judge для автоматичної оцінки та A/B-тестування з реальними користувачами.

Чи можна використовувати Agentic RAG з локальними моделями?

Так, LangGraph і LangChain підтримують будь-які LLM через уніфікований інтерфейс. Для локального розгортання підійдуть Ollama або vLLM з моделями Llama 3, Mistral чи Phi-3. Для маршрутизації та оцінки документів вистачить моделі з 7-8B параметрами, а для генерації краще брати від 70B. Це знімає залежність від зовнішніх API і дає повний контроль над даними.