Multi-Agent Orchestration with LangGraph: Patterns and Pitfalls

Multi-agent systems sound compelling on paper: specialized agents, parallel execution, complex reasoning split across focused workers. In practice, most teams discover that the hardest part isn't making agents work — it's keeping them from spiraling. LangGraph gives you the primitives to build these systems as explicit graphs, and that explicitness is both its strength and the place where most mistakes happen.

What LangGraph actually models

LangGraph represents a workflow as a directed graph. Nodes are Python functions (or LLM calls). Edges are transitions between them. State is a typed dictionary that flows through every node and accumulates across steps.

Unlike a simple chain, a LangGraph graph can loop, branch conditionally, and fan out in parallel. The state is visible at every step. This beats "call the model in a loop and hope it decides to stop."

The core structure for a supervisor-worker pattern:

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

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next: str
    steps: int

def supervisor(state: AgentState) -> AgentState:
    # Hard exit limit — always required
    if state["steps"] >= 10:
        return {"next": "FINISH", "steps": state["steps"]}
    response = llm.invoke([
        SystemMessage("Route to: researcher, coder, or FINISH"),
        *state["messages"],
    ])
    return {
        "next": parse_routing(response.content),
        "steps": state["steps"] + 1,
    }

def researcher(state: AgentState) -> AgentState:
    result = search_tool.run(state["messages"][-1].content)
    return {"messages": [AIMessage(f"Research: {result}")]}

def coder(state: AgentState) -> AgentState:
    code = llm.invoke([
        SystemMessage("Write working Python code for the task."),
        *state["messages"],
    ])
    return {"messages": [code]}

builder = StateGraph(AgentState)
builder.add_node("supervisor", supervisor)
builder.add_node("researcher", researcher)
builder.add_node("coder", coder)
builder.set_entry_point("supervisor")
builder.add_conditional_edges(
    "supervisor",
    lambda s: s["next"],
    {"researcher": "researcher", "coder": "coder", "FINISH": END},
)
builder.add_edge("researcher", "supervisor")
builder.add_edge("coder", "supervisor")
app = builder.compile()

The steps counter is not optional. Without it, a supervisor that never outputs "FINISH" will exhaust your API budget before you notice.

Pattern: parallel workers with a join

Some tasks decompose into independent subtasks you can run concurrently. LangGraph's Send primitive handles this without external tooling.

from langgraph.constants import Send

class PlanState(TypedDict):
    query: str
    results: Annotated[list, operator.add]
    final: str

def planner(state: PlanState) -> list:
    subtasks = decompose(state["query"])  # returns list of strings
    return [Send("worker", {"task": t}) for t in subtasks]

def worker(state: dict) -> dict:
    result = llm.invoke(state["task"])
    return {"results": [result.content]}

def aggregator(state: PlanState) -> PlanState:
    merged = "\n---\n".join(state["results"])
    final = llm.invoke(f"Synthesize:\n{merged}")
    return {"final": final.content}

builder = StateGraph(PlanState)
builder.add_node("planner", planner)
builder.add_node("worker", worker)
builder.add_node("aggregator", aggregator)
builder.set_entry_point("planner")
builder.add_edge("worker", "aggregator")
builder.add_edge("aggregator", END)
app = builder.compile()

Annotated[list, operator.add] is what makes the join work: results from parallel workers are concatenated before the aggregator runs. Using a plain list annotation instead silently discards all but the last result.

The five pitfalls that actually bite teams

1. Unbounded loops. Already covered — always add a step counter and force-exit. Ten iterations is a reasonable default for most supervisor workflows.

2. State explosion. Every node receives the full state dictionary. If you append to messages indefinitely, a supervisor running 20 steps sends thousands of accumulated tokens per node call. Add a periodic summarization step or slice the messages list to the last N entries before passing state forward.

3. No error handling in nodes. LangGraph does not catch exceptions for you. A tool call that raises TimeoutError crashes the graph at that node with no recovery path. Wrap external calls explicitly:

def safe_search(state: AgentState) -> AgentState:
    try:
        result = search_api.call(state["query"])
        return {"messages": [AIMessage(result)]}
    except Exception as e:
        return {
            "messages": [AIMessage(f"Search unavailable: {e}. Continuing.")],
            "next": "fallback",
        }

4. Non-deterministic routing. Using a language model to decide which node to visit next is inconsistent: the same input can produce different routing decisions across runs depending on temperature and model sampling. For workflows where correctness matters, prefer explicit rule-based routing — pattern-match on output strings or check state fields directly — and reserve model-driven routing for tasks where adaptability matters more than reproducibility.

5. Skipping checkpointing. LangGraph supports built-in state persistence via SqliteSaver or PostgresSaver. Most teams skip it during development. They regret it the first time they need to debug a 20-step graph that fails on step 18 with no way to replay from a saved point. Enable it from the start:

from langgraph.checkpoint.sqlite import SqliteSaver

checkpointer = SqliteSaver.from_conn_string("agent_checkpoints.db")
app = builder.compile(checkpointer=checkpointer)

# Identify each run by a thread_id
config = {"configurable": {"thread_id": "run-42"}}
result = app.invoke(initial_state, config=config)

# Inspect any saved checkpoint
for cp in checkpointer.list(config):
    print(cp.metadata)

Thread IDs let you pause, inspect, and replay any run — invaluable for long workflows with tool calls that cost real money to re-execute.

When LangGraph is and isn't the right choice

Use it when your workflow has genuine conditional branching on intermediate results, multiple tools that need coordination, or a requirement to inspect and replay execution state. These are the cases where the graph model pays for itself.

Skip it when the task is a simple sequential pipeline: prompt → tool → parse → return. For that, a plain Python function is faster to write, easier to read, and produces a cleaner stack trace when something fails.

One note on agent security: when you give agents access to real external tools — file systems, databases, HTTP endpoints — tool call inputs become a direct attack surface. The security hardening checklists we publish include a section on agentic systems covering input validation and sandboxing for production tool use.

The takeaway

Multi-agent orchestration with LangGraph is tractable when you treat the graph as software, not magic. The patterns here — supervisor with hard step limits, parallel fan-out with explicit annotation-based joins, defensive tool wrappers, early checkpointing — cover the majority of production use cases.

The most common failure mode isn't a hard bug. It's a prototype that works on happy-path inputs and then fails quietly in production: the supervisor loops, the state grows past 50k tokens, and there's no checkpoint to resume from when a tool call fails on step 14. These are easy fixes. They're also easy to defer until they cost you.

---

I run AYI NEDJIMI Consultants, a cybersecurity consulting firm. We publish free security hardening checklists — PDF and Excel.