The earlier RAG note split quality into retrieval and generation. A reflective agent turns that distinction into control flow: decide whether current information is needed, search when necessary, write an answer, then determine whether the evidence or the answer needs another pass.
LangGraph is useful here because the loop is explicit. Nodes perform work; edges decide what runs next; shared state records what the system already tried. “Agentic” becomes a graph you can inspect instead of a vibe emitted by a while loop.
State is a retention policy #
The state carries conversation messages, the current answer, search queries, search results, attempt count, and improvement advice. Different fields need different update behavior:
class GraphState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
response: Optional[str]
search_queries: Annotated[list[str], operator.add]
search_results: Annotated[list[SearchResult], merge_search_results]
attempt: int
search_improvement_advice: Optional[str]
answer_improvement_advice: Optional[str]Conversation history accumulates. Search queries also accumulate so the model can avoid repeating the same query. Search results, however, keep only the latest attempt while merging parallel results. Keeping every fetched page forever would turn the context window into a recycling bin.
Give each node one decision #
The graph contains six responsibilities:
- Decide whether the question requires current web information.
- Generate one or more search queries.
- Execute searches in parallel and retrieve page content.
- Generate an evidence-grounded answer, or answer directly when search is unnecessary.
- Evaluate the answer and its supporting search results.
- Finish, retry retrieval, or retry generation.
The evaluation node returns structured output:
class AnswerEvaluation(BaseModel):
is_satisfactory: bool
need: Literal["search", "generate"] | None
reason: str
feedback: str | NoneIf the evidence does not contain what the question needs, route to query generation. If the evidence is sufficient but the answer ignores or mangles it, route only to answer generation. Re-running search for a prose problem wastes time; rewriting prose over missing evidence merely produces a more polished absence.
Bound the reflection loop #
The evaluator checks search relevance first, answer use of evidence second, and overall clarity last. It produces concrete feedback for the next node. The implementation stops after three attempts even if the evaluator remains dissatisfied.
That limit is essential. Self-reflection is another model call, not self-awareness. The evaluator can be wrong, disagree with itself, or request an impossible improvement. Bounded retries convert that uncertainty into predictable latency and cost.
The answer prompt also enforces useful constraints: rely on retrieved material, lead with a direct answer, preserve conversation context, include concrete names and dates, and say when evidence is incomplete.
Wire and run the graph #
graph = StateGraph(GraphState)
graph.add_node(should_web_search)
graph.add_node(generate_search_queries)
graph.add_node(execute_search)
graph.add_node(generate_answer_from_search)
graph.add_node(generate_answer)
graph.add_node(evaluate_answer)
graph.add_edge(START, "should_web_search")
app = graph.compile()
result = await app.ainvoke({
"messages": [HumanMessage(content="What is today's weather in Tokyo?")],
"attempt": 0,
})For a current-weather question, the agent chooses search, creates a dated query, retrieves weather sources, writes an answer, and evaluates whether the claims are supported. The architecture does not guarantee truth. It gives failures somewhere specific to go.