Реализация Agentic RAG с автономным поиском информации

Реализация Agentic RAG с автономным поиском информации

Agentic RAG — архитектура, в которой LLM-агент самостоятельно решает: нужен ли поиск, сколько раз искать, какие запросы формулировать, и достаточно ли найденной информации для ответа. В отличие от стандартного RAG с фиксированным one-shot retrieval, агент итеративно исследует базу знаний до получения достаточного контекста.

Стандартный RAG vs Agentic RAG

Стандартный RAG:

1. Запрос → Retrieval (один раз) → Генерация
2. Нет контроля достаточности контекста
3. Нет адаптации стратегии поиска

Agentic RAG:

1. Запрос → Агент анализирует задачу
2. Агент формулирует поисковый запрос
3. Retrieval → Агент оценивает результат
4. Если контекст недостаточен → новый поиск с другим запросом
5. Повторение до достаточного контекста
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: маршрутизация по сложности запроса

Не все вопросы требуют агентного подхода. 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)

Практический кейс: аналитический ассистент для инвестора

Задача: ответы на аналитические вопросы по корпусу финансовых отчётов 200 компаний.

Примеры вопросов:

• «Как изменилась рентабельность компании X за 3 года?» → iterative (3 поиска по годам)
• «Какие компании в секторе имеют EBITDA margin выше 25%?» → iterative (несколько поисков + агрегация)
• «Каков P/E компании X?» → single shot

Результаты Agentic vs Single-Shot RAG:

Agentic RAG критически улучшает complex queries (+218% для кросс-компания) при умеренной деградации latency (×2.4 среднее).

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 неделя
• Реализация iterative retrieval: 1–2 недели
• Adaptive routing: 1 неделя
• Тестирование и оценка: 2 недели
• Итого: 5–7 недель