Installer avec Codex ou Claude Copiez ce prompt, collez-le dans Codex, Claude ou un autre assistant, puis laissez-le vérifier la page du skill et l'installer pour vous.
Une commande directe contourne le prompt de vérification. Examinez la source avant de l'exécuter.
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.
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.
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.
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+)
# 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
Platform
Deploy graphs as managed APIs with persistence, streaming, and multi-tenancy.
