langgraph（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 (Q1 2026) — new in this bump: - Deferred nodes (defer=True on add_node) — the node runs only after all *other* upstream nodes for the current super-step have completed, which makes "aggregate once everyone else is done" patterns a one-liner instead of a custom reducer. - Pre/post model hooks on create_react_agent(...) and ToolNode — inject compression, summarization, or PII redaction without subclassing. - 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

Total: 37 rules across 11 categories

State Management

State schemas determine how data flows between nodes. Wrong schemas cause silent data loss.

Routing & Branching

Control flow between nodes. Always include END fallback to prevent hangs.

Parallel Execution

Run independent nodes concurrently. Use Annotated[list, add] to accumulate results.

Supervisor Patterns

Central coordinator routes to specialized workers. Workers return to supervisor.

Tool Calling

Integrate function calling into LangGraph agents. Keep tools under 10 per agent.

Checkpointing

Persist workflow state for recovery and debugging.

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.cache import CachePolicy, 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 Hooks (1.2+)

# defer=True — node waits for every other upstream node at this super-step
graph.add_node("aggregate", aggregate_fn, defer=True)

# Pre/post model hooks — no subclassing required
from langgraph.prebuilt import create_react_agent

agent = create_react_agent(
    model=model,
    tools=tools,
    pre_model_hook=compress_history,   # e.g. summarize if > N tokens
    post_model_hook=redact_pii,        # e.g. scrub emails/SSNs before persist
)

Human-in-Loop

Pause workflows for human intervention. Requires checkpointer for state persistence.

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().

Subgraphs

Compose modular, reusable workflow components with nested graphs.

Functional API

Build workflows using @entrypoint and @task decorators instead of explicit graph construction.

Platform

Deploy graphs as managed APIs with persistence, streaming, and multi-tenancy.

Quick Start Example

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()

2026 Key Patterns

• Streaming v2 (LG 1.1): Use version="v2" for type-safe streaming — fully typed stream() and astream() returns. Default remains "v1" for backwards compat.
• Command API: Use Command(update=..., goto=...) when updating state AND routing together
• context_schema: Pass runtime config (temperature, provider) without polluting state
• CachePolicy: Cache expensive node results with TTL via InMemoryCache
• RemainingSteps: Proactively handle recursion limits
• Store: Cross-thread memory separate from Checkpointer (thread-scoped)
• interrupt(): Dynamic interrupts inside node logic (replaces interrupt_before for conditional cases)
• add_edge(START, node): Not set_entry_point() (deprecated)
• LTS release: LangGraph 1.x is LTS — will remain ACTIVE until v2.0

Key Decisions

Common Mistakes

1. Forgetting add reducer (overwrites instead of accumulates)
2. Mutating state in place (breaks checkpointing)
3. No END fallback in routing (workflow hangs)
4. Infinite retry loops (no max counter)
5. Side effects in router functions
6. Too many tools per agent (context overflow)
7. Raising exceptions in tools (crashes agent loop)
8. No checkpointer in production (lose progress on crash)
9. Wrapping interrupt() in try/except (breaks the mechanism)
10. Not transforming state at subgraph boundaries
11. Forgetting .result() on Functional API tasks
12. Using set_entry_point() (deprecated, use add_edge(START,...))

Evaluations

See test-cases.json for consolidated test cases across all categories.

Related Skills

• ork:agent-orchestration - Higher-level multi-agent coordination, ReAct loop patterns, and framework comparisons
• temporal-io - Durable execution alternative
• ork:llm-integration - General LLM function calling
• type-safety-validation - Pydantic model patterns