LangGraph

Production-grade stateful graphs for agents. Built on LangChain, used by Replit, Uber, Elastic, Cognition.

Core Concepts

StateGraph

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

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # Auto-merge
    user_input: str
    context: dict
    iterations: int

graph = StateGraph(AgentState)

# Add nodes
graph.add_node("reasoning", reasoning_node)
graph.add_node("tools", tool_node)

# Add edges
graph.add_edge(START, "reasoning")
graph.add_conditional_edges("reasoning", should_continue)
graph.add_edge("tools", "reasoning")

# Compile
app = graph.compile(checkpointer=MemorySaver())

Key Components

State Design

Reducers (Merge Strategies)

from typing import Annotated
import operator

class State(TypedDict):
    # Append (default for lists)
    messages: Annotated[list[BaseMessage], operator.add]

    # Replace (last write wins)
    user_input: str

    # Custom reducer
    context: Annotated[dict, lambda old, new: {**old, **new}]

    # Accumulate (numeric)
    tokens_used: Annotated[int, operator.add]

Best Practices

• Minimal state — Only what's needed for decisions
• Immutable updates — Return partial dict, not mutated state
• Serializable — JSON-compatible for checkpointing
• Versioned — Add schema_version for migrations

Control Flow Patterns

1. Sequential

graph.add_edge("node_a", "node_b")
graph.add_edge("node_b", "node_c")

2. Branching (Conditional)

def route(state) -> str:
    if state["needs_tool"]:
        return "tools"
    return "final"

graph.add_conditional_edges("reasoning", route, {
    "tools": "tools",
    "final": "final_answer"
})

3. Parallel (Fan-out/Fan-in)

# Fan-out: single node -> multiple parallel
graph.add_edge("start", "parallel_a")
graph.add_edge("start", "parallel_b")

# Fan-in: multiple -> single (using reducer)
# Both write to same state key with operator.add
graph.add_edge("parallel_a", "join")
graph.add_edge("parallel_b", "join")

4. Loops

def should_continue(state) -> str:
    if state["iterations"] >= 5:
        return "final"
    return "reasoning"

graph.add_conditional_edges("tools", should_continue)

5. Dynamic Routing (LLM-based)

def router(state) -> str:
    decision = llm.invoke(ROUTER_PROMPT.format(state=state))
    return parse_action(decision)

graph.add_conditional_edges("reasoning", router)

Human-in-the-Loop (Interrupt)

Basic Pattern

from langgraph.graph import interrupt

def human_review(state):
    # This PAUSES execution
    feedback = interrupt({
        "question": "Approve this action?",
        "context": state["proposed_action"]
    })
    # Execution resumes HERE after resume()
    return {"human_feedback": feedback}

graph.add_node("human_review", human_review)

Resume

# From client
config = {"configurable": {"thread_id": "123"}}
app.invoke(None, config=config)  # Runs until interrupt

# Human provides feedback
app.resume(config, {"approved": True, "notes": "Looks good"})

Approval UI Integration

# Slack bot example
@app.route("/slack/interactive", methods=["POST"])
def slack_interactive():
    payload = json.loads(request.form["payload"])
    thread_id = payload["actions"][0]["value"]
    approved = payload["actions"][0]["action_id"] == "approve"

    # Resume the graph
    app.resume(
        {"configurable": {"thread_id": thread_id}},
        {"approved": approved}
    )
    return "", 200

Checkpointing & Persistence

Checkpointer Options

Configuration

from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    conninfo="postgresql://user:pass@host/db",
    max_size=20,
    kwargs={"autocommit": True}
)

checkpointer = PostgresSaver(pool)
checkpointer.setup()  # Creates tables

app = graph.compile(checkpointer=checkpointer)

Thread Management

# Each conversation = thread
config = {"configurable": {"thread_id": "user-123-session-456"}}

# Resume from checkpoint
state = app.get_state(config)
print(state.values)  # Current state
print(state.next)    # Next node to execute
print(state.tasks)   # Pending tasks

# Time travel
history = list(app.get_state_history(config))
app.update_state(config, {"messages": [...]} )  # Fork from past

Migration

# Add schema_version to state
class State(TypedDict):
    schema_version: int
    # ... other fields

def migrate_state(state):
    if state.get("schema_version", 0) < 2:
        state["new_field"] = default_value
        state["schema_version"] = 2
    return state

# Apply on resume
def load_state(config):
    state = checkpointer.get(config)
    return migrate_state(state)

Subgraphs (Multi-Agent)

Subgraph as Node

# Sub-agent graph
subgraph = StateGraph(SubState)
subgraph.add_node("researcher", researcher)
subgraph.add_node("writer", writer)
subgraph.add_edge(START, "researcher")
subgraph.add_edge("researcher", "writer")
subgraph.add_edge("writer", END)
compiled_sub = subgraph.compile()

# Main graph uses subgraph as node
main = StateGraph(State)
main.add_node("research_team", compiled_sub)  # Treated as single node
main.add_edge(START, "research_team")

Inter-Agent Communication (send)

from langgraph.graph import send

def coordinator(state):
    # Send to researcher subgraph
    send("researcher", {"task": "Find X", "thread_id": "research-1"})
    # Send to coder subgraph
    send("coder", {"task": "Implement Y", "thread_id": "code-1"})
    return {"status": "delegated"}

# Subgraphs receive via special input
def researcher(state):
    task = state.get("task")  # From send()
    return {"result": do_research(task)}

Streaming

Token Streaming

async for chunk in app.astream({"input": "Hello"}, config, stream_mode="messages"):
    if chunk[0].content:
        print(chunk[0].content, end="", flush=True)

Stream Modes

Custom Streaming

def streaming_node(state):
    for token in llm.stream(state["prompt"]):
        yield {"token": token}  # Custom stream mode

Observability (LangSmith)

Automatic Tracing

import langsmith
client = langsmith.Client()

# Automatic with LangGraph
app = graph.compile(checkpointer=checkpointer)
# Traces automatically sent if LANGSMITH_API_KEY set

Custom Metadata

config = {
    "configurable": {"thread_id": "123"},
    "metadata": {
        "user_id": "user-123",
        "session_type": "coding",
        "model": "gpt-4o"
    },
    "tags": ["production", "v2.1"]
}

Evaluation Integration

from langsmith.evaluation import evaluate

def target(inputs):
    return app.invoke(inputs, config={"configurable": {"thread_id": "eval"}})

results = evaluate(
    target,
    data="my-dataset",
    evaluators=[accuracy, latency, cost],
    experiment_prefix="agent-v2"
)

Deployment

LangGraph Platform (Managed)

# langgraph.yaml
dependencies: ["."]
graphs:
  my_agent: "./agent.py:graph"
env:
  - OPENAI_API_KEY
  - LANGSMITH_API_KEY

Self-Hosted (Docker)

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8000"]

FastAPI Server

from fastapi import FastAPI
from langgraph.pregel.remote import RemoteGraph

app = FastAPI()

# Expose compiled graph as API
graph = RemoteGraph("my_agent", url="http://langgraph-server:8000")

@app.post("/invoke")
async def invoke(input: dict, thread_id: str):
    return await graph.ainvoke(input, config={"configurable": {"thread_id": thread_id}})

@app.post("/stream")
async def stream(input: dict, thread_id: str):
    async for chunk in graph.astream(input, config={"configurable": {"thread_id": thread_id}}):
        yield chunk

Testing

Unit Testing Nodes

def test_reasoning_node():
    state = {"messages": [HumanMessage("Hello")], "context": {}}
    result = reasoning_node(state)
    assert "messages" in result
    assert len(result["messages"]) > 0

Integration Testing

def test_full_graph():
    app = graph.compile(checkpointer=MemorySaver())
    config = {"configurable": {"thread_id": "test"}}
    result = app.invoke({"input": "Test question"}, config)
    assert "answer" in result

Time-Travel Testing

def test_replay():
    # Run to checkpoint
    app.invoke({"input": "Step 1"}, config)
    state = app.get_state(config)

    # Fork and test alternative
    app.update_state(config, {"messages": [AlternativeMessage()]})
    result = app.invoke(None, config)
    assert result["path"] == "alternative"

Performance Tips

1. Minimize state size — Only essential data
2. Batch LLM calls — Use abatch for parallel
3. Cache checkpointer — Reuse connection pools
4. Stream early — stream_mode="messages" for UX
5. Async everywhere — ainvoke, astream, abatch

Common Patterns

ReAct in LangGraph

def reasoning_node(state):
    response = llm.bind_tools(tools).invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state):
    tool_calls = state["messages"][-1].tool_calls
    results = []
    for tc in tool_calls:
        tool = tools_by_name[tc["name"]]
        results.append(tool.invoke(tc["args"]))
    return {"messages": results}

def should_continue(state):
    return "tools" if state["messages"][-1].tool_calls else "end"

Plan-and-Execute

def planner(state):
    plan = llm.invoke(PLAN_PROMPT.format(goal=state["goal"]))
    return {"plan": parse_plan(plan), "step": 0}

def executor(state):
    step = state["plan"][state["step"]]
    result = execute_step(step)
    return {"results": [result], "step": state["step"] + 1}

def should_replan(state):
    if state["step"] >= len(state["plan"]):
        return "final"
    if step_failed(state):
        return "replan"
    return "executor"

Related Notes

• Framework Comparison — vs other frameworks
• Control Flow — Patterns implemented
• Human in the Loop — interrupt() deep dive
• Checkpointing — Persistence patterns
• Coordination Patterns — Subgraphs, send()
• Observability — LangSmith integration
• Deployment Architecture — Deployment patterns
• Agent Benchmarks — Testing with LangGraph

Sources: LangGraph docs (langchain-ai.github.io/langgraph), LangGraph Platform docs, LangSmith docs, GitHub langgraph repo, production case studies (Replit, Uber, Elastic, Cognition), LangChain blog, LangGraph Academy.