Реалізація Agentic RAG з автономним пошуком інформації

Впровадження Agentic RAG з автономним пошуком інформації

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

Стандартний RAG проти 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 критично поліпшує складні запити (+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 тиждень
• Впровадження ітеративного retrieval: 1–2 тижні
• Адаптивна маршрутизація: 1 тиждень
• Тестування та оцінка: 2 тижні
• Всього: 5–7 тижнів