Building Your First Stateful Agent with LangGraph

This is the second post in our LangGraph fundamentals series, building on why graphs instead of chains.

The previous post's example was intentionally simple — a straight line, to introduce nodes, edges, and state without any real complexity yet. This post builds the workflow that actually motivates reaching for LangGraph in the first place: generate an answer, evaluate whether it's actually good enough, and loop back to regenerate if it isn't — a genuine loop with real state carried across every attempt, which a linear chain has no natural way to express.

Defining state that accumulates across a loop

from typing import TypedDict
from langgraph.graph import StateGraph, END
 
class ResearchState(TypedDict):
    question: str
    draft_answer: str
    quality_score: int
    attempts: int

attempts here is the detail that matters: this field needs to persist and increment across multiple passes through the same nodes — exactly what a shared, persistent state object (rather than each step's output only feeding the immediately next step, as in a plain chain) is actually for.

Writing the nodes

def draft(state: ResearchState) -> ResearchState:
    response = model.invoke(f"Answer this question: {state['question']}")
    state["draft_answer"] = response.content
    state["attempts"] += 1
    return state
 
def evaluate(state: ResearchState) -> ResearchState:
    eval_prompt = (
        f"Rate this answer's quality from 1-10, respond with just the number.\n\n"
        f"Question: {state['question']}\nAnswer: {state['draft_answer']}"
    )
    response = model.invoke(eval_prompt)
    state["quality_score"] = int(response.content.strip())
    return state

Each node is an ordinary function taking the current state and returning an updated version of it — nothing about a node itself is LangGraph-specific; it's the graph structure connecting them that provides the actual looping behavior.

The conditional edge: where the actual loop happens

def should_continue(state: ResearchState) -> str:
    if state["quality_score"] >= 7:
        return "end"
    if state["attempts"] >= 3:
        return "end"  # give up after 3 attempts, don't loop forever
    return "retry"
 
graph = StateGraph(ResearchState)
graph.add_node("draft", draft)
graph.add_node("evaluate", evaluate)
 
graph.set_entry_point("draft")
graph.add_edge("draft", "evaluate")
graph.add_conditional_edges(
    "evaluate",
    should_continue,
    {"retry": "draft", "end": END},
)
 
app = graph.compile()
result = app.invoke({"question": "Explain quantum computing simply.", "draft_answer": "", "quality_score": 0, "attempts": 0})
print(f"Final answer (after {result['attempts']} attempts): {result['draft_answer']}")

Final answer (after 2 attempts): Quantum computing uses quantum bits,
or qubits, which can represent both 0 and 1 simultaneously...

add_conditional_edges is the actual mechanism making this a graph rather than a chain: after evaluate runs, should_continue inspects the current state and returns a string key, which determines which node runs next — draft again (a genuine loop back to an earlier step) or END (finishing the graph entirely). This is precisely the shape a plain LCEL chain has no clean way to express: the next step depends on a runtime decision based on the accumulated state, not a fixed position in a fixed sequence.

The safeguard that matters as much as the loop itself

Notice should_continue checks attempts >= 3 before checking whether quality is still insufficient. Without an explicit cap like this, a workflow where quality genuinely never reaches the target threshold loops indefinitely — the exact same "bound your loop before you rely on it converging" instinct that matters for the retry-with-backoff logic covered in our OpenAI API post, or the max_iterations safeguard on a LangChain agent, covered in our previous series. Any loop driven by a runtime, model-judged condition needs an explicit, unconditional exit, not just an optimistic assumption that the condition will eventually be satisfied.

Inspecting state at each step for debugging

Because a stateful loop can run an unpredictable number of times, it's worth being able to inspect exactly what happened on each pass, not just the final result:

for step in app.stream({"question": "Explain quantum computing simply.", "draft_answer": "", "quality_score": 0, "attempts": 0}):
    node_name = list(step.keys())[0]
    print(f"After '{node_name}': attempts={step[node_name].get('attempts')}, score={step[node_name].get('quality_score')}")

After 'draft': attempts=1, score=0
After 'evaluate': attempts=1, score=5
After 'draft': attempts=2, score=5
After 'evaluate': attempts=2, score=8

app.stream() (rather than app.invoke()) yields the state after each individual node runs, not just the final result — genuinely useful for confirming a loop is actually converging as expected, and for catching a case where quality_score never actually improves across attempts, which would otherwise only show up as "it took the maximum number of attempts and still wasn't great," with no visibility into whether each retry was making genuine progress.

Why this genuinely needed a graph

Try expressing this same "draft, evaluate, conditionally loop back" workflow as a plain LCEL chain, and there's no clean way to do it — a chain has no built-in concept of "go back to an earlier step based on this step's result." An AgentExecutor's loop (from our LangChain series) could approximate something similar by treating "regenerate" as a tool the agent chooses to call, but that overloads the tool-calling mechanism for something that's really a structural control-flow decision, not an action taken in the world. LangGraph's conditional edges express this pattern directly, as the actual structural feature it is.

Passing configuration into nodes without threading it through state

Not every value a node needs belongs in the workflow's actual state — a model instance, an API client, or a configuration flag is often better passed through LangGraph's config mechanism, kept separate from the data the graph is actually reasoning about:

from langgraph.graph import StateGraph, END
from typing import TypedDict
 
class AgentState(TypedDict):
    question: str
    draft_answer: str
 
def draft(state: AgentState, config: dict) -> AgentState:
    model_name = config["configurable"].get("model_name", "gpt-4o")
    response = invoke_model(model_name, state["question"])
    state["draft_answer"] = response
    return state
 
app = graph.compile()
result = app.invoke(
    {"question": "Explain Kubernetes.", "draft_answer": ""},
    config={"configurable": {"model_name": "gpt-4o-mini"}},
)

This keeps AgentState focused on the actual data flowing through the workflow — the question, the draft, the quality score — while runtime configuration (which model to use, which API credentials, a feature flag) travels through a separate channel, the same "don't mix configuration into state that shouldn't have to know about it" instinct as keeping Terraform variables and Kubernetes ConfigMaps distinct from the actual application data they configure.

Debugging a stuck or unexpectedly long-running graph

A graph that seems to hang is usually either a conditional edge routing back to retry more times than expected, or a node itself blocking on a slow external call. app.stream(), introduced above, is the first diagnostic step — but for a graph that's genuinely stuck rather than just slow, checking the current state directly (app.get_state(config).values) shows exactly what data the graph has accumulated so far, which node it's currently on, and — combined with the attempts counter from this post's own example — whether it's looping as expected or stuck in a pattern that never actually satisfies its exit condition.

Persisting a stateful loop's progress across restarts

The retry loop in this post lives entirely in memory during a single invoke() call — fine for a short-lived loop, but the same checkpointing mechanism covered in depth in the human-in-the-loop post later in this series applies equally here: compiling the graph with a checkpointer means a long-running retry loop's progress survives a process restart, resumable from exactly the attempt it was on rather than starting over from zero.

What to actually remember from this post

• A shared state object is what lets attempts (and other accumulated values) persist correctly across multiple loop iterations — not something a plain chain naturally supports.
• add_conditional_edges is the actual loop mechanism — a function inspects state and returns a key determining which node runs next, including looping back to an earlier one.
• Always cap a runtime-driven loop with an explicit, unconditional exit — an "evaluate and retry" loop with no attempt limit can run indefinitely if the quality bar is never met.
• app.stream() exposes state after every individual node, not just the final result — genuinely useful for confirming a loop is converging rather than just repeating without improvement.
• This pattern (generate, evaluate, conditionally retry) is a genuine example of something a plain chain can't cleanly express — the actual justification for reaching for a graph here, not just added complexity for its own sake.

Next in the series: LangGraph Conditional Edges and Control Flow, where we go deeper on branching patterns beyond the simple retry loop shown here.