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.
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
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.
1---2name: langgraph-patterns3description: 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.4---56# LangGraph patterns (langgraph 1.2.11 / langchain 1.4.2)78## Overview910Every 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).1112Read `reference.md` when you need exact signatures (`create_agent` kwargs, `ModelRequest` fields, `ToolRuntime`, `StreamMode`).1314## Quick reference1516| Task | Import | Call |17|---|---|---|18| Build agent | `from langchain.agents import create_agent` | `create_agent(model, tools, system_prompt=, middleware=[...], checkpointer=, context_schema=)` → `CompiledStateGraph` |19| 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) |20| **Short-circuit to END** | same | `@hook_config(can_jump_to=["end"])` on the hook, then `return {"jump_to": "end", "messages": [AIMessage(...)]}` |21| 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=...))` |22| 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` |23| Checkpointer (dev/test) | `from langgraph.checkpoint.memory import InMemorySaver` | `create_agent(..., checkpointer=InMemorySaver())` |24| 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(...)` |25| Thread | — | `config={"configurable": {"thread_id": "..."}}` on every invoke/stream; omitting it with a checkpointer raises `ValueError: Checkpointer requires ... 'thread_id'` |26| 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 |27| 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)` |28| 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"}]})` |29| 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`) |30| Stream events | — | `async for ev in graph.astream_events(input, config=..., version="v2")` |31| 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`) |32| 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` |33| 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 |3435## Middleware recipe — fast-path that skips the model3637```python38from typing import Any39from langchain.agents.middleware import AgentMiddleware, AgentState, hook_config40from langchain_core.messages import AIMessage, HumanMessage41from langgraph.runtime import Runtime4243class FilterFastPathMiddleware(AgentMiddleware):44 @hook_config(can_jump_to=["end"]) # REQUIRED — without this, jump_to is never read45 async def abefore_agent(self, state: AgentState, runtime: Runtime) -> dict[str, Any] | None:46 last = state["messages"][-1]47 if isinstance(last, HumanMessage) and last.text.startswith("filter:"):48 return {"jump_to": "end", "messages": [AIMessage(content="...")]}49 return None50```5152- `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).53- `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.54- Sync and async hooks both work; the factory checks both for `can_jump_to`. Prefer `a`-prefixed in an async app.55- Custom state: set `state_schema = MyState` on the class where `class MyState(AgentState): my_field: NotRequired[int]`.5657## Streaming — `astream_events(version="v2")`5859Filter on `ev["event"]`:6061| Event | `ev["data"]` | Use |62|---|---|---|63| `on_chat_model_stream` | `["chunk"]` is `AIMessageChunk` — use `.text` (property) not `.content` (a list for Anthropic) | token deltas |64| `on_chat_model_end` | `["output"]` full `AIMessage`, `.tool_calls` | decide next step |65| `on_tool_start` / `on_tool_end` / `on_tool_error` | `["input"]` / `["output"]` / `["error"]`; `ev["name"]` is the tool | tool lifecycle |66| `on_custom_event` | `ev["name"]`, `ev["data"]` — from `adispatch_custom_event` | structured payloads to the UI |6768`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.6970## Traps7172| Model reaches for | Reality in these versions |73|---|---|74| `{"jump_to": "end"}` without `@hook_config` | Silently ignored — model still runs. Every jump needs `can_jump_to`. |75| `from langgraph.prebuilt import create_react_agent` | Deprecated since 1.0 → `from langchain.agents import create_agent`. `AgentState` moved too. |76| `from langgraph.checkpoint.memory import MemorySaver` | No such name. It's `InMemorySaver`. |77| `AsyncSqliteSaver` / `AsyncPostgresSaver` "from langgraph" | Separate packages: `langgraph-checkpoint-sqlite`, `langgraph-checkpoint-postgres`. Not installed by `langgraph`. |78| `Command(goto=...)` from `wrap_model_call` | Raises `NotImplementedError`. Use `jump_to` from `before_model`/`after_model`. |79| `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`. |80| `chunk.content` as a string | Anthropic content is a list of blocks. Use `chunk.text`. |81| `InjectedState` / `InjectedStore` annotations | Still work but `ToolRuntime` is the 1.x way — one param, everything on it. |82| `interrupt()` outside a checkpointed thread | Raises — interrupts need a checkpointer and a `thread_id`. |83| `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. |84| Python 3.10 async config propagation | Pass `config=` explicitly to `adispatch_custom_event`. Not an issue on 3.11+ (contextvars propagate). |8586## ChatAnthropic (langchain-anthropic 1.7.2) — verified kwargs8788- `ChatAnthropic(model=, max_tokens=, temperature=, thinking={"type": "enabled", "budget_tokens": N}, betas=[...], stream_usage=True, context_management=...)`89- `.bind_tools(tools, tool_choice=None | "auto" | "any" | "<name>" | {"type": "tool", "name": ...}, parallel_tool_calls=, strict=)` — no `"none"` option exists90- 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.
andreasbloomquist