langgraph

What this skill does

# 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 (Q1 2026) — new in this bump:**
>
> - **Deferred nodes** (`defer=True` on `add_node`) — the node runs only after all *other* upstream nodes have completed, so its execution is deferred until the run is about to end. This makes "aggregate once everyone else is done" patterns a one-liner instead of a custom reducer.
> - **Model middleware** (`before_model` / `after_model`) on `create_agent(...)` (from `langchain.agents`) — inject compression, summarization, or PII redaction without subclassing. Note: the legacy `pre_model_hook` / `post_model_hook` params only existed on the now-deprecated `create_react_agent`; the current equivalent is middleware on `create_agent`.
> - **Node-level caching** via `CachePolicy(ttl=..., key_func=...)` with `SqliteCache` and `RedisCache` backends (pluggable via `graph.compile(cache=...)`). Idempotent nodes skip recomputation on replay.

## Quick Reference

| Category | Rules | Impact | When to Use |
|----------|-------|--------|-------------|
| [State Management](#state-management) | 4 | CRITICAL | Designing workflow state schemas, accumulators, reducers |
| [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: 37 rules across 11 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 |

## 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.1 introduces `version="v2"`** — 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 progres