2. StateGraph patterns for agents

Overview

In guide #9 you learned StateGraph as a general-purpose tool: nodes, edges, compile, invoke. You know how it works. Now the question is different: how do you use StateGraph specifically to control agents? We're not repeating fundamentals — we're applying them to a concrete domain where nodes represent agent capabilities, edges represent flow control, and the state is the agent's cognitive context.

The key difference is this: in a generic workflow, nodes are "processing steps" (extract data, transform, save). In an agent, nodes are cognitive capabilities — plan, research, analyze, synthesize. Edges are not "step 1 → step 2" but decisions: "do I need more information, or can I answer already?" That distinction changes how you design the graph.

This capsule sets up the fundamental patterns you'll use throughout the module: how to map the perceive-reason-act architecture (which you implemented by hand in M1) to the nodes and edges of a StateGraph, how to design nodes as pure functions that take state and return partial state, and when to choose a custom StateGraph vs create_react_agent. By the end, you'll have the mental foundation to design the Research Agent's state machine in capsule 08.

From general LangGraph to agent architecture

The mental shift

From guide #9 you bring StateGraph, add_node, add_edge, compile, invoke. The API doesn't change. What changes is what the nodes and edges represent when you model an agent.

Mapping perceive-reason-act to StateGraph

In M1 you implemented the ReAct loop by hand with a for. Now that same loop becomes a graph:

Manual loop (M1):                    StateGraph (M4):

for i in range(max_iter):     →      START → reason
    response = llm(messages)                    ↓
    if no tool_calls:                    tool_calls?
        return response          No →    END
    execute tools                Yes →   tools → reason

The behavior is identical. The advantage of the graph is that you can:

• Visualize it — draw_mermaid_png() generates a diagram of your agent
• Add nodes — insert planning, analysis, validation without rewriting the loop
• Complex conditional routing — route by multiple conditions without nested ifs
• Checkpointing — save the agent's state between iterations (M6)
• Subgraphs — encapsulate parts of the agent as reusable modules (capsule 07)

That's the point of this module: moving from an imperative loop to a declarative architecture that scales.

Nodes as agent capabilities

Principle: one node = one capability

In an agent modeled as a StateGraph, each node represents a functional capability. It's not "step 1, step 2" — it's "this node knows how to reason", "this node knows how to execute tools", "this node knows how to plan".

The typical capabilities of an agent:

Nodes as pure functions

Each node is a function that receives the full state and returns a partial state — only the fields that change. LangGraph does the merge automatically.

from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

def reasoning_node(state: AgentState) -> dict:
    """Capability: reasoning. Invokes the LLM with the current context."""
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

Notice the pattern:

1. Receives state — the agent's full context
2. Does ONE thing — reason, execute tools, plan
3. Returns a partial — only {"messages": [response]}, not the whole state

This pattern is universal for every node of an agent. You never modify the state directly — you return what changed and LangGraph applies the reducers.

The tool execution node

from langchain_core.messages import ToolMessage

def tool_node(state: AgentState) -> dict:
    last_message = state["messages"][-1]
    results = []
    for tool_call in last_message.tool_calls:
        tool_fn = tools_by_name[tool_call["name"]]
        result = tool_fn.invoke(tool_call["args"])
        results.append(ToolMessage(content=str(result), tool_call_id=tool_call["id"]))
    return {"messages": results}

This node doesn't reason — it only executes. The reason node decides what to do, the tools node does it. That separation lets you insert validation, logging, or rate limiting between the two.

Preview: the Research Agent's nodes

When you build the Research Agent (capsule 08), your nodes will follow this same pattern but specialized: planning_node generates a research plan, research_node runs searches, analysis_node evaluates results, and synthesis_node combines everything into an answer. Each one is a pure function with the same signature: receives state, returns a partial. You can test them in isolation, replace them, or add new ones without affecting the others.

Edges as flow control

Static edges: fixed flow

When you know one step is always followed by another, you use add_edge:

from langgraph.graph import StateGraph, START, END

graph = StateGraph(AgentState)
graph.add_node("plan", planning_node)
graph.add_node("research", research_node)
graph.add_node("analyze", analysis_node)
graph.add_node("synthesize", synthesis_node)

graph.add_edge(START, "plan")
graph.add_edge("plan", "research")
graph.add_edge("research", "analyze")
graph.add_edge("analyze", "synthesize")
graph.add_edge("synthesize", END)

agent = graph.compile()

This is a linear agent: plan → research → analyze → synthesize → done. No cycles, no decisions. Useful as a starting point, but limited — a real agent needs to decide whether it already has enough information or needs another research round.

Conditional edges: decision points

Agents make decisions. add_conditional_edges models those decisions:

def should_continue(state: AgentState) -> str:
    """Decides whether the agent needs to keep acting or can finish."""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return "end"

graph = StateGraph(AgentState)
graph.add_node("reason", reasoning_node)
graph.add_node("tools", tool_node)

graph.add_edge(START, "reason")
graph.add_conditional_edges(
    "reason",
    should_continue,
    {"tools": "tools", "end": END}
)
graph.add_edge("tools", "reason")

agent = graph.compile()

The routing function receives the state and returns a string that maps to a destination node. It's the same logic as the if not response.tool_calls you had in the manual loop of M1, but declared as part of the graph's architecture.

Multiple edges from one node

A node can have several possible destinations:

def route_after_analysis(state: AgentState) -> str:
    """After analyzing, decide the next step."""
    last = state["messages"][-1]
    
    if "NEED MORE DATA" in last.content:
        return "research"
    if "CONTRADICTION DETECTED" in last.content:
        return "verify"
    return "synthesize"

graph.add_conditional_edges(
    "analyze",
    route_after_analysis,
    {
        "research": "research",
        "verify": "verify",
        "synthesize": "synthesize"
    }
)

This is the equivalent of an if/elif/else in your loop, but expressed as part of the graph — visible, debuggable, and extensible. Capsule 05 goes deeper into conditional routing with more advanced patterns.

The minimal agent as a StateGraph

The reason-tools pattern

The simplest possible agent has two nodes and one conditional edge. Putting together everything you've seen (state, nodes, edges):

from dotenv import load_dotenv
load_dotenv()

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage, ToolMessage
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def search(query: str) -> str:
    """Search the web for information."""
    return f"Search results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression."""
    try:
        return str(eval(expression))
    except Exception as e:
        return f"Error: {e}"

tools = [search, calculator]
tools_by_name = {t.name: t for t in tools}
model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

def reasoning_node(state: AgentState) -> dict:
    return {"messages": [model_with_tools.invoke(state["messages"])]}

def tool_node(state: AgentState) -> dict:
    last = state["messages"][-1]
    results = []
    for tc in last.tool_calls:
        result = tools_by_name[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return {"messages": results}

def should_continue(state: AgentState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return "end"

graph = StateGraph(AgentState)
graph.add_node("reason", reasoning_node)
graph.add_node("tools", tool_node)
graph.add_edge(START, "reason")
graph.add_conditional_edges("reason", should_continue, {"tools": "tools", "end": END})
graph.add_edge("tools", "reason")

agent = graph.compile()
result = agent.invoke({"messages": [("user", "What is 25 * 17?")]})
print(result["messages"][-1].content)

Step-by-step flow

When you run agent.invoke(...):

1. START → "reason"
   - reasoning_node receives: [HumanMessage("What is 25 * 17?")]
   - LLM returns: AIMessage with tool_calls=[calculator("25 * 17")]

2. should_continue → "tools" (because there are tool_calls)
   - tool_node runs: calculator("25 * 17") → "425"
   - Returns: [ToolMessage(content="425")]

3. "tools" → "reason"
   - reasoning_node receives: [HumanMessage, AIMessage, ToolMessage("425")]
   - LLM returns: AIMessage(content="25 × 17 = 425")

4. should_continue → "end" (there are no tool_calls)
   - The graph finishes

This flow is exactly the ReAct loop from M1, but expressed as a graph. The logic is the same — the representation changed.

Why the graph if it's the same thing?

The fair question: if the behavior is identical to the manual loop, why complicate it with a graph?

The answer is in what comes next:

The minimal agent doesn't justify the graph on its own. The graph justifies itself when you evolve the agent — and that's exactly what you'll do from module 4 to module 10.

Agent StateGraph vs create_react_agent

What create_react_agent does internally

When you write create_react_agent(model, tools), LangGraph builds a StateGraph almost identical to the minimal agent: START → agent (reasoning) → should_continue → tools or END, with tools → agent for the cycle. It's the same reason-tools pattern — create_react_agent is a convenience function that wraps it.

When to use each

Practical rule: if your agent only needs reason → tools → respond, use create_react_agent. If you need planning, analysis, complex routing, or custom nodes, use StateGraph.

Migrating from one to the other

If you start with create_react_agent and need more control, the migration is direct: you decompose the prebuilt into its equivalent nodes and edges (like the minimal agent you already saw). You lose nothing — you gain control. The custom StateGraph is a superset of create_react_agent.

Integrated example: an agent with planning

Let's look at an agent that goes beyond the minimal pattern — with 4 specialized nodes:

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage, ToolMessage
from langgraph.graph.message import add_messages
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool

@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    return f"[Results for '{query}': relevant information found]"

tools = [web_search]
tools_by_name = {t.name: t for t in tools}
model = init_chat_model("openai:gpt-4.1-mini")
model_with_tools = model.bind_tools(tools)

class ResearchState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

def planner(state: ResearchState) -> dict:
    planning_messages = [
        SystemMessage(content="You are a planner. Generate a 2-3 step plan to research the question."),
        state["messages"][-1]
    ]
    return {"messages": [model.invoke(planning_messages)]}

def researcher(state: ResearchState) -> dict:
    return {"messages": [model_with_tools.invoke(state["messages"])]}

def executor(state: ResearchState) -> dict:
    last = state["messages"][-1]
    results = []
    for tc in last.tool_calls:
        result = tools_by_name[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return {"messages": results}

def synthesizer(state: ResearchState) -> dict:
    msgs = state["messages"] + [HumanMessage(content="Synthesize the information into a clear answer.")]
    return {"messages": [model.invoke(msgs)]}

def route_after_research(state: ResearchState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "execute"
    return "synthesize"

graph = StateGraph(ResearchState)
graph.add_node("plan", planner)
graph.add_node("research", researcher)
graph.add_node("execute", executor)
graph.add_node("synthesize", synthesizer)

graph.add_edge(START, "plan")
graph.add_edge("plan", "research")
graph.add_conditional_edges("research", route_after_research, {"execute": "execute", "synthesize": "synthesize"})
graph.add_edge("execute", "research")
graph.add_edge("synthesize", END)

research_agent = graph.compile()
result = research_agent.invoke({
    "messages": [HumanMessage(content="What are the current trends in AI agents?")]
})
print(result["messages"][-1].content)

This agent's flow

START → plan → research → tool_calls?
                              │
                    Yes → execute → research (cycle)
                    No  → synthesize → END

Four nodes, each with a clear responsibility. The research↔execute cycle allows multiple search rounds. The synthesizer is only invoked when the researcher no longer needs more tools.

This pattern is the foundation of the Research Agent you'll build as the project (capsule 08). The difference will be a richer state (typed state — capsule 03) and more sophisticated routing (capsule 05).

Visualization with draw_mermaid_png

Why visualizing is mandatory

When you design an agent as a StateGraph, visualization isn't optional — it's your main debugging and communication tool. A graph with 4+ nodes and conditional edges becomes hard to reason about in your head. draw_mermaid_png() shows you exactly what nodes exist, how they connect, and where the conditions are.

Generating the visualization

from IPython.display import Image, display

graph_image = research_agent.get_graph().draw_mermaid_png()
display(Image(graph_image))

If you're not in a notebook:

with open("agent_graph.png", "wb") as f:
    f.write(research_agent.get_graph().draw_mermaid_png())

print("Graph saved to agent_graph.png")

Reading the diagram

The diagram shows: each box is a node, the arrows are edges, the forks indicate conditional edges, and __start__/__end__ are the entry and exit points. At a glance you see the complete architecture: how many nodes, where the cycles are, where the decisions are.

When to visualize

Tip: visualize before compiling

You can call graph.get_graph().draw_mermaid_png() before graph.compile(). If the diagram doesn't look like you expected, you fix it before compiling — cheaper than debugging unexpected behavior at runtime.

Connection with the project

Module 4 — Research Agent

In capsule 08 you'll build an AI Research Agent as a StateGraph. This capsule's patterns are the direct foundation:

• Nodes as capabilities: Your Research Agent will have planning, research, analysis, and synthesis nodes — each one a pure function that receives state and returns a partial
• Edges as flow control: Conditional edges will decide whether the agent needs more research or can move on to synthesis
• Visualization: You'll generate your agent's diagram to verify the architecture

Later modules

The state machine you design here will be extended in every module:

Nodes with clear responsibilities and well-typed state allow clean extension. Coupled nodes force rewrites.

Troubleshooting

Problem 1: The agent finishes immediately without executing tools

Cause: The LLM returns a message with no tool_calls on the first iteration. The should_continue function sends it to END.

Solution: Check that the model has the tools bound correctly:

model_with_tools = model.bind_tools(tools)

response = model_with_tools.invoke([HumanMessage(content="Search for information about X")])
print(f"tool_calls: {response.tool_calls}")
print(f"content: {response.content}")

If tool_calls is empty, the prompt may need adjusting so the model uses the available tools.

Problem 2: Error "Node 'X' not found" when compiling

Cause: An edge references a node that wasn't added with add_node.

Solution: Check that every node referenced in an edge exists:

graph.add_node("reason", reasoning_node)
graph.add_node("tools", tool_node)

graph.add_conditional_edges("reason", should_continue, {
    "tools": "tools",
    "end": END
})

The mapping "tools": "tools" requires graph.add_node("tools", ...) to exist. The START and END strings are special and don't need add_node.

Problem 3: The graph enters an infinite loop

Cause: The routing function never returns the end condition, or the LLM always generates tool_calls.

Solution: Add an iteration limit as a parameter of compile():

agent = graph.compile()
result = agent.invoke(
    {"messages": [("user", "question")]},
    {"recursion_limit": 10}
)

recursion_limit is the maximum number of graph steps. If your cycle has 2 nodes (reason + tools), each iteration consumes 2 steps, so recursion_limit=10 allows ~5 agent iterations.

Problem 4: The tool_node fails with a "KeyError" on tools_by_name

Cause: The LLM generates a tool name that doesn't exist in your tools dictionary.

Solution: Validate the name before executing:

def tool_node(state: AgentState) -> dict:
    last = state["messages"][-1]
    results = []
    for tc in last.tool_calls:
        if tc["name"] not in tools_by_name:
            results.append(ToolMessage(
                content=f"Error: tool '{tc['name']}' not available",
                tool_call_id=tc["id"]
            ))
            continue
        result = tools_by_name[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return {"messages": results}

Problem 5: draw_mermaid_png fails or shows nothing

Cause: The grandalf dependency is missing, or the environment doesn't render properly.

Solution: pip install grandalf and save to a file: open("graph.png", "wb").write(agent.get_graph().draw_mermaid_png()). If it persists, use print(agent.get_graph().draw_ascii()) as an alternative.

Exercises

Exercise 1: Identify the pattern (Easy)

Given this code, identify: how many nodes does the agent have? How many static vs conditional edges? What is the cycle?

graph = StateGraph(AgentState)
graph.add_node("reason", reasoning_node)
graph.add_node("tools", tool_node)
graph.add_edge(START, "reason")
graph.add_conditional_edges("reason", should_continue, {"tools": "tools", "end": END})
graph.add_edge("tools", "reason")
agent = graph.compile()

Exercise 2: Add a validation node (Medium)

Modify the minimal agent to add a validate node between tools and reason. The validation node checks that the tool results don't contain "ERROR". If it detects an error, it adds a message reporting the problem.

def validate_node(state: AgentState) -> dict:
    """Validates tool results before passing them to reason."""
    last = state["messages"][-1]
    if isinstance(last, ToolMessage) and "ERROR" in last.content:
        warning = HumanMessage(
            content=f"Warning: the tool result contains an error: {last.content}. Try a different strategy."
        )
        return {"messages": [warning]}
    return {"messages": []}

graph = StateGraph(AgentState)
graph.add_node("reason", reasoning_node)
graph.add_node("tools", tool_node)
graph.add_node("validate", validate_node)

graph.add_edge(START, "reason")
graph.add_conditional_edges("reason", should_continue, {"tools": "tools", "end": END})
graph.add_edge("tools", "validate")
graph.add_edge("validate", "reason")

agent = graph.compile()

Exercise 3: Build a 3-node agent (Medium)

Create an agent with three nodes: receive (adds a system message to the context), reason (reasons and can call tools, with a cycle to an execute node), and respond (generates the final answer). The flow should be: receive → reason ↔ execute (cycle) → respond → END.

from langchain_core.messages import SystemMessage

def receive_node(state: AgentState) -> dict:
    system = SystemMessage(content="You are a research assistant. Use the available tools.")
    return {"messages": [system]}

def reason_node(state: AgentState) -> dict:
    return {"messages": [model_with_tools.invoke(state["messages"])]}

def execute_node(state: AgentState) -> dict:
    last = state["messages"][-1]
    results = []
    for tc in last.tool_calls:
        result = tools_by_name[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return {"messages": results}

def respond_node(state: AgentState) -> dict:
    msgs = state["messages"] + [HumanMessage(content="Generate a concise final answer.")]
    return {"messages": [model.invoke(msgs)]}

def route_reasoning(state: AgentState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "execute"
    return "respond"

graph = StateGraph(AgentState)
graph.add_node("receive", receive_node)
graph.add_node("reason", reason_node)
graph.add_node("execute", execute_node)
graph.add_node("respond", respond_node)

graph.add_edge(START, "receive")
graph.add_edge("receive", "reason")
graph.add_conditional_edges("reason", route_reasoning, {"execute": "execute", "respond": "respond"})
graph.add_edge("execute", "reason")
graph.add_edge("respond", END)

agent = graph.compile()
result = agent.invoke({"messages": [HumanMessage(content="What is 42 * 58?")]})
print(result["messages"][-1].content)

Exercise 4: Visualize and compare (Medium)

Generate the Mermaid visualization for the minimal agent (2 nodes) and for the research_agent from the integrated example (4 nodes). Compare: how many nodes, cycles, and decision points does each one have? What extra information do you get from the 4-node diagram that isn't obvious in the code?

from IPython.display import Image, display

display(Image(agent.get_graph().draw_mermaid_png()))
display(Image(research_agent.get_graph().draw_mermaid_png()))

Exercise 5: From create_react_agent to a custom StateGraph (Hard)

Take this prebuilt agent and convert it into an equivalent custom StateGraph. Add a log node that prints the iteration number before each reasoning step. Use an iteration: int field in the state.

from langgraph.prebuilt import create_react_agent

prebuilt_agent = create_react_agent(model, tools)
result = prebuilt_agent.invoke({"messages": [("user", "What is 100/4?")]})

class LoggedState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    iteration: int

def log_node(state: LoggedState) -> dict:
    current = state.get("iteration", 0) + 1
    print(f"[LOG] Iteration {current} — {len(state['messages'])} messages")
    return {"iteration": current}

def reasoning_logged(state: LoggedState) -> dict:
    return {"messages": [model_with_tools.invoke(state["messages"])]}

def tools_logged(state: LoggedState) -> dict:
    last = state["messages"][-1]
    results = []
    for tc in last.tool_calls:
        result = tools_by_name.get(tc["name"], lambda **_: "Error: tool does not exist").invoke(tc["args"]) if tc["name"] in tools_by_name else f"Error: '{tc['name']}' not available"
        results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return {"messages": results}

def should_continue_logged(state: LoggedState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return "end"

graph = StateGraph(LoggedState)
graph.add_node("log", log_node)
graph.add_node("reason", reasoning_logged)
graph.add_node("tools", tools_logged)

graph.add_edge(START, "log")
graph.add_edge("log", "reason")
graph.add_conditional_edges("reason", should_continue_logged, {"tools": "tools", "end": END})
graph.add_edge("tools", "log")

custom_agent = graph.compile()
result = custom_agent.invoke({"messages": [("user", "What is 100/4?")], "iteration": 0})
print(result["messages"][-1].content)

Summary

In this capsule you learned:

• StateGraph for agents is no different in API from general StateGraph — the difference is in what the nodes (cognitive capabilities) and edges (agent decisions) represent
• Nodes as capabilities: each node is a pure function that does one thing — reason, execute tools, plan, analyze, synthesize. It receives state, returns a partial
• Edges as flow control: static edges for a fixed sequence, conditional edges for decision points where the agent picks its path
• The minimal agent is a StateGraph with 2 nodes (reason, tools) and one conditional edge — internally identical to create_react_agent
• create_react_agent vs custom: prebuilt for speed, custom for control. Custom is a superset of prebuilt
• Visualization with draw_mermaid_png() isn't optional — it's your debugging, verification, and communication tool
• The graph justifies itself through evolution: the minimal agent doesn't need a graph, but adding planning, memory, and multi-agent does

Next capsule: Typed State Design — you'll learn to design the agent's TypedDict with rich fields (plan, iteration_count, quality_score) and custom reducers so the state supports the features of modules 5-10.

Additional resources

1. LangGraph StateGraph Concepts — Official documentation for nodes, edges, and state
2. LangGraph create_react_agent Reference — API of the prebuilt agent and its parameters
3. LangGraph Visualization Guide — How to generate and customize graph visualizations
4. LangGraph Tutorials — Step-by-step tutorials from the LangGraph team
5. ReAct Paper — The paper that formalized the reason-act pattern you now express as a graph