Over 60% of enterprises piloting LLM applications in 2024 reported that single-agent chains hit a ceiling on complex reasoning tasks, according to LangChain's State of AI Agents survey. Most teams start with linear prompt chains, only to discover that real-world workflows — research, coding, customer support — demand parallel execution, shared memory, and dynamic handoffs between specialized agents. LangGraph, released to general availability on May 14, 2025, solves this by modeling agent interactions as stateful graphs instead of linear pipes. This guide walks you through building production-ready multi-agent systems with LangGraph, from graph definition to checkpointed deployment, with working code patterns you can adapt today.
Quick Answer: Build autonomous multi-agent systems in LangGraph by defining a StateGraph with typed state, adding specialized agent nodes that read/write shared state, connecting them with conditional edges for dynamic routing, enabling checkpointing via SqliteSaver or PostgresSaver for persistence, and compiling the graph for streaming or batch execution. Each agent runs as a node; the graph orchestrates handoffs, loops, and human-in-the-loop interrupts automatically.
Why LangGraph for Multi-Agent Systems
Graphs Beat Chains for Agent Coordination
Traditional LLM chains (LCEL, LangChain Expression Language) execute linearly: input → step A → step B → output. Multi-agent workflows need branching, cycles, and shared memory — a researcher agent loops until satisfied, then hands off to a writer agent, which may trigger a fact-checker that routes back to research. LangGraph models this as a directed graph where nodes are agents or tools and edges encode routing logic. The StateGraph class maintains a single mutable state dictionary that every node reads and writes, eliminating the context-window stuffing that plagues chain-based approaches.
Checkpointing Enables Long-Running Autonomy
Autonomous agents run for hours or days. LangGraph's checkpointing — backed by SqliteSaver for development or PostgresSaver for production — snapshots the full graph state after every node. This enables time-travel debugging, human-in-the-loop approval gates, and crash recovery without losing progress. The May 2025 LangGraph Platform launch added managed infrastructure for exactly this use case, but the open-source library handles it locally with three lines of code.
Native Streaming and Interrupts
LangGraph streams token-by-token from any node and supports interrupt() for human approval mid-graph. A code-review agent can pause before merging a PR, wait for a developer's click in a UI, then resume from the exact checkpoint. No external queue or database required — the graph state is the queue.
Core Architecture: State, Nodes, Edges
Define Typed State with Pydantic
Start with a Pydantic model that captures everything agents need to share: messages, intermediate results, configuration, and control flags. This becomes your single source of truth.
from typing import Annotated, List, Literal
from langgraph.graph import StateGraph, add_messages
from pydantic import BaseModel, Field
class AgentState(BaseModel):
messages: Annotated[List[BaseMessage], add_messages] = Field(default_factory=list)
research_findings: List[str] = Field(default_factory=list)
current_draft: str = ""
review_status: Literal["pending", "approved", "rejected"] = "pending"
iteration_count: int = 0
max_iterations: int = 3
The add_messages reducer merges message lists automatically — critical when multiple agents append to the conversation history concurrently.
Build Specialized Agent Nodes
Each node is a Python function (or LangChain Runnable) that accepts AgentState and returns a partial state update. Keep nodes single-purpose: one researches, one writes, one reviews.
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def researcher(state: AgentState) -> dict:
query = state.messages[-1].content
prompt = f"Research this topic thoroughly. Return 3-5 key findings as a JSON list: {query}"
response = llm.invoke([SystemMessage(content=prompt)])
findings = eval(response.content) # In production, use structured output
return {"research_findings": findings, "iteration_count": state.iteration_count + 1}
def writer(state: AgentState) -> dict:
findings = "\n".join(state.research_findings)
prompt = f"Write a concise article based on these findings:\n{findings}"
draft = llm.invoke([SystemMessage(content=prompt)]).content
return {"current_draft": draft}
def reviewer(state: AgentState) -> dict:
prompt = f"Review this draft for accuracy and clarity. Return 'approved' or 'rejected' with reason: {state.current_draft}"
verdict = llm.invoke([SystemMessage(content=prompt)]).content
status = "approved" if "approved" in verdict.lower() else "rejected"
return {"review_status": status}
Wire Conditional Edges for Dynamic Routing
Edges determine the next node. Use add_conditional_edges with a routing function that inspects state — this is where autonomy lives.
def route_after_review(state: AgentState) -> str:
if state.review_status == "approved" or state.iteration_count >= state.max_iterations:
return "end"
return "researcher" # Loop back for another research pass
graph = StateGraph(AgentState)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)
graph.add_node("reviewer", reviewer)
graph.set_entry_point("researcher")
graph.add_edge("researcher", "writer")
graph.add_edge("writer", "reviewer")
graph.add_conditional_edges("reviewer", route_after_review, {"researcher": "researcher", "end": END})
app = graph.compile(checkpointer=SqliteSaver.from_conn_string("sqlite:///checkpoints.db"))
The graph loops researcher → writer → reviewer until approval or max iterations. Each loop increments iteration_count in shared state.
Production Patterns: Persistence, Streaming, Human-in-the-Loop
Postgres Checkpointing for Horizontal Scaling
Swap SqliteSaver for PostgresSaver when deploying multiple workers. The schema is managed automatically; you only provide a connection string.
from langgraph.checkpoint.postgres import PostgresSaver
from psycopg_pool import ConnectionPool
pool = ConnectionPool("postgresql://user:pass@localhost:5432/langgraph", max_size=20)
checkpointer = PostgresSaver(pool)
checkpointer.setup() # Creates tables if missing
app = graph.compile(checkpointer=checkpointer)
Each thread (conversation) gets a unique thread_id. Pass it in config on every invoke:
config = {"configurable": {"thread_id": "user-123-session-456"}}
for chunk in app.stream({"messages": [HumanMessage(content="AI safety research")]}, config):
print(chunk)
Streaming Token-by-Token from Any Node
Wrap node LLMs with stream_mode="messages" to yield tokens as they generate. The graph streams chunks with metadata identifying the emitting node.
for chunk, metadata in app.stream(input_data, config, stream_mode="messages"):
if metadata["langgraph_node"] == "writer":
print(chunk.content, end="", flush=True)
Human-in-the-Loop with interrupt()
Insert interrupt() before irreversible actions. The graph pauses, serializes state, and waits for a resume signal with updated state.
from langgraph.types import interrupt
def publisher(state: AgentState) -> dict:
decision = interrupt({"draft": state.current_draft, "action": "publish?"})
# Execution resumes here after human responds
if decision.get("approve"):
# Call publish API
return {"published": True}
return {"published": False}
graph.add_node("publisher", publisher)
graph.add_edge("reviewer", "publisher")
# Resume later:
app.invoke(Command(resume={"approve": True}), config)
Real Example: Autonomous Code Review Agent
This complete example builds a three-agent code reviewer that fetches a PR diff, analyzes for bugs, suggests fixes, and posts comments — all as a single LangGraph application.
# Full working example at github.com/langchain-ai/langgraph-examples/code-review-agent
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
import requests
@tool
def fetch_pr_diff(repo: str, pr_number: int) -> str:
"""Fetch diff from GitHub API."""
url = f"https://api.github.com/repos/{repo}/pulls/{pr_number}"
headers = {"Accept": "application/vnd.github.v3.diff"}
return requests.get(url, headers=headers).text
@tool
def post_review_comment(repo: str, pr_number: int, body: str) -> dict:
"""Post review comment to GitHub."""
url = f"https://api.github.com/repos/{repo}/pulls/{pr_number}/reviews"
return requests.post(url, json={"body": body, "event": "COMMENT"}).json()
class CodeReviewState(BaseModel):
messages: Annotated[List[BaseMessage], add_messages] = Field(default_factory=list)
repo: str = ""
pr_number: int = 0
diff: str = ""
findings: List[str] = Field(default_factory=list)
review_body: str = ""
llm = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools([fetch_pr_diff, post_review_comment])
def fetch_diff(state: CodeReviewState) -> dict:
diff = fetch_pr_diff.invoke({"repo": state.repo, "pr_number": state.pr_number})
return {"diff": diff}
def analyze_code(state: CodeReviewState) -> dict:
prompt = f"Analyze this diff for bugs, security issues, and style violations. Return findings as JSON list:\n{state.diff[:8000]}"
response = llm.invoke([SystemMessage(content=prompt)])
findings = eval(response.content)
return {"findings": findings}
def compose_review(state: CodeReviewState) -> dict:
findings_md = "\n".join(f"- {f}" for f in state.findings)
body = f"## Automated Code Review\n\n{findings_md}\n\n*Generated by LangGraph agent*"
return {"review_body": body}
def publish_review(state: CodeReviewState) -> dict:
post_review_comment.invoke({"repo": state.repo, "pr_number": state.pr_number, "body": state.review_body})
return {"messages": [SystemMessage(content="Review posted")]}
graph = StateGraph(CodeReviewState)
graph.add_node("fetch", fetch_diff)
graph.add_node("analyze", analyze_code)
graph.add_node("compose", compose_review)
graph.add_node("publish", publish_review)
graph.set_entry_point("fetch")
graph.add_edge("fetch", "analyze")
graph.add_edge("analyze", "compose")
graph.add_edge("compose", "publish")
graph.add_edge("publish", END)
app = graph.compile(checkpointer=SqliteSaver.from_conn_string("sqlite:///code-review.db"))
# Run:
config = {"configurable": {"thread_id": "pr-review-42"}}
app.invoke({"repo": "myorg/myrepo", "pr_number": 42}, config)
This runs autonomously on a webhook trigger. Add interrupt() before publish for human approval.
LangGraph vs. Alternatives: Multi-Agent Frameworks Compared
Choosing a multi-agent framework depends on state management, deployment model, and ecosystem lock-in. The table below compares LangGraph against the three most cited alternatives in production surveys.
| Capability | LangGraph | CrewAI | AutoGen | LlamaIndex Agents |
|---|---|---|---|---|
| State model | Single mutable StateGraph (Pydantic) | Shared context dict per crew | Conversation history + agent memory | Workflow state + agent state |
| Persistence | Built-in checkpointers (SQLite, Postgres, Redis) | Manual (custom serialization) | Manual (conversation history) | Manual (workflow serialization) |
| Human-in-the-loop | Native interrupt() + resume |
Callback hooks only | User proxy agent pattern | Event hooks |
| Streaming | Token-level, per-node metadata | Full response only | Token-level via callbacks | Token-level |
| Graph cycles / loops | First-class (conditional edges) | Limited (sequential tasks) | Supported (group chat) | Supported (workflow cycles) |
| Deployment | LangGraph Platform (managed) or self-hosted | Self-hosted only | Self-hosted only | Self-hosted or LlamaCloud |
| LangChain ecosystem | Native (same team) | Integrations available | Integrations available | Native (same team) |
LangGraph wins when you need production-grade persistence, human-in-the-loop gates, and fine-grained streaming — common in customer-facing agents. CrewAI is faster for quick prototypes with predefined roles. AutoGen excels at research-style group chats. LlamaIndex Agents suit RAG-heavy workflows where retrieval is the primary orchestration.
Common Mistakes and How to Fix Them
Mistake: Stuffing Everything Into One Giant State
Why It Hurts: A monolithic state model couples unrelated agents, bloats checkpoints, and makes debugging impossible. One agent's schema change breaks everyone.
Fix: Use namespaced state keys (research.findings, writer.draft) and Pydantic's model_config = ConfigDict(extra="allow") for extensibility. Split independent subgraphs with graph.compile() and invoke them as tools from a parent graph.
Mistake: Skipping Checkpointing in Development
Why It Hurts: Without checkpoints, you lose the ability to pause, inspect, and resume. Every graph restart re-runs expensive LLM calls.
Fix: Always compile with SqliteSaver — even locally. It's one import and one line. The .db file gives you instant time-travel debugging.
Mistake: Using Global Variables for Shared Data
Why It Hurts: Global state breaks with concurrent threads, makes testing flaky, and prevents horizontal scaling.
Fix: All shared data lives in AgentState. Pass configuration via config["configurable"] at invoke time. Inject API clients through node closures or LangChain's RunnableConfig.
Mistake: Hardcoding Model Calls Inside Nodes
Why It Hurts: Swapping models (GPT-4o → Claude 3.5) requires editing every node. Testing with a mock LLM becomes painful.
Fix: Define nodes as LangChain Runnables that accept an llm parameter. Bind the model at graph compile time: graph.compile(llm=ChatOpenAI(...)). This enables A/B testing and drop-in mocks.
Pro Tips
- Use structured output: Replace
eval()withllm.with_structured_output(PydanticModel)for guaranteed schema compliance. - Batch independent nodes: Add
graph.add_edge("node_a", "node_b")without ordering — LangGraph runs them in parallel when possible. - Version your graphs: Store compiled graph JSON (
app.get_graph().draw_mermaid()) in version control. Diff graph structure like code. - Monitor with LangSmith: Add
tracing_v2=Trueto capture every node input/output, latency, and token count automatically. - Test nodes in isolation: Each node is a pure function. Unit test with
node(test_state)before wiring into the graph.
FAQ
What is the difference between LangGraph and LangChain Expression Language (LCEL)?
LCEL defines linear or branched chains declaratively with the pipe operator (|). LangGraph models workflows as stateful graphs with cycles, shared mutable state, and built-in checkpointing. Use LCEL for simple pipelines; switch to LangGraph when you need loops, human-in-the-loop, or multi-agent coordination.
Can I run LangGraph without the LangGraph Platform?
Yes. The open-source langgraph package (PyPI, MIT license) includes StateGraph, checkpointers (SQLite, Postgres, Redis), streaming, and interrupts. The Platform adds managed hosting, horizontal scaling, a visual debugger, and team collaboration features — optional for production teams.
How do I handle authentication and user context in multi-tenant deployments?
Pass a user_id in config["configurable"] at invoke time. Use it as a namespace prefix for thread IDs (f"{user_id}:{session_id}") and as a filter in Postgres row-level security policies. Never store secrets in graph state; inject clients via RunnableConfig.
Why does my graph loop infinitely?
Infinite loops happen when a conditional edge always returns the same node without a terminating condition in state. Add a counter (iteration_count) to state, increment it in the looping node, and route to END when it exceeds max_iterations. Always set a hard limit.
What are the emerging best practices for multi-agent evaluation?
Evaluate at two levels: node-level (unit test each agent's output against golden datasets) and graph-level (end-to-end scenarios with LangSmith's evaluation harness). Track pass/fail rates per node, token cost per run, and human approval rates for interrupt gates. The 2025 LangGraph Platform adds built-in evaluation pipelines for this.
Conclusion
LangGraph shifts multi-agent development from fragile prompt chains to durable, inspectable graphs. The pattern is consistent: define typed state, write single-purpose nodes, wire conditional edges, enable checkpointing, and compile. The result runs locally today and scales to managed infrastructure tomorrow. Start with the researcher-writer-reviewer loop in this guide, add a Postgres checkpointer, and you have a production skeleton that handles crashes, human reviews, and concurrent users without rewrites. The graph is not just orchestration — it is your application's source of truth.
- Model agent workflows as StateGraphs with Pydantic state — shared mutable state replaces context stuffing.
- Always compile with a checkpointer (SqliteSaver for dev, PostgresSaver for prod) to enable persistence, interrupts, and time-travel debugging.
- Use conditional edges for dynamic routing; keep nodes pure and testable in isolation.
- Adopt structured output, parallel edges, and LangSmith tracing from day one to avoid retrofit pain.
0 comments:
Post a Comment