langchain-langgraph-basics

What this skill does

# 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