Нужен ли исследовательскому агенту цикл самопроверки: testing a bounded Research–Reflect loop in LangGraph

Нужен ли исследовательскому агенту цикл самопроверки
Temporary fallback cover; replace in editorial pass.

Updated · Level: intermediate · Reading and setup time: about 60 minutes

A linear research pipeline — plan queries, search, write — always produces an answer, including when the evidence it collected does not support one. The usual fix is to add a self-check step that sends the agent back for more searching. That fix has a cost: every extra round means more model calls, more latency, and, if nothing caps it, a loop that may never end. This article does not settle the question with opinion. You will build both versions on the same model and corpus, give the loop four explicit stop conditions, and produce a comparison table from your own runs.

What you will have at the end

The real problem: two failure modes pulling in opposite directions

Failure mode 1 — the confident linear writer. In a linear pipeline the writer step runs no matter what retrieval returned. Suppose the planner's first queries find a document saying "release 4.2 shipped on the date recorded in ticket NV-118", but never find the ticket itself. A writer told to "answer the question" can do one of three things: say it does not know, quote the indirection without resolving it, or fill the gap with a plausible date — a hallucination. The pipeline has no step where the gap is spotted and acted on.

Failure mode 2 — the unbounded reviewer. Add a step that asks "is this evidence enough?" and routes back to search when the answer is no, and you have bought a new set of problems:

The useful question, then, is not "reflection: yes or no?". It is: does a reflection loop with hard bounds improve answers enough on your questions to justify the calls and time it adds? The rest of the article builds a test that answers that.

The concrete case: a synthetic corpus with planted gaps

The corpus is fictional: an internal data pipeline called "Kestrel" at a made-up company "Norvale". We use invented facts on purpose. The model cannot know them from pretraining, so the only way to answer correctly is to retrieve them. On public topics, a correct answer could come from the model's memory, and you would be measuring that memory instead of the loop.

The corpus is built so that some answers are spread across several documents linked by references (release → ticket → date), and one question has no answer in the corpus at all.

Step 0. Environment

You need Python 3.10+ and a local model served by Ollama. Running locally means no API keys and makes repeated runs cheap. If you prefer a hosted model, swap the chat model class in llm.py. Keep everything else the same so the comparison stays fair.

mkdir reflection-test && cd reflection-test
python -m venv .venv
source .venv/bin/activate          # Windows: .venv\Scripts\activate
pip install langgraph langchain-core langchain-ollama pydantic pytest
pip freeze > requirements.lock.txt # record the exact versions you tested

# any instruction-tuned model that supports structured output will do;
# this tag is an example, not a recommendation backed by our measurements
ollama pull qwen2.5:7b-instruct

Write down the model tag, the version list in requirements.lock.txt and your hardware. You cannot interpret the time column without them.

Step 1. Corpus, questions and a deterministic search tool

corpus.json:

[
  {"id": "k-01", "text": "Kestrel is an internal data pipeline at Norvale. Kestrel replaced the Heron scheduler in release 4.2."},
  {"id": "k-02", "text": "Release 4.2 of Kestrel shipped on the date recorded in change ticket NV-118."},
  {"id": "k-03", "text": "Change ticket NV-118 was approved by the platform board. Deployment window: March 14."},
  {"id": "k-04", "text": "The Heron scheduler was retired because its retry policy caused duplicate exports."},
  {"id": "k-05", "text": "Kestrel retry policy: at most 3 retries with exponential backoff starting at 30 seconds."},
  {"id": "k-06", "text": "The Kestrel on-call rotation is owned by the Ingest team."},
  {"id": "k-07", "text": "Ingest team staffing is tracked in a separate staffing sheet that is not part of this archive."},
  {"id": "k-08", "text": "Duplicate exports were detected by the reconciliation job named Tally."}
]

questions.json:

[
  {"id": "q1-single", "question": "What is Kestrel's retry policy?",
   "required_facts": ["3 retries", "30 seconds"], "answerable": true},
  {"id": "q2-two-hop", "question": "Why was the Heron scheduler retired, and which job detected the problem?",
   "required_facts": ["duplicate exports", "tally"], "answerable": true},
  {"id": "q3-three-hop", "question": "On what date was the Kestrel release that replaced Heron deployed?",
   "required_facts": ["4.2", "march 14"], "answerable": true},
  {"id": "q4-unanswerable", "question": "Who is the lead of the team that owns the Kestrel on-call rotation?",
   "required_facts": [], "answerable": false}
]

The search tool scores by keyword overlap and breaks ties by document id. It is deliberately crude: it returns the same results for the same query every time, so any difference between runs comes from the agent and not from the retriever.

search.py:

import json
import re
from pathlib import Path

DOCS = json.loads(Path(__file__).with_name("corpus.json").read_text(encoding="utf-8"))


def tokenize(text: str) -> set[str]:
    return set(re.findall(r"[a-z0-9]+", text.lower()))


def search(query: str, k: int = 3) -> list[dict]:
    q = tokenize(query)
    scored = []
    for doc in DOCS:
        overlap = len(q & tokenize(doc["text"]))
        if overlap:
            scored.append((overlap, doc))
    scored.sort(key=lambda pair: (-pair[0], pair[1]["id"]))
    return [doc for _, doc in scored[:k]]

Check it before adding any model:

python -c "from search import search; print([d['id'] for d in search('NV-118 deployment')])"

Given the corpus above, k-03 should appear first, since it shares the most tokens with the query. If it does not, fix the corpus or the tokenizer now. Debugging retrieval through an agent is much harder.

Step 2. Shared state with counters

All nodes read and write one state object. Counters live in the state rather than in logs, so every run reports its own cost. Fields annotated with operator.add are reducers: when a node returns {"llm_calls": 1}, LangGraph adds 1 to the stored value instead of replacing it.

state.py:

import operator
from typing import Annotated, TypedDict


class ResearchState(TypedDict):
    question: str
    pending_queries: list[str]
    queries_run: Annotated[list[str], operator.add]
    evidence: Annotated[list[dict], operator.add]
    new_evidence_last_round: int
    iteration: int
    llm_calls: Annotated[int, operator.add]
    search_calls: Annotated[int, operator.add]
    verdict: str
    missing: list[str]
    stop_reason: str
    answer: str


def initial_state(question: str) -> dict:
    return {
        "question": question,
        "pending_queries": [],
        "queries_run": [],
        "evidence": [],
        "new_evidence_last_round": 0,
        "iteration": 0,
        "llm_calls": 0,
        "search_calls": 0,
        "verdict": "",
        "missing": [],
        "stop_reason": "",
        "answer": "",
    }

llm.py:

import os

from langchain_ollama import ChatOllama

MODEL = os.environ.get("AGENT_MODEL", "qwen2.5:7b-instruct")
llm = ChatOllama(model=MODEL, temperature=0)

temperature=0 cuts down run-to-run variation but does not remove it. That is why the harness repeats every question.

Step 3. Nodes: Plan, Research, Reflect, Write

Both modes use the same Plan, Research and Write nodes. The only difference is whether a Reflect node sits between Research and Write. Without that, the comparison would also be measuring prompt differences.

The reviewer returns structured output (a Pydantic model), so routing is decided by code and not by parsing free text. Reflect also chooses the stop reason, so routing reduces to a single if.

nodes.py:

import os

from langchain_core.messages import HumanMessage, SystemMessage
from pydantic import BaseModel, Field

from llm import llm
from search import search

MAX_ITERATIONS = int(os.environ.get("MAX_ITERATIONS", "3"))
MAX_LLM_CALLS = int(os.environ.get("MAX_LLM_CALLS", "8"))
MAX_QUERIES_PER_ROUND = 3


class Plan(BaseModel):
    queries: list[str] = Field(description="1-3 short keyword search queries")


class Reflection(BaseModel):
    sufficient: bool = Field(description="True only if the evidence answers every part of the question")
    missing: list[str] = Field(default_factory=list, description="What is still unknown")
    next_queries: list[str] = Field(default_factory=list, description="Up to 3 new keyword queries")


def format_evidence(evidence: list[dict]) -> str:
    return "\n".join(f"[{e['doc_id']}] {e['text']}" for e in evidence) or "(no evidence)"


def plan(state: dict) -> dict:
    messages = [
        SystemMessage("You plan searches over a small internal document archive. "
                      "Return 1-3 short keyword queries."),
        HumanMessage(state["question"]),
    ]
    try:
        queries = llm.with_structured_output(Plan).invoke(messages).queries[:MAX_QUERIES_PER_ROUND]
    except Exception:
        queries = []
    return {"pending_queries": queries or [state["question"]], "llm_calls": 1}


def research(state: dict) -> dict:
    already = set(state["queries_run"])
    seen = {e["doc_id"] for e in state["evidence"]}
    ran, new_evidence = [], []
    for query in state["pending_queries"]:
        key = query.strip().lower()
        if key in already:
            continue
        already.add(key)
        ran.append(key)
        for doc in search(query):
            if doc["id"] not in seen:
                seen.add(doc["id"])
                new_evidence.append({"doc_id": doc["id"], "text": doc["text"], "query": query})
    return {
        "evidence": new_evidence,
        "queries_run": ran,
        "search_calls": len(ran),
        "iteration": state["iteration"] + 1,
        "new_evidence_last_round": len(new_evidence),
    }


def reflect(state: dict) -> dict:
    calls_after = state["llm_calls"] + 1
    messages = [
        SystemMessage("You check whether the evidence fully answers the question. Judge only by the "
                      "evidence text, not by your own knowledge. If something is missing, name it and "
                      "propose up to 3 new keyword queries that differ from the queries already run. "
                      "Follow references: if a document points to a ticket, release or name, search for it."),
        HumanMessage(f"Question: {state['question']}\n\n"
                     f"Queries already run: {state['queries_run']}\n\n"
                     f"Evidence:\n{format_evidence(state['evidence'])}"),
    ]
    try:
        review = llm.with_structured_output(Reflection).invoke(messages)
    except Exception:
        review = Reflection(sufficient=False, missing=["reviewer output could not be parsed"])

    run_before = set(state["queries_run"])
    fresh = [q for q in review.next_queries if q.strip().lower() not in run_before][:MAX_QUERIES_PER_ROUND]

    if review.sufficient:
        stop = "sufficient"
    elif state["iteration"] >= MAX_ITERATIONS:
        stop = "max_iterations"
    elif calls_after + 2 > MAX_LLM_CALLS:      # one more Reflect and the final Write must still fit
        stop = "budget"
    elif not fresh:
        stop = "no_new_queries"
    elif state["iteration"] > 1 and state["new_evidence_last_round"] == 0:
        stop = "stagnation"
    else:
        stop = ""

    return {
        "llm_calls": 1,
        "verdict": "sufficient" if review.sufficient else "insufficient",
        "missing": review.missing,
        "pending_queries": fresh,
        "stop_reason": stop,
    }


def route_after_reflect(state: dict) -> str:
    return "write" if state["stop_reason"] else "research"


def write(state: dict) -> dict:
    gaps = ""
    if state.get("verdict") == "insufficient":
        gaps = "\n\nA reviewer found these gaps: " + "; ".join(state["missing"])
    messages = [
        SystemMessage("Answer using only the evidence. Cite every claim with its [doc-id]. If the evidence "
                      "does not answer part of the question, write 'Insufficient evidence:' and name what is "
                      "missing. Do not guess."),
        HumanMessage(f"Question: {state['question']}\n\n"
                     f"Evidence:\n{format_evidence(state['evidence'])}{gaps}"),
    ]
    return {"answer": llm.invoke(messages).content, "llm_calls": 1}

Why four stop conditions, and in this order

  1. sufficient — the normal exit.
  2. max_iterations — a hard cap on research rounds. It guards against a reviewer that is never satisfied.
  3. budget — a cap on total LLM calls. It reserves room for one more review and the final write, so the agent never runs out of budget before it has produced an answer.
  4. no_new_queries / stagnation — catch a reviewer that rephrases the same query, or a corpus with nothing more to give. These are the cheapest exits and fire most often on unanswerable questions.

A parse failure in the reviewer does not loop indefinitely. It produces an empty next_queries, which leads to no_new_queries, which leads to Write. That is deliberate.

With the defaults (MAX_ITERATIONS=3, MAX_LLM_CALLS=8), the worst case for the reflective mode is 1 Plan + 3 Reflect + 1 Write = 5 LLM calls and at most 9 search calls. The linear mode always makes 2 LLM calls (Plan + Write) and at most 3 search calls. You can work out these ceilings before running anything; the measurements then show how often real runs get close to them.

Step 4. Two graphs from the same nodes

graph.py:

from langgraph.graph import END, START, StateGraph

from nodes import plan, reflect, research, route_after_reflect, write
from state import ResearchState


def build_linear():
    g = StateGraph(ResearchState)
    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("research", "write")
    g.add_edge("write", END)
    return g.compile()


def build_reflective():
    g = StateGraph(ResearchState)
    g.add_node("plan", plan)
    g.add_node("research", research)
    g.add_node("reflect", reflect)
    g.add_node("write", write)
    g.add_edge(START, "plan")
    g.add_edge("plan", "research")
    g.add_edge("research", "reflect")
    g.add_conditional_edges("reflect", route_after_reflect, {"research": "research", "write": "write"})
    g.add_edge("write", END)
    return g.compile()


if __name__ == "__main__":
    print(build_reflective().get_graph().draw_mermaid())

Run python graph.py and check the printed Mermaid diagram: there should be exactly one cycle, research → reflect → research, and exactly one way out of it, reflect → write.

When invoking, set a recursion limit as a second, independent safeguard. With three iterations, the reflective graph takes about 8 steps. A limit of 25 should never be reached, so if you see GraphRecursionError, your own stop logic has a bug. A limit set close to the expected step count would instead hide that bug.

Step 5. Prove the loop terminates — without a model

The termination guarantee is the main point of this design, so test it with a stub model that behaves in the ways you are worried about. These tests run in under a second and need no Ollama.

test_loop_bounds.py:

import itertools

import nodes
from graph import build_reflective
from state import initial_state


class StubLLM:
    def __init__(self, make_reflection):
        self.make_reflection = make_reflection

    def with_structured_output(self, schema):
        outer = self

        class Runner:
            def invoke(self, _messages):
                if schema is nodes.Plan:
                    return nodes.Plan(queries=["kestrel"])
                return outer.make_reflection()

        return Runner()

    def invoke(self, _messages):
        class Message:
            content = "Insufficient evidence: stub answer"
        return Message()


def run_with(make_reflection, monkeypatch):
    monkeypatch.setattr(nodes, "llm", StubLLM(make_reflection))
    return build_reflective().invoke(initial_state("stub question"), config={"recursion_limit": 25})


def test_satisfied_reviewer_stops_after_one_round(monkeypatch):
    final = run_with(lambda: nodes.Reflection(sufficient=True), monkeypatch)
    assert final["stop_reason"] == "sufficient"
    assert final["iteration"] == 1
    assert final["llm_calls"] == 3  # plan + reflect + write


def test_reviewer_repeating_itself_stops(monkeypatch):
    final = run_with(lambda: nodes.Reflection(sufficient=False, next_queries=["Kestrel"]), monkeypatch)
    assert final["stop_reason"] == "no_new_queries"


def test_never_satisfied_reviewer_is_capped(monkeypatch):
    counter = itertools.count(1)
    final = run_with(
        lambda: nodes.Reflection(sufficient=False, next_queries=[f"kestrel variant {next(counter)}"]),
        monkeypatch,
    )
    assert final["stop_reason"] in {"max_iterations", "budget", "stagnation"}
    assert final["iteration"] <= nodes.MAX_ITERATIONS
    assert final["llm_calls"] <= nodes.MAX_LLM_CALLS
pytest -q test_loop_bounds.py

The second test covers the case where a case-variant of an executed query ("Kestrel" vs "kestrel") is correctly treated as a repeat. The third covers the never-satisfied reviewer. Depending on what the stub queries retrieve, it can exit through stagnation or max_iterations, so it asserts on the bounds and does not require one particular reason.

Step 6. Evaluation harness

The evaluation is deterministic and uses no LLM judge. Each answer gets these scores:

run_eval.py:

import csv
import json
import re
import sys
import time
from pathlib import Path

from graph import build_linear, build_reflective
from state import initial_state

REPEATS = int(sys.argv[1]) if len(sys.argv) > 1 else 3
QUESTIONS = json.loads(Path("questions.json").read_text(encoding="utf-8"))
GRAPHS = {"linear": build_linear(), "reflective": build_reflective()}
CITATION = re.compile(r"\[(k-\d+)\]")


def score(question: dict, final: dict) -> dict:
    answer = final["answer"].lower()
    cited = set(CITATION.findall(final["answer"]))
    retrieved = {e["doc_id"] for e in final["evidence"]}
    row = {"citations": len(cited), "citations_valid": int(cited <= retrieved)}
    if question["answerable"]:
        hits = sum(fact.lower() in answer for fact in question["required_facts"])
        row["fact_recall"] = round(hits / len(question["required_facts"]), 3)
        row["abstained"] = ""
    else:
        row["fact_recall"] = ""
        row["abstained"] = int("insufficient evidence" in answer)
    return row


rows = []
with open("answers.jsonl", "w", encoding="utf-8") as log:
    for repeat in range(REPEATS):
        for question in QUESTIONS:
            for mode, graph in GRAPHS.items():
                started = time.perf_counter()
                final = graph.invoke(initial_state(question["question"]), config={"recursion_limit": 25})
                seconds = round(time.perf_counter() - started, 2)
                row = {
                    "mode": mode, "question": question["id"], "repeat": repeat, "seconds": seconds,
                    "llm_calls": final["llm_calls"], "search_calls": final["search_calls"],
                    "iterations": final["iteration"], "stop_reason": final["stop_reason"] or "linear",
                    **score(question, final),
                }
                rows.append(row)
                log.write(json.dumps({**row, "answer": final["answer"],
                                      "queries_run": final["queries_run"]}, ensure_ascii=False) + "\n")
                print(row)

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

Inside each repeat, the two modes alternate per question. If the model server warms up over time or your machine slows down, both modes are affected about equally. Run a warm-up call before timing anything — the first request to Ollama often includes model loading time:

ollama run qwen2.5:7b-instruct "ok" > /dev/null
python run_eval.py 5

aggregate.py:

import csv
from collections import Counter, defaultdict
from statistics import mean, median

rows = list(csv.DictReader(open("results.csv", encoding="utf-8")))
by_mode = defaultdict(list)
for r in rows:
    by_mode[r["mode"]].append(r)

print("| Metric | " + " | ".join(by_mode) + " |")
print("|---|" + "---|" * len(by_mode))


def line(name, fn):
    print(f"| {name} | " + " | ".join(fn(rs) for rs in by_mode.values()) + " |")


line("Mean fact recall (answerable)",
     lambda rs: f"{mean(float(r['fact_recall']) for r in rs if r['fact_recall']):.2f}")
line("Abstention rate (unanswerable)",
     lambda rs: f"{mean(int(r['abstained']) for r in rs if r['abstained'] != ''):.2f}")
line("Citation validity rate", lambda rs: f"{mean(int(r['citations_valid']) for r in rs):.2f}")
line("Median seconds per question", lambda rs: f"{median(float(r['seconds']) for r in rs):.1f}")
line("Median LLM calls", lambda rs: f"{median(int(r['llm_calls']) for r in rs):g}")
line("Max LLM calls", lambda rs: str(max(int(r['llm_calls']) for r in rs)))
line("Median search calls", lambda rs: f"{median(int(r['search_calls']) for r in rs):g}")
line("Stop reasons", lambda rs: ", ".join(f"{k}: {v}" for k, v in Counter(r["stop_reason"] for r in rs).items()))
python aggregate.py

Step 7. The comparison table

This is the table aggregate.py produces. The cells are empty on purpose: fill them in from your own run, and record the model tag, repeat count and hardware next to the table.

Linear vs. bounded reflective agent — fill in from python aggregate.py
Metriclinearreflective
Mean fact recall (answerable)——
Abstention rate (unanswerable)——
Citation validity rate——
Median seconds per question——
Median LLM calls2 (fixed by design)—
Max LLM calls2 (fixed by design)— (ceiling 5 with defaults)
Median search calls— (ceiling 3)— (ceiling 9)
Stop reasonslinear—

Run details to record: model ___, Ollama version ___, repeats ___, CPU/GPU ___, date ___.

How to read your table

Do not stop at the averages. Look at recall per question in results.csv:

A decision rule you can apply

Before you run anything, write down the minimum improvement that would justify the extra cost — for example, "recall on multi-hop questions must improve by at least X, and median latency may grow by no more than Y". Then compare. If you set the threshold after looking at the numbers, any result can be made to look like a win. If reflection only helps on multi-hop questions, a reasonable design is to keep the linear path by default and let a cheap classifier, or the first reviewer verdict, decide whether to enter the loop.

Verification checklist

  1. pytest -q passes: all three termination tests are green.
  2. python graph.py shows a single cycle with a single exit.
  3. In results.csv, no reflective row exceeds MAX_LLM_CALLS or MAX_ITERATIONS, and every linear row has llm_calls == 2.
  4. Every row has a non-empty stop_reason. An empty one in reflective mode would mean Write was reached without a decision, which the graph should make impossible.
  5. Read at least five answers from each mode in answers.jsonl by hand. Substring matching can be fooled — "not 3 retries" contains "3 retries" — so the score should agree with what you read.
  6. Rerun with MAX_ITERATIONS=1. The reflective mode should then behave like linear plus one review. That is a quick check that the cap is actually wired in.

Failure cases and what they look like

The rubber-stamp reviewer

Symptom: almost every reflective run stops with sufficient after round 1, and multi-hop recall equals linear. The reflective mode has added one call and changed nothing. Fix: make the reviewer prompt require a list of each sub-question with the doc-id that answers it, and set sufficient to true only when every item has one. A clause like "Follow references" in the prompt matters on this corpus.

The never-satisfied reviewer

Symptom: many max_iterations stops, including on q1. Fix: the caps are doing their job, but you are paying for it. Tighten the reviewer's criteria, or lower MAX_ITERATIONS to 2 and measure again.

Query paraphrasing

Symptom: queries_run contains "kestrel release date", "release date of kestrel", "kestrel date released". Exact-match deduplication will not catch these. stagnation eventually will, because they return the same documents, but only after a wasted round. Fix: dedupe on sorted token sets instead of raw strings, or reject a query whose top-k results are already all in evidence.

Structured output parse failures

Symptom: missing contains "reviewer output could not be parsed". Small local models fail at JSON schemas more often than large hosted ones. Fix: count these failures separately — add a field if they are frequent — and treat a high rate as a reason to change the model, not something to hide in the averages.

The hedging writer

Symptom: once the reviewer's gap list reaches the writer, answers to answerable questions start with "Insufficient evidence:" even though all the facts were retrieved. Recall may still score well, because the facts are present, but the answer is worse to read. Fix: pass gaps to the writer only when stop_reason is not sufficient (the code already does this), and check by hand.

GraphRecursionError

Symptom: an exception instead of an answer. With a limit of 25 and these caps, this should not happen. If it does, a stop condition is not being reached — usually because iteration is not being incremented, or route_after_reflect reads a field that the node did not set.

Limitations of this test

So: does a research agent need a self-check loop?

Not by default. A bounded loop is worth adding only if your own table shows a real gain on the kind of questions your users ask — typically multi-hop questions over documents that refer to one another — at a latency and call cost you have agreed to in advance. Whichever way you decide, the parts you should keep are the bounds: an iteration cap, a call budget that reserves room for the final answer, and an exit when there is no new evidence. An unbounded loop answers the question no better than a linear pipeline, and it costs more.

Where to go next

We publish what works for us—and implement the same solutions for your business. We design AI automation, Telegram bots, chats, and AI agents for real-world processes. Discuss your project →