LangGraph Workflow Patterns

rules/

Quick Reference

State Management

rules/state-typeddict.md

TypedDict

Annotated[list, add]

rules/state-pydantic.md

BaseModel

rules/state-messages.md

MessagesState

add_messages

rules/state-reducers.md

Annotated[T, reducer_fn]

Routing & Branching

rules/routing-conditional.md

add_conditional_edges

rules/routing-retry-loops.md

rules/routing-semantic.md

Command

Parallel Execution

Annotated[list, add]

rules/parallel-fanout-fanin.md

Send

rules/parallel-map-reduce.md

asyncio.gather

rules/parallel-error-isolation.md

return_exceptions=True

Supervisor Patterns

rules/supervisor-basic.md

Command

rules/supervisor-priority.md

rules/supervisor-round-robin.md

agents_completed

Tool Calling

rules/tools-bind.md

model.bind_tools(tools)

tool_choice

rules/tools-toolnode.md

ToolNode(tools)

rules/tools-dynamic.md

rules/tools-interrupts.md

interrupt()

Checkpointing

rules/checkpoints-setup.md

MemorySaver

PostgresSaver

rules/checkpoints-recovery.md

thread_id

get_state_history

rules/checkpoints-store.md

Store

Human-in-Loop

rules/human-in-loop-interrupt.md

interrupt()

Command(resume=)

rules/human-in-loop-approval.md

interrupt_before

rules/human-in-loop-feedback.md

Streaming

rules/streaming-modes.md

rules/streaming-tokens.md

messages

rules/streaming-custom-events.md

get_stream_writer()

Subgraphs

rules/subgraphs-invoke.md

rules/subgraphs-add-as-node.md

add_node(name, compiled_graph)

rules/subgraphs-state-mapping.md

Functional API

@entrypoint

@task

rules/functional-entrypoint.md

rules/functional-task.md

.result()

rules/functional-migration.md

StateGraph

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

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

Key Decisions

add_edge(START, node)

set_entry_point()

Annotated[list[T], add]

interrupt()

["updates", "custom"]

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

test-cases.json

Related Skills

• agent-orchestration

  - Higher-level multi-agent coordination, ReAct loop patterns, and framework comparisons

• temporal-io

  - Durable execution alternative

• llm-integration

  - General LLM function calling

• type-safety-validation

  - Pydantic model patterns