langgraph

# LangGraph Workflow Patterns Comprehensive patterns for building production LangGraph workflows. **LangGraph 1.x is LTS** (Long Term Support) — the first stable major release, powering agents at Uber, LinkedIn, and Klarna. Each category has individual rule files in `rules/` loaded on-demand. > **LangGraph 1.2 (shipped 2026-05-12) — the fault-tolerance release.** Everything below is on > `StateGraph.add_node(...)` unless noted: > > - **Per-node timeouts** — `timeout=` accepts `float | timedelta | TimeoutPolicy`. > `TimeoutPolicy(run_timeout=, idle_timeout=, refresh_on="auto"|"heartbeat")` separates a hard > wall-clock cap from an idle cap that progress refreshes. On expiry LangGraph raises > `NodeTimeoutError` (carrying `kind="idle"|"run"` and `elapsed`), drops that attempt's writes, and > defers to the retry policy. Cooperative: it rides asyncio cancellation, so a node blocking the > GIL is *not* interrupted. See `rules/resilience-node-timeouts.md`. > - **Node error handlers** — `error_handler=` registers a recovery node that runs once the retry > budget is exhausted. It receives failure context by declaring a parameter typed `NodeError` > (fields `node`, `error`) and returns a `Command` to update state and reroute. > See `rules/resilience-error-handlers.md`. > - **`RunControl`** (`langgraph.runtime`) — cooperative graceful shutdown. `request_drain(reason)` > from any thread; nodes poll `runtime.drain_requested` and stop at a checkpoint boundary, leaving > a resumable thread instead of a half-applied superstep. See `rules/resilience-graceful-drain.md`. > - **`DeltaChannel`** (`langgraph.channels.delta`, **beta**) — checkpoints store only incremental > writes and replay them through a batch reducer, with a snapshot every `snapshot_frequency` > updates. Fixes checkpoint cost growing with thread length. Its reducer takes a *batch* and must > be batching-invariant. See `rules/state-delta-channel.md`. > - **`runtime.heartbeat()`** — explicit progress signal, the only one that refreshes an idle timeout > under `refresh_on="heartbeat"`. > > **Landed earlier, in 1.1 — not 1.2** (they are current and supported; only their release > attribution was wrong in prior versions of this skill): deferred nodes (`defer=True`), node-level > caching (`CachePolicy` + `graph.compile(cache=...)`), and model middleware > (`before_model` / `after_model`) on `create_agent`. ## Quick Reference | Category | Rules | Impact | When to Use | |----------|-------|--------|-------------| | [State Management](#state-management) | 5 | CRITICAL | Designing workflow state schemas, accumulators, reducers, delta channels | | [Resilience](#resilience) | 3 | CRITICAL | Node timeouts, error handlers, graceful drain (1.2+) | | [Routing & Branching](#routing--branching) | 4 | HIGH | Dynamic routing, retry loops, semantic routing, cross-graph | | [Parallel Execution](#parallel-execution) | 3 | HIGH | Fan-out/fan-in, map-reduce, concurrent agents | | [Supervisor Patterns](#supervisor-patterns) | 3 | HIGH | Central coordinators, round-robin, priority dispatch | | [Tool Calling](#tool-calling) | 4 | CRITICAL | Binding tools, ToolNode, dynamic selection, approvals | | [Checkpointing](#checkpointing) | 3 | HIGH | Persistence, recovery, cross-thread Store memory | | [Human-in-Loop](#human-in-loop) | 3 | MEDIUM | Approval gates, feedback loops, interrupt/resume | | [Streaming](#streaming) | 3 | MEDIUM | Real-time updates, token streaming, custom events | | [Subgraphs](#subgraphs) | 3 | MEDIUM | Modular composition, nested graphs, state mapping | | [Functional API](#functional-api) | 3 | MEDIUM | @entrypoint/@task decorators, migration from StateGraph | | [Platform](#platform) | 3 | HIGH | Deployment, RemoteGraph, double-texting strategies | **Total: 41 rules across 12 categories** ## State Management State schemas determine how data flows between nodes. Wrong schemas cause silent data loss. | Rule | File | Key Pattern | |------|------|-------------| | TypedDict State | `rules/state-typeddict.md` | `TypedDict` + `Annotated[list, add]` for accumulators | | Pydantic Validation | `rules/state-pydantic.md` | `BaseModel` at boundaries, TypedDict internally | | MessagesState | `rules/state-messages.md` | `MessagesState` or `add_messages` reducer | | Custom Reducers | `rules/state-reducers.md` | `Annotated[T, reducer_fn]` for merge/overwrite | | Delta Channels (1.2, beta) | `rules/state-delta-channel.md` | `DeltaChannel(reducer, snapshot_frequency=)` for large accumulators | ## Resilience Fault tolerance for nodes that talk to the outside world. New in 1.2 — before it, the only lever was `retry_policy`, which cannot help a node that never fails because it never returns. | Rule | File | Key Pattern | |------|------|-------------| | Node Timeouts | `rules/resilience-node-timeouts.md` | `add_node(..., timeout=TimeoutPolicy(run_timeout=, idle_timeout=))` | | Error Handlers | `rules/resilience-error-handlers.md` | `add_node(..., error_handler=)` + param typed `NodeError` → `Command` | | Graceful Drain | `rules/resilience-graceful-drain.md` | `RunControl().request_drain()` + `runtime.drain_requested` | ```python from langgraph.types import RetryPolicy, TimeoutPolicy from langgraph.errors import NodeError builder.add_node( "call_vendor", call_vendor, timeout=TimeoutPolicy(run_timeout=300, idle_timeout=30), retry_policy=RetryPolicy(max_attempts=3), error_handler=lambda state, error: Command( update={"failure": f"{error.node}: {error.error}"}, goto="degraded_path" ), ) ``` ## Routing & Branching Control flow between nodes. Always include END fallback to prevent hangs. | Rule | File | Key Pattern | |------|------|-------------| | Conditional Edges | `rules/routing-conditional.md` | `add_conditional_edges` with explicit mapping | | Retry Loops | `rules/routing-retry-loops.md` | Loop-back edges with max retry counter | | Semantic Routing | `rules/routing-semantic.md` | Embedding similarity or `Command` API routing | | Cross-Graph Navigation | `rules/routing-cross-graph.md` | `Command(graph=Command.PARENT)` for parent/sibling routing | ## Parallel Execution Run independent nodes concurrently. Use `Annotated[list, add]` to accumulate results. | Rule | File | Key Pattern | |------|------|-------------| | Fan-Out/Fan-In | `rules/parallel-fanout-fanin.md` | `Send` API for dynamic parallel branches | | Map-Reduce | `rules/parallel-map-reduce.md` | `asyncio.gather` + result aggregation | | Error Isolation | `rules/parallel-error-isolation.md` | `return_exceptions=True` + per-branch timeout | ## Supervisor Patterns Central coordinator routes to specialized workers. Workers return to supervisor. | Rule | File | Key Pattern | |------|------|-------------| | Basic Supervisor | `rules/supervisor-basic.md` | `Command` API for state update + routing | | Priority Routing | `rules/supervisor-priority.md` | Priority dict ordering agent execution | | Round-Robin | `rules/supervisor-round-robin.md` | Completion tracking with `agents_completed` | ## Tool Calling Integrate function calling into LangGraph agents. Keep tools under 10 per agent. | Rule | File | Key Pattern | |------|------|-------------| | Tool Binding | `rules/tools-bind.md` | `model.bind_tools(tools)` + `tool_choice` | | ToolNode Execution | `rules/tools-toolnode.md` | `ToolNode(tools)` prebuilt parallel executor | | Dynamic Selection | `rules/tools-dynamic.md` | Embedding-based tool relevance filtering | | Tool Interrupts | `rules/tools-interrupts.md` | `interrupt()` for approval gates on tools | ## Checkpointing Persist workflow state for recovery and debugging. | Rule | File | Key Pattern | |------|------|-------------| | Checkpointer Setup | `rules/checkpoints-setup.md` | `MemorySaver` dev / `PostgresSaver` prod | | State Recovery | `rules/checkpoints-recovery.md` | `thread_id` resume + `get_state_history` | | Cross-Thread Store | `rules/checkpoints-store.md` | `Store` for long-term memory across threads | ## Node-Level Caching (1.2+) Independent of checkpointing. Cache individual node output so re-runs with identical inputs skip execution entirely. ```python from langgraph.graph import StateGraph from langgraph.types import CachePolicy from langgraph.cache.sqlite import SqliteCache graph = StateGraph(State) graph.add_node( "expensive_fetch", fetch_fn, cache_policy=CachePolicy(ttl=3600, key_func=lambda s: s["query"]), ) # RedisCache(url=...) for distributed workers compiled = graph.compile(cache=SqliteCache("cache.db")) ``` Use when a node is idempotent and expensive (embeddings, external APIs). Do **not** use for nodes whose output depends on wall-clock time or mutable external state unless `key_func` captures that variance. ## Deferred Nodes & Model Middleware (1.2+) ```python # defer=True — node execution is deferred until the run is about to end, # i.e. after every other upstream node has completed graph.add_node("aggregate", aggregate_fn, defer=True) # Model middleware — no subclassing required. # create_react_agent is @deprecated since v1.0; use create_agent from langchain.agents. # The legacy pre_model_hook/post_model_hook are now before_model/after_model middleware. from langchain.agents import create_agent agent = create_agent( model=model, tools=tools, middleware=[compress_history, redact_pii], # before_model / after_model hooks system_prompt="...", # prompt= renamed to system_prompt ) ``` ## Human-in-Loop Pause workflows for human intervention. Requires checkpointer for state persistence. | Rule | File | Key Pattern | |------|------|-------------| | Interrupt/Resume | `rules/human-in-loop-interrupt.md` | `interrupt()` function + `Command(resume=)` | | Approval Gate | `rules/human-in-loop-approval.md` | `interrupt_before` + state update + resume | | Feedback Loop | `rules/human-in-loop-feedback.md` | Iterative interrupt until approved | ## Streaming Real-time updates and progress tracking for workflows. **LangGraph 1.2 supports `version="v2"`** (introduced in 1.1), an opt-in streaming format with full type safety on `stream()`, `astream()`, `invoke()`, and `ainvoke()`. | Rule | File | Key Pattern | |------|------|-------------| | Stream Modes | `rules/streaming-modes.md` | 5 modes: values, updates, messages, custom, debug | | Token Streaming | `rules/streaming-tokens.md` | `messages` mode with node/tag filtering | | Custom Events | `rules/streaming-custom-events.md` | `get_stream_writer()` for progress events | | Streaming v2 | `rules/streaming-v2-format.md` | `version="v2"` for typed streaming (LG 1.1+) | ## Subgraphs Compose modular, reusable workflow components with nested graphs. | Rule | File | Key Pattern | |------|------|-------------| | Invoke from Node | `rules/subgraphs-invoke.md` | Different schemas, explicit state mapping | | Add as Node | `rules/subgraphs-add-as-node.md` | Shared state, `add_node(name, compiled_graph)` | | State Mapping | `rules/subgraphs-state-mapping.md` | Boundary transforms between parent/child | ## Functional API Build workflows using `@entrypoint` and `@task` decorators instead of explicit graph construction. | Rule | File | Key Pattern | |------|------|-------------| | @entrypoint | `rules/functional-entrypoint.md` | Workflow entry point with optional checkpointer | | @task | `rules/functional-task.md` | Returns futures, `.result()` to block | | Migration | `rules/functional-migration.md` | `StateGraph` to Functional API conversion |