LangGraph - State Graphs for Agentic Workflows

LCEL chains are powerful but limited—they can’t loop, branch dynamically, or maintain complex state between steps. LangGraph solves this by modeling agent workflows as state machines: graphs where nodes are processing steps and edges define control flow. This explicit structure enables cycles, conditional routing, and persistent state that production agents require.

From Chains to Graphs: Why the Shift Matters

The previous post showed how LCEL creates elegant linear pipelines. But consider what happens when an agent needs to:

1. Call a tool, examine the result, and decide whether to call another tool
2. Retry a failed operation with modified parameters
3. Wait for human approval before proceeding
4. Remember what it did in previous conversation turns

These patterns share a common structure—they require cycles (revisiting previous steps) and conditional branching (different paths based on runtime state). LCEL’s linear composition can’t express these directly.

LangGraph provides the missing abstraction: directed graphs where nodes represent processing steps and edges represent transitions. This isn’t just a different syntax—it’s a fundamentally different computational model.

graph LR
    subgraph LCEL["LCEL Chains"]
        A1[Prompt] --> B1[Model] --> C1[Parser]
    end

    subgraph LangGraph["LangGraph"]
        A2[Start] --> B2[Process]
        B2 --> C2{Condition}
        C2 -->|Yes| D2[Action A]
        C2 -->|No| E2[Action B]
        D2 --> F2[Evaluate]
        E2 --> F2
        F2 -->|Retry| B2
        F2 -->|Done| G2[End]
    end

    classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff
    classDef orangeClass fill:#F39C12,stroke:#333,stroke-width:2px,color:#fff

    class A1,B1,C1 blueClass
    class A2,B2,C2,D2,E2,F2,G2 orangeClass

State Machines in Computing

State machines are one of computer science’s oldest abstractions, dating to the 1950s. They model systems as:

• A set of states (configurations the system can be in)
• Transitions between states (triggered by events or conditions)
• Actions that occur during transitions or within states

LangGraph applies this model to LLM workflows. The “state” is the accumulated data (messages, tool results, intermediate computations). “Transitions” are edges between processing nodes. “Actions” are the node functions themselves.

This framing provides important guarantees. Every execution follows a well-defined path through the graph. State changes are explicit and traceable. Loops are bounded by the graph structure rather than hidden in recursive code.

The Three Pillars: State, Nodes, and Edges

LangGraph workflows rest on three concepts that map directly to state machine theory:

Component	Purpose	State Machine Analog
State	Data container flowing through the graph	Machine’s memory/configuration
Nodes	Functions that process and transform state	State transition actions
Edges	Connections defining allowed transitions	Transition rules

State: The Memory of Your Agent

State in LangGraph is a typed data structure—typically a TypedDict or Pydantic model—that carries all information the agent needs. Unlike LCEL where data flows linearly from output to input, LangGraph state persists and accumulates across the entire execution.

from typing import TypedDict, Annotated
from operator import add

class AgentState(TypedDict):
    messages: Annotated[list, add]  # Conversation history (accumulates)
    current_task: str               # What we're working on (overwrites)
    tool_results: Annotated[list, add]  # Results from tools (accumulates)
    iteration: int                  # Loop counter (overwrites)

The Annotated syntax with add (or any binary function) creates reducers—rules for how to combine old and new values. Without a reducer, new values simply overwrite old ones. With add, lists concatenate, enabling patterns like message history that grows with each turn.

This design choice has deep implications. Accumulating state means nodes don’t need to pass everything through return values. A tool execution node can add its results to state; a later summarization node can access those results directly.

Nodes: Processing Steps

Nodes are pure functions that receive the current state and return updates. They don’t modify state directly—they return dictionaries describing what should change.

def process_query(state: AgentState) -> dict:
    """Classify and route the user's query."""
    query = state["messages"][-1].content

    # LLM decides the category
    response = llm.invoke(f"Classify this query: {query}")

    return {
        "current_task": response.content,
        "iteration": state.get("iteration", 0) + 1
    }

This immutable approach enables:

• Debugging: You can inspect state before and after each node
• Replay: Re-run nodes with identical inputs for testing
• Checkpointing: Save state at any point for resumption

Edges: Control Flow

Edges connect nodes and come in two varieties:

Fixed edges always transition to a specific next node:

graph.add_edge("process", "respond")  # Always go from process to respond

Conditional edges evaluate state and choose among multiple targets:

def should_continue(state: AgentState) -> str:
    if state["iteration"] >= 3:
        return "give_up"
    if state.get("success"):
        return "respond"
    return "retry"

graph.add_conditional_edges("evaluate", should_continue, {
    "retry": "process",
    "respond": "respond",
    "give_up": "error_handler"
})

The routing function receives current state and returns a string key. The mapping dictionary translates keys to target node names. This indirection allows renaming nodes without changing routing logic.

Building Your First Graph

Let’s construct a simple workflow that classifies queries and routes them to specialized handlers.

graph TD
    A[START] --> B[Classify]
    B --> C{Route}
    C -->|technical| D[Tech Handler]
    C -->|billing| E[Billing Handler]
    C -->|general| F[General Handler]
    D --> G[END]
    E --> G
    F --> G

    classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff
    classDef orangeClass fill:#F39C12,stroke:#333,stroke-width:2px,color:#fff
    classDef greenClass fill:#27AE60,stroke:#333,stroke-width:2px,color:#fff

    class A,G greenClass
    class B,C orangeClass
    class D,E,F blueClass

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class SupportState(TypedDict):
    query: str
    category: str
    response: str

def classify(state: SupportState) -> dict:
    """Determine query category."""
    query = state["query"].lower()
    if "password" in query or "error" in query:
        return {"category": "technical"}
    elif "bill" in query or "charge" in query:
        return {"category": "billing"}
    return {"category": "general"}

def route(state: SupportState) -> str:
    """Return the handler name based on category."""
    return state["category"]

def tech_handler(state: SupportState) -> dict:
    return {"response": "Technical support: Let me help you with that issue..."}

def billing_handler(state: SupportState) -> dict:
    return {"response": "Billing team: I can look into your account..."}

def general_handler(state: SupportState) -> dict:
    return {"response": "Thanks for reaching out! How can I help?"}

# Construct the graph
graph = StateGraph(SupportState)
graph.add_node("classify", classify)
graph.add_node("technical", tech_handler)
graph.add_node("billing", billing_handler)
graph.add_node("general", general_handler)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route, {
    "technical": "technical",
    "billing": "billing",
    "general": "general"
})
graph.add_edge("technical", END)
graph.add_edge("billing", END)
graph.add_edge("general", END)

app = graph.compile()

The compile() step transforms the graph definition into an executable Runnable. It validates that all edges point to valid nodes, that there’s a path from START to every node, and that every node has a path to END.

The Tool-Calling Agent Loop

The most common LangGraph pattern implements the agent loop: call model → check for tools → execute tools → repeat.

graph TD
    A[START] --> B[Call Model]
    B --> C{Has Tool Calls?}
    C -->|Yes| D[Execute Tools]
    D --> B
    C -->|No| E[Finish]
    E --> F[END]

    classDef blueClass fill:#4A90E2,stroke:#333,stroke-width:2px,color:#fff
    classDef orangeClass fill:#F39C12,stroke:#333,stroke-width:2px,color:#fff
    classDef greenClass fill:#27AE60,stroke:#333,stroke-width:2px,color:#fff

    class A,F greenClass
    class B,D blueClass
    class C,E orangeClass

This graph has a cycle—execute_tools loops back to call_model. LCEL can’t express this directly, but LangGraph handles it naturally. The cycle continues until the conditional edge routes to finish instead of back to tools.

The key insight is that tool execution changes state, adding ToolMessage objects that the model sees on the next iteration. This accumulated context lets the model decide whether it needs more information or can answer.

State Accumulation in Action

Consider an agent answering “What’s the weather in the city where Apple HQ is located?”

Initial state:

{"messages": [HumanMessage("What's the weather...")]}

After first model call:

{
    "messages": [HumanMessage(...), AIMessage(tool_calls=[{"name": "search", ...}])],
    "pending_tool_calls": [{"name": "search", "args": {"query": "Apple headquarters location"}}]
}

After tool execution:

{
    "messages": [..., AIMessage(...), ToolMessage("Cupertino, California")],
    "pending_tool_calls": []
}

After second model call:

{
    "messages": [..., AIMessage(tool_calls=[{"name": "weather", ...}])],
    "pending_tool_calls": [{"name": "weather", "args": {"city": "Cupertino"}}]
}

Each iteration adds to state rather than replacing it. The model sees its full history, enabling coherent multi-step reasoning.

Checkpointing: Persistence and Time Travel

Production agents need to survive restarts, support long-running conversations, and enable debugging. LangGraph’s checkpointing system addresses all three.

from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent = graph.compile(checkpointer=checkpointer)

# Each thread ID maintains separate state
config = {"configurable": {"thread_id": "user-123"}}

result = agent.invoke({"messages": [HumanMessage("Hello")]}, config=config)
# State saved automatically

# Later, continue the same conversation
result = agent.invoke({"messages": [HumanMessage("What did I just say?")]}, config=config)
# Agent has access to the "Hello" message from before

The MemorySaver stores state in memory (good for development). Production systems use SqliteSaver or Redis-backed implementations for durability.

Time Travel Debugging

Checkpointing enables time travel—inspecting or resuming from any previous state:

# Get full execution history
history = list(agent.get_state_history(config))

for i, snapshot in enumerate(history):
    print(f"Step {i}: {snapshot.values['messages'][-1].content[:50]}...")

# Resume from step 2
earlier_state = history[2]
result = agent.invoke({"messages": [HumanMessage("Try a different approach")]},
                      config=earlier_state.config)

This capability transforms debugging. Instead of adding print statements and re-running, you can examine the exact state at any point and branch off with different inputs.

MessagesState: The Common Case

Most agents are conversational, maintaining message history as their primary state. LangGraph provides MessagesState as a convenience:

from langgraph.graph import MessagesState

# Equivalent to:
# class MessagesState(TypedDict):
#     messages: Annotated[list, add]

graph = StateGraph(MessagesState)

This pattern is so common that LangGraph pre-defines it. Your nodes receive state with a messages key, and any messages you return are appended automatically.

Streaming: Real-Time Feedback

For interactive applications, waiting for the complete response is too slow. LangGraph supports streaming at multiple levels:

Values mode: Emit complete state after each node

for state in agent.stream({"messages": [HumanMessage("Tell me a story")]},
                          stream_mode="values"):
    print(f"Current state: {state}")

Updates mode: Emit only the changes from each node

for update in agent.stream({"messages": [HumanMessage("Tell me a story")]},
                           stream_mode="updates"):
    for node_name, changes in update.items():
        print(f"{node_name} produced: {changes}")

For LLM token streaming (word-by-word output), you combine LangGraph streaming with LangChain’s model streaming—a pattern we’ll cover in production deployment.

Visualization and Introspection

LangGraph graphs are inspectable. You can generate visual representations for documentation or debugging:

# Get Mermaid diagram syntax
mermaid_code = app.get_graph().draw_mermaid()
print(mermaid_code)

# Save as PNG image (requires graphviz)
app.get_graph().draw_mermaid_png(output_file_path="my_agent.png")

This capability closes the loop on documentation. The graph you define in code is the documentation—there’s no separate diagram to keep in sync.

Key Takeaways

1. Graphs extend what’s possible: Cycles, conditional branching, and complex state management require graph structures that LCEL can’t express.

2. State is explicit and typed: TypedDict or Pydantic schemas define exactly what data flows through your agent, with reducers controlling accumulation.

3. Nodes are pure functions: They receive state and return updates. This immutability enables debugging, replay, and checkpointing.

4. Edges control flow explicitly: Fixed edges for sequential steps, conditional edges for dynamic routing. The routing logic is separated from the processing logic.

5. Checkpointing enables production features: Persistence, conversation resumption, and time-travel debugging come from treating state as first-class.

6. The tool loop is a graph pattern: Call model → route on tool calls → execute tools → loop back. This cycle is the foundation of most agents.

---

Next: Connecting LangGraph Agents to APIs and Databases - We’ll integrate external systems, build SQL agents, and address security considerations for real-world deployments.