LangGraph Agents skill for Codex

Done well means: state is a small typed schema with explicit reducers, every node is a pure-ish function returning a partial update, side effects are idempotent because nodes can re-run, durable checkpoints are on in production, risky tools pause with interrupt, and the graph is covered by tests that run without a live model.

Before you write a graph

• Read AGENTS.md, check pip show langgraph langchain, and look for an existing graph (StateGraph( usages) and checkpointer config.
• Write down the control flow as a short list or Mermaid diagram first: nodes, edges, where the model decides, where a human decides, where it ends. If it is a straight line, use a plain chain ($langchain-apps), not a graph.
• For a standard tool-calling agent, langchain.agents.create_agent (built on LangGraph) is enough. Hand-build a StateGraph when you need custom routing, extra state, approval steps, parallel branches or subgraphs.

State design

from typing import Annotated, Literal, TypedDict
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage

class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]   # append + update by message id
    findings: Annotated[list[dict], add]                 # parallel branches concatenate
    plan: list[str]                                       # last write wins
    approved: bool | None

Rules:

• Without a reducer, a key is overwritten; two parallel nodes writing the same key without a reducer raise an error. Give every key written by fan-out a reducer.
• Keep large blobs (files, full documents) out of state; store a reference. State is serialised into every checkpoint.
• Nodes return only the keys they change: return {"plan": steps}.

A tool loop with an approval gate

from langgraph.prebuilt import ToolNode
from langgraph.types import interrupt, Command

RISKY = {"send_refund", "delete_account"}
tools = [lookup_order, send_refund]
model_with_tools = model.bind_tools(tools)

def call_model(state: State) -> dict:
    return {"messages": [model_with_tools.invoke(state["messages"])]}

def route(state: State) -> Literal["approve", "tools", "__end__"]:
    last = state["messages"][-1]
    if not getattr(last, "tool_calls", None):
        return END
    return "approve" if any(tc["name"] in RISKY for tc in last.tool_calls) else "tools"

def approve(state: State) -> Command[Literal["tools", "agent"]]:
    calls = state["messages"][-1].tool_calls
    decision = interrupt({"question": "Approve these actions?", "tool_calls": calls})
    if decision.get("approved"):
        return Command(goto="tools")
    rejection = [{"role": "tool", "tool_call_id": tc["id"], "content": "Rejected by reviewer."} for tc in calls]
    return Command(goto="agent", update={"messages": rejection})

builder = StateGraph(State)
builder.add_node("agent", call_model)
builder.add_node("approve", approve)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", route)
builder.add_edge("tools", "agent")

Resume after the human answers, on the same thread:

config = {"configurable": {"thread_id": ticket_id}, "recursion_limit": 25}
result = graph.invoke({"messages": [("user", text)]}, config)
if "__interrupt__" in result:
    payload = result["__interrupt__"][0].value      # show to reviewer
    result = graph.invoke(Command(resume={"approved": True}), config)

interrupt re-runs the whole node on resume. Anything before the interrupt() call executes twice, so keep approval nodes free of side effects and put the action in the next node.

Checkpointers

Environment	Checkpointer
Unit tests, notebooks	from langgraph.checkpoint.memory import InMemorySaver
Single-process local tool	SQLite (langgraph-checkpoint-sqlite)
Production, multiple workers	Postgres (langgraph-checkpoint-postgres)

from langgraph.checkpoint.postgres import PostgresSaver

with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()                 # creates tables; run once, e.g. in a migration step
    graph = builder.compile(checkpointer=checkpointer)
    graph.invoke(inputs, {"configurable": {"thread_id": thread_id}})

Use AsyncPostgresSaver (langgraph.checkpoint.postgres.aio) in async servers. Without a checkpointer, interrupt, multi-turn memory and resume-after-crash do not work.

• thread_id = one conversation or job. Derive it from your domain ID, never a random value you then lose.
• graph.get_state(config) shows current values and the next node; get_state_history lists past checkpoints; update_state edits state (use for operator fixes, logged).
• Long-term, cross-thread memory belongs in a store, not the checkpointer: $agent-memory.

Fan-out with Send

from langgraph.types import Send

def fan_out(state: State):
    return [Send("research", {"topic": t}) for t in state["plan"]]

builder.add_conditional_edges("planner", fan_out, ["research"])
builder.add_edge("research", "synthesize")

Each research call receives its own input and returns {"findings": [...]}; the add reducer merges them before synthesize runs. Cap fan-out width in the planner (e.g. 8) and see $multi-agent-orchestration for budgets and partial-failure handling.

Streaming

stream_mode	Emits	Use
"updates"	Per-node partial updates	Progress UI, logs
"values"	Full state after each step	Debugging
"messages"	LLM tokens with metadata	Chat token streaming
"custom"	Whatever nodes write via get_stream_writer()	Tool progress events

Pass a list (stream_mode=["updates", "messages"]) to get tuples of (mode, chunk).

Limits and failure handling

• Set recursion_limit in config deliberately; hitting it raises GraphRecursionError. Catch it, persist the state, and return a “stopped after N steps” result instead of a 500.
• Node-level retries: builder.add_node("fetch", fetch, retry_policy=RetryPolicy(max_attempts=3)) (from langgraph.types import RetryPolicy) for transient tool or API failures.
• Tool errors: ToolNode returns an error ToolMessage to the model by default so it can recover; make messages actionable (“order not found; IDs look like ORD-1234”).
• Side effects must be idempotent (idempotency keys on payments, upserts not inserts) because a crash after the effect but before the checkpoint replays the node.

Testing

from langgraph.checkpoint.memory import InMemorySaver

def test_refund_requires_approval(fake_model_returning_refund_call):
    graph = build_graph(model=fake_model_returning_refund_call).compile(checkpointer=InMemorySaver())
    cfg = {"configurable": {"thread_id": "t1"}}
    out = graph.invoke({"messages": [("user", "refund ORD-1")]}, cfg)
    assert "__interrupt__" in out
    assert graph.get_state(cfg).next == ("approve",)
    out = graph.invoke(Command(resume={"approved": False}), cfg)
    assert refunds_api.calls == []

Test routing functions directly as plain functions; they are where most bugs live. Use a fake chat model that emits scripted tool calls (GenericFakeChatModel with AIMessage(tool_calls=...)).

Gotchas

• Returning the whole state from a node with an add reducer duplicates lists. Return deltas.
• add_messages matches by message ID; editing a message means returning it with the same ID.
• Interrupt payloads and resume values must be JSON-serialisable for Postgres checkpointers.
• Changing the State schema breaks resuming old threads. Add optional keys with defaults; migrate or drain in-flight threads before renaming keys.
• Thread history grows forever: trim or summarise messages inside the graph and prune old checkpoints with a retention job (after confirming retention rules with the user).
• graph.get_graph().draw_mermaid() prints the compiled topology; paste it into the PR so reviewers can see the edges.

Verification

• Tests cover each routing branch, the approve and reject paths, and the recursion limit.
• Run one real conversation, kill the process mid-run, restart, and resume from the same thread_id; state must continue from the last checkpoint.
• Trace a run ($llm-observability) and confirm the node sequence matches the diagram.

Output summary template

**Graph:** agent -> (approve?) -> tools -> agent; planner fans out to research x<=8 -> synthesize
**State keys:** messages (add_messages), findings (add), plan, approved
**Persistence:** AsyncPostgresSaver, thread_id = support ticket ID
**Human-in-the-loop:** interrupt before send_refund, delete_account
**Tests:** 11 passing (routing, approve/reject, resume after restart, recursion cap)