# LangGraph — Components Deep Dive

*A study guide to LangGraph 1.x (Python), taken one component at a time: state, nodes, edges, compile and run, persistence, interrupts, fault tolerance, subgraphs, workflow patterns, the Functional API and streaming. Each part says what it is, shows the code, says when to use it, and lists the traps.*

> **How to read this guide.** Each component has (1) a mental model, (2) the code you will type, (3) a *when to use / when not to* verdict, (4) traps, and (5) 📖 doc links. Section 12 is the decision table for **“when do I use which kind of memory?”** and section 9 for **“which workflow pattern?”**. The companion page is *LangChain — Components Deep Dive* (tools, middleware, memory from the agent side). **Version note:** written 17 Oct 2026 from docs.langchain.com. ⚑ marks features that need a minimum version (table in §14). The Functional API section (§10) draws on the graph-API docs and my prior knowledge more than on a full re-read, so verify it against the linked pages.

---

## 0. The map: what LangGraph is

> **Analogy: a factory floor with a conveyor and a clipboard.** The *clipboard* is the shared state. Each *station* (node) reads the clipboard, does a job, and writes updates. *Signs on the floor* (edges) say which station comes next. A *supervisor with a camera* (the checkpointer) photographs the clipboard after every shift, so if the power fails you restart from the last photo, not from the start.

- **Three primitives:** **State** (shared snapshot plus the rules for merging updates), **Nodes** (functions that do the work and return updates), **Edges** (functions or fixed links that choose what runs next). *Nodes do the work, edges say what to do next.* Nodes and edges are plain functions: they may contain an LLM or just ordinary code.
- **Execution model (Pregel-inspired message passing):** the program runs in discrete **super-steps**. Nodes that run in parallel share a super-step; sequential nodes are separate ones. A node becomes active when it receives a message on an incoming edge, runs, and emits updates. The run ends when every node is inactive and nothing is in transit. **Checkpoints are written at super-step boundaries**, which explains most of the behaviour in §5-7.
- **LangGraph is a low-level runtime.** It gives you durable execution, streaming, human-in-the-loop and persistence. LangChain's `create_agent` is built on it, so you can start high-level and drop down. The docs' guidance: use LangGraph for fine-grained orchestration control, long-running stateful work, workflows mixing deterministic and agentic steps, and production-grade deployment.
- **Two APIs:** the **Graph API** (`StateGraph`, explicit nodes and edges; this page's main subject) and the **Functional API** (`@entrypoint` and `@task` on ordinary Python control flow; §10).

<table id="bkmrk-you-need%E2%80%A6usea-model-"><thead><tr><th>You need…</th><th>Use</th></tr></thead><tbody><tr><td>A model plus tools in a loop, memory, approvals</td><td>`create_agent` (LangChain)</td></tr><tr><td>Branching, fan-out/fan-in, explicit loops, deterministic steps between model calls</td><td>**LangGraph Graph API**</td></tr><tr><td>Mostly linear Python with loops and ifs, but you want checkpointing and interrupts</td><td>**Functional API**</td></tr><tr><td>Planning, files and subagents out of the box for long autonomous tasks</td><td>Deep Agents (built on LangGraph)</td></tr></tbody></table>

📖 [Overview](https://docs.langchain.com/oss/python/langgraph/overview) · [Thinking in LangGraph](https://docs.langchain.com/oss/python/langgraph/thinking-in-langgraph) · [Graph API overview](https://docs.langchain.com/oss/python/langgraph/graph-api) · [Choosing between Graph and Functional](https://docs.langchain.com/oss/python/langgraph/choosing-apis) · [LangGraph runtime](https://docs.langchain.com/oss/python/langgraph/pregel)

---

## 1. State

> **Analogy: a shared whiteboard with one marker per column.** Each column (state key) has its own rule for what happens when someone writes: overwrite the old value, or append to it.

### 1.1 Schemas

- **`TypedDict`** is the main documented choice. Use a **`dataclass`** if you want default values. A **Pydantic `BaseModel`** adds recursive validation but is slower. (`create_agent` in LangChain only supports TypedDict state.)
- **Input/output/private schemas.** By default input and output equal the graph schema. Pass `input_schema=` and `output_schema=` to constrain what the graph accepts and returns; nodes can also write to extra *private* channels declared by a schema they use.

```
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class InputState(TypedDict):
    user_input: str
class OutputState(TypedDict):
    graph_output: str
class OverallState(TypedDict):
    foo: str
    user_input: str
    graph_output: str
class PrivateState(TypedDict):
    bar: str

def node_1(state: InputState) -> OverallState:  return {'foo': state['user_input'] + ' name'}
def node_2(state: OverallState) -> PrivateState: return {'bar': state['foo'] + ' is'}
def node_3(state: PrivateState) -> OutputState:  return {'graph_output': state['bar'] + ' Lance'}

b = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
b.add_node('node_1', node_1); b.add_node('node_2', node_2); b.add_node('node_3', node_3)
b.add_edge(START, 'node_1'); b.add_edge('node_1', 'node_2'); b.add_edge('node_2', 'node_3'); b.add_edge('node_3', END)
b.compile().invoke({'user_input': 'My'})      # {'graph_output': 'My name is Lance'}
```

**Trap: private is not secret.** Schemas constrain what a node *reads* and what `invoke` *returns*. When streaming with `stream_mode='values'` the graph emits **all** channels, private ones included. Pass `output_keys=[...]` to restrict, or use `stream_mode='updates'`.

### 1.2 Reducers

Each key has its own reducer: a binary function `new = reducer(left=current, right=update)`. With no reducer, **the update overwrites**.

```
from operator import add
from typing import Annotated
class State(TypedDict):
    foo: int                          # overwrite
    bar: Annotated[list[str], add]    # append
# state {'foo':1,'bar':['hi']}; a node returns {'foo':2} then {'bar':['bye']} -> {'foo':2,'bar':['hi','bye']}
```

- **Messages:** use `Annotated[list[AnyMessage], add_messages]`, or subclass `MessagesState` (`class State(MessagesState): documents: list[str]`). `add_messages` tracks message *IDs*, so updating an existing message replaces it and `RemoveMessage` deletes one; `operator.add` would just append duplicates. It also deserialises dicts (`{'type': 'human', 'content': ...}`) into message objects, so read attributes with dot notation.
- **Parallel writes:** two nodes writing the same key in one super-step *need* a reducer, or the update is ambiguous.
- **Overwrite:** a merging reducer cannot be cleared by returning an empty value (`[]` is merged in). Wrap the update: `{'errors': Overwrite([])}` (from `langgraph.types`). This matters for error buffers and retry counters.
- **RemainingSteps:** a managed value, `remaining_steps: RemainingSteps`, tells a node how many super-steps are left before the recursion limit (§4).

**When to put something in state:** if more than one node needs it or it must survive an interrupt or crash. **When not to:** connections, clients, secrets and per-run config belong in runtime *context* (§2), because state is checkpointed and must be serialisable.

📖 [State](https://docs.langchain.com/oss/python/langgraph/graph-api#state) · [Use the graph API (how-tos)](https://docs.langchain.com/oss/python/langgraph/use-graph-api)

---

## 2. Nodes

> **Analogy: a station on the floor.** It receives the clipboard, does one job, writes back only the lines it changed.

A node is a sync or async Python function. It can accept `state`, then optionally `config` (a `RunnableConfig` with `thread_id`, tags) and `runtime` (a `Runtime[Context]`). A node returns a **partial update**, not the whole state.

```
from dataclasses import dataclass
from langgraph.runtime import Runtime

@dataclass
class Context:
    user_id: str

def my_node(state: State, runtime: Runtime[Context]):
    print(runtime.context.user_id, runtime.execution_info.thread_id)
    return {'results': 'Hello, {}!'.format(state['input'])}

builder = StateGraph(State, context_schema=Context)
builder.add_node('my_node', my_node)      # name defaults to the function name if omitted
```

<table id="bkmrk-runtime-fieldusecont"><thead><tr><th>`Runtime` field</th><th>Use</th></tr></thead><tbody><tr><td>`context`</td><td>Per-run immutable config (`context_schema`; pass `context=` to invoke)</td></tr><tr><td>`store`</td><td>Long-term memory store</td></tr><tr><td>`stream_writer`</td><td>Emit custom stream events</td></tr><tr><td>`execution_info` ⚑</td><td>`thread_id`, `run_id`, `checkpoint_id`, `task_id`, `node_attempt` (1-indexed), `node_first_attempt_time`</td></tr><tr><td>`server_info` ⚑</td><td>Assistant/graph ID and authenticated user when running on LangGraph Server</td></tr><tr><td>`heartbeat()` ⚑</td><td>Reset an idle timeout manually (§7)</td></tr><tr><td>`control` / `drain_requested` ⚑</td><td>Graceful-shutdown signals (§7)</td></tr></tbody></table>

### 2.1 Re-execution and idempotency (the rule behind many bugs)

Checkpoints are saved at super-step boundaries, **not mid-function**. After an interrupt or a retry the affected node **runs again from its first line**, so any side effect before the pause runs again. Design nodes so a second run is harmless: upserts instead of inserts, idempotency keys, read-before-write. You may freely add or remove nodes and edges for existing threads; determinism rules apply only to *tasks* and *interrupts* inside a node.

### 2.2 Tasks inside a node

If a node does several expensive or non-idempotent operations, wrap each in a `@task` (from `langgraph.func`). Task results are checkpointed, so on resume completed tasks are skipped instead of re-run.

```
from langgraph.func import task
@task
def fetch(url: str): return requests.get(url).text[:100]
def call_api(state):
    futures = [fetch(u) for u in state['urls']]      # run concurrently
    return {'results': [f.result() for f in futures]}
```

### 2.3 Node caching

```
from langgraph.types import CachePolicy
from langgraph.cache.memory import InMemoryCache
builder.add_node('expensive', expensive, cache_policy=CachePolicy(ttl=3))   # ttl seconds; key_func defaults to a hash of the input
graph = builder.compile(cache=InMemoryCache())
```

Use caching for pure, expensive functions (embedding a fixed text, a slow lookup). Do **not** cache nodes with side effects or time-dependence. The result carries `__metadata__: {'cached': True}` on a hit.

📖 [Nodes](https://docs.langchain.com/oss/python/langgraph/graph-api#nodes)

---

## 3. Edges, Send and Command

> **Analogy: signs on the factory floor.** Some arrows are painted on (normal edges). Some depend on what is on the clipboard (conditional edges). A foreman can also say 'make one copy of this job per item on the list' (Send) or 'do this update *and* go there next' (Command).

### 3.1 Edge types

```
graph.add_edge('a', 'b')                                          # always a -> b
graph.add_edge(START, 'a'); graph.add_edge('b', END)              # entry / finish
graph.add_conditional_edges('a', route_fn)                        # route_fn(state) returns a node name or list of names
graph.add_conditional_edges('a', route_fn, {True: 'b', False: 'c'})   # optional path map
graph.add_conditional_edges(START, route_fn)                      # conditional entry point
```

- **Fan-out is automatic:** a node with several outgoing edges runs *all* destinations in parallel in the next super-step. A conditional edge that returns a list does the same.
- **One routing mechanism per node.** Use normal edges *or* dynamic routing (conditional edges, `Command`, tool-driven `goto`), never both. `Command` only *adds* dynamic edges; static edges still fire, so you would run both destinations.

### 3.2 Command

`Command(update=..., goto=..., graph=..., resume=...)`. Return it from a node to update state and route in one step. You **must annotate the return type** with the possible destinations, `-> Command[Literal['next_node']]`, so rendering and validation know where it can go.

```
def decide(state: State) -> Command[Literal['approve', 'reject']]:
    return Command(update={'checked': True}, goto='approve' if state['score'] > 5 else 'reject')
```

- **`graph=Command.PARENT`** navigates from inside a subgraph to a node in the parent (the basis of multi-agent handoffs). If the key is shared by parent and subgraph schemas, the parent needs a reducer for it.
- **Input to `invoke`:** `Command(resume=...)` is the *only* Command intended as run input. Never send `Command(update=...)` to continue a conversation: any Command input resumes from the latest checkpoint, so a finished graph looks stuck. Send a plain input dict (`{'messages': [...]}`) to start a new turn on an existing thread.
- **Use `Command` versus a conditional edge:** Command when you must update state *and* route; conditional edge when you only route.

### 3.3 Send (map-reduce)

> **Analogy:** a teacher hands each student a different page of the exam to mark, then collects the marks into one pile.

```
from langgraph.types import Send
def assign_workers(state: State):
    return [Send('write_section', {'section': s}) for s in state['sections']]   # one worker per item, each with its own input

class State(TypedDict):
    sections: list[Section]
    completed: Annotated[list, operator.add]     # workers write here in parallel, so it needs a reducer
graph.add_conditional_edges('orchestrator', assign_workers, ['write_section'])
```

Use `Send` when the *number* of parallel branches is unknown until runtime and each branch needs its own input. A `Send` can carry its own `timeout=` ⚑ that overrides the target node's default.

📖 [Edges](https://docs.langchain.com/oss/python/langgraph/graph-api#edges) · [Command](https://docs.langchain.com/oss/python/langgraph/graph-api#command) · [Map-reduce and Send how-to](https://docs.langchain.com/oss/python/langgraph/use-graph-api)

---

## 4. Compile and run

You **must** `compile()` before use. Compilation checks structure (no orphan nodes) and is where you attach runtime pieces.

```
graph = builder.compile(
    checkpointer=InMemorySaver(),      # thread persistence, HITL, time travel
    store=my_store,                    # cross-thread memory
    cache=InMemoryCache(),             # node caching
    interrupt_before=['node_a'],       # static breakpoints (debugging only)
)
config = {'configurable': {'thread_id': 't1'}, 'recursion_limit': 50}
result = graph.invoke({'messages': [...]}, config, context={'llm_provider': 'anthropic'})
```

- **`thread_id`** lives in `configurable`. **`recursion_limit`** is a *top-level* key of config, not inside `configurable`. **Context** is a separate `context=` argument.
- **Recursion limit** = max super-steps in one run. Default 1000 since LangGraph 1.0.6; exceeding it raises `GraphRecursionError`. The step counter is `config['metadata']['langgraph_step']`.
- **Proactive handling (recommended):** a `RemainingSteps` state field plus a router that sends the graph to a fallback node when `remaining_steps <= 2`, so the run ends gracefully with partial results and checkpoints. **Reactive:** catch `GraphRecursionError` outside; simpler, but the run is terminated.
- **Other metadata in `config['metadata']`:** `langgraph_node`, `langgraph_triggers`, `langgraph_path`, `langgraph_checkpoint_ns`.
- **Visualise:** `graph.get_graph().draw_mermaid_png()` (use `xray=True` to expand subgraphs).
- **v2 output ⚑:** `invoke(..., version='v2')` returns an object with `.value` (state) and `.interrupts`, cleanly separating interrupts from state.

📖 [Recursion limit](https://docs.langchain.com/oss/python/langgraph/graph-api#recursion-limit) · [Quickstart](https://docs.langchain.com/oss/python/langgraph/quickstart) · [Run a local server](https://docs.langchain.com/oss/python/langgraph/local-server)

---

## 5. Persistence: checkpointers, threads, stores

> **Analogy: save slots in a video game.** A *thread* is one save file. A *checkpoint* is a save point written at the end of every super-step. Loading a slot continues from the last save; you can also load an older save and branch (*time travel*). The *store* is the shared hall of fame that every save file can read.

- **Checkpointer:** compile with one and LangGraph snapshots state at every super-step, organised in threads. `thread_id` is the **primary key**: without it nothing is saved and nothing can resume after an interrupt.
- **Pending writes:** if one node fails mid-super-step, successful sibling nodes' writes are stored, so a retry does not redo them.
- **What persistence unlocks:** human-in-the-loop, conversational (short-term) memory, time-travel replay and forking, and fault-tolerant execution.
- **Backends:** `InMemorySaver` (dev), SQLite, Postgres (`PostgresSaver`, `AsyncPostgresSaver`), MongoDB, Azure Cosmos DB. Call `.setup()` once for SQL backends. On the Agent Server all of this is provisioned for you.
- **Inspect:** `graph.get_state(config)`; add `subgraphs=True` to see nested state (only works when the subgraph is statically discoverable: added as a node or called inside a node, not when hidden inside a tool).

### 5.1 Stores (cross-thread)

A store holds JSON documents under a **namespace tuple** and a **key**, accessible from any thread. It is the right place for user preferences and learned facts.

```
from langgraph.store.memory import InMemoryStore       # PostgresStore / MongoDB / Redis in production
store = InMemoryStore(index={'embed': embeddings, 'dims': 1536, 'fields': ['food_preference', '$']})
ns = (user_id, 'memories')
store.put(ns, str(uuid7()), {'food_preference': 'I like pizza'}, index=['food_preference'])   # index=False: stored, not searchable
store.search(ns, query='What does the user like to eat?', limit=3)
store.list_namespaces(prefix=(user_id,), max_depth=2)

graph = builder.compile(checkpointer=checkpointer, store=store)
async def call_model(state: MessagesState, runtime: Runtime[Context]):
    hits = await runtime.store.asearch((runtime.context.user_id, 'memories'), query=state['messages'][-1].content, limit=3)
```

- **`search` traps:** `namespace_prefix` matches by *prefix* (`('alice',)` also returns `('alice','memories')`); results beyond `limit` (default 10) are **silently dropped**, so paginate with `offset`; ordering differs by backend (Postgres: most recently updated first; in-memory: insertion order), so sort client-side if order matters.
- **Custom store:** subclass `BaseStore` and implement the five async methods (`aput`, `aget`, `adelete`, `asearch`, `alist_namespaces`). No conformance suite exists, so test against `InMemoryStore`.

**Checkpointer versus store:** the checkpointer saves the *whole graph state for one thread*; the store saves *chosen facts for many threads*. You normally want both.

**Time travel** (replay a past execution, or fork from a past checkpoint to explore another path) has its own how-to page; I did not re-read it for this guide.

📖 [Persistence](https://docs.langchain.com/oss/python/langgraph/persistence) · [Checkpointers](https://docs.langchain.com/oss/python/langgraph/checkpointers) · [Stores](https://docs.langchain.com/oss/python/langgraph/stores) · [Time travel](https://docs.langchain.com/oss/python/langgraph/use-time-travel) · [Add memory](https://docs.langchain.com/oss/python/langgraph/add-memory)

---

## 6. Interrupts and human-in-the-loop

> **Analogy: a bookmark with a note in the margin.** The graph stops, marks its place, and waits as long as needed. When you hand back an answer, it re-reads the chapter from the start of the page and carries on.

Call `interrupt(value)` inside a node. LangGraph saves state, surfaces `value` (any JSON-serialisable object), and waits. Resume by invoking again with `Command(resume=answer)` on the **same thread\_id**; `answer` becomes the return value of `interrupt()`. You need a checkpointer and a thread ID.

```
from langgraph.types import interrupt, Command

def approval(state) -> Command[Literal['proceed', 'cancel']]:
    ok = interrupt({'question': 'Proceed?', 'details': state['action']})
    return Command(goto='proceed' if ok else 'cancel')

stream = graph.stream_events(inputs, config, version='v3')
_ = stream.output                       # drive the run
if stream.interrupted: print(stream.interrupts)      # tuple of Interrupt(value=..., id=...)
graph.stream_events(Command(resume=True), config, version='v3').output
```

### 6.1 The rules (each one is a classic bug)

1. **The node restarts from its first line on resume.** Everything before `interrupt()` runs again.
2. **Never wrap `interrupt()` in a bare `try/except`.** It pauses by raising a special exception; swallowing it silently disables the pause. Catch specific exception types only.
3. **Do not reorder, skip or loop interrupts non-deterministically.** Resume values are matched by *index* within the node. No conditional skipping, no loops over data that may change.
4. **No `while True` validation loops with `interrupt()` in one node.** Each resume replays earlier iterations. Instead: call `interrupt()` once, store a `pending_question` in state on invalid input, and use a **conditional edge** to loop back.
5. **Payloads must be serialisable** (strings, numbers, dicts of simple values). No functions or class instances.
6. **Side effects before `interrupt()` must be idempotent**: upsert, not insert; or move the side effect after the interrupt or into a separate node.

### 6.2 Patterns

<table id="bkmrk-patternshapeapprove-"><thead><tr><th>Pattern</th><th>Shape</th></tr></thead><tbody><tr><td>Approve or reject</td><td>`interrupt` returns bool; route with `Command(goto=...)`</td></tr><tr><td>Review and edit state</td><td>`edited = interrupt({'content': state['draft']})`; return `{'draft': edited}`</td></tr><tr><td>Interrupt inside a tool</td><td>The tool calls `interrupt({...})` before acting; resume with `{'action': 'approve', 'subject': 'new subject'}` to approve with edits</td></tr><tr><td>Validate human input</td><td>One interrupt per node call plus a conditional-edge loop (rule 4)</td></tr><tr><td>Several parallel interrupts</td><td>Resume with `Command(resume={interrupt.id: answer, ...})` for all pending IDs at once</td></tr><tr><td>Typed answers ⚑</td><td>`interrupt(payload, response_schema=Decision)` (Pydantic, TypedDict or dataclass); invalid resume raises `ValidationError` and you can resend to the same thread. A raw JSON-schema dict is exposed to clients but *not* validated. Needs langgraph 1.2.12.</td></tr></tbody></table>

- **Static breakpoints** (`interrupt_before`/`interrupt_after` at compile or run time, resume with `invoke(None, config)`) are for **debugging**, not for human-in-the-loop. Use `interrupt()` for that.
- **Subgraph called as a function:** both the parent node and the subgraph node restart from their beginnings on resume.
- **Drive a UI loop:** call `stream_events` in a loop, stream tokens from `stream.messages`, check `stream.interrupted`, ask the user, then resume with `Command(resume=...)`, until nothing is interrupted.

📖 [Interrupts](https://docs.langchain.com/oss/python/langgraph/interrupts) · [HITL concepts](https://docs.langchain.com/oss/python/concepts/human-in-the-loop)

---

## 7. Fault tolerance

> **Analogy: a delivery driver with three escalating responses.** Knock again (retry). Don't wait forever at one door (timeout). If the house is truly unreachable, follow the 'undeliverable' procedure (error handler): leave a note and return to depot.

⚑ Per-node timeouts, error handlers, `set_node_defaults` and graceful drain need **langgraph 1.2+**. The order is fixed: an attempt raises (including a timeout) → the retry policy decides → only after retries are exhausted does the error handler run.

### 7.1 Retries

```
from langgraph.types import RetryPolicy, default_retry_on
builder.add_node('call_api', call_api, retry_policy=RetryPolicy(max_attempts=3))
# custom: retry_on can be an exception type, a sequence of types, or a callable(exc) -> bool
def retry_on(exc): return False if isinstance(exc, MyBug) else default_retry_on(exc)
```

<table id="bkmrk-parameterdefaultmax_"><thead><tr><th>Parameter</th><th>Default</th></tr></thead><tbody><tr><td>`max_attempts` (includes the first)</td><td>3</td></tr><tr><td>`initial_interval` / `backoff_factor` / `max_interval`</td><td>0.5 s / 2.0 / 128 s</td></tr><tr><td>`jitter`</td><td>True</td></tr><tr><td>`retry_on`</td><td>`default_retry_on`</td></tr></tbody></table>

**Default is conservative:** it retries any exception *except* `ValueError`, `TypeError`, `ArithmeticError`, `ImportError`, `LookupError`, `NameError`, `SyntaxError`, `RuntimeError`, `ReferenceError`, `StopIteration`, `StopAsyncIteration`, `OSError` (programming errors). For `requests` and `httpx` only 5xx are retried. `NodeTimeoutError` is retryable. Inside a node, `runtime.execution_info.node_attempt` lets you switch to a fallback API on attempt 2+.

### 7.2 Timeouts

```
from langgraph.types import TimeoutPolicy
builder.add_node('call_model', call_model, timeout=60)                                  # seconds, wall clock
builder.add_node('call_model', call_model, timeout=TimeoutPolicy(run_timeout=120, idle_timeout=30))
```

- **Async nodes only.** A sync node with a timeout is rejected at compile time. Wrap blocking I/O with `asyncio.to_thread` inside an async node.
- `run_timeout`: hard cap per attempt, never refreshed. `idle_timeout`: fires only if the node stops making progress; the clock resets on state writes, stream chunks, child-task scheduling, stream-writer calls and any LangChain callback event (tokens, tool calls). `refresh_on='heartbeat'` restricts refresh to explicit `runtime.heartbeat()` calls for long work with no natural progress signal.
- On timeout: `NodeTimeoutError` (with `node`, `elapsed`, `kind`), writes from that attempt are cleared, then the retry policy decides. The clock resets per attempt.

### 7.3 Error handlers (compensation, the Saga pattern)

```
from langgraph.errors import NodeError
def payment_failed(state: State, error: NodeError) -> Command:
    return Command(update={'status': 'compensated: ' + str(error.error)}, goto='finalize')

builder.add_node('charge_payment', charge_payment,
                 retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
                 error_handler=payment_failed)
```

- The handler gets the state and, if you annotate `error: NodeError`, the failing `node` and `error`. It may also take `runtime` or `config`. Return a state update or a `Command` to route elsewhere.
- **Failure provenance is checkpointed:** if the process crashes after the failure but before the handler finishes, the handler sees the same `NodeError` on resume.
- **`interrupt()` bypasses retries and handlers** (it is control flow, not failure). A failing *subgraph* surfaces to the parent node's handler. If the handler itself raises, the error propagates as if there were no handler. One handler per node.

### 7.4 Graph-wide defaults

```
graph = (StateGraph(State)
    .set_node_defaults(retry_policy=RetryPolicy(max_attempts=3), error_handler=mark_failed, timeout=TimeoutPolicy(run_timeout=30))
    .add_node('step_a', step_a)
    .add_node('charge', charge, error_handler=refund)     # per-node value overrides the default
    .compile())
```

<table id="bkmrk-defaultregular-nodes"><thead><tr><th>Default</th><th>Regular nodes</th><th>Error-handler nodes</th></tr></thead><tbody><tr><td>`retry_policy`</td><td>Yes</td><td>Yes</td></tr><tr><td>`timeout`</td><td>Yes</td><td>Yes</td></tr><tr><td>`error_handler`</td><td>Yes</td><td>**No** (a handler must not catch itself)</td></tr><tr><td>`cache_policy`</td><td>Yes</td><td>**No** (caching a handler is unsafe)</td></tr></tbody></table>

Defaults are **not inherited by subgraphs**. The same `timeout=` and `retry_policy=` exist on `@task` and `@entrypoint` in the Functional API.

### 7.5 Graceful shutdown (drain)

```
import signal
from langgraph.runtime import RunControl
from langgraph.errors import GraphDrained
control = RunControl()
signal.signal(signal.SIGTERM, lambda *_: control.request_drain('sigterm'))
try:
    graph.invoke(inputs, config, control=control)
except GraphDrained as e:
    pass                                   # checkpoint saved; resume later
graph.invoke(None, config)                 # resume with the same thread_id
```

Drain is cooperative and works **between super-steps**: a running node (including one mid-retry) finishes first. It does not cancel asyncio tasks or kill threads; pair it with a hard timeout. If the graph finishes on the same tick, it just returns normally (check `control.drain_requested`). Inside a node, `runtime.drain_requested` lets you skip expensive work.

### 7.6 Changing a deployed graph (migrations)

- Threads *at the end* of the graph: any topology change is fine.
- Threads *interrupted*: any change except **renaming or removing a node** the thread is about to enter.
- State keys: adding or removing is safe both ways; **renaming a key loses its saved value** in existing threads; incompatible type changes can break old threads.
- Changes that are technically compatible but alter behaviour (a new tool set, a restructured flow): pin a behavioural version in state so old threads keep the old path while new threads take the new one.

**Traps.** Retrying a non-idempotent node duplicates side effects. A timeout on a sync node fails at compile time. Putting the handler on every node via defaults and then *also* expecting handler nodes to be covered (they are not).

📖 [Fault tolerance](https://docs.langchain.com/oss/python/langgraph/fault-tolerance) · [Backward compatibility](https://docs.langchain.com/oss/python/langgraph/backward-compatibility)

---

## 8. Subgraphs

> **Analogy: a department inside the company.** The company chart (parent graph) treats it as one box. Inside, it has its own stations. Whether the department remembers last week's work depends on its filing policy (checkpointer mode).

### 8.1 How parent and subgraph talk

<table id="bkmrk-patternuse-whenhowca"><thead><tr><th>Pattern</th><th>Use when</th><th>How</th></tr></thead><tbody><tr><td>**Call inside a node**</td><td>Different state schemas (no shared keys), or you must transform state</td><td>A wrapper node maps parent state to subgraph input, calls `subgraph.invoke(...)`, maps the result back</td></tr><tr><td>**Add as a node**</td><td>Shared keys (for example both use `messages`)</td><td>`builder.add_node('agent', compiled_subgraph)`; no wrapper</td></tr></tbody></table>

```
# different schemas
def call_subgraph(state: Parent):
    out = subgraph.invoke({'bar': state['foo']})
    return {'foo': out['bar']}
# shared schema
builder.add_node('node_2', compiled_subgraph)
```

### 8.2 Persistence mode: the `checkpointer` argument of `compile()`

<table id="bkmrk-modecheckpointer%3Dbeh"><thead><tr><th>Mode</th><th>`checkpointer=`</th><th>Behaviour</th><th>Pick it for</th></tr></thead><tbody><tr><td>**Per-invocation** (default)</td><td>`None`</td><td>Each call starts fresh; inherits the parent's checkpointer so interrupts and durable execution work within the call</td><td>Most apps; subagents handling independent requests</td></tr><tr><td>**Per-thread**</td><td>`True`</td><td>State accumulates across calls on the same thread</td><td>A subagent that needs multi-turn memory</td></tr><tr><td>**Stateless**</td><td>`False`</td><td>No checkpointing; runs like a function; no interrupts, no durability</td><td>Cheap pure helpers; accept that a crash means rerun</td></tr></tbody></table>

- **Per-thread caveats:** no parallel calls to the same per-thread subgraph (checkpoint conflict); several *different* per-thread subgraphs called inside one node get namespaces by *call order*, so reordering mixes up their states. Give each its own wrapper `StateGraph` with a unique node name for a stable namespace. Subgraphs added as nodes already get name-based namespaces.
- The **parent must have a checkpointer** for interrupts, state inspection and per-thread memory to work in a subgraph.
- **Inspect:** `graph.get_state(config, subgraphs=True).tasks[0].state`. Per-invocation shows only the current call; per-thread shows accumulated state.
- **Stream:** `stream_events(..., version='v3')` and iterate `stream.subgraphs` (each has `graph_name`, `path`, `messages`, `values`); or classic `stream(..., subgraphs=True)`. Without that flag, `stream_mode='messages'` on a parent does not emit tokens from an inner agent.
- **Navigate up:** `Command(goto='other', graph=Command.PARENT)`.

📖 [Subgraphs](https://docs.langchain.com/oss/python/langgraph/use-subgraphs)

---

## 9. Workflow and agent patterns, one by one

Workflows follow predetermined code paths; agents choose their own path. All of these are just graphs. Pick the simplest that fits.

<table id="bkmrk-patternshapeuse-when"><thead><tr><th>Pattern</th><th>Shape</th><th>Use when</th><th>Avoid when</th></tr></thead><tbody><tr><td>**Prompt chaining**</td><td>A → gate → B → C; each call consumes the previous output</td><td>Well-defined task splittable into verifiable steps (translate, then verify)</td><td>Steps are not really sequential</td></tr><tr><td>**Parallelization**</td><td>START fans out to N nodes; all feed an aggregator</td><td>Independent subtasks for speed, or repeated runs for confidence</td><td>Branches depend on each other</td></tr><tr><td>**Routing**</td><td>Classifier node (structured output) → conditional edge → specialised node</td><td>Distinct input types needing different handling</td><td>Categories overlap heavily</td></tr><tr><td>**Orchestrator-worker**</td><td>Plan → `Send` one worker per subtask → synthesiser</td><td>Subtasks cannot be predefined (edit N unknown files)</td><td>Subtasks are fixed (use parallelization)</td></tr><tr><td>**Evaluator-optimizer**</td><td>Generate → evaluate → loop back with feedback until accepted</td><td>Clear success criteria that need iteration</td><td>No reliable evaluator; unbounded cost (add an iteration cap)</td></tr><tr><td>**Agent**</td><td>LLM node ↔ tool node loop until no tool calls</td><td>Open-ended problems where steps are unpredictable</td><td>You know the steps in advance</td></tr></tbody></table>

```
# Agent loop in the Graph API
def llm_call(state: MessagesState):
    return {'messages': [llm_with_tools.invoke([SystemMessage('You are a calculator.')] + state['messages'])]}
def tool_node(state: MessagesState):
    out = []
    for tc in state['messages'][-1].tool_calls:
        out.append(ToolMessage(content=tools_by_name[tc['name']].invoke(tc['args']), tool_call_id=tc['id']))
    return {'messages': out}
def should_continue(state) -> Literal['tool_node', '__end__']:
    return 'tool_node' if state['messages'][-1].tool_calls else END

b = StateGraph(MessagesState)
b.add_node('llm_call', llm_call); b.add_node('tool_node', tool_node)
b.add_edge(START, 'llm_call')
b.add_conditional_edges('llm_call', should_continue, ['tool_node', END])
b.add_edge('tool_node', 'llm_call')
agent = b.compile()
```

```
# Routing: structured output as the router
class Route(BaseModel):
    step: Literal['poem', 'story', 'joke']
router = llm.with_structured_output(Route)
def route_node(state): return {'decision': router.invoke(state['input']).step}
graph.add_conditional_edges('route_node', lambda s: {'story': 'write_story', 'joke': 'write_joke', 'poem': 'write_poem'}[s['decision']])
```

### 9.1 ToolNode

`ToolNode` (from `langgraph.prebuilt`) is the prebuilt node that runs tool calls: it handles **parallel tool execution, error handling and state injection**. Add it with `builder.add_node('tools', ToolNode([search, calculator]))`. Tools read graph state and run context through the injected `ToolRuntime` argument. **Tools only see the state passed to the ToolNode:** if you call a ToolNode manually from another node, pass the full state (`tool_node.invoke(state)`), not just `{'messages': ...}`, or custom state fields are invisible. A tool returning `Command` must include a matching `ToolMessage`.

**Hand-written loop versus `create_agent`:** write the loop yourself only when you need control the middleware cannot give (custom routing, extra nodes between model and tools). Otherwise use `create_agent` and embed it as a node.

📖 [Workflows and agents](https://docs.langchain.com/oss/python/langgraph/workflows-agents) · [Agentic RAG](https://docs.langchain.com/oss/python/langgraph/agentic-rag) · [SQL agent](https://docs.langchain.com/oss/python/langgraph/sql-agent)

---

## 10. The Functional API

> **Analogy: a recipe written as ordinary steps** instead of a wiring diagram. You write normal Python (`if`, `while`, `for`); the runtime still saves progress after each marked step.

```
from langgraph.func import entrypoint, task

@task
def generate(topic: str): return llm.invoke('Write a joke about ' + topic).content
@task
def improve(joke: str):  return llm.invoke('Make this funnier: ' + joke).content

@entrypoint(checkpointer=InMemorySaver())
def workflow(topic: str):
    joke = generate(topic).result()          # tasks return futures; .result() waits
    return improve(joke).result()
```

- **Tasks are the checkpoint unit.** Their results are saved, so a resumed run skips completed tasks. Launch several before calling `.result()` to run in parallel.
- **Determinism and idempotency** rules are stricter here, because the whole entrypoint re-runs on resume: keep task and `interrupt` order stable, and keep side effects inside tasks.
- `@task` and `@entrypoint` accept `retry_policy=` and `timeout=`. An entrypoint compiles to a single node.
- **Choose it when** the flow reads naturally as code, you do not need a drawn graph, and you want minimal state plumbing. **Choose the Graph API when** you want explicit visualisation, fine-grained state reducers, parallel branches via edges, or per-node policies. The two can be mixed (a task inside a graph node, §2.2).

📖 [Functional API](https://docs.langchain.com/oss/python/langgraph/functional-api) · [Use the functional API](https://docs.langchain.com/oss/python/langgraph/use-functional-api)

---

## 11. Streaming

<table id="bkmrk-modeyieldsvaluesfull"><thead><tr><th>Mode</th><th>Yields</th></tr></thead><tbody><tr><td>`values`</td><td>Full state after each step (includes private channels)</td></tr><tr><td>`updates`</td><td>Only what each node changed</td></tr><tr><td>`messages`</td><td>`(token, metadata)` from any LLM call inside nodes</td></tr><tr><td>`custom`</td><td>Whatever you write with `runtime.stream_writer` / `get_stream_writer()`</td></tr></tbody></table>

- **Classic:** `graph.stream(inp, config, stream_mode=['updates', 'messages'], version='v2')` yields dicts with `type`, `ns`, `data`. Interrupts appear under `__interrupt__` in `updates`.
- **Event streaming (recommended for new apps):** `stream = graph.stream_events(inp, config, version='v3')` gives typed projections you consume independently: `stream.messages`, `stream.values`, `stream.subgraphs`, `stream.interrupted`, `stream.interrupts`, `stream.output`. Pass `output_keys=[...]` to restrict streamed state.
- Inside a node, `get_stream_writer()` only works within a LangGraph execution context.
- Disable token streaming for one model with `streaming=False` (or `disable_streaming=True`); useful on the platform for models whose output must not reach the client.

📖 [Streaming](https://docs.langchain.com/oss/python/langgraph/streaming) · [Event streaming](https://docs.langchain.com/oss/python/langgraph/event-streaming) · [Frontend](https://docs.langchain.com/oss/python/langgraph/frontend/overview)

---

## 12. Memory in LangGraph: when to use which

> **Analogy:** the checkpointer is the *save file of this playthrough*; the store is the *profile that follows the player across playthroughs*; state is the *HUD you see right now*.

<table id="bkmrk-needusewhere-it-live"><thead><tr><th>Need</th><th>Use</th><th>Where it lives</th></tr></thead><tbody><tr><td>Continue the same conversation later</td><td>Checkpointer + `thread_id`</td><td>Thread (all graph state)</td></tr><tr><td>Pause for a human, survive a crash</td><td>Checkpointer</td><td>Thread</td></tr><tr><td>Keep history from overflowing the context window</td><td>Trim, `RemoveMessage` or a summarise node</td><td>State (rewritten) or model input (transient)</td></tr><tr><td>Remember facts about a user across threads</td><td>Store (`runtime.store`)</td><td>Cross-thread, namespaced</td></tr><tr><td>Find memories by meaning</td><td>Store with `index` config and `search(query=...)`</td><td>Cross-thread</td></tr><tr><td>Per-request configuration or clients</td><td>Runtime `context`</td><td>Not persisted</td></tr><tr><td>Data that should not be checkpointed (live connections)</td><td>Keep out of state; use context or closures</td><td>n/a</td></tr></tbody></table>

**Pattern for long-term memory in a graph:** a `call_model` node reads relevant memories from the store and adds them to the prompt; a separate `update_memory` node (or background job) writes new ones. Keep the write path in its own node so retries and interrupts do not duplicate writes, and use user-scoped namespaces. Memory design from the agent side (tools, middleware, summarisation) is on the LangChain page.

---

## 13. LangChain agents are LangGraph graphs

- `create_agent` returns a compiled graph. Embed it as a node or subgraph, and its middleware (HITL, summarisation, PII, retries) travels with it.
- As a subgraph, tokens from the inner model only appear in the parent's `messages` stream with `subgraphs=True` (or `stream.subgraphs`). Give each agent a `name` to see `lc_agent_name` in metadata.
- Multi-agent handoffs use `Command(goto=..., graph=Command.PARENT)` under the hood; the per-thread subagent rules in §8.2 apply.
- Deep Agents adds planning, a filesystem and subagents on top of this same runtime.

---

## 14. Production, version gates and the trap list

- **Deploy:** LangSmith Cloud or the Agent Server provisions checkpointer and store automatically and exposes the graph through an API; locally you pass them yourself. Use LangSmith Studio to inspect state, set breakpoints and replay runs. Trace with LangSmith.
- **Test:** unit-test nodes as plain functions, and graphs with in-memory persistence; use `get_state` to assert on checkpoints.
- **Operate:** graceful drain on SIGTERM, per-node timeouts and retries, a default error handler that marks the external job failed, recursion limits with proactive fallback.

<table id="bkmrk-%E2%9A%91-featureminimumdefa"><thead><tr><th>⚑ Feature</th><th>Minimum</th></tr></thead><tbody><tr><td>Default recursion limit 1000</td><td>langgraph 1.0.6</td></tr><tr><td>v2 stream/invoke format (`GraphOutput`)</td><td>1.1</td></tr><tr><td>`runtime.execution_info`, `server_info`</td><td>1.1.5</td></tr><tr><td>Per-node timeouts, node error handlers, `set_node_defaults`, `RunControl` drain, `runtime.heartbeat`</td><td>1.2</td></tr><tr><td>`interrupt(..., response_schema=...)`</td><td>1.2.12</td></tr><tr><td>Event streaming (`stream_events(version='v3')`)</td><td>langchain 1.3-era releases; check your install</td></tr></tbody></table>

**Trap checklist:**

1. No checkpointer yet using `interrupt`, threads or time travel; or no `thread_id`.
2. Non-idempotent code before an `interrupt()` or in a retried node.
3. Bare `try/except` around `interrupt()`; `while True` prompt loops; reordered interrupts.
4. Parallel writes to one key with no reducer; expecting an empty list to clear a reducer field.
5. Mixing static edges with `Command`/conditional routing from the same node.
6. A `Command` return annotation missing; a tool returning `Command` without a `ToolMessage`.
7. Sending `Command(update=...)` as input to continue a chat (the graph looks stuck).
8. `recursion_limit` placed inside `configurable`; no proactive fallback.
9. Timeout on a sync node; relying on defaults being inherited by subgraphs.
10. Renaming a state key or a node while threads are in flight.
11. Treating private state channels as hidden in `values` streams.
12. `store.search` silently truncated at the default limit of 10.

---

## 15. Exercises

1. Build the agent loop from §9 by hand with `MessagesState`, compile it with `InMemorySaver`, and show that two `invoke` calls on one `thread_id` share history.
2. Add a node that calls `interrupt()` before a tool runs. Resume with approval, then repeat and reject. Then deliberately put a counter-incrementing database call *before* the interrupt and observe the double run.
3. Make a state key `errors: Annotated[list[str], add]`. Write a node that tries to clear it by returning `[]`, see that nothing changes, then fix it with `Overwrite`.
4. Add a flaky async node with `RetryPolicy(max_attempts=3)`, a `TimeoutPolicy(run_timeout=2)`, and an `error_handler` that routes to a `finalize` node. Make it fail on every attempt and read the final state.
5. Fan out with `Send` over a list of 5 items into a worker node writing to `results: Annotated[list, add]`, then synthesise. Add a `RemainingSteps` guard.

<details id="bkmrk-solution-notesthe-ch"><summary>Solution notes</summary>

1. The checkpointer reloads the whole state for that `thread_id`; the `messages` reducer appends the new turn.
2. On resume the node restarts from its first line, so the pre-interrupt side effect runs twice. Fix by upserting, or by moving the call after the interrupt or into its own node.
3. A merging reducer merges the empty list into the old value; `{'errors': Overwrite([])}` bypasses the reducer.
4. The order is attempts → retry policy → handler. The handler sees `NodeError` with the node name and last exception; status is whatever your handler wrote. Remember timeouts need an async node.
5. Each `Send` gets its own worker input; the reducer on `results` merges parallel writes. Without the reducer the parallel writes conflict.

</details>---

**Sources:** docs.langchain.com (Python): [LangGraph overview](https://docs.langchain.com/oss/python/langgraph/overview), [Graph API](https://docs.langchain.com/oss/python/langgraph/graph-api), [persistence](https://docs.langchain.com/oss/python/langgraph/persistence), [stores](https://docs.langchain.com/oss/python/langgraph/stores), [interrupts](https://docs.langchain.com/oss/python/langgraph/interrupts), [fault tolerance](https://docs.langchain.com/oss/python/langgraph/fault-tolerance), [subgraphs](https://docs.langchain.com/oss/python/langgraph/use-subgraphs), [workflows and agents](https://docs.langchain.com/oss/python/langgraph/workflows-agents), [runtimes, frameworks, harnesses](https://docs.langchain.com/oss/python/concepts/products), [LangGraph v1 release notes](https://docs.langchain.com/oss/python/releases/langgraph-v1). Time travel and the Functional API were not fully re-read (see scope notes above).