LangGraph Enterprise Patterns

LangGraph’s StateGraph API looks simple in tutorials: a few nodes, a few edges, and a compiled graph. But enterprise workloads expose sharp edges in that simplicity: concurrent state writes that silently overwrite each other, human review steps that need to survive server restarts, and streaming pipelines that must fan out to dozens of dynamic subgraphs. This lesson maps the gap between LangGraph’s getting-started examples and the patterns that actually hold up in production.

StateGraph Design with TypedDict Reducers

The foundation of every LangGraph workflow is the state schema. In enterprise systems, getting this schema wrong costs you correctness silently, no exceptions, just missing data.

The critical insight is that LangGraph fields need reducers whenever more than one node can write to them concurrently. Without a reducer, concurrent writes use last-write-wins semantics, meaning all but one agent’s output is discarded.

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

# WHY TypedDict instead of dataclass: LangGraph's state merging logic
# works with dict-like structures. TypedDict gives static type checking
# while remaining compatible with LangGraph's internal state diffing.
class ReviewState(TypedDict):
    # WHY Annotated[list, operator.add]: when 10 jurisdiction agents
    # concurrently append their findings, operator.add merges all lists
    # instead of overwriting. Without this, you lose 9 of 10 results.
    findings: Annotated[list[dict], operator.add]

    # WHY plain str here: only one node (the orchestrator) writes the
    # final summary, so last-write-wins is safe and correct.
    summary: str

    # WHY Annotated[int, operator.add]: numeric aggregations (token
    # counts, cost tallies) need additive reducers for the same reason
    # as lists: concurrent increments would otherwise overwrite.
    total_tokens_used: Annotated[int, operator.add]

    # WHY bool with no reducer: a single supervisor sets approval status.
    # Multiple writers would be a design error we want to catch early.
    approved: bool

The mermaid below shows how reducers protect state integrity during parallel fan-out:

flowchart TD
    A["Orchestrator\n(assigns jurisdictions)"] -->|"Send() × N"| B["Agent: US"]
    A -->|"Send() × N"| C["Agent: EU"]
    A -->|"Send() × N"| D["Agent: APAC"]
    B -->|"findings += [us_result]"| E["State Merge\n(operator.add reducer)"]
    C -->|"findings += [eu_result]"| E
    D -->|"findings += [apac_result]"| E
    E --> F["Aggregator Node"]
    style A fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style B fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style C fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style D fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style E fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style F fill:#EEF0F7,stroke:#6366F1,color:#0F172A

Parallel Execution with Send() and Subgraphs

Send() is LangGraph’s mechanism for dynamic fan-out, creating a variable number of parallel node invocations at runtime, each with its own state payload. This is essential when you don’t know at graph-compile time how many agents you need.

from langgraph.types import Send

def route_to_jurisdiction_agents(state: ReviewState):
    # WHY return a list of Send() objects instead of a single next node:
    # each Send() launches an independent parallel invocation of the
    # target node with a scoped state payload. LangGraph runs all of
    # them concurrently and merges results via reducers once all complete.
    return [
        Send("jurisdiction_agent", {"jurisdiction": j, "doc": state["doc"]})
        for j in state["jurisdictions"]
    ]

# WHY compile with checkpointer at the builder level, not per-invocation:
# the PostgresSaver connection pool is expensive to create. Building it
# once and passing it to compile() reuses connections across all runs.
with PostgresSaver.from_conn_string(DATABASE_URL) as checkpointer:
    checkpointer.setup()  # creates checkpoint tables if they don't exist
    graph = builder.compile(checkpointer=checkpointer)

Subgraphs vs flat graphs is a design decision that matters at scale. Use subgraphs when a group of nodes forms a reusable, independently testable unit (e.g., a “document enrichment pipeline” used by multiple parent workflows). Use flat graphs when the overhead of subgraph state translation would add latency with no reuse benefit. Subgraph state schemas are isolated, the parent and child graphs can have different state shapes, with explicit handoff nodes translating between them.

Human-in-the-Loop with interrupt_before

The interrupt_before parameter halts execution before a specified node, persists the full graph state, and hands control back to the calling process. This is not a sleep, the process can exit entirely and resume days later.

sequenceDiagram
    participant W as Workflow
    participant DB as PostgresSaver
    participant Op as Human Operator
    participant G as LangGraph

    W->>G: invoke(state, config={thread_id: "run-42"})
    G->>DB: checkpoint before "approve_contract" node
    G-->>W: raises GraphInterrupt exception
    W->>Op: notify via Slack/email with thread_id
    Op->>G: get_state(config={thread_id: "run-42"})
    Op->>G: update_state(config, {"approved": True, "edits": [...]})
    Op->>G: invoke(None, config={thread_id: "run-42"})
    G->>DB: load checkpoint, apply edits, resume
    G-->>W: final state

# WHY interrupt_before rather than a conditional node that calls input():
# interrupt_before is process-safe. The process can die between the
# interrupt and the resume: PostgresSaver holds the state. input() would
# block a worker thread and fail on any async or serverless deployment.
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["approve_contract", "send_to_regulator"],
)

# Operator resumes with edits injected into state
graph.update_state(
    config={"configurable": {"thread_id": thread_id}},
    values={"approved": True, "analyst_notes": "Reviewed. Clause 4.2 amended."},
)
# Passing None as input signals: resume from checkpoint, don't re-run from start
final_state = graph.invoke(None, config={"configurable": {"thread_id": thread_id}})

Streaming Agent Outputs Across Boundaries

For long-running workflows, streaming intermediate results keeps operators informed and enables early error detection. LangGraph’s .astream_events() emits fine-grained events: node start/end, LLM token deltas, and tool call completions.

flowchart TD
    A["astream_events()"] --> B{"event kind?"}
    B -->|"on_chat_model_stream"| C["Forward token to UI\n(real-time display)"]
    B -->|"on_tool_end"| D["Log tool result\n(audit trail)"]
    B -->|"on_chain_end + node name"| E["Update progress tracker\n(% complete)"]
    C --> F["WebSocket / SSE"]
    D --> G["Structured Logger"]
    E --> H["Dashboard"]
    style A fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style B fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style C fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style D fill:#f0fdf9,stroke:#0D9488,color:#0F172A
    style E fill:#f0fdf9,stroke:#0D9488,color:#0F172A

In enterprise deployments, filter streaming events by tags assigned at node compile time so you only pay the serialization cost for events your consumers actually need. Combine streaming with a distributed tracing system (OpenTelemetry) by emitting a span for each on_chain_start/on_chain_end pair, this gives you waterfall views of which agents took the longest.

The patterns in this lesson: typed reducers, dynamic Send(), durable interrupts, and event streaming, form the backbone of any LangGraph system that needs to scale beyond a single developer’s laptop. The next lesson covers AutoGen’s conversational multi-agent model, which takes a fundamentally different approach to orchestration through structured agent dialogue.