OverviewHistoryStatsSecurity
npx skills add ...
Documentation
SKILL.md
npx skills add yonatangross/orchestkit --skill langgraph
LangGraph 1.x (LTS) Python workflow patterns for state management, delta channels, resilience (node timeouts, error handlers, graceful drain), routing, parallel execution, supervisor-worker, tool calling, checkpointing, human-in-loop, streaming (v2 format), subgraphs, and functional API. Use when building LangGraph pipelines, multi-agent systems, or AI workflows.
npx skills add yonatangross/orchestkit --skill langgraph
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=acceptsfloat | 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 raisesNodeTimeoutError(carryingkind="idle"|"run"andelapsed), 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. Seerules/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 typedNodeError(fieldsnode,error) and returns aCommandto update state and reroute. Seerules/resilience-error-handlers.md.RunControl(langgraph.runtime) — cooperative graceful shutdown.request_drain(reason)from any thread; nodes pollruntime.drain_requestedand stop at a checkpoint boundary, leaving a resumable thread instead of a half-applied superstep. Seerules/resilience-graceful-drain.md.DeltaChannel(langgraph.channels.delta, beta) — checkpoints store only incremental writes and replay them through a batch reducer, with a snapshot everysnapshot_frequencyupdates. Fixes checkpoint cost growing with thread length. Its reducer takes a batch and must be batching-invariant. Seerules/state-delta-channel.md.runtime.heartbeat()— explicit progress signal, the only one that refreshes an idle timeout underrefresh_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) oncreate_agent.
| Category | Rules | Impact | When to Use |
|---|---|---|---|
| State Management | 5 | CRITICAL | Designing workflow state schemas, accumulators, reducers, delta channels |
| Resilience | 3 | CRITICAL | Node timeouts, error handlers, graceful drain (1.2+) |
| Routing & Branching | 4 | HIGH | Dynamic routing, retry loops, semantic routing, cross-graph |
| Parallel Execution | 3 | HIGH | Fan-out/fan-in, map-reduce, concurrent agents |
| Supervisor Patterns | 3 | HIGH | Central coordinators, round-robin, priority dispatch |
| Tool Calling | 4 | CRITICAL | Binding tools, ToolNode, dynamic selection, approvals |
| Checkpointing | 3 | HIGH | Persistence, recovery, cross-thread Store memory |
| Human-in-Loop | 3 | MEDIUM | Approval gates, feedback loops, interrupt/resume |
| Streaming | 3 | MEDIUM | Real-time updates, token streaming, custom events |
| Subgraphs | 3 | MEDIUM | Modular composition, nested graphs, state mapping |
| Functional API | 3 | MEDIUM | @entrypoint/@task decorators, migration from StateGraph |
| Platform | 3 | HIGH | Deployment, RemoteGraph, double-texting strategies |
Total: 41 rules across 12 categories
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 |
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 |
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 |
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 |
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 |
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 |
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 |
Independent of checkpointing. Cache individual node output so re-runs with identical inputs skip execution entirely.
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.
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 |
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+) |
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 |
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 |
Deploy graphs as managed APIs with persistence, streaming, and multi-tenancy.
| Rule | File | Key Pattern |
|---|---|---|
| Deployment | rules/platform-deployment.md | langgraph.json + CLI + Assistants API |
| RemoteGraph | rules/platform-remote-graph.md | RemoteGraph for calling deployed graphs |
| Double Texting | rules/platform-double-texting.md | 4 strategies: reject, rollback, enqueue, interrupt |
version="v2" for type-safe streaming — fully typed stream() and astream() returns. Default remains "v1" for backwards compat.Command(update=..., goto=...) when updating state AND routing togetherSqliteCache (prod) or InMemoryCache from langgraph.cache.memory (dev)interrupt_before for conditional cases)set_entry_point() (deprecated)| Decision | Recommendation |
|---|---|
| State type | TypedDict internally, Pydantic at boundaries |
| Entry point | add_edge(START, node) not set_entry_point() |
| Routing + state update | Command API |
| Routing only | Conditional edges |
| Accumulators | Annotated[list[T], add] always |
| Dev checkpointer | MemorySaver |
| Prod checkpointer | PostgresSaver |
| Short-term memory | Checkpointer (thread-scoped) |
| Long-term memory | Store (cross-thread, namespaced) |
| Max parallel branches | 5-10 concurrent |
| Tools per agent | 5-10 max (dynamic selection for more) |
| Approval gates | interrupt() for high-risk operations |
| Stream modes | ["updates", "custom"] for most UIs |
| Subgraph pattern | Invoke for isolation, Add-as-Node for shared state |
| Functional vs Graph | Functional for simple flows, Graph for complex topology |
add reducer (overwrites instead of accumulates)interrupt() in try/except (breaks the mechanism).result() on Functional API tasksset_entry_point() (deprecated, use add_edge(START, ...))See test-cases.json for consolidated test cases across all categories.
ork:agent-orchestration - Higher-level multi-agent coordination, ReAct loop patterns, and framework comparisonstemporal-io - Durable execution alternativeork:llm-integration - General LLM function callingtype-safety-validation - Pydantic model patternsfrom 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"))# 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
)from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing import TypedDict, Annotated, Literal
from operator import add
class State(TypedDict):
input: str
results: Annotated[list[str], add]
def supervisor(state) -> Command[Literal["worker", END]]:
if not state.get("results"):
return Command(update={"input": state["input"]}, goto="worker")
return Command(goto=END)
def worker(state) -> dict:
return {"results": [f"Processed: {state['input']}"]}
graph = StateGraph(State)
graph.add_node("supervisor", supervisor)
graph.add_node("worker", worker)
graph.add_edge(START, "supervisor")
graph.add_edge("worker", "supervisor")
app = graph.compile()