Нужен ли исследовательскому агенту цикл самопроверки

Типичный исследовательский агент устроен как линейный пайплайн: спланировать запросы, найти материалы, написать ответ. Проблема в том, что последний шаг выполняется всегда, даже когда найденного не хватает. Модель уверенно заполняет пробелы догадками, и читатель не может понять, где факт, а где домысел.

Очевидное решение — дать агенту перечитать собранное и решить, нужен ли ещё поиск. Но у неограниченной рефлексии свои проблемы: каждый проход стоит лишних вызовов модели, а агент, который никогда не бывает «доволен», может крутиться бесконечно. Ниже соберём обе версии на LangGraph, поставим циклу жёсткие ограничения и сравним их на одном наборе вопросов.

Что получится: рабочий граф с циклом Research–Reflect не больше чем в два прохода, линейная версия для контроля и скрипт, который записывает качество, время и число запросов в CSV.

Как устроен эксперимент

• plan — модель составляет до двух поисковых запросов (1 вызов LLM).
• research — поиск по локальному корпусу, без LLM.
• reflect — модель оценивает, хватает ли заметок, и предлагает новые запросы (1 вызов LLM). Есть только в версии с циклом.
• write — итоговый ответ (1 вызов LLM). Если проверка не прошла, ответ обязан начинаться с пометки «НЕДОСТАТОЧНО ДАННЫХ».

Цикл останавливается при выполнении любого из трёх условий: рефлексия признала данные достаточными; лимит в MAX_REFLECTIONS = 2 исчерпан; новых, ещё не использованных запросов нет, то есть прогресса не будет. Подстраховкой служит recursion_limit самого LangGraph.

Чтобы эксперимент воспроизводился без сети и платных API, поиск идёт по маленькому учебному корпусу о вымышленном сервисе Ledger-7. Это пример, а не реальная система: все «факты» о нём придуманы для теста.

Шаг 1. Окружение

Нужны Python 3.10+ и локально установленная Ollama. Модель ниже — только пример, подойдёт любая, которая умеет возвращать JSON.

python -m venv .venv
source .venv/bin/activate
pip install -U langgraph langchain-ollama
ollama pull qwen2.5:7b

Шаг 2. Граф агента

Сохраните файл как agent.py. В обеих версиях одни и те же узлы, разница только в рёбрах.

# agent.py
import json
import re
from typing import List, TypedDict

from langchain_ollama import ChatOllama
from langgraph.graph import END, START, StateGraph

MODEL = "qwen2.5:7b"  # пример, подставьте свою модель
MAX_REFLECTIONS = 2
llm = ChatOllama(model=MODEL, temperature=0)

# Учебный корпус: Ledger-7 — вымышленный сервис, факты придуманы для теста.
CORPUS = {
    "d1": "Ledger-7 хранит сообщения в журнале и реплицирует каждое на три узла.",
    "d2": "Запись в Ledger-7 считается подтверждённой, когда её сохранили два узла из трёх.",
    "d3": "Ledger-7 хранит сообщения семь дней, после чего журнал сжимается.",
    "d4": "При сбое отправки клиенты Ledger-7 повторяют попытку с экспоненциальной задержкой.",
}


def search(query: str, k: int = 2) -> List[str]:
    words = set(re.findall(r"\w+", query.lower()))
    scored = []
    for doc_id, text in CORPUS.items():
        score = len(words & set(re.findall(r"\w+", text.lower())))
        if score:
            scored.append((score, doc_id, text))
    scored.sort(reverse=True)
    return [f"[{d}] {t}" for _, d, t in scored[:k]]


class State(TypedDict, total=False):
    question: str
    queries: List[str]
    used_queries: List[str]
    notes: List[str]
    reflections: int
    sufficient: bool
    llm_calls: int
    search_calls: int
    answer: str


def ask(prompt: str) -> str:
    return llm.invoke(prompt).content


def parse_json(text: str):
    match = re.search(r"\{.*\}", text, re.S)
    if not match:
        return None
    try:
        return json.loads(match.group(0))
    except json.JSONDecodeError:
        return None


def plan(state: State) -> State:
    raw = ask(
        "Составь не больше двух коротких поисковых запросов для вопроса. "
        'Верни только JSON вида {"queries": ["..."]}.\n'
        f"Вопрос: {state['question']}"
    )
    data = parse_json(raw) or {}
    queries = [q for q in data.get("queries", []) if isinstance(q, str)][:2]
    return {
        "queries": queries or [state["question"]],
        "used_queries": [],
        "notes": [],
        "reflections": 0,
        "sufficient": False,
        "llm_calls": 1,
        "search_calls": 0,
    }


def research(state: State) -> State:
    notes = list(state["notes"])
    used = list(state["used_queries"])
    for q in state["queries"]:
        used.append(q)
        for hit in search(q):
            if hit not in notes:
                notes.append(hit)
    return {
        "notes": notes,
        "used_queries": used,
        "search_calls": state["search_calls"] + len(state["queries"]),
    }


def reflect(state: State) -> State:
    notes = "\n".join(state["notes"]) or "(заметок нет)"
    raw = ask(
        "Ты проверяешь, хватает ли заметок для полного ответа на вопрос. "
        "Не используй знания вне заметок. Верни только JSON: "
        '{"sufficient": true|false, "missing": ["..."], "new_queries": ["..."]}.\n'
        f"Вопрос: {state['question']}\nЗаметки:\n{notes}"
    )
    data = parse_json(raw)
    update = {
        "reflections": state["reflections"] + 1,
        "llm_calls": state["llm_calls"] + 1,
    }
    if data is None:
        # Не смогли разобрать ответ: не считаем данные достаточными и не продолжаем цикл.
        return {**update, "sufficient": False, "queries": []}
    new = [
        q for q in data.get("new_queries", [])
        if isinstance(q, str) and q not in state["used_queries"]
    ][:2]
    return {**update, "sufficient": bool(data.get("sufficient")), "queries": new}


def route(state: State) -> str:
    if state["sufficient"]:
        return "write"
    if state["reflections"] >= MAX_REFLECTIONS:
        return "write"
    if not state["queries"]:
        return "write"  # новых запросов нет, следующий проход ничего не даст
    return "research"


def write(state: State) -> State:
    notes = "\n".join(state["notes"]) or "(заметок нет)"
    checked = state.get("reflections", 0) > 0
    rule = ""
    if checked and not state["sufficient"]:
        rule = ("Проверка показала, что данных не хватает. Начни ответ со строки "
                "'НЕДОСТАТОЧНО ДАННЫХ:' и перечисли, чего не хватает.\n")
    answer = ask(
        "Ответь на вопрос только по заметкам, ссылаясь на [id] источников.\n"
        f"{rule}Вопрос: {state['question']}\nЗаметки:\n{notes}"
    )
    return {"answer": answer, "llm_calls": state["llm_calls"] + 1}


def build(reflective: bool):
    g = StateGraph(State)
    g.add_node("plan", plan)
    g.add_node("research", research)
    g.add_node("write", write)
    g.add_edge(START, "plan")
    g.add_edge("plan", "research")
    g.add_edge("write", END)
    if reflective:
        g.add_node("reflect", reflect)
        g.add_edge("research", "reflect")
        g.add_conditional_edges("reflect", route, {"research": "research", "write": "write"})
    else:
        g.add_edge("research", "write")
    return g.compile()

Обратите внимание на две детали. Во-первых, неразобранный ответ рефлексии останавливает цикл, но не делает вид, что всё в порядке: итоговый ответ получит пометку о нехватке данных. Во-вторых, повторяющиеся запросы отбрасываются, поэтому модель не может бесконечно «искать то же самое ещё раз».

Шаг 3. Скрипт сравнения

Сохраните как compare.py. В наборе три вопроса: два отвечаемых, где нужны факты из разных документов, и один, ответа на который в корпусе нет.

# compare.py
import csv
import statistics
import time

from agent import build

CASES = [
    {"q": "Сколько узлов хранят копию сообщения в Ledger-7 и когда запись подтверждается?",
     "facts": ["три", "два"], "answerable": True},
    {"q": "Сколько Ledger-7 хранит сообщения и что делают клиенты при сбое отправки?",
     "facts": ["сем", "экспоненциал"], "answerable": True},
    {"q": "Под какой лицензией распространяется Ledger-7?",
     "facts": [], "answerable": False},
]


def score(answer: str, case: dict) -> float:
    text = answer.lower()
    flagged = text.lstrip().startswith("недостаточно данных")
    if not case["answerable"]:
        return 1.0 if flagged else 0.0
    return sum(f in text for f in case["facts"]) / len(case["facts"])


rows = []
for mode, reflective in [("linear", False), ("reflect", True)]:
    app = build(reflective)
    for case in CASES:
        start = time.perf_counter()
        out = app.invoke({"question": case["q"]}, config={"recursion_limit": 12})
        rows.append({
            "mode": mode,
            "question": case["q"],
            "quality": score(out["answer"], case),
            "seconds": round(time.perf_counter() - start, 2),
            "llm_calls": out["llm_calls"],
            "search_calls": out["search_calls"],
        })

with open("results.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=rows[0].keys())
    writer.writeheader()
    writer.writerows(rows)

for mode in ("linear", "reflect"):
    part = [r for r in rows if r["mode"] == mode]
    print(mode,
          "quality=%.2f" % statistics.mean(r["quality"] for r in part),
          "sec=%.2f" % statistics.mean(r["seconds"] for r in part),
          "llm=%.1f" % statistics.mean(r["llm_calls"] for r in part),
          "search=%.1f" % statistics.mean(r["search_calls"] for r in part))

python compare.py

Шаг 4. Таблица сравнения

Число вызовов следует прямо из устройства графа, поэтому его границы известны заранее. Качество и время зависят от модели и железа, и мы их не придумываем: заполните эти строки значениями из своего results.csv.

Чего стоит ожидать по устройству графа. На третьем вопросе линейная версия в принципе не может набрать балл: у неё нет шага, который поставил бы пометку о нехватке данных. Версия с циклом тратит на вопрос максимум на два вызова LLM больше. Помогает ли второй проход на первых двух вопросах, зависит от того, насколько удачные запросы составит модель на шаге plan. Именно это и показывает замер.

Как проверить, что всё работает

1. В results.csv у строк linear поле llm_calls равно 2, у строк reflect — 3 или 4. Любое другое значение означает, что граф собран не так.
2. Посмотрите путь выполнения: for step in app.stream({"question": "..."}): print(list(step)). Узел reflect должен встретиться не больше двух раз.
3. Временно поставьте MAX_REFLECTIONS = 0. Версия с циклом должна сразу уходить в write и ставить пометку, так как проверка не пройдена.
4. Запустите сравнение 2–3 раза. Даже при temperature=0 локальные модели не всегда детерминированы, поэтому смотрите на разброс, а не на один прогон.

Типовые ошибки

• Цикл без счётчика. Условие «пока модель не скажет sufficient» рано или поздно зациклится. Нужны и счётчик, и проверка на прогресс.
• Надежда только на recursion_limit. При превышении LangGraph выбрасывает GraphRecursionError, и ответа нет вовсе. Это аварийный тормоз, а не механизм остановки.
• Молчаливый fallback. Если при ошибке разбора JSON считать данные достаточными, цикл формально работает, но ничего не проверяет.
• Рефлексия знает больше заметок. Без запрета на внешние знания модель отвечает «данных достаточно», опираясь на свою память, и проверка теряет смысл.
• Изменяемые списки в состоянии. Возвращайте новые списки, а не вызывайте append на state["notes"], иначе отладка через стрим будет показывать путаницу.

Ограничения

• Метрика качества грубая: она проверяет вхождение подстрок и может ошибиться в обе стороны, например засчитать «два» внутри «двадцать». Для серьёзных выводов нужна ручная разметка или отдельный оценщик.
• Три вопроса и игрушечный корпус демонстрируют механику, но не дают статистически значимого вывода. Расширьте набор под свою задачу.
• Рефлексия оценивает полноту заметок, но не их истинность. Если источник ошибается, цикл этого не заметит.
• В реальном поиске каждый дополнительный проход стоит сетевых запросов и, возможно, денег. Границы из таблицы помогают оценить худший случай заранее.

Вывод

Так нужен ли исследовательскому агенту цикл самопроверки? По устройству графа одно можно сказать без замеров: с ограниченным циклом агент получает возможность честно сказать «данных недостаточно», а линейный пайплайн такой возможности лишён. Цена фиксирована и известна заранее: в нашем графе это максимум два лишних вызова LLM на вопрос. Даёт ли второй проход прирост качества на отвечаемых вопросах, покажет только ваш замер. Начните с MAX_REFLECTIONS = 1 и увеличивайте лимит, только если CSV показывает пользу.

Смежные материалы собраны в разделе гайдов, а определения терминов — в глоссарии.