RetrievalState and the state graph

LangGraph is a state machine library. You define a TypedDict that holds everything the graph needs to know, then define nodes that read from it and return partial updates. The library handles the transitions. For our retrieval agent, the state captures the query, the current node, the path taken, and the retrieved content.

retriever.py
python
from typing import Annotated, List, Optional, TypedDict
import operator
from tree import TreeNode

class RetrievalState(TypedDict):
    query: str
    current_node: Optional[TreeNode]
    tree: TreeNode
    path_taken: Annotated[List[str], operator.add]
    retrieved_content: Annotated[List[str], operator.add]
    reasoning: str
    confidence: float
    should_descend: bool
    target_child_id: Optional[str]
    depth: int
    final_answer: Optional[str]
    call_log: Annotated[List[dict], operator.add]

Note the Annotated fields with operator.add. That tells LangGraph to append new values rather than overwrite. Every time a node returns a list, it gets concatenated onto the running state. This is how we accumulate the path and the call log.

The Annotated syntax is LangGraph's reducer pattern. Unannotated fields are overwritten on every update. Annotated fields are combined using the supplied function. For path_taken and retrieved_content, we want append semantics, so operator.add does the job.

Retrieval state graph

Four nodes, conditional routing, and a loop back to analyze when the agent descends.

retriever.py
python
from langgraph.graph import StateGraph, END

def _build_graph(client, model):
    workflow = StateGraph(RetrievalState)

    workflow.add_node("analyze",  _make_analyze(client, model))
    workflow.add_node("descend",  _make_descend())
    workflow.add_node("retrieve", _make_retrieve())
    workflow.add_node("generate", _make_generate(client, model))

    workflow.set_entry_point("analyze")
    workflow.add_conditional_edges(
        "analyze", _route,
        {"descend": "descend", "retrieve": "retrieve", "end": END},
    )
    workflow.add_edge("descend",  "analyze")
    workflow.add_edge("retrieve", "generate")
    workflow.add_edge("generate", END)

    return workflow.compile()

Graph assembly is declarative. Register nodes, set the entry point, add edges. The conditional edge from analyze calls a router function that picks the next node based on the current state.

Quiz: Quiz

Loading practice…