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