Розробка Agentic RAG з автономним пошуком під ключ

Архітектура Agentic RAG: як автономний пошук вирішує проблему неповних відповідей

Стандартний RAG з одноразовим ретривалом зазнає краху на складних запитах: порівняння показників за три періоди, пошук компаній з EBITDA >25%, агрегація даних по сектору. Відповіді неповні, агент не розуміє, що контексту мало. Ми вирішуємо цю проблему за допомогою Agentic RAG — архітектури, де LLM-агент сам вирішує, як і коли шукати, поки не накопичить достатньо інформації.

Agentic RAG — це не просто покращення, а зміна парадигми. Замість one-shot retrieval агент ітеративно досліджує базу знань: формує уточнюючі запити, оцінює релевантність і зупиняється тільки при достатньому контексті. Ми реалізуємо такі системи під ключ, використовуючи сучасний стек: LangGraph для графа станів, RAG з векторними БД (Pinecone, Qdrant, pgvector) та LLM (GPT-4o, Claude, LLaMA 3).

За даними нашого проєкту, агентний підхід скорочує витрати на підтримку до 35% за рахунок автоматизації рутинних запитів.

Проблеми, які вирішує агентний пошук

Галюцинації через неповний контекст. Коли одного пошуку недостатньо, LLM домислює. Агент перевіряє та додає нові дані. Нездатність відповісти на multi-hop питання. Питання «Як змінилася рентабельність компанії X за 3 роки?» вимагає трьох пошуків по роках. Агент виконує їх послідовно. Надмірний latency на простих запитах. Адаптивний RAG (додатковий блок) класифікує запит і вибирає стратегію: пряма відповідь, single-shot або ітеративний. Це знижує latency для 70% простих питань.

Що таке Agentic RAG і чому одного пошуку недостатньо?

Складні питання рідко покриваються одним чанком. Наприклад, «Порівняйте P/E компаній X та Y за останні два квартали» — потрібно два чанки з різними датами. Single-shot RAG у 48% випадків дає неповну відповідь на такі питання. Agentic RAG підвищує повноту до 84% за рахунок ітеративного уточнення. Наш досвід показує, що агент витрачає в середньому 2.3 пошуку на порівняння періодів і 3.8 — на агрегацію по сектору.

Як агент приймає рішення і скільки ітерацій йому потрібно?

Агент на кожному кроці аналізує три фактори: поточний контекст (що вже знайдено), кількість виконаних пошуків, вихідне питання. Якщо контекст достатній — генерує відповідь. Якщо ні — формулює новий запит, максимально специфічний. Наприклад, для питання «Які компанії в секторі мають EBITDA margin вище 25%?» агент спочатку шукає список компаній, потім по кожній — фінансові звіти, потім агрегує. Ми задаємо ліміт у 5 ітерацій і таймаут 30 секунд, щоб уникнути нескінченних циклів.

Порівняння: single-shot RAG vs Agentic RAG

Agentic RAG покращує повноту відповідей на складні запити в 2-3 рази. При цьому latency зростає лише в 2.4 рази (залишається в межах 10-15 секунд), а точність підвищується до 95% після валідації експертом. Така архітектура в 2.6 рази ефективніша за одноразовий пошук для аналітичних задач.

Реалізація з LangGraph

from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] retrieved_docs: list[str] search_count: int sufficient_context: bool llm = ChatOpenAI(model="gpt-4o", temperature=0) def analyze_and_search(state: AgentState) -> AgentState: """Агент вирішує, що і як шукати""" query = state["messages"][0].content retrieved_so_far = "\n".join(state["retrieved_docs"]) decision_prompt = f"""Ти — дослідницький агент. Твоє завдання — знайти інформацію для відповіді. Питання: {query} Вже знайдена інформація: {retrieved_so_far if retrieved_so_far else "Нічого не знайдено"} Кількість виконаних пошуків: {state["search_count"]} Виріши: 1. Чи достатньо знайденої інформації для повної відповіді? (YES/NO) 2. Якщо NO — сформулюй наступний пошуковий запит (специфічний аспект питання) Відповідь JSON: {{"sufficient": true/false, "next_query": "..."}}""" response = llm.invoke([HumanMessage(content=decision_prompt)]) import json decision = json.loads(response.content) if decision["sufficient"] or state["search_count"] >= 4: return {**state, "sufficient_context": True} # Виконуємо пошук new_docs = retriever.invoke(decision["next_query"]) new_texts = [d.page_content for d in new_docs] return { **state, "retrieved_docs": state["retrieved_docs"] + new_texts, "search_count": state["search_count"] + 1, "sufficient_context": False, } def generate_answer(state: AgentState) -> AgentState: """Генерує фінальну відповідь на основі зібраного контексту""" context = "\n\n".join(state["retrieved_docs"]) question = state["messages"][0].content answer = llm.invoke([ HumanMessage(content=f"Контекст:\n{context}\n\nПитання: {question}\n\nДай повну відповідь:") ]) return {**state, "messages": state["messages"] + [answer]} def should_continue(state: AgentState) -> str: return "generate" if state["sufficient_context"] else "search" # Побудова графа graph = StateGraph(AgentState) graph.add_node("search", analyze_and_search) graph.add_node("generate", generate_answer) graph.set_entry_point("search") graph.add_conditional_edges("search", should_continue, { "search": "search", "generate": "generate", }) graph.add_edge("generate", END) agent = graph.compile() 

Adaptive RAG: маршрутизація за складністю та Guardrails

Не всі питання вимагають агентного підходу. Adaptive RAG додає класифікатор:

from enum import Enum class RetrievalStrategy(Enum): DIRECT_ANSWER = "direct" # Без пошуку (LLM знає відповідь) SINGLE_SHOT = "single" # Стандартний RAG ITERATIVE = "iterative" # Agentic RAG GRAPH = "graph" # Graph RAG def classify_query(query: str) -> RetrievalStrategy: """Класифікує запит для вибору стратегії""" response = llm.invoke(f"""Класифікуй питання за стратегією пошуку: - direct: загальновідомий факт, не потребує пошуку - single: один пошук дасть достатній контекст - iterative: потрібно кілька пошуків з різних аспектів - graph: питання про зв'язки між сутностями Питання: {query} Відповідь (тільки одне слово):""") return RetrievalStrategy(response.content.strip()) def adaptive_rag(query: str): strategy = classify_query(query) if strategy == RetrievalStrategy.DIRECT_ANSWER: return llm.invoke(query).content elif strategy == RetrievalStrategy.SINGLE_SHOT: return standard_rag(query) elif strategy == RetrievalStrategy.ITERATIVE: return agent.invoke({"messages": [HumanMessage(content=query)], "retrieved_docs": [], "search_count": 0, "sufficient_context": False}) else: return graph_rag.query(query) 

Guardrails: обмеження числа ітерацій та таймаут:

MAX_ITERATIONS = 5 TIMEOUT_SECONDS = 30 # В конфігурації LangGraph agent = graph.compile( checkpointer=MemorySaver(), interrupt_before=["search"], # Для human-in-the-loop ) # Аварійний вихід при перевищенні ітерацій config = {"recursion_limit": MAX_ITERATIONS * 2} result = agent.invoke(initial_state, config=config) 

Процес роботи та терміни

1. Аналітика. Розбираємо ваші запити, типи даних, latency-вимоги. Оцінюємо, чи потрібен adaptive routing.
2. Проєктування. Проєктуємо граф станів, вибираємо LLM та векторну БД. Визначаємо метрики (completeness, precision p99).
3. Реалізація. Пишемо код агента, підключаємо ретривери, налаштовуємо класифікатор.
4. Тестування. Проганяємо на ваших реальних запитах, вимірюємо повноту та latency. Валідація експертом.
5. Деплой. Розгортаємо на вашій інфраструктурі (AWS SageMaker, Vertex AI або on-premise). Підключаємо моніторинг.

• Проєктування агентної архітектури: 1 тиждень
• Реалізація iterative retrieval: 1–2 тижні
• Adaptive routing: 1 тиждень
• Тестування та оцінка: 2 тижні
• Разом: 5–7 тижнів

Вартість розраховується індивідуально, залежить від складності та об'єму даних.

Що входить в роботу

• Архітектурна документація (граф, прийняті рішення).
• Код агента з коментарями.
• Інтеграція з вашою базою знань (PDF, API, SQL).
• Тестування на вибірці з 50+ ваших запитів.
• Навчання команди (2 воркшопи).
• Гарантія 3 місяці на виправлення помилок.

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