LangGraph Deep Dive

LangGraph models an agent as a state machine: a shared state object flows through nodes (functions) connected by edges (fixed or conditional). Persistence, streaming and human-in-the-loop come built in.

Setup

python -m venv .venv && source .venv/bin/activate
pip install -U langgraph langchain "langchain[anthropic]" langgraph-checkpoint-postgres
export ANTHROPIC_API_KEY=sk-ant-...
# optional tracing
export LANGSMITH_TRACING=true LANGSMITH_API_KEY=lsv2_...

init_chat_model("provider:model") lets you swap providers (openai:, google_genai:, bedrock_converse:, azure_openai:) without touching graph code — handy when each cloud prefers its own model endpoint.

Core concepts

State & reducers

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]  # append / update by message id
    steps: Annotated[int, add]               # sum numeric updates
    category: str                            # no reducer → last write wins

Without a reducer, a node returning {"category": "billing"} overwrites the value. With add_messages, returning {"messages": [msg]} appends (or replaces a message with the same id). MessagesState is a ready-made state with just the messages key.

Your first graph

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

class S(TypedDict):
    text: str
    shout: bool

def clean(state: S):
    return {"text": state["text"].strip()}

def route(state: S) -> str:
    return "upper" if state["shout"] else "done"

def upper(state: S):
    return {"text": state["text"].upper() + "!"}

g = StateGraph(S)
g.add_node("clean", clean)
g.add_node("upper", upper)
g.add_edge(START, "clean")
g.add_conditional_edges("clean", route, {"upper": "upper", "done": END})
g.add_edge("upper", END)
app = g.compile()

print(app.invoke({"text": "  hello agents ", "shout": True}))
# {'text': 'HELLO AGENTS!', 'shout': True}
print(app.get_graph().draw_mermaid())   # visualise

A tool-calling (ReAct) agent by hand

from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition

@tool
def multiply(a: float, b: float) -> float:
    """Multiply two numbers."""
    return a * b

@tool
def fx_rate(base: str, quote: str) -> float:
    """Get the FX rate from base to quote currency, e.g. USD->EUR."""
    return {"USDEUR": 0.92, "EURUSD": 1.09}.get(base + quote, 1.0)

tools = [multiply, fx_rate]
llm = init_chat_model("anthropic:claude-sonnet-5-5").bind_tools(tools)

def agent(state: MessagesState):
    return {"messages": [llm.invoke(state["messages"])]}

g = StateGraph(MessagesState)
g.add_node("agent", agent)
g.add_node("tools", ToolNode(tools))
g.add_edge(START, "agent")
g.add_conditional_edges("agent", tools_condition)  # → "tools" or END
g.add_edge("tools", "agent")                        # loop back
graph = g.compile()

out = graph.invoke({"messages": [("user", "What is 250 USD in EUR, times 3?")]})
print(out["messages"][-1].content)

Shortcut: LangChain 1.x ships langchain.agents.create_agent(model, tools, system_prompt=...) (the successor to langgraph.prebuilt.create_react_agent), which builds this exact graph for you and supports middleware. Build it by hand at least once — interviewers ask what's inside.

Persistence & memory (checkpointers)

from langgraph.checkpoint.memory import InMemorySaver
graph = g.compile(checkpointer=InMemorySaver())

cfg = {"configurable": {"thread_id": "user-42"}}
graph.invoke({"messages": [("user", "Hi, I'm Priya.")]}, cfg)
r = graph.invoke({"messages": [("user", "What's my name?")]}, cfg)
print(r["messages"][-1].content)   # → "Your name is Priya."

# inspect / time-travel
snapshot = graph.get_state(cfg)
history = list(graph.get_state_history(cfg))

For production use a durable checkpointer:

from langgraph.checkpoint.postgres import PostgresSaver   # or .aio.AsyncPostgresSaver

with PostgresSaver.from_conn_string(DB_URI) as saver:
    saver.setup()                      # creates tables once
    graph = g.compile(checkpointer=saver)

Long-term memory across threads uses a Store (InMemoryStore, PostgresStore) passed via compile(store=...), organised by namespaces like ("users", user_id, "preferences"), optionally with semantic search over embeddings.

Human-in-the-loop with interrupt

from langgraph.types import interrupt, Command

def approve_refund(state):
    decision = interrupt({                 # pauses the graph, state is checkpointed
        "action": "refund",
        "amount": state["amount"],
        "question": "Approve this refund?",
    })
    if decision["approved"]:
        return Command(goto="issue_refund")
    return Command(goto="explain_denial", update={"note": decision.get("reason", "")})

# 1st call runs until the interrupt
result = graph.invoke(inputs, cfg)
print(result["__interrupt__"])           # show the payload to a human

# later — maybe hours later, from another process
graph.invoke(Command(resume={"approved": True}), cfg)

Streaming

# node-level updates
for chunk in graph.stream(inputs, cfg, stream_mode="updates"):
    print(chunk)

# LLM tokens as they are generated
for msg_chunk, meta in graph.stream(inputs, cfg, stream_mode="messages"):
    if meta["langgraph_node"] == "agent" and msg_chunk.content:
        print(msg_chunk.content, end="", flush=True)

# async + multiple modes at once
async for mode, data in graph.astream(inputs, cfg, stream_mode=["updates", "messages"]):
    ...

Other modes: "values" (full state after each step), "custom" (your own events via get_stream_writer()), "debug".

Parallelism with Send (map-reduce)

from langgraph.types import Send

class Overall(TypedDict):
    topics: list[str]
    summaries: Annotated[list[str], add]

def fan_out(state: Overall):
    return [Send("summarise", {"topic": t}) for t in state["topics"]]

def summarise(s: dict):
    return {"summaries": [llm.invoke(f"One line on {s['topic']}").content]}

g = StateGraph(Overall)
g.add_node("summarise", summarise)
g.add_conditional_edges(START, fan_out, ["summarise"])
g.add_edge("summarise", END)

Subgraphs & multi-agent supervisor

A compiled graph can be added as a node in another graph (a subgraph). Combine that with Command(goto=...) and you get a supervisor:

from typing import Literal
from pydantic import BaseModel
from langgraph.graph import END
from langgraph.types import Command
from langchain.agents import create_agent

class Route(BaseModel):
    next: Literal["researcher", "writer", "FINISH"]

router = init_chat_model("anthropic:claude-haiku-5-5").with_structured_output(Route)

def supervisor(state: MessagesState) -> Command[Literal["researcher", "writer", "__end__"]]:
    choice = router.invoke([("system", "Pick the next worker or FINISH.")] + state["messages"])
    return Command(goto=END if choice.next == "FINISH" else choice.next)

research_agent = create_agent(model, tools=[web_search], name="researcher")
writer_agent   = create_agent(model, tools=[], name="writer")

g = StateGraph(MessagesState)
g.add_node("supervisor", supervisor)
g.add_node("researcher", research_agent)   # subgraph as a node
g.add_node("writer", writer_agent)
g.add_edge(START, "supervisor")
g.add_edge("researcher", "supervisor")
g.add_edge("writer", "supervisor")
team = g.compile()

The langgraph-supervisor and langgraph-swarm packages provide prebuilt versions of these patterns.

Production checklist

• Recursion limit: config={"recursion_limit": 25} caps runaway loops.
• Retries: g.add_node("tools", node, retry_policy=RetryPolicy(max_attempts=3)).
• Async everywhere in web servers: ainvoke/astream + AsyncPostgresSaver.
• Durable checkpointer (Postgres) so any replica can resume any thread — keeps containers stateless.
• Tracing: LangSmith env vars or OpenTelemetry.
• Deploy options: self-host behind FastAPI (what we do next), or LangGraph Platform / LangSmith Deployment using langgraph.json.