Multi-agent systems built on top of language models promise a lot: parallel reasoning, specialization, longer task horizons. LangGraph makes this accessible in Python, but the patterns that work in demos break under real workloads in ways that are not obvious until you are already in production.
This is a practical guide to the patterns that hold up, the ones that look attractive but cause pain, and the failure modes you should handle before they surprise you.
What LangGraph Actually Gives You
LangGraph is a graph execution engine for stateful, multi-step LLM workflows. Nodes are Python callables. Edges define control flow, including conditional branches. State flows through nodes as a typed dict that each node can read and modify.
The core loop looks like this:
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
next_agent: str
final_answer: str
def researcher(state: AgentState) -> AgentState:
# Call an LLM, add to messages
result = call_llm(state["messages"], system="You are a research agent.")
return {"messages": [result], "next_agent": "analyst"}
def analyst(state: AgentState) -> AgentState:
result = call_llm(state["messages"], system="You are an analyst. Summarize findings.")
return {"messages": [result], "final_answer": result.content}
def router(state: AgentState) -> str:
return state.get("next_agent", END)
builder = StateGraph(AgentState)
builder.add_node("researcher", researcher)
builder.add_node("analyst", analyst)
builder.set_entry_point("researcher")
builder.add_conditional_edges("researcher", router)
builder.add_edge("analyst", END)
graph = builder.compile()
result = graph.invoke({"messages": [], "next_agent": "", "final_answer": ""})
The Annotated[list, operator.add] on messages tells LangGraph to merge lists across concurrent branches rather than overwrite them. Miss this and you lose messages when nodes run in parallel.
Patterns That Hold Up in Production
Explicit state transitions. Each node should set next_agent (or equivalent routing state) explicitly, not leave it implied. Implicit routing based on message content requires the LLM to produce structured routing decisions consistently—under load, under unusual inputs, and after a partial failure. Explicit is cheaper and more reliable.
Bounded loops. When you build a loop where one agent checks the work of another, add a step counter to state and enforce a hard maximum:
class AgentState(TypedDict):
messages: Annotated[list, operator.add]
iterations: int
done: bool
def checker(state: AgentState) -> AgentState:
if state["iterations"] >= 5:
# Force completion regardless of quality
return {"done": True, "iterations": state["iterations"]}
result = call_llm(state["messages"], system="Check the work. Reply DONE or REVISE.")
done = "DONE" in result.content
return {
"messages": [result],
"done": done,
"iterations": state["iterations"] + 1,
}
def should_continue(state: AgentState) -> str:
return END if state["done"] else "worker"
Without the cap, a loop where agents disagree spins until you hit a rate limit or token budget. Five iterations is usually enough; anything that needs more is a sign the task decomposition is wrong.
Typed state with validation. Define your state as a strict TypedDict and validate inputs before the graph runs. An agent that receives malformed state produces confusing output that looks like LLM error rather than a data error. Pydantic models as intermediate state validators catch these early.
Patterns That Look Good but Cause Pain
Passing full message history through every node. It feels natural—every agent has full context. In practice, message history grows across node calls, and by iteration three you are paying for tokens that earlier nodes added and later nodes do not need. Instead, maintain a separate context field in state that is explicitly managed, and pass summaries rather than full transcripts to nodes that do not need the complete history.
Supervisor patterns with open-ended delegation. A supervisor node that tells agents "do whatever is needed" requires the agent to produce a valid routing decision as structured output on every call. When it does not—and it will not, under adversarial or unexpected inputs—the router fails. Constrain what agents can request. Use an enum of allowed next steps, validated at the routing layer, not inferred from free text.
Parallel fanout without result validation. LangGraph supports parallel execution via Send(). When five nodes run in parallel and one fails silently—returns an empty dict, exceeds context, hits a rate limit—the merged state is incomplete. The downstream node sees partial data with no indication of what is missing. Add a validation node after any parallel fanout that checks required fields are present before continuing.
Handling Failures Without Restarting Everything
The most expensive failure in a multi-agent system is restarting from scratch because one node failed late in a long run. LangGraph's checkpointing prevents this:
from langgraph.checkpoint.sqlite import SqliteSaver
# Persist state to SQLite after each node
checkpointer = SqliteSaver.from_conn_string("/tmp/agent-checkpoints.db")
graph = builder.compile(checkpointer=checkpointer)
# First run
config = {"configurable": {"thread_id": "run-001"}}
result = graph.invoke(initial_state, config=config)
# Resume from last checkpoint if it failed mid-way
result = graph.invoke(None, config=config) # None = resume from checkpoint
With a checkpointer, a failure at node 7 of 10 resumes from node 7, not node 1. For tasks that take minutes and make external calls, this is not optional.
For rate limit handling, wrap each LLM call with a retry decorator that uses exponential backoff. LangGraph does not retry nodes automatically—that is the caller's responsibility:
import time
from functools import wraps
def with_retry(max_attempts: int = 3, base_delay: float = 1.0):
def decorator(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return fn(*args, **kwargs)
except RateLimitError:
if attempt == max_attempts - 1:
raise
time.sleep(base_delay * (2 ** attempt))
return wrapper
return decorator
@with_retry(max_attempts=3)
def call_llm(messages, system=""):
# your LLM call here
pass
Observability: What You Actually Need
The standard Python logging is too coarse for multi-agent graphs. You need per-node timing and state diffs. Add a simple wrapper:
import time
import json
def traced_node(name: str, fn):
def wrapper(state: AgentState) -> AgentState:
start = time.monotonic()
result = fn(state)
elapsed = time.monotonic() - start
print(json.dumps({
"node": name,
"elapsed_ms": round(elapsed * 1000),
"state_keys_modified": list(result.keys()),
}))
return result
return wrapper
# Wrap nodes at registration time
builder.add_node("researcher", traced_node("researcher", researcher))
This gives you per-run traces you can grep. For production, pipe this to your structured log sink. You will use it the first time a graph hangs and you need to know at which node.
You can find a security hardening checklist for agentic AI systems covering sandboxing, tool call controls, and audit logging — useful when you move these graphs to production infrastructure.
The Takeaway
LangGraph is a solid foundation for multi-agent work. The patterns that hold up are: explicit state transitions, bounded loops, typed and validated state, and checkpointing from the start. The patterns that cause pain are: growing message history passed through every node, open-ended supervisor delegation, and parallel fanout without result validation.
Build the observability layer before you need it, cap every loop, and checkpoint early. The rest is mostly prompt engineering and iteration.
I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.
Top comments (0)