LangGraph Core Concepts

Why State Machines for Agents?

Before LangGraph, agent loops were typically written as imperative Python while loops, while not done: call_llm(); if tool_call: run_tool(). This works for simple cases but creates three problems at scale:

1. No visibility: You can’t see where in the workflow a run is without adding print statements
2. No persistence: If the process crashes, the entire run is lost
3. No control: Adding features like human approval or parallel branches requires significant restructuring

LangGraph solves all three by representing the agent workflow as an explicit directed graph. Every state is trackable, serializable to a database, and independently testable.

The Four Core Concepts

flowchart LR
    ST["STATE\nTyped data structure\nthat flows through the graph"]
    ND["NODE\nPython function, receives\nState and returns updates"]
    ED["EDGE\nDirected connection between\nnodes, can be conditional"]
    GR["GRAPH\nStateGraph wiring all\nnodes and edges together"]
    GR --> ST
    GR --> ND
    GR --> ED
    ND -- "reads / writes" --> ST
    style ST fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style ND fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style ED fill:#EEF0F7,stroke:#818CF8,color:#0F172A
    style GR fill:#fff7ed,stroke:#f59e0b,color:#0F172A

State

State is a TypedDict or dataclass annotated with reducer functions that control how updates merge.

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

class AgentState(TypedDict):
    # Annotated[list[BaseMessage], add_messages]: the reducer appends new messages
    # instead of replacing the whole list; this is how conversation history accumulates
    messages: Annotated[list[BaseMessage], add_messages]
    
    # Standard field: each update replaces the previous value (no reducer = overwrite)
    task_status: str
    
    # Optional fields with defaults
    tool_call_count: int
    final_answer: str | None
    error: str | None

Key rule: Fields without a reducer are replaced by node updates. Fields with add_messages or a custom reducer are merged.

Nodes

Nodes are plain Python functions (sync or async). They receive the full state and return a dict of fields to update:

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage

llm = ChatOpenAI(model="gpt-5.6-luna", temperature=0)

def call_model(state: AgentState) -> dict:
    """Call the LLM with the current message history."""
    # state["messages"] contains the full conversation so far
    response = llm.invoke(state["messages"])
    # returning a partial dict: only listed fields get updated, others are untouched
    return {
        "messages": [response],  # add_messages reducer appends this to existing history
        "tool_call_count": state["tool_call_count"] + 1,  # increment the counter
    }

def call_tools(state: AgentState) -> dict:
    """Execute the tool calls from the last message."""
    # The last message is always the most recent AIMessage after call_model runs
    last_message = state["messages"][-1]
    tool_messages = []
    
    for tool_call in last_message.tool_calls:
        # execute_tool looks up the function by tool_call["name"] and runs it
        result = execute_tool(tool_call)
        tool_messages.append(result)
    
    # ToolMessages get appended to history via the add_messages reducer
    return {"messages": tool_messages}

Edges

Conditional edges are functions that inspect state and return the name of the next node to visit. They are the branching mechanism of the graph.

from langgraph.graph import END

def should_continue(state: AgentState) -> str:
    """Conditional edge: route based on whether the model made tool calls."""
    last_message = state["messages"][-1]
    
    # tool_calls is non-empty when the LLM wants to invoke a tool
    if last_message.tool_calls:
        return "call_tools"
    
    # Guard against runaway loops: if iteration limit hit, force termination
    if state["tool_call_count"] >= 10:
        return END
    
    # No tool calls and under limit: model produced a final answer
    return END

Assembling the Graph

Once nodes and edges are defined, StateGraph wires them into a runnable workflow. The compiled agent object is what you call to run the graph.

from langgraph.graph import StateGraph, END

# Create the graph: StateGraph(AgentState) ensures all nodes share the same typed state
graph_builder = StateGraph(AgentState)

# Add nodes: string name is how edges reference this node
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("call_tools", call_tools)

# Set entry point: this node runs first when graph.invoke() is called
graph_builder.set_entry_point("call_model")

# Add edges
graph_builder.add_conditional_edges(
    "call_model",       # source node
    should_continue,    # routing function, returns the name of the next node
    {
        # map routing function return values to actual node names
        "call_tools": "call_tools",
        END: END,
    }
)
graph_builder.add_edge("call_tools", "call_model")  # loop back after tool execution

# compile() locks the graph structure and returns a runnable object
agent = graph_builder.compile()

Visualizing the Graph

flowchart TD
    START([START]) --> CM[call_model]
    CM -->|has tool calls| CT[call_tools]
    CT --> CM
    CM -->|no tool calls| END([END])
    style CM fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style CT fill:#f0fdf9,stroke:#0D9488,color:#0F172A

This is the classic ReAct loop as an explicit state machine. The model runs, if it requests tools those tools run and the result goes back to the model, and this continues until the model produces a final answer or the iteration limit is hit.

Running the Compiled Graph

from langchain_core.messages import HumanMessage

# Initial state must include all required TypedDict fields
initial_state = {
    "messages": [HumanMessage(content="What is the population of Tokyo?")],
    "task_status": "running",
    "tool_call_count": 0,
    "final_answer": None,
    "error": None,
}

# Synchronous execution: blocks until the graph reaches END
result = agent.invoke(initial_state)
print(result["messages"][-1].content)

# Streaming execution: yields an event dict after every node completes
# useful for showing progress in a UI or logging intermediate steps
for event in agent.stream(initial_state):
    for node_name, node_output in event.items():
        print(f"--- {node_name} ---")
        if "messages" in node_output:
            print(node_output["messages"][-1].content[:200])

State Persistence with Checkpointers

LangGraph can serialize graph state to a database after each node, enabling:

• Fault tolerance, resume from where you left off after a crash
• Human-in-the-loop, pause for approval, then resume
• Time travel, replay the graph from any historical state

from langgraph.checkpoint.memory import MemorySaver  # dev/testing
from langgraph.checkpoint.postgres import PostgresSaver  # production

# Development: in-memory checkpointer: state is lost when the process exits
checkpointer = MemorySaver()
agent = graph_builder.compile(checkpointer=checkpointer)

# thread_id scopes the checkpoint: same thread_id resumes the same conversation
config = {"configurable": {"thread_id": "user-123-task-456"}}

result = agent.invoke(initial_state, config=config)

# Later: resume from where we left off
# (the state is loaded from the checkpointer automatically: no need to pass initial_state again)
continued = agent.invoke({"messages": [HumanMessage("Continue")]}, config=config)

Production Checkpointing with Postgres

For production, use a real database so state survives process restarts and scales across multiple workers.

# For production, use a real database
import asyncpg
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async def create_agent_with_db_persistence():
    conn = await asyncpg.connect(settings.database_url)
    checkpointer = AsyncPostgresSaver(conn)
    # setup() creates the langgraph_checkpoints table if it doesn't exist
    await checkpointer.setup()  # creates the checkpoint tables
    
    return graph_builder.compile(checkpointer=checkpointer)

With Postgres persistence, every agent run is durably stored. You can query which runs are in progress, which have completed, and replay any run step by step in LangSmith.