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

📖 Overview · Thinking in LangGraph · Graph API overview · Choosing between Graph and Functional · LangGraph runtime

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 · Use the graph API (how-tos)

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

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

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 · Command · Map-reduce and Send how-to

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 · Quickstart · Run a 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 · Checkpointers · Stores · Time travel · 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

• 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 · HITL concepts

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)

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

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 · 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

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

• 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

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.

# 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 · Agentic RAG · 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 · Use the functional API

11. Streaming

• 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 · Event streaming · Frontend

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.

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.

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.

Sources: docs.langchain.com (Python): LangGraph overview, Graph API, persistence, stores, interrupts, fault tolerance, subgraphs, workflows and agents, runtimes, frameworks, harnesses, LangGraph v1 release notes. Time travel and the Functional API were not fully re-read (see scope notes above).