# Langgraph Patterns

> Use when writing or reviewing LangGraph / langchain 1.x agent code — create_agent, AgentMiddleware hooks, checkpointers, interrupt/Command human-in-the-loop, astream_events, ChatAnthropic — pinned to langgraph 1.2.11, langchain 1.4.2, langchain-core 1.6.3, langchain-anthropic 1.7.2. Also use when an agent "runs but the middleware didn't short-circuit", "jump_to is ignored", or an import from langgraph.prebuilt warns as deprecated.

- Skill: `andreasbloomquist/langgraph-patterns` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add andreasbloomquist/langgraph-patterns`
- Raw SKILL.md: https://api.skillmd.com/api/skills/andreasbloomquist/langgraph-patterns/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: andreasbloomquist (https://skillmd.com/u/andreasbloomquist)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/andreasbloomquist/langgraph-patterns

---


# LangGraph patterns (langgraph 1.2.11 / langchain 1.4.2)

## Overview

Every symbol below was verified against the installed source, not docs or memory. The model's default recall of this API is ~95% right and the 5% fails **silently** — see Traps. When in doubt, `grep` the installed package; signatures live in [reference.md](reference.md).

Read `reference.md` when you need exact signatures (`create_agent` kwargs, `ModelRequest` fields, `ToolRuntime`, `StreamMode`).

## Quick reference

| Task | Import | Call |
|---|---|---|
| Build agent | `from langchain.agents import create_agent` | `create_agent(model, tools, system_prompt=, middleware=[...], checkpointer=, context_schema=)` → `CompiledStateGraph` |
| Middleware base | `from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config` | subclass; override `before_agent` / `before_model` / `after_model` / `after_agent` / `wrap_model_call` / `wrap_tool_call` (+ `a`-prefixed async twins) |
| **Short-circuit to END** | same | `@hook_config(can_jump_to=["end"])` on the hook, then `return {"jump_to": "end", "messages": [AIMessage(...)]}` |
| Modify the model request | `from langchain.agents.middleware import ModelRequest, ModelResponse` | `awrap_model_call(self, request, handler)` → `return await handler(request.override(system_message=..., tools=..., messages=...))` |
| Intercept a tool call | `from langchain.agents.middleware import ToolCallRequest` | `awrap_tool_call(self, request, handler)`; `request.tool_call["name"]`, `request.tool_call["args"]`; skip `handler` to short-circuit, return a `ToolMessage` |
| Checkpointer (dev/test) | `from langgraph.checkpoint.memory import InMemorySaver` | `create_agent(..., checkpointer=InMemorySaver())` |
| Checkpointer (prod) | `langgraph-checkpoint-sqlite` / `langgraph-checkpoint-postgres` — **separate pip packages**, not in `langgraph-checkpoint` 4.2.0 | `AsyncSqliteSaver.from_conn_string(...)` / `AsyncPostgresSaver.from_conn_string(...)` |
| Thread | — | `config={"configurable": {"thread_id": "..."}}` on every invoke/stream; omitting it with a checkpointer raises `ValueError: Checkpointer requires ... 'thread_id'` |
| Runtime context (per-request deps) | `create_agent(..., context_schema=MyCtx)` | `graph.ainvoke(input, config=..., context=MyCtx(...))`; read via `runtime.context` in hooks, `runtime.context` on `ToolRuntime` in tools |
| Human-in-the-loop (inside a tool) | `from langgraph.types import interrupt, Command` | `answer = interrupt({"question": ...})` pauses; `ainvoke` caller sees `result["__interrupt__"][0].value`; a **streaming** caller gets no result dict — use `(await graph.aget_state(config)).tasks[0].interrupts[0].value`; resume with `graph.ainvoke(Command(resume=answer), config=same_thread)` |
| Human-in-the-loop (declarative) | `from langchain.agents.middleware import HumanInTheLoopMiddleware` | `HumanInTheLoopMiddleware(interrupt_on={"book_flight": {"allowed_decisions": ["approve","reject"]}})`; resume with `Command(resume={"decisions": [{"type": "approve"}]})` |
| Route to a node | `from langgraph.types import Command` | `return Command(goto="node_name", update={...})` from a **graph node** — not supported from `wrap_model_call` (raises `NotImplementedError`; use `jump_to`) |
| Stream events | — | `async for ev in graph.astream_events(input, config=..., version="v2")` |
| Emit a custom event from a tool | `from langchain_core.callbacks import adispatch_custom_event` | `await adispatch_custom_event("panel_state", payload)` → surfaces as `on_custom_event` (this is what AG-UI forwards — see `ag-ui-protocol`) |
| Tool-side access to state/store/context | `from langchain.tools import ToolRuntime` | `def my_tool(x: str, runtime: ToolRuntime) -> str:` — auto-injected, hidden from the schema; `runtime.state`, `runtime.store`, `runtime.context`, `runtime.tool_call_id`, `runtime.stream_writer` |
| Tool error handling | `from langgraph.prebuilt import ToolNode` | `ToolNode(tools, handle_tool_errors=True \| str \| callable \| (ExcType, ...))` — `create_agent` builds its own; use `ToolErrorMiddleware` / `ToolRetryMiddleware` instead |

## Middleware recipe — fast-path that skips the model

```python
from typing import Any
from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config
from langchain_core.messages import AIMessage, HumanMessage
from langgraph.runtime import Runtime

class FilterFastPathMiddleware(AgentMiddleware):
    @hook_config(can_jump_to=["end"])          # REQUIRED — without this, jump_to is never read
    async def abefore_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
        last = state["messages"][-1]
        if isinstance(last, HumanMessage) and last.text.startswith("filter:"):
            return {"jump_to": "end", "messages": [AIMessage(content="...")]}
        return None
```

- `before_agent` runs once per run; `before_model` runs before *every* model call in the loop. Use `before_agent` for "is this whole turn a fast path"; `before_model` for per-iteration checks (round budgets).
- `can_jump_to` values: `"end"`, `"model"`, `"tools"`. The decorator sets `__can_jump_to__`, and only then does the factory build a conditional edge; otherwise it adds a plain edge and `jump_to` is silently ignored.
- Sync and async hooks both work; the factory checks both for `can_jump_to`. Prefer `a`-prefixed in an async app.
- Custom state: set `state_schema = MyState` on the class where `class MyState(AgentState): my_field: NotRequired[int]`.

## Streaming — `astream_events(version="v2")`

Filter on `ev["event"]`:

| Event | `ev["data"]` | Use |
|---|---|---|
| `on_chat_model_stream` | `["chunk"]` is `AIMessageChunk` — use `.text` (property) not `.content` (a list for Anthropic) | token deltas |
| `on_chat_model_end` | `["output"]` full `AIMessage`, `.tool_calls` | decide next step |
| `on_tool_start` / `on_tool_end` / `on_tool_error` | `["input"]` / `["output"]` / `["error"]`; `ev["name"]` is the tool | tool lifecycle |
| `on_custom_event` | `ev["name"]`, `ev["data"]` — from `adispatch_custom_event` | structured payloads to the UI |

`astream(stream_mode=...)` accepts `"values" | "updates" | "messages" | "custom" | "checkpoints" | "tasks" | "debug"` or a list. `stream_mode="custom"` receives `runtime.stream_writer(...)` writes — a **different channel** from `adispatch_custom_event`; the AG-UI adapter reads only the latter.

## Traps

| Model reaches for | Reality in these versions |
|---|---|
| `{"jump_to": "end"}` without `@hook_config` | Silently ignored — model still runs. Every jump needs `can_jump_to`. |
| `from langgraph.prebuilt import create_react_agent` | Deprecated since 1.0 → `from langchain.agents import create_agent`. `AgentState` moved too. |
| `from langgraph.checkpoint.memory import MemorySaver` | No such name. It's `InMemorySaver`. |
| `AsyncSqliteSaver` / `AsyncPostgresSaver` "from langgraph" | Separate packages: `langgraph-checkpoint-sqlite`, `langgraph-checkpoint-postgres`. Not installed by `langgraph`. |
| `Command(goto=...)` from `wrap_model_call` | Raises `NotImplementedError`. Use `jump_to` from `before_model`/`after_model`. |
| `get_stream_writer()` to send data to the browser | Goes to `stream_mode="custom"`, which `ag_ui_langgraph` does not read. Use `adispatch_custom_event`. |
| `chunk.content` as a string | Anthropic content is a list of blocks. Use `chunk.text`. |
| `InjectedState` / `InjectedStore` annotations | Still work but `ToolRuntime` is the 1.x way — one param, everything on it. |
| `interrupt()` outside a checkpointed thread | Raises — interrupts need a checkpointer and a `thread_id`. |
| `adispatch_custom_event(...)` without a parent run | `RuntimeError: Unable to dispatch an adhoc event without a parent run id` — only call from inside a tool/node the graph is running. |
| Python 3.10 async config propagation | Pass `config=` explicitly to `adispatch_custom_event`. Not an issue on 3.11+ (contextvars propagate). |

## ChatAnthropic (langchain-anthropic 1.7.2) — verified kwargs

- `ChatAnthropic(model=, max_tokens=, temperature=, thinking={"type": "enabled", "budget_tokens": N}, betas=[...], stream_usage=True, context_management=...)`
- `.bind_tools(tools, tool_choice=None | "auto" | "any" | "<name>" | {"type": "tool", "name": ...}, parallel_tool_calls=, strict=)` — no `"none"` option exists
- Prompt caching: put `{"type": "text", "text": ..., "cache_control": {"type": "ephemeral"}}` blocks in the `SystemMessage` content list, or tag the last tool dict with `"cache_control"`. There's a stock `AnthropicPromptCachingMiddleware` in `langchain_anthropic.middleware` — check what it tags before assuming it matches your cache boundary.