API 速查表

本篇把 LangGraph 0.2+ 常用 API 浓缩成速查表，按模块分类，每项给签名 + 一句话说明。写代码时忘了参数名、想确认返回值结构时随手翻一翻。配合常见问题 FAQ 使用效率最高。

一、StateGraph 构建与编译

图构建

API	签名	说明
`StateGraph`	`StateGraph(state_schema, input=None, output=None)`	创建状态图，state_schema 是 TypedDict
`add_node`	`add_node(key: str, action: Callable)`	添加节点，key 是节点名，action 是节点函数
`add_edge`	`add_edge(start_key, end_key)`	添加普通边，从 start 节点到 end 节点
`add_conditional_edges`	`add_conditional_edges(start_key, condition_fn, path_map=None)`	添加条件边，condition_fn 返回字符串决定下一节点
`add_sequence`	`add_sequence(nodes: list[Callable])`	0.2+ 新增，按顺序串联多个节点
`set_entry_point`	`set_entry_point(key)`	设置入口节点（等价于 `add_edge(START, key)`）
`compile`	`compile(checkpointer=None, interrupt_before=None, interrupt_after=None, name=None)`	编译图，返回可执行的 `CompiledStateGraph`

python

from langgraph.graph import StateGraph, START, END

builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_edge(START, "a")
builder.add_conditional_edges("a", lambda s: "b" if s["ok"] else END)
builder.add_edge("b", END)
graph = builder.compile(checkpointer=saver, interrupt_before=["b"])

起止常量

常量	说明
`START`	图的虚拟入口节点
`END`	图的虚拟出口节点

二、State 定义与 Reducer

API	说明
`TypedDict` (typing)	定义 state schema 的基础类型
`Annotated[T, reducer]`	给字段绑定 reducer 函数
`add_messages`	消息列表 reducer，按 id 追加/覆盖/删除
`operator.add`	列表拼接 reducer（list 用 `+`）

python

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
import operator

class State(TypedDict):
    messages: Annotated[list, add_messages]   # 消息按 id 去重合并
    history: Annotated[list, operator.add]    # 列表拼接
    count: int                                 # 默认覆盖

内置 State

State	说明
`MessagesState`	预置 state，含 `messages: Annotated[list, add_messages]`

python

from langgraph.graph import MessagesState
builder = StateGraph(MessagesState)  # 等价于自带 messages 字段

三、执行：invoke / stream

API	签名	说明
`invoke`	`invoke(input, config=None, *, stream_mode="values")`	同步执行，返回最终 state
`ainvoke`	`await ainvoke(input, config=None, *, stream_mode="values")`	异步执行
`stream`	`stream(input, config=None, *, stream_mode="values")`	同步流式，返回迭代器
`astream`	`astream(input, config=None, *, stream_mode="values")`	异步流式

stream_mode 取值

模式	吐出内容
`values`	每步完整 state
`updates`	每步 state 增量 `{node: {field: val}}`
`messages`	LLM token + 元数据，元组 `(chunk, metadata)`
`custom`	节点内 `get_stream_writer()` 写的自定义事件
`debug`	调试信息

多模式聚合：stream_mode=["messages", "updates"]，迭代得到 (mode, chunk) 元组。

config 常用字段

python

config = {
    "configurable": {"thread_id": "t1"},  # 线程标识，决定会话隔离
    "recursion_limit": 25,                # 最大步数
    "metadata": {"user_id": "u1"},        # 透传到 LangSmith 的标签
}

四、状态查询与控制

API	签名	说明
`get_state`	`get_state(config)`	取当前线程的 state 快照
`aget_state`	`await aget_state(config)`	异步取快照
`get_state_history`	`get_state_history(config)`	取历史所有快照（时间旅行）
`update_state`	`update_state(config, values, as_node=None)`	手动更新 state（HITL）
`aupdate_state`	`await aupdate_state(...)`	异步更新

python

snapshot = await graph.aget_state(config)
print(snapshot.values)        # 当前 state
print(snapshot.next)          # 下一步要执行的节点
await graph.aupdate_state(config, {"messages": [ai_msg]}, as_node="human")

五、Checkpointer 持久化

Saver	导入	说明
`MemorySaver`	`from langgraph.checkpoint.memory import MemorySaver`	内存，开发用
`SqliteSaver`	`from langgraph.checkpoint.sqlite import SqliteSaver`	SQLite，单机
`AsyncSqliteSaver`	`from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver`	SQLite 异步
`PostgresSaver`	`from langgraph.checkpoint.postgres import PostgresSaver`	Postgres 同步
`AsyncPostgresSaver`	`from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver`	生产异步

python

from langgraph.checkpoint.memory import MemorySaver
saver = MemorySaver()
graph = builder.compile(checkpointer=saver)

六、预置智能体

create_react_agent

参数	类型	说明
`model`	BaseChatModel	LLM，需已 bind_tools
`tools`	list[tool]	工具列表
`prompt`	str / SystemMessage / Callable	系统提示词
`state_schema`	type	自定义 state schema
`state_modifier`	Callable	state 预处理函数
`checkpointer`	BaseCheckpointSaver	持久化
`store`	BaseStore	长期记忆存储
`pre_model_hook`	Callable	调模型前钩子
`post_model_hook`	Callable	调模型后钩子

python

from langgraph.prebuilt import create_react_agent
agent = create_react_agent(llm, tools, prompt="你是助手", checkpointer=saver)

ToolNode 与 tools_condition

API	说明
`ToolNode(tools)`	工具执行节点，自动按 AIMessage.tool_calls 调对应工具
`tools_condition(state)`	条件函数：有 tool_calls 返回 `"tools"`，否则返回 `END`

python

from langgraph.prebuilt import ToolNode, tools_condition
builder.add_node("tools", ToolNode(tools))
builder.add_conditional_edges("agent", tools_condition, "tools")

七、人机交互（HITL）

API	签名	说明
`interrupt`	`interrupt(value)`	节点内调用，暂停图并返回 value 给调用方
`Command`	`Command(update=None, goto=None, resume=None, graph=None)`	控制流对象，可更新 state 并跳转
`Send`	`Send(node, state)`	动态分发，向某节点发送指定 state（map-reduce）

python

from langgraph.types import interrupt, Command

def human_review(state):
    decision = interrupt({"need_review": True, "draft": state["draft"]})
    return {"decision": decision}  # resume 时传入

# 恢复执行
await graph.ainvoke(None, config, command=Command(resume="approved"))

python

# Send：动态扇出
def fanout(state):
    return [Send("worker", {"item": i}) for i in state["items"]]
builder.add_conditional_edges("split", fanout, ["worker"])

八、Store（长期记忆）

API	签名	说明
`InMemoryStore`	`InMemoryStore()`	内存 store
`put`	`await put(namespace, key, value, index=None)`	写入
`get`	`await get(namespace, key)`	读取
`search`	`await search(namespace, query=None, limit=10)`	搜索（支持向量）
`delete`	`await delete(namespace, key)`	删除

python

from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
await store.put(("mem", "u1"), "fact1", {"text": "喜欢猫"})
item = await store.get(("mem", "u1"), "fact1")
results = await store.search(("mem", "u1"), query="宠物")

九、langchain_core 消息类型

类型	说明
`HumanMessage`	用户消息
`AIMessage`	AI 回复（可含 tool_calls）
`SystemMessage`	系统提示
`ToolMessage`	工具返回结果，需带 `tool_call_id`
`AIMessageChunk`	流式 token 片段
`RemoveMessage`	删除消息（按 id）
`trim_messages`	裁剪历史消息，控制 token

python

from langchain_core.messages import (
    HumanMessage, AIMessage, SystemMessage, ToolMessage, RemoveMessage
)

十、模型绑定工具

API	说明
`ChatOpenAI(model, temperature, timeout, max_retries)`	OpenAI 兼容模型客户端
`llm.bind_tools(tools)`	绑定工具，返回能发 tool_calls 的模型
`llm.with_structured_output(schema)`	强制输出结构化 JSON

python

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, timeout=30, max_retries=2)
llm = llm.bind_tools(tools)

十一、工具定义

API	说明
`@tool`	装饰器，把函数变工具，docstring 是工具描述
`@tool(args_schema=MyModel)`	指定参数 schema（Pydantic BaseModel）
`InjectedToolArg`	标记参数由运行时注入而非模型生成

python

from langchain_core.tools import tool
from pydantic import BaseModel, Field

class SearchInput(BaseModel):
    query: str = Field(description="搜索关键词")

@tool(args_schema=SearchInput)
def search(query: str) -> str:
    """根据关键词搜索文档"""
    return do_search(query)

十二、配置与全局

API	说明
`get_config()`	节点内取当前 config
`get_stream_writer()`	节点内取流式写入器（stream_mode="custom"）

python

from langgraph.config import get_stream_writer
writer = get_stream_writer()
writer({"progress": 0.5})  # 前端 stream_mode="custom" 能收到

版本提示

本表基于 langgraph 0.2+。API 仍在演进，个别签名可能微调，以官方文档为准：https://langchain-ai.github.io/langgraph/

如需对比 Java 版 API，见与 LangGraph4j 对比。

API 速查表 ​

一、StateGraph 构建与编译 ​

图构建 ​

起止常量 ​

二、State 定义与 Reducer ​

内置 State ​

三、执行：invoke / stream ​

stream_mode 取值 ​

config 常用字段 ​

四、状态查询与控制 ​

五、Checkpointer 持久化 ​

六、预置智能体 ​

create_react_agent ​

ToolNode 与 tools_condition ​

七、人机交互（HITL） ​

八、Store（长期记忆） ​

九、langchain_core 消息类型 ​

十、模型绑定工具 ​

十一、工具定义 ​

十二、配置与全局 ​

API 速查表

一、StateGraph 构建与编译

图构建

起止常量

二、State 定义与 Reducer

内置 State

三、执行：invoke / stream

stream_mode 取值

config 常用字段

四、状态查询与控制

五、Checkpointer 持久化

六、预置智能体

create_react_agent

ToolNode 与 tools_condition

七、人机交互（HITL）

八、Store（长期记忆）

九、langchain_core 消息类型

十、模型绑定工具

十一、工具定义

十二、配置与全局