Multi-Agent Workflows with LangGraph: State, Nodes and Knowing When Not ToMulti-Agent Workflows with LangGraph: State, Nodes and Knowing When Not To

The pitch for autonomous agents is that you describe a goal and the model figures out the steps. In practice you get a system that works 70% of the time, fails in a different way each run, and is essentially impossible to debug — because the control flow lives inside a model's reasoning rather than in your code.

LangGraph's answer: you own the control flow; the model owns the decisions inside each step. That's a much better division of labour, and it's why I reach for it over a free-form agent loop.

The model: state, nodes, edges#

Three concepts, and they're all you need.

State is a typed object that flows through the graph. Every node receives it and returns a partial update.

Nodes are plain functions. Some call a model, some hit an API, some are pure logic. There's nothing special about them.

Edges decide what runs next — fixed, or conditional on the state.

from typing import Annotated, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END
 
class AgentState(TypedDict):
    query: str
    crop: str | None
    weather: dict | None
    soil: dict | None
    # Annotated with a reducer: parallel nodes append instead of overwriting
    findings: Annotated[list[str], add]
    answer: str | None
    confidence: float

That Annotated[list[str], add] is the detail that matters most and gets skipped in most tutorials. By default, a node returning a key replaces it. With a reducer, returns are merged — so two nodes running in parallel both contribute to findings instead of the second one silently clobbering the first.

If you take one thing from this: the reducer is where parallel agent output either accumulates correctly or gets silently lost.

Nodes are just functions#

def classify(state: AgentState) -> dict:
    """Work out what the user is actually asking about."""
    result = llm.with_structured_output(CropQuery).invoke(state["query"])
    return {"crop": result.crop}          # partial update, merged into state
 
 
def fetch_weather(state: AgentState) -> dict:
    data = weather_api.forecast(state["crop"], days=14)
    return {
        "weather": data,
        "findings": [f"14-day forecast: {data['summary']}"],   # appended via the reducer
    }
 
 
def fetch_soil(state: AgentState) -> dict:
    data = soil_api.profile(state["crop"])
    return {"soil": data, "findings": [f"Soil profile: {data['summary']}"]}

Two properties worth preserving deliberately: nodes take state and return a dict, so they're trivially unit-testable without the graph; and they're small enough that when one misbehaves you can read it in ten seconds.

Wiring the graph#

builder = StateGraph(AgentState)
 
builder.add_node("classify", classify)
builder.add_node("weather", fetch_weather)
builder.add_node("soil", fetch_soil)
builder.add_node("synthesize", synthesize)
builder.add_node("clarify", ask_for_clarification)
 
builder.add_edge(START, "classify")
 
# Fan out — both run concurrently
builder.add_edge("classify", "weather")
builder.add_edge("classify", "soil")
 
# Fan in — synthesize waits for both
builder.add_edge("weather", "synthesize")
builder.add_edge("soil", "synthesize")
 
graph = builder.compile()

The fan-out/fan-in is implicit: two edges from classify run in parallel, and synthesize only fires once both have completed. Their findings merge because of the reducer. Sequential code would have paid for both API calls one after the other.

Conditional edges are where the intelligence goes#

def route_after_classify(state: AgentState) -> str:
    if state["crop"] is None:
        return "clarify"           # ambiguous — ask instead of guessing
    return "gather"
 
builder.add_conditional_edges(
    "classify",
    route_after_classify,
    {"clarify": "clarify", "gather": "weather"},
)

The router is a Python function, not a prompt. It's deterministic, testable, and you can read it. The model decides what the crop is; your code decides what happens next. That separation is the whole design.

Loops need explicit budgets#

Retry and refinement loops are natural in a graph — and are also how you accidentally build something that burns $40 in tokens on one request:

def should_retry(state: AgentState) -> str:
    if state["confidence"] >= 0.7:
        return "done"
    if state["attempts"] >= 3:          # ← hard ceiling, always
        return "done"
    return "retry"
 
builder.add_conditional_edges(
    "synthesize",
    should_retry,
    {"retry": "weather", "done": END},
)

Also compile with a recursion limit as a backstop:

graph = builder.compile()
graph.invoke(inputs, config={"recursion_limit": 25})

Two independent limits, because the loop counter is the one you'll forget to increment.

Checkpointing: the feature that makes it production-ready#

A checkpointer persists state after every node. This buys three things that are genuinely hard to build yourself:

from langgraph.checkpoint.postgres import PostgresSaver
 
checkpointer = PostgresSaver.from_conn_string(os.environ["DATABASE_URL"])
graph = builder.compile(checkpointer=checkpointer)
 
config = {"configurable": {"thread_id": f"user:{user_id}:session:{session_id}"}}
result = graph.invoke({"query": "Is it a good week to sow wheat?"}, config)

Resume after failure. If the soil API times out at node four, you re-run from node four — not from the start, not paying for the first three model calls again.

Conversation memory. Same thread_id, state carries over. Multi-turn without a bespoke session store.

Time-travel debugging. Inspect state at any checkpoint and re-run from there with different inputs. This is the thing I missed most in hand-rolled agent loops — being able to ask "what did state look like right before it went wrong" and get an actual answer.

Human in the loop#

Some actions shouldn't happen without approval. interrupt_before pauses the graph and persists it:

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_purchase"],
)
 
state = graph.invoke(inputs, config)      # pauses before execute_purchase
# … present state to the user, get approval …
graph.invoke(None, config)                # resumes from the interrupt

Passing None means "continue from where you stopped." The pause can last a millisecond or a day — the state is in Postgres.

Streaming, so the UI isn't a spinner#

for event in graph.stream(inputs, config, stream_mode="updates"):
    for node, update in event.items():
        yield {"node": node, "update": update}

Surface node names in the UI — "Checking weather…", "Analysing soil…". A 12-second request feels acceptable when the user can see progress and completely broken when they can't. Cheapest UX win in the entire stack.

When not to use this#

Be honest about it. If your workflow is:

retrieve → generate → return

that's a function. Two awaits. Adding a graph framework buys you nothing but a dependency and an abstraction to explain.

LangGraph earns its complexity when you have branching that depends on intermediate results, parallel work that must be merged, loops with real termination conditions, or steps that need to survive a crash or wait for a human.

Under that bar, write the two function calls.

What I'd tell myself at the start#

• Define state first. Getting the fields and reducers right resolves most later design questions.
• Keep nodes small and pure. If a node needs a docstring to explain its branches, it's two nodes.
• Put routing in Python, not in prompts. Deterministic and testable beats clever.
• Add the checkpointer on day one, not when debugging gets painful.
• Cap every loop twice.
• Stream node names from the first version.

The reason to reach for a graph isn't that it makes agents smarter. It's that it makes them inspectable — and an AI system you can't inspect is one you can't ship.