--- name: langchain-langgraph-basics description: "Build a correct LangGraph 1.0 StateGraph \u2014 typed TypedDict state\ \ with reducers,\nnodes, edges, compile, and recursion budgets \u2014 without hitting\ \ the silent-termination\nand state-replacement traps. Use when writing your first\ \ LangGraph StateGraph,\ndiagnosing why a graph halted without reaching END, or\ \ picking recursion_limit.\nTrigger with \"langgraph statgraph\", \"langgraph basics\"\ , \"GraphRecursionError\",\n\"langgraph conditional edges\".\n" allowed-tools: Read, Write, Edit, Bash(python:*) version: 2.0.0 license: MIT author: Jeremy Longshore tags: - saas - langchain - langgraph - python - langchain-1.0 - statgraph compatibility: Designed for Claude Code, also compatible with Codex --- # LangChain LangGraph Basics (Python) ## Overview A conditional edge whose router returns a string that is not in `path_map` halts the graph without reaching `END`. No exception. No log line. The invocation just returns whatever state existed at the halt point — pain-catalog entry P56, and the single most common reason a newly wired `StateGraph` "almost works." The sibling pain: `Command(update={"messages": [msg]})` wipes the prior message history because `messages` was declared as a plain `list[AnyMessage]` instead of `Annotated[list[AnyMessage], add_messages]` — the reducer is what turns `update` into "append" instead of "replace" (P18). Two more gotchas this skill defuses: - P55 — `GraphRecursionError: Recursion limit of 25 reached` fires on graphs that never loop, because `recursion_limit` counts **supersteps** (one step per synchronous batch of node executions), not loop iterations. A planner + executor + validator + summarizer can hit 25 without any cycle. - P20 — Upgrading `langgraph` silently reads old `PostgresSaver` checkpoints as empty state. Checkpoint schemas evolve; `PostgresSaver.setup()` must be rerun after every version bump before production traffic. This skill walks through a minimal `StateGraph` end to end: a `TypedDict` state with reducers on every list field, node functions that return partial-state dicts, edges and defensive conditional edges with `END` as a fallback in `path_map`, compilation with a checkpointer, `recursion_limit` sizing, and invocation with an explicit `thread_id`. Pin: `langgraph 1.0.x`, `langchain-core 1.0.x`. Pain-catalog anchors: P16, P18, P20, P55, P56. ## Prerequisites - Python 3.10+ - `pip install langgraph>=1.0,<2.0 langchain-core>=1.0,<2.0` - A chat model (see `langchain-model-inference`), or a pure-logic graph with no LLM - For persistence beyond a single process: `pip install langgraph-checkpoint-postgres` and a Postgres 14+ instance ## Instructions ### Step 1 — Define state as a `TypedDict` with reducers on list fields Every list-shaped field in state needs a reducer. Without one, `Command(update=...)` and node returns *replace* the field. The message-history reducer lives in `langgraph.graph.message`: ```python from typing import Annotated, TypedDict from langchain_core.messages import AnyMessage from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # Reducer "add_messages" appends + dedupes by message id (P18) messages: Annotated[list[AnyMessage], add_messages] # Plain list field also needs a reducer — use operator.add to concat scratchpad: Annotated[list[str], operator.add] # Scalars don't need a reducer; update replaces them step_count: int done: bool ``` If you forget the reducer on `messages`, a resume with `Command(update={"messages": [new_msg]})` will overwrite the entire prior history. Validate reducers are in place with `graph.get_graph().draw_mermaid()` — annotated fields render with their reducer name. See [State Reducers](references/state-reducers.md) for the built-in list (`add_messages`, `operator.add`, `max`, `min`) and how to write a custom merger for non-trivial merge logic. ### Step 2 — Write nodes as functions that return partial-state dicts A node takes the full state and returns only the keys it wants to update. The reducer handles merge: ```python def plan(state: AgentState) -> dict: # Returning a dict means "update these fields" return { "messages": [("assistant", "Plan: step 1, step 2, step 3")], "scratchpad": ["planned_at_step_1"], "step_count": state["step_count"] + 1, } def execute(state: AgentState) -> dict: return { "messages": [("assistant", f"Executed {state['step_count']} steps")], "done": state["step_count"] >= 3, } ``` Nodes must be deterministic on their inputs — LangGraph re-runs them during time-travel replay, and a side-effecting node (DB write without idempotency key) will double-fire. Push side effects to the checkpointer boundary or tool calls. ### Step 3 — Wire edges and conditional edges defensively ```python from typing import Literal from langgraph.graph import StateGraph, START, END # Router MUST return a value in the path_map keyset (P56) def should_continue(state: AgentState) -> Literal["execute", "end"]: if state["done"] or state["step_count"] >= 10: return "end" return "execute" builder = StateGraph(AgentState) builder.add_node("plan", plan) builder.add_node("execute", execute) builder.add_edge(START, "plan") # path_map ALWAYS includes END as a fallback — if the router returns anything # else, the graph reaches END instead of halting silently (P56) builder.add_conditional_edges( "plan", should_continue, path_map={"execute": "execute", "end": END}, ) builder.add_edge("execute", "plan") # loop back to plan ``` The `Literal` return annotation on `should_continue` is a static guard — mypy catches typos before runtime. `path_map={"execute": "execute", "end": END}` is the spelled-out form; the compact form `path_map=["execute", END]` also works when router return values match node names directly. See [Conditional Edges](references/conditional-edges.md) for all four `add_conditional_edges` signatures, the `path` vs `path_map` distinction, and a pytest pattern that asserts every router return value hits a known route. ### Step 4 — Compile with a checkpointer ```python from langgraph.checkpoint.memory import MemorySaver # MemorySaver is in-process — use PostgresSaver in production (P20) checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer) ``` For production, swap to `langgraph.checkpoint.postgres.PostgresSaver`. After every `langgraph` version bump, run `PostgresSaver.setup()` in staging before prod traffic — the schema evolves and old rows are silently read as empty state. ### Step 5 — Pick `recursion_limit` for the graph's superstep count `recursion_limit` defaults to **25**. It is not a loop counter; it counts **total supersteps**, and a superstep is one synchronous round of node executions (parallel branches in the same step count as one). Typical shapes: | Graph shape | Supersteps per run | Suggested `recursion_limit` | |---|---|---| | Simple ReAct agent (plan → tool → observe → done) | 6-12 | 15 | | Planner + executor + validator | 12-25 | 30 | | Deep agent with sub-plans, reflection, branch merge | 30-60 | 75 | | Fan-out with N parallel branches that re-join | N + merge steps | 2 × max depth | ```python config = { "configurable": {"thread_id": "user-42"}, # required for checkpointing (P16) "recursion_limit": 30, } result = graph.invoke({"messages": [], "scratchpad": [], "step_count": 0, "done": False}, config) ``` If you hit `GraphRecursionError` on a graph that clearly isn't looping (P55), add `print(state["step_count"])` at the entry of each node to see the actual superstep count, then either raise the limit or restructure with a subgraph so each subgraph gets its own budget. See [Recursion Limits](references/recursion-limits.md) for the full derivation and a diagnostic script that traces superstep count at runtime. ### Step 6 — Invoke with `thread_id` in `config["configurable"]` Every invocation against a checkpointer-backed graph needs a `thread_id` in `config["configurable"]`. Without it, each call gets a fresh state with no warning (P16). Enforce it at your application boundary: ```python def run_agent(user_id: str, user_message: str) -> dict: config = { "configurable": {"thread_id": user_id}, "recursion_limit": 30, } assert config["configurable"].get("thread_id"), "thread_id required" return graph.invoke( {"messages": [("user", user_message)], "scratchpad": [], "step_count": 0, "done": False}, config, ) ``` See [First Graph Walkthrough](references/first-graph-walkthrough.md) for a line-by-line annotation of a minimal 3-node graph that demonstrates typed state, a reducer, and a conditional edge to `END`. ## Decision Tree ``` Is your field a list you want to append? -> Annotate with add_messages (for messages) or operator.add (for plain lists) Is your router adding a new output string? -> Add that string to path_map BEFORE deploying; include END as a fallback Hitting GraphRecursionError on a non-looping graph? -> supersteps != iterations; raise recursion_limit to 50 or split into subgraphs Upgraded langgraph minor version? -> Re-run PostgresSaver.setup() in staging before routing prod traffic Multi-turn agent forgets between calls? -> thread_id missing from config["configurable"] — enforce at boundary ``` ## Output - `TypedDict` state with reducer annotations on every list field - Node functions returning partial-state dicts, free of hidden side effects - Conditional edges with `Literal`-typed routers and `END` in every `path_map` - Compiled graph with a checkpointer matched to the deployment shape - `recursion_limit` sized from the superstep count table, not the default 25 - Invocations with explicit `thread_id` validated at the app boundary ## Error Handling | Error | Cause | Fix | |-------|-------|-----| | `GraphRecursionError: Recursion limit of 25 reached` | Supersteps counter, not loops (P55) | Raise `recursion_limit` or split into subgraphs; add per-node logging to count actual steps | | Graph halts without reaching `END`, no error | Router returned a value not in `path_map` (P56) | Type router as `Literal[...]`; include `END` as a default key in `path_map` | | `Command(update={"messages": [msg]})` wipes history | Missing reducer on list field (P18) | Annotate as `Annotated[list[AnyMessage], add_messages]` | | Multi-turn memory resets between calls, no warning | Missing `thread_id` in config (P16) | Assert `config["configurable"]["thread_id"]` at app boundary | | Old checkpoints read as empty state after upgrade | Schema change; `PostgresSaver` doesn't auto-migrate (P20) | Run `PostgresSaver.setup()` in staging after every `langgraph` bump | | `TypeError: Object of type datetime is not JSON serializable` at interrupt | Non-primitive in state (P17) | Keep state primitives-only; serialize complex types to ISO strings | | Node runs twice during replay | Time-travel re-executes deterministic nodes | Push side effects to tools or the checkpointer write boundary | ## Examples ### Building a minimal linear graph (no loop) Three nodes (`plan` → `execute` → `summarize`), no conditionals, one reducer on `messages`. Runs in 4 supersteps (START counts as one). Safe at default `recursion_limit=25`, but set it to 10 explicitly so readers see the budget. See [First Graph Walkthrough](references/first-graph-walkthrough.md) for the line-by-line annotation and the `graph.get_graph().draw_mermaid()` output. ### Adding a conditional loop with a bounded retry budget A validator node that either completes (`"end"` → `END`) or retries (`"retry"` → back to executor), bounded by `step_count >= 3`. The router is `Literal["retry", "end"]` and `path_map` maps both. Caps at 7 supersteps, so `recursion_limit=15` is plenty of headroom. See [Conditional Edges](references/conditional-edges.md) for the full example and the pytest that iterates every `Literal` branch. ### Diagnosing a mystery `GraphRecursionError` A planner that fans out to 4 parallel executors, then a validator, then a summarizer. Looks linear in the mermaid diagram, hits `GraphRecursionError: Recursion limit of 25 reached` on 10% of runs. Cause: each parallel executor counts as its own superstep branch when the merge node is conditional. Fix: raise to 50 or wrap the fan-out in a subgraph. See [Recursion Limits](references/recursion-limits.md) for the diagnostic script and the subgraph refactor. ## Resources - [LangGraph: Low-level concepts (StateGraph, reducers, nodes, edges)](https://langchain-ai.github.io/langgraph/concepts/low_level/) - [LangGraph: Persistence and checkpointing](https://langchain-ai.github.io/langgraph/concepts/persistence/) - [LangGraph: `add_messages` reducer](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.message.add_messages) - [LangGraph: `add_conditional_edges` API](https://langchain-ai.github.io/langgraph/reference/graphs/#langgraph.graph.state.StateGraph.add_conditional_edges) - [LangGraph: Recursion limit and supersteps](https://langchain-ai.github.io/langgraph/how-tos/recursion-limit/) - [LangChain 1.0 release notes](https://blog.langchain.com/langchain-langgraph-1dot0/) - Pack pain catalog: `docs/pain-catalog.md` (entries P16, P17, P18, P20, P55, P56)