Building a Multi-Agent System With LangGraph

Picture this: you're a developer who spent a day building a LangGraph multi-agent pipeline. Three nodes, conditional edges, everything working in your notebook. Then you add a second agent that writes to the same state field, and suddenly your pipeline is merging lists instead of replacing them, or silently dropping one agent's output entirely. You read the LangGraph documentation on state reducers, realize you missed a critical concept, and spend two more hours untangling state management that you thought was handled automatically.

State reducers are the hardest part of LangGraph multi-agent systems — and the part most tutorials gloss over. This guide explains what they do, why getting them wrong breaks pipelines silently, and how to design state schemas that multiple agents can write to safely.

The Problem: State Collisions in a Content Pipeline

A content marketing team built a LangGraph multi-agent system to automate blog post production. The pipeline had three agents: a researcher that gathered key facts, a drafter that wrote the post, and a reviewer that flagged issues.

Their state schema looked like this:

[object Object], typing ,[object Object], TypedDict, ,[object Object],, ,[object Object],

,[object Object], ,[object Object],(,[object Object],):
    task: ,[object Object],
    research: ,[object Object],[,[object Object],]
    draft: ,[object Object],
    review_notes: ,[object Object],[,[object Object],]
    status: ,[object Object],

The pipeline worked for two weeks. Then they added a second researcher agent to gather different types of sources in parallel. Both researchers wrote to research. The second researcher's output consistently overwrote the first's — not because of a bug in their code, but because of how LangGraph handles state updates when multiple nodes write to the same key without a reducer.

Debugging took three hours because the system never threw an error. The first researcher's work simply disappeared.

The Wrong Approach: Ignoring State Reducers

[object Object], langgraph.graph ,[object Object], StateGraph, END
,[object Object], langchain_anthropic ,[object Object], ChatAnthropic
,[object Object], typing ,[object Object], TypedDict, ,[object Object],

llm = ChatAnthropic(model=,[object Object],)

,[object Object],
,[object Object], ,[object Object],(,[object Object],):
    topic: ,[object Object],
    research: ,[object Object],[,[object Object],]      ,[object Object],
    draft: ,[object Object],
    final: ,[object Object],

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    response = llm.invoke(,[object Object],)
    ,[object Object], {,[object Object],: [,[object Object],]}   ,[object Object],

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    response = llm.invoke(,[object Object],)
    ,[object Object], {,[object Object],: [,[object Object],]}  ,[object Object],

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    ,[object Object],
    response = llm.invoke(,[object Object],)
    ,[object Object], {,[object Object],: response.content}

,[object Object],
graph = StateGraph(BadState)
graph.add_node(,[object Object],, researcher_a)
graph.add_node(,[object Object],, researcher_b)
graph.add_node(,[object Object],, drafter)
graph.set_entry_point(,[object Object],)
graph.add_edge(,[object Object],, ,[object Object],)
graph.add_edge(,[object Object],, ,[object Object],)
graph.add_edge(,[object Object],, END)

app = graph.,[object Object],()

What this does: When researcher_a returns {"research": ["Web: ..."]}, LangGraph replaces the research field in state. When researcher_b then returns {"research": ["Academic: ..."]}, it replaces research again. The drafter sees only the academic research. The web research is permanently lost. No error, no warning.

⚠️ Common mistake: Assuming LangGraph automatically merges lists when multiple nodes write to the same list field. It doesn't — it replaces by default. You must explicitly define a reducer function for any state field that multiple nodes will write to. Omitting reducers on shared fields is the most common source of silent data loss in LangGraph pipelines.

The Correct Approach: State Reducers and Namespaced Fields

[object Object], langgraph.graph ,[object Object], StateGraph, END
,[object Object], langchain_anthropic ,[object Object], ChatAnthropic
,[object Object], typing ,[object Object], TypedDict, ,[object Object],, ,[object Object],, Annotated
,[object Object], operator

llm = ChatAnthropic(model=,[object Object],, temperature=,[object Object],)

,[object Object],
,[object Object], ,[object Object],(,[object Object],):
    topic: ,[object Object],
    ,[object Object],
    research_sources: Annotated[,[object Object],[,[object Object],], operator.add]
    draft: ,[object Object],
    review_notes: Annotated[,[object Object],[,[object Object],], operator.add]
    final_output: ,[object Object],[,[object Object],]
    approved: ,[object Object],

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    response = llm.invoke(
        ,[object Object],
        ,[object Object],
    )
    ,[object Object],
    ,[object Object], {,[object Object],: [,[object Object],]}

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    response = llm.invoke(
        ,[object Object],
        ,[object Object],
    )
    ,[object Object], {,[object Object],: [,[object Object],]}

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    ,[object Object],
    all_research = ,[object Object],.join(state[,[object Object],])
    response = llm.invoke(
        ,[object Object],
        ,[object Object],
    )
    ,[object Object], {,[object Object],: response.content, ,[object Object],: ,[object Object],}

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    response = llm.invoke(
        ,[object Object],
        ,[object Object],
    )
    ,[object Object], json
    ,[object Object],:
        review_data = json.loads(response.content)
        ,[object Object], {
            ,[object Object],: review_data.get(,[object Object],, []),
            ,[object Object],: review_data.get(,[object Object],, ,[object Object],)
        }
    ,[object Object], json.JSONDecodeError:
        ,[object Object], {,[object Object],: [response.content], ,[object Object],: ,[object Object],}

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    ,[object Object],
    ,[object Object], ,[object Object], ,[object Object], state.get(,[object Object],, ,[object Object],) ,[object Object], ,[object Object],

,[object Object],
graph = StateGraph(ContentState)
graph.add_node(,[object Object],, web_researcher)
graph.add_node(,[object Object],, academic_researcher)
graph.add_node(,[object Object],, drafter)
graph.add_node(,[object Object],, reviewer)

graph.set_entry_point(,[object Object],)
graph.add_edge(,[object Object],, ,[object Object],)
graph.add_edge(,[object Object],, ,[object Object],)
graph.add_edge(,[object Object],, ,[object Object],)
graph.add_conditional_edges(
    ,[object Object],,
    should_revise,
    {,[object Object],: END, ,[object Object],: ,[object Object],}
)

app = graph.,[object Object],()

,[object Object],
result = app.invoke({
    ,[object Object],: ,[object Object],,
    ,[object Object],: [],
    ,[object Object],: [],
    ,[object Object],: ,[object Object],,
    ,[object Object],: ,[object Object],,
    ,[object Object],: ,[object Object],
})
,[object Object],(,[object Object],)
,[object Object],(,[object Object],)
,[object Object],(,[object Object],)

What this does: The Annotated[List[str], operator.add] declaration tells LangGraph to use operator.add (list concatenation) as the reducer for research_sources. When both researchers return a list, LangGraph appends them rather than replacing. The drafter receives all sources from both researchers. The conditional edge routes back to the drafter if the reviewer doesn't approve, enabling iterative improvement with a built-in loop.

Results and What Changed

After implementing proper state reducers and the conditional review loop, the content team saw concrete improvements:

Research completeness: Articles that previously drew from only one source type now consistently integrated both web and academic sources. Human reviewers rated completeness 28% higher.

Auto-revision success: The reviewer-drafter loop resolved 65% of quality issues automatically without human intervention, reducing average human review time from 15 minutes to 5 minutes per article.

Debugging clarity: With explicit state schemas and reducer annotations, the team could inspect the complete state at any node using LangGraph's built-in state snapshot functionality. Previously, debugging required adding print statements throughout; now they called app.get_state() at any checkpoint.

⚡ Pro tip: LangGraph's checkpointer feature lets you save state snapshots at every node execution. For production systems, use SqliteSaver or PostgresSaver as your checkpointer — this gives you automatic state persistence, the ability to resume interrupted pipelines from their last checkpoint, and a complete audit trail of how state evolved through the graph. It's one of LangGraph's most powerful features and is underused in introductory tutorials.

How to Apply the LangGraph Multi-Agent Pattern to Your Situation

The langgraph multi agent pattern works best when your task has natural conditional branching — places where the next step depends on the output of the previous step. If your pipeline is always A → B → C with no branching, a simple sequential script is easier to maintain. LangGraph earns its overhead when you need loops (retry logic, iterative improvement), conditional routing (different agents for different input types), or parallel fan-out with state merging.

Three questions to answer before choosing LangGraph:

Does your pipeline need conditional edges? If the next step is always the same regardless of the previous step's output, LangGraph adds complexity without value.

Do multiple agents need to write to overlapping state fields? If yes, state reducers are critical — and LangGraph handles them more cleanly than rolling your own coordination logic.

Do you need persistence across interruptions? LangGraph's checkpoint system is excellent for long-running tasks where network failures or process crashes would otherwise lose all progress.

Next Steps

Add the langgraph package to your project, implement the schema with proper reducer annotations, and run the pipeline on your actual task. Pay attention to the review_notes accumulation — if your reviewer flags the same issue across multiple iterations, you need to add a "don't repeat previous feedback" instruction to your reviewer's system prompt.

Conditional Edges and Dynamic Routing

State reducers handle the "what do nodes write" question. Conditional edges handle the "which node runs next" question. Together they are the two mechanisms that make LangGraph genuinely different from a sequential pipeline.

[object Object], langgraph.graph ,[object Object], StateGraph, END
,[object Object], typing ,[object Object], TypedDict, Annotated, ,[object Object],
,[object Object], operator

,[object Object], ,[object Object],(,[object Object],):
    query: ,[object Object],
    research: Annotated[,[object Object],[,[object Object],], operator.add]
    analysis: ,[object Object],
    quality_score: ,[object Object],
    revision_count: ,[object Object],

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    ,[object Object],
    ,[object Object], state[,[object Object],] >= ,[object Object],:
        ,[object Object], ,[object Object],
    ,[object Object], state[,[object Object],] >= ,[object Object],:
        ,[object Object], ,[object Object],  ,[object Object],
    ,[object Object],:
        ,[object Object], ,[object Object],

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    ,[object Object], anthropic ,[object Object], Anthropic
    client = Anthropic()
    response = client.messages.create(
        model=,[object Object],, max_tokens=,[object Object],,
        messages=[{,[object Object],: ,[object Object],, ,[object Object],: ,[object Object],}]
    )
    ,[object Object], {,[object Object],: [response.content[,[object Object],].text]}

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    ,[object Object], anthropic ,[object Object], Anthropic
    client = Anthropic()
    combined_research = ,[object Object],.join(state[,[object Object],])
    response = client.messages.create(
        model=,[object Object],, max_tokens=,[object Object],,
        messages=[{,[object Object],: ,[object Object],, ,[object Object],: ,[object Object],}]
    )
    ,[object Object], {,[object Object],: response.content[,[object Object],].text, ,[object Object],: state.get(,[object Object],, ,[object Object],)}

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    analysis = state[,[object Object],]
    score = ,[object Object],(,[object Object],, ,[object Object],(analysis.split()) / ,[object Object],)  ,[object Object],
    ,[object Object], {,[object Object],: score}

,[object Object], ,[object Object],(,[object Object],) -> ,[object Object],:
    count = state.get(,[object Object],, ,[object Object],) + ,[object Object],
    ,[object Object], {,[object Object],: state[,[object Object],] + ,[object Object],, ,[object Object],: count}

,[object Object],
workflow = StateGraph(ResearchState)
workflow.add_node(,[object Object],, research_node)
workflow.add_node(,[object Object],, analysis_node)
workflow.add_node(,[object Object],, quality_check_node)
workflow.add_node(,[object Object],, revision_node)

workflow.set_entry_point(,[object Object],)
workflow.add_edge(,[object Object],, ,[object Object],)
workflow.add_edge(,[object Object],, ,[object Object],)
workflow.add_conditional_edges(
    ,[object Object],,
    quality_router,
    {,[object Object],: END, ,[object Object],: ,[object Object],}
)
workflow.add_edge(,[object Object],, ,[object Object],)  ,[object Object],

graph = workflow.,[object Object],()
result = graph.invoke({,[object Object],: ,[object Object],, ,[object Object],: ,[object Object],})
,[object Object],(result[,[object Object],])

What this does: The quality_router function inspects the current state and returns a string — "publish" or "revise" — that maps to the next node. After quality check, the graph either ends (publish) or routes back to the analysis node (revise). The revision_count field in state prevents infinite revision loops by forcing publication after three revisions regardless of quality score. This cycle — analyze, check, revise, re-check — is impossible to express cleanly in a sequential pipeline or simple orchestrator.

⚡ Pro tip: Always include a loop-breaking condition in any LangGraph cycle. A conditional edge that routes back to an earlier node can loop indefinitely if the exit condition is never met. Common loop-breakers: a maximum iteration counter in state, a confidence threshold with a forced exit after N attempts, or a timestamp that exits after a time limit. Test the loop-breaking condition explicitly by designing inputs that will never naturally reach the exit condition.

LangGraph vs a Custom Orchestrator

LangGraph earns its setup cost when your pipeline needs two or more of these: state that persists across multiple agent calls, conditional branching that depends on intermediate outputs, cycles and revision loops, or parallel fan-out with state merging.

For pipelines that are purely sequential with no branching — Agent A → Agent B → Agent C — LangGraph adds state schema definition overhead without providing meaningful benefit over a three-function sequential script. The framework investment pays off in proportion to the routing complexity of the task.

⚡ Pro tip: Start with LangGraph's StateGraph in "debug" mode during development: graph.compile(debug=True). This prints each node's input state and output state as the graph executes, making it straightforward to identify which node produced an unexpected state value. In production, compile without debug mode; the extra output is useful in development and noisy in production.

Checkpointing for Production Reliability

LangGraph's checkpointer feature saves the full graph state after each node execution. For long-running pipelines — research tasks that make multiple external calls, workflows that take more than a few seconds — checkpointing enables resume-from-failure:

[object Object], langgraph.checkpoint.sqlite ,[object Object], SqliteSaver

,[object Object], SqliteSaver.from_conn_string(,[object Object],) ,[object Object], checkpointer:
    graph = workflow.,[object Object],(checkpointer=checkpointer)
    
    config = {,[object Object],: {,[object Object],: ,[object Object],}}
    result = graph.invoke({,[object Object],: ,[object Object],, ,[object Object],: ,[object Object],}, config)

What this does: The SqliteSaver stores state snapshots in an SQLite database after every node execution. If the pipeline fails at node 4 of 7, you can resume from the checkpoint at node 3 by invoking the graph with the same thread_id. For production, replace ":memory:" with a file path or use PostgresSaver for distributed systems.

The LangGraph Mental Model

LangGraph requires a different mental model than most orchestration frameworks. The key shift: in LangGraph, your pipeline is a graph, not a list. Tasks are nodes. The flow between tasks is edges. The entire execution state is a single typed dictionary that every node can read from and write to.

This means before writing any code, you need to answer: what fields does my state dictionary need? What does each node read? What does each node write? Which fields might multiple nodes write to (requiring a reducer)? This upfront design work feels like overhead but prevents the class of bugs — state collisions, missing fields, unexpected overwrites — that plagued the unstructured shared state approaches that LangGraph was designed to replace.

The payoff is a system where the execution flow is inspectable at every step (via checkpointing), the state transitions are explicit (via typed state), and the routing logic is testable (via the conditional edge functions). For complex pipelines that would otherwise be a tangle of shared dictionaries and if-else routing logic, LangGraph produces genuinely more maintainable code.

The langgraph multi agent pattern is most worth its setup cost for pipelines that will evolve significantly over time — adding new nodes, changing routing logic, supporting new task types. The explicit graph structure makes these changes surgical rather than risky.Once you've tuned your agent system prompts for the LangGraph context, save them to PromptABCD. LangGraph prompts often need iteration because the conversation history and state context that agents receive differ from standard single-turn prompts — the working versions are worth preserving.