Разработка 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 дня. Свяжитесь с нами, чтобы обсудить ваш проект.