LangGraph: Cyclical State Machines & StateGraph

1. Why Chains Aren't Enough: The Case for Cyclic Graphs

Standard pipelines (like LangChain Expression Language or LlamaIndex query engines) operate as Directed Acyclic Graphs (DAGs). Data flows strictly in one direction from Step A → Step B → Step C.

Real-world agents are fundamentally cyclic state machines:

• An agent calls a tool → the tool returns data → the graph loops back to the agent.
• If the data is insufficient, the agent calls a second tool → loops back again.
• Only when the agent satisfies its stopping condition does the graph break the cycle and route to END.

LangGraph is built from the ground up to orchestrate these cyclical, stateful agent graphs.

2. Core Building Blocks: State, Nodes & Edges

3. Understanding State Reducers (add_messages)

In LangGraph, nodes do not replace the entire state dictionary; they return incremental updates. Without a reducer, returning {"messages": [new_msg]} would wipe out all preceding conversation history!

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

# The AgentState definition
class AgentState(TypedDict):
    # 'add_messages' is the reducer: it appends new messages and updates existing ones by ID
    messages: Annotated[list[BaseMessage], add_messages]
    sender_id: str
    retry_count: int

4. Hands-On Python: End-to-End Tool-Calling Agent

Here is a complete, working LangGraph script implementing a robust agent with built-in tool execution and conditional routing:

from typing import TypedDict, Annotated, Literal
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

# 1. Define Tools
@tool
def calculate_vat(amount: float, country_code: str) -> str:
    """Calculates VAT for a specific European country code."""
    vat_rates = {"DE": 0.19, "FR": 0.20, "UK": 0.20}
    rate = vat_rates.get(country_code.upper(), 0.15)
    tax = amount * rate
    return f"VAT for {country_code.upper()} on USD {amount:.2f} is USD {tax:.2f} (Rate: {rate*100}%)"

tools = [calculate_vat]
tool_node = ToolNode(tools)

# 2. Define State Schema
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

# 3. Define the Agent Node (LLM with tools bound)
def call_agent_model(state: AgentState):
    # Mocking LLM response or binding tools via: llm.bind_tools(tools)
    messages = state["messages"]
    # Simulated response: Model calls calculate_vat
    return {"messages": messages}

# 4. Define Conditional Routing Function
def should_continue(state: AgentState) -> Literal["tools", "__end__"]:
    last_message = state["messages"][-1]
    # If the model requested tool calls, route to "tools"
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    # Otherwise, return final answer to the user
    return "__end__"

# 5. Assemble the StateGraph
workflow = StateGraph(AgentState)

# Add Nodes
workflow.add_node("agent", call_agent_model)
workflow.add_node("tools", tool_node)

# Set Entry Point
workflow.set_entry_point("agent")

# Add Edges
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "__end__": END
    }
)

# Cycle from tools back to agent!
workflow.add_edge("tools", "agent")

# 6. Compile Graph
app = workflow.compile()
print("LangGraph Agent compiled successfully!")