zoujindouGithub - StateGraph：构建有状态图的主类
- Nodes（节点）：执行任务并更新状态的函数
- Edges（边）：定义执行顺序（静态或条件式）
- START/END：标记入口和退出点的特殊节点
- 带 Reducer 的 State：控制状态更新的合并方式
图在执行前必须先进行 compile()。
设计 LangGraph 应用程序
构建新图时，请遵循以下 5 个步骤：
- 梳理离散步骤 —— 绘制工作流的流程图。每个步骤都将成为一个节点。
- 明确每个步骤的职责 —— 划分节点类型：LLM 步骤、数据步骤、动作步骤或用户输入步骤。针对每个步骤，确定其静态上下文（Prompt）、动态上下文（来自 State）、重试策略以及预期结果。
- 设计状态 (State) —— 状态是所有节点共享的内存。存储原始数据，并在节点内按需格式化 Prompt。
- 构建节点 —— 将每个步骤实现为一个接收状态并返回部分更新的函数。
- 串联成图 —— 用边连接节点，添加条件路由，并在需要时配合 checkpointer 进行编译。
| 适合使用 LangGraph 的场景 | 适合使用替代方案的场景 |
|---|---|
| 需要对 Agent 编排进行细粒度控制 | 快速原型开发 → LangChain agents |
| 构建包含分支/循环的复杂工作流 | 简单的无状态工作流 → 直接使用 LangChain |
| 需要人机协作 (Human-in-the-loop)、持久化存储 | 开箱即用型功能特性 → Deep Agents |
状态管理 (State Management)
| 需求 | 解决方案 | 示例 |
|---|---|---|
| 覆盖原有值 | 无 Reducer（默认） | 简单字段，如计数器 |
| 追加到列表 | Reducer (operator.add / concat) |
消息历史、日志 |
| 自定义逻辑 | 自定义 Reducer 函数 | 复杂合并逻辑 |
from typing_extensions import TypedDict, Annotated
import operator
class State(TypedDict):
name: str # 默认：更新时直接覆盖
messages: Annotated[list, operator.add] # 追加到列表中
total: Annotated[int, operator.add] # 整数求和累加
import { StateSchema, ReducedValue, MessagesValue } from "@langchain/langgraph";
import { z } from "zod";
const State = new StateSchema({
name: z.string(), // 默认：直接覆盖
messages: MessagesValue, // 用于消息的内置类型
items: new ReducedValue(
z.array(z.string()).default(() => []),
{ reducer: (current, update) => current.concat(update) }
),
});
# 错误写法：列表将被覆盖 (OVERWRITTEN)
class State(TypedDict):
messages: list # 没有 Reducer！
# 节点 1 返回：{"messages": ["A"]}
# 节点 2 返回：{"messages": ["B"]}
# 最终结果：{"messages": ["B"]} # "A" 丢失了！
# 正确写法：使用带有 operator.add 的 Annotated
from typing import Annotated
import operator
class State(TypedDict):
messages: Annotated[list, operator.add]
# 最终结果：{"messages": ["A", "B"]}
// 错误写法：数组将被覆盖
const State = new StateSchema({
items: z.array(z.string()), // 没有 Reducer！
});
// 节点 1：{ items: ["A"] }，节点 2：{ items: ["B"] }
// 最终结果：{ items: ["B"] } // A 丢失了！
// 正确写法：使用 ReducedValue
const State = new StateSchema({
items: new ReducedValue(
z.array(z.string()).default(() => []),
{ reducer: (current, update) => current.concat(update) }
),
});
// 最终结果：{ items: ["A", "B"] }
# 错误写法：返回整个状态对象
def my_node(state: State) -> State:
state["field"] = "updated"
return state # 切勿直接修改并返回！
# 正确写法：仅返回包含更新项的字典
def my_node(state: State) -> dict:
return {"field": "updated"}
// 错误写法：返回整个状态
const myNode = async (state: typeof State.State) => {
state.field = "updated";
return state; // 切勿这样做！
};
// 正确写法：返回部分更新
const myNode = async (state: typeof State.State) => {
return { field: "updated" };
};
节点 (Nodes)
节点函数支持接收以下参数：
| 签名 | 适用场景 |
|---|---|
def node(state: State) |
仅需访问状态的简单节点 |
def node(state: State, config: RunnableConfig) |
需要 thread_id、标签 (tags) 或可配置参数 |
def node(state: State, runtime: Runtime[Context]) |
需要运行时上下文、store 或 stream_writer |
from langchain_core.runnables import RunnableConfig
from langgraph.runtime import Runtime
def plain_node(state: State):
return {"results": "done"}
def node_with_config(state: State, config: RunnableConfig):
thread_id = config["configurable"]["thread_id"]
return {"results": f"Thread: {thread_id}"}
def node_with_runtime(state: State, runtime: Runtime[Context]):
user_id = runtime.context.user_id
return {"results": f"User: {user_id}"}
| 签名 | 适用场景 |
|---|---|
(state) => {...} |
仅需访问状态的简单节点 |
(state, config) => {...} |
需要 thread_id、标签 (tags) 或可配置参数 |
import { GraphNode, StateSchema } from "@langchain/langgraph";
const plainNode: GraphNode<typeof State> = (state) => {
return { results: "done" };
};
const nodeWithConfig: GraphNode<typeof State> = (state, config) => {
const threadId = config?.configurable?.thread_id;
return { results: `Thread: ${threadId}` };
};
边 (Edges)
| 需求 | 边类型 | 适用场景 |
|---|---|---|
| 始终流向同一节点 | add_edge() |
固定、确定性的流程 |
| 基于状态进行路由 | add_conditional_edges() |
动态分支 |
| 同时更新状态并路由 | Command |
在单个节点内合并逻辑 |
| 扇出到多个节点 | Send |
带有动态输入的并行处理 |
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
class State(TypedDict):
input: str
output: str
def process_input(state: State) -> dict:
return {"output": f"Processed: {state['input']}"}
def finalize(state: State) -> dict:
return {"output": state["output"].upper()}
graph = (
StateGraph(State)
.add_node("process", process_input)
.add_node("finalize", finalize)
.add_edge(START, "process")
.add_edge("process", "finalize")
.add_edge("finalize", END)
.compile()
)
result = graph.invoke({"input": "hello"})
print(result["output"]) # "PROCESSED: HELLO"
import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import { z } from "zod";
const State = new StateSchema({
input: z.string(),
output: z.string().default(""),
});
const processInput = async (state: typeof State.State) => {
return { output: `Processed: ${state.input}` };
};
const finalize = async (state: typeof State.State) => {
return { output: state.output.toUpperCase() };
};
const graph = new StateGraph(State)
.addNode("process", processInput)
.addNode("finalize", finalize)
.addEdge(START, "process")
.addEdge("process", "finalize")
.addEdge("finalize", END)
.compile();
const result = await graph.invoke({ input: "hello" });
console.log(result.output); // "PROCESSED: HELLO"
from typing import Literal
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
query: str
route: str
result: str
def classify(state: State) -> dict:
if "weather" in state["query"].lower():
return {"route": "weather"}
return {"route": "general"}
def route_query(state: State) -> Literal["weather", "general"]:
return state["route"]
graph = (
StateGraph(State)
.add_node("classify", classify)
.add_node("weather", lambda s: {"result": "Sunny, 72F"})
.add_node("general", lambda s: {"result": "General response"})
.add_edge(START, "classify")
.add_conditional_edges("classify", route_query, ["weather", "general"])
.add_edge("weather", END)
.add_edge("general", END)
.compile()
)
import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import { z } from "zod";
const State = new StateSchema({
query: z.string(),
route: z.string().default(""),
result: z.string().default(""),
});
const classify = async (state: typeof State.State) => {
if (state.query.toLowerCase().includes("weather")) {
return { route: "weather" };
}
return { route: "general" };
};
const routeQuery = (state: typeof State.State) => state.route;
const graph = new StateGraph(State)
.addNode("classify", classify)
.addNode("weather", async () => ({ result: "Sunny, 72F" }))
.addNode("general", async () => ({ result: "General response" }))
.addEdge(START, "classify")
.addConditionalEdges("classify", routeQuery, ["weather", "general"])
.addEdge("weather", END)
.addEdge("general", END)
.compile();
Command
Command 将状态更新和路由逻辑整合在单个返回值中。字段说明：
update：要应用的状态更新（类似于从节点返回字典）goto：下一步要跳转的目标节点名称（可为单个或多个）resume：在interrupt()之后恢复执行时传入的值 —— 参见人机协作 (HITL) 技能
from langgraph.types import Command
from typing import Literal
class State(TypedDict):
count: int
result: str
def node_a(state: State) -> Command[Literal["node_b", "node_c"]]:
"""在单次返回中更新状态并决定下一个跳转的节点。"""
new_count = state["count"] + 1
if new_count > 5:
return Command(update={"count": new_count}, goto="node_c")
return Command(update={"count": new_count}, goto="node_b")
graph = (
StateGraph(State)
.add_node("node_a", node_a)
.add_node("node_b", lambda s: {"result": "B"})
.add_node("node_c", lambda s: {"result": "C"})
.add_edge(START, "node_a")
.add_edge("node_b", END)
.add_edge("node_c", END)
.compile()
)
import { StateGraph, StateSchema, START, END, Command } from "@langchain/langgraph";
import { z } from "zod";
const State = new StateSchema({
count: z.number().default(0),
result: z.string().default(""),
});
const nodeA = async (state: typeof State.State) => {
const newCount = state.count + 1;
if (newCount > 5) {
return new Command({ update: { count: newCount }, goto: "node_c" });
}
return new Command({ update: { count: newCount }, goto: "node_b" });
};
const graph = new StateGraph(State)
.addNode("node_a", nodeA, { ends: ["node_b", "node_c"] })
.addNode("node_b", async () => ({ result: "B" }))
.addNode("node_c", async () => ({ result: "C" }))
.addEdge(START, "node_a")
.addEdge("node_b", END)
.addEdge("node_c", END)
.compile();
Python：使用 Command[Literal["node_a", "node_b"]] 作为返回类型注解，以声明合法的 goto 跳转目标。
TypeScript：将 { ends: ["node_a", "node_b"] } 作为第三个参数传给 addNode，以声明合法的 goto 跳转目标。
警告：Command 仅添加动态边 —— 使用 add_edge / addEdge 定义的静态边仍会照常执行。如果 node_a 返回了 Command(goto="node_c")，同时图中存在 graph.add_edge("node_a", "node_b")，那么 node_b 和 node_c 都会被执行。
Send API
使用 Send 实现扇出 (Fan-out)：从条件边返回 [Send("worker", {...})] 以派生并行 Worker。这要求在结果字段上配置 Reducer。
from langgraph.types import Send
from typing import Annotated
import operator
class OrchestratorState(TypedDict):
tasks: list[str]
results: Annotated[list, operator.add]
summary: str
def orchestrator(state: OrchestratorState):
"""将任务分发给 Worker。"""
return [Send("worker", {"task": task}) for task in state["tasks"]]
def worker(state: dict) -> dict:
return {"results": [f"Completed: {state['task']}"]}
def synthesize(state: OrchestratorState) -> dict:
return {"summary": f"Processed {len(state['results'])} tasks"}
graph = (
StateGraph(OrchestratorState)
.add_node("worker", worker)
.add_node("synthesize", synthesize)
.add_conditional_edges(START, orchestrator, ["worker"])
.add_edge("worker", "synthesize")
.add_edge("synthesize", END)
.compile()
)
result = graph.invoke({"tasks": ["Task A", "Task B", "Task C"]})
import { Send, StateGraph, StateSchema, ReducedValue, START, END } from "@langchain/langgraph";
import { z } from "zod";
const State = new StateSchema({
tasks: z.array(z.string()),
results: new ReducedValue(
z.array(z.string()).default(() => []),
{ reducer: (curr, upd) => curr.concat(upd) }
),
summary: z.string().default(""),
});
const orchestrator = (state: typeof State.State) => {
return state.tasks.map((task) => new Send("worker", { task }));
};
const worker = async (state: { task: string }) => {
return { results: [`Completed: ${state.task}`] };
};
const synthesize = async (state: typeof State.State) => {
return { summary: `Processed ${state.results.length} tasks` };
};
const graph = new StateGraph(State)
.addNode("worker", worker)
.addNode("synthesize", synthesize)
.addConditionalEdges(START, orchestrator, ["worker"])
.addEdge("worker", "synthesize")
.addEdge("synthesize", END)
.compile();
# 错误写法：没有 Reducer - 最后一个 Worker 会覆盖其他结果
class State(TypedDict):
results: list
# 正确写法
class State(TypedDict):
results: Annotated[list, operator.add] # 自动累加聚合
// 错误写法：没有 Reducer
const State = new StateSchema({ results: z.array(z.string()) });
// 正确写法
const State = new StateSchema({
results: new ReducedValue(z.array(z.string()).default(() => []), { reducer: (curr, upd) => curr.concat(upd) }),
});
运行图：Invoke 与 Stream
调用 graph.invoke(input, config) 会运行图直到执行完毕，并返回最终状态。
result = graph.invoke({"input": "hello"})
# 携带 config 配置（用于持久化、标签等）
result = graph.invoke({"input": "hello"}, {"configurable": {"thread_id": "1"}})
const result = await graph.invoke({ input: "hello" });
// 携带 config 配置
const result = await graph.invoke({ input: "hello" }, { configurable: { thread_id: "1" } });
| 模式 | 流式传输内容 | 适用场景 |
|---|---|---|
values |
每一步执行后的完整状态 | 监控完整状态 |
updates |
状态增量变更 (deltas) | 追踪增量更新 |
messages |
LLM Token 与元数据 | 聊天 UI 界面 |
custom |
用户自定义数据 | 进度指示器 |
for chunk in graph.stream(
{"messages": [HumanMessage("Hello")]},
stream_mode="messages"
):
token, metadata = chunk
if hasattr(token, "content"):
print(token.content, end="", flush=True)
for await (const chunk of graph.stream(
{ messages: [new HumanMessage("Hello")] },
{ streamMode: "messages" }
)) {
const [token, metadata] = chunk;
if (token.content) {
process.stdout.write(token.content);
}
}
from langgraph.config import get_stream_writer
def my_node(state):
writer = get_stream_writer()
writer("Processing step 1...")
# 执行任务
writer("Complete!")
return {"result": "done"}
for chunk in graph.stream({"data": "test"}, stream_mode="custom"):
print(chunk)
import { getWriter } from "@langchain/langgraph";
const myNode = async (state: typeof State.State) => {
const writer = getWriter();
writer("Processing step 1...");
// 执行任务
writer("Complete!");
return { result: "done" };
};
for await (const chunk of graph.stream({ data: "test" }, { streamMode: "custom" })) {
console.log(chunk);
}
错误处理 (Error Handling)
将错误类型与正确的处理策略相匹配：
| 错误类型 | 解决方 | 策略 | 示例 |
|---|---|---|---|
| 瞬态错误（网络抖动、速率限制） | 系统 | RetryPolicy(max_attempts=3) |
add_node(..., retry_policy=...) |
| LLM 可恢复错误（工具调用失败） | LLM | ToolNode(tools, handle_tool_errors=True) |
错误作为 ToolMessage 返回 |
| 用户可修复错误（缺失信息） | 人工 | interrupt({"message": ...}) |
收集缺失数据（参见 HITL 技能） |
| 非预期错误 | 开发者 | 向外冒泡 | raise |
from langgraph.types import RetryPolicy
workflow.add_node(
"search_documentation",
search_documentation,
retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0)
)
workflow.addNode(
"searchDocumentation",
searchDocumentation,
{
retryPolicy: { maxAttempts: 3, initialInterval: 1.0 },
},
);
from langgraph.prebuilt import ToolNode
tool_node = ToolNode(tools, handle_tool_errors=True)
workflow.add_node("tools", tool_node)
import { ToolNode } from "@langchain/langgraph/prebuilt";
const toolNode = new ToolNode(tools, { handleToolErrors: true });
workflow.addNode("tools", toolNode);
常见问题修复
# 错误写法
builder.invoke({"input": "test"}) # 抛出 AttributeError！
# 正确写法
graph = builder.compile()
graph.invoke({"input": "test"})
// 错误写法
await builder.invoke({ input: "test" });
// 正确写法
const graph = builder.compile();
await graph.invoke({ input: "test" });
# 错误写法：无限循环
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", "node_a")
# 正确写法
def should_continue(state):
return END if state["count"] > 10 else "node_b"
builder.add_conditional_edges("node_a", should_continue)
// 错误写法：无限循环
builder.addEdge("node_a", "node_b").addEdge("node_b", "node_a");
// 正确写法
builder.addConditionalEdges("node_a", (state) => state.count > 10 ? END : "node_b");
# 路由函数返回的节点名称必须已存在于图中
builder.add_node("my_node", func) # 在边中引用之前必须先添加节点
builder.add_conditional_edges("node_a", router, ["my_node"])
# Python 中 Command 的返回类型需要用 Literal 标注合法的路由目标
def node_a(state) -> Command[Literal["node_b", "node_c"]]:
return Command(goto="node_b")
# START 仅作为入口使用 —— 不能将边路由回 START
builder.add_edge("node_a", START) # 错误！
builder.add_edge("node_a", "entry") # 请使用具名的入口节点代替
# Reducer 要求类型匹配
return {"items": ["item"]} # 对于列表 Reducer 请传入 List，而不是普通字符串
// 始终 await graph.invoke() —— 它返回一个 Promise
const result = await graph.invoke({ input: "test" });
// TS 中的 Command 节点需要传入 { ends } 来声明路由目标
builder.addNode("router", routerFn, { ends: ["node_b", "node_c"] });
- 直接修改状态 (Mutate state) —— 务必从节点返回部分更新字典
- 将边路由回
START——START仅作为入口，请改用具名节点 - 遗漏列表字段的 Reducer —— 没有 Reducer 时，最后一次写入会覆盖先前的值
- 在未理解其均会执行的情况下混用静态边与
Command.goto