langgraph-fundamentals

LangGraph 将 Agent 工作流建模为**有向图**：

• StateGraph：用于构建有状态图的主类
• 节点：执行工作并更新状态的函数
• 边：定义执行顺序（静态或条件式）
• START/END：标记入口点和出口点的特殊节点
• 带 Reducer 的状态：控制状态更新的合并方式

执行前必须先对图调用 compile()。

设计 LangGraph 应用

构建新图时，请遵循以下 5 个步骤：

1. 规划离散步骤 — 绘制工作流流程图。每个步骤都对应一个节点。
2. 明确每个步骤的作用 — 对节点进行分类：大模型步骤、数据步骤、操作步骤或用户输入步骤。对于每个步骤，确定静态上下文（提示词）、动态上下文（来自状态）、重试策略和预期结果。
3. 设计状态 — 状态是所有节点共享的内存。存储原始数据，并在节点内按需格式化提示词。
4. 构建节点 — 将每个步骤实现为一个接收状态并返回部分更新的函数。
5. 连接各部分 — 使用边连接节点，添加条件路由，并在需要时通过 checkpointer 编译。

适用 LangGraph 的场景	适用替代方案的场景
需要对 Agent 编排进行精细控制	快速原型开发 → LangChain agents
构建带有分支/循环的复杂工作流	简单的无状态工作流 → LangChain direct
需要人在回路中参与、持久化	需要开箱即用的功能 → Deep Agents

---

状态管理

需求	解决方案	示例
覆盖值	不使用 reducer（默认）	计数器等简单字段
追加到列表	使用 reducer（operator.add / concat）	消息历史记录、日志
自定义逻辑	自定义 reducer 函数	复杂合并

使用 reducer 定义状态架构，以累积列表并对整数求和。 ```python from typing_extensions import TypedDict, Annotated import operator

class State(TypedDict): name: str # Default: overwrites on update messages: Annotated[list, operator.add] # Appends to list total: Annotated[int, operator.add] # Sums integers

</python>
<typescript>
使用 StateSchema 和 ReducedValue 累积数组。
```typescript
import { StateSchema, ReducedValue, MessagesValue } from "@langchain/langgraph";
import { z } from "zod";

const State = new StateSchema({
  name: z.string(),  // Default: overwrites
  messages: MessagesValue,  // Built-in for messages
  items: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (current, update) => current.concat(update) }
  ),
});

如果没有 reducer，返回列表会覆盖之前的值。 ```python # WRONG: List will be OVERWRITTEN class State(TypedDict): messages: list # No reducer!

Node 1 returns: {"messages": ["A"]}

Node 2 returns: {"messages": ["B"]}

Final: {"messages": ["B"]} # "A" is LOST!

CORRECT: Use Annotated with operator.add

from typing import Annotated import operator

class State(TypedDict): messages: Annotated[list, operator.add]

Final: {"messages": ["A", "B"]}

</python>
<typescript>
如果没有 ReducedValue，数组会被覆盖，而不是追加。
```typescript
// WRONG: Array will be overwritten
const State = new StateSchema({
  items: z.array(z.string()),  // No reducer!
});
// Node 1: { items: ["A"] }, Node 2: { items: ["B"] }
// Final: { items: ["B"] }  // A is lost!

// CORRECT: Use ReducedValue
const State = new StateSchema({
  items: new ReducedValue(
    z.array(z.string()).default(() => []),
    { reducer: (current, update) => current.concat(update) }
  ),
});
// Final: { items: ["A", "B"] }

节点必须返回部分更新，不能修改并返回完整状态。 ```python # WRONG: Returning entire state object def my_node(state: State) -> State: state["field"] = "updated" return state # Don't mutate and return!

CORRECT: Return dict with only the updates

def my_node(state: State) -> dict: return {"field": "updated"}

</python>
<typescript>
仅返回部分更新，不要返回完整状态对象。
```typescript
// WRONG: Returning entire state
const myNode = async (state: typeof State.State) => {
  state.field = "updated";
  return state;  // Don't do this!
};

// CORRECT: Return partial updates
const myNode = async (state: typeof State.State) => {
  return { field: "updated" };
};

---

节点

节点函数接收以下参数：

函数签名	适用场景
def node(state: State)	仅需状态的简单节点
def node(state: State, config: RunnableConfig)	需要 thread_id、标签或可配置值
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、标签或可配置值

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}` };
};

---

边

需求	边类型	使用场景
始终前往同一节点	add_edge()	固定的确定性流程
根据状态路由	add_conditional_edges()	动态分支
更新状态并路由	Command	在单个节点中合并逻辑
扇出到多个节点	Send	使用动态输入进行并行处理

包含两个节点和线性边的简单图。 ```python 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"

</python>
<typescript>
使用 addEdge 串联节点，并在调用前进行编译。
```typescript
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"

使用条件边，根据状态路由到不同节点。 ```python 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() )

</python>
<typescript>
addConditionalEdges 根据函数的返回值进行路由。
```typescript
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() 后恢复执行的值——参见 human-in-the-loop 技能

Command 让你可以在一次返回中更新状态并选择下一个节点。 ```python 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"]]: """Update state AND decide next node in one return.""" 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() )

</python>
<typescript>
返回包含 update 和 goto 的 Command，以合并状态变更和路由。
```typescript
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 扇出：从条件边返回 [Send("worker", {...})]，以启动并行工作节点。结果字段必须配置 reducer。

使用 Send API 将任务扇出到并行工作节点，并汇总结果。 ```python 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): """Fan out tasks to workers.""" 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"]})

</python>
<typescript>
使用 Send API 将任务扇出到并行工作节点，并汇总结果。
```typescript
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 累积并行工作节点的结果（否则最后一个工作节点会覆盖之前的结果）。 ```python # WRONG: No reducer - last worker overwrites class State(TypedDict): results: list

CORRECT

class State(TypedDict): results: Annotated[list, operator.add] # Accumulates

</python>
<typescript>
使用 ReducedValue 累积并行工作节点的结果。
```typescript
// WRONG: No reducer
const State = new StateSchema({ results: z.array(z.string()) });

// CORRECT
const State = new StateSchema({
  results: new ReducedValue(z.array(z.string()).default(() => []), { reducer: (curr, upd) => curr.concat(upd) }),
});

---

运行图：Invoke 和 Stream

调用 graph.invoke(input, config) 运行图直至完成，并返回最终状态。

```python result = graph.invoke({"input": "hello"}) # With config (for persistence, tags, etc.) result = graph.invoke({"input": "hello"}, {"configurable": {"thread_id": "1"}}) ``` ```typescript const result = await graph.invoke({ input: "hello" }); // With config const result = await graph.invoke({ input: "hello" }, { configurable: { thread_id: "1" } }); ```

模式	流式内容	使用场景
values	每个步骤后的完整状态	监控完整状态
updates	状态增量	跟踪增量更新
messages	LLM 令牌 + 元数据	聊天 UI
custom	用户定义的数据	进度指示器

实时流式传输 LLM 令牌，以便在聊天 UI 中显示。 ```python for chunk in graph.stream( {"messages": [HumanMessage("Hello")]}, stream_mode="messages" ): token, metadata = chunk if hasattr(token, "content"): print(token.content, end="", flush=True) ``` 实时流式传输 LLM 令牌，以便在聊天 UI 中显示。 ```typescript 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); } } ``` 使用 stream writer 在节点内部发出自定义进度更新。 ```python from langgraph.config import get_stream_writer

def my_node(state): writer = get_stream_writer() writer("Processing step 1...") # Do work writer("Complete!") return {"result": "done"}

for chunk in graph.stream({"data": "test"}, stream_mode="custom"): print(chunk)

</python>
<typescript>
使用 stream writer 在节点内部发出自定义进度更新。
```typescript
import { getWriter } from "@langchain/langgraph";

const myNode = async (state: typeof State.State) => {
  const writer = getWriter();
  writer("Processing step 1...");
  // Do work
  writer("Complete!");
  return { result: "done" };
};

for await (const chunk of graph.stream({ data: "test" }, { streamMode: "custom" })) {
  console.log(chunk);
}

---

错误处理

将错误类型匹配到正确的处理程序：

错误类型	由谁修复	策略	示例
瞬时错误（网络、速率限制）	系统	RetryPolicy(max_attempts=3)	add_node(..., retry_policy=...)
大模型可恢复错误（工具故障）	大模型	ToolNode(tools, handle_tool_errors=True)	错误作为 ToolMessage 返回
用户可修复错误（信息缺失）	人工	interrupt({"message": ...})	收集缺失的数据（参见 HITL 技能）
意外错误	开发者	让错误向上冒泡	raise

对瞬时错误（网络问题、速率限制）使用 RetryPolicy。 ```python from langgraph.types import RetryPolicy

workflow.add_node( "search_documentation", search_documentation, retry_policy=RetryPolicy(max_attempts=3, initial_interval=1.0) )

</python>
<typescript>
对瞬时错误使用 retryPolicy。
```typescript
workflow.addNode(
  "searchDocumentation",
  searchDocumentation,
  {
    retryPolicy: { maxAttempts: 3, initialInterval: 1.0 },
  },
);

使用 langgraph.prebuilt 中的 ToolNode 处理工具执行和错误。将 handle_tool_errors=True 时，错误会作为 ToolMessages 返回，以便大模型恢复。 ```python from langgraph.prebuilt import ToolNode

tool_node = ToolNode(tools, handle_tool_errors=True)

workflow.add_node("tools", tool_node)

</python>
<typescript>
使用 @langchain/langgraph/prebuilt 中的 ToolNode 处理工具执行和错误。将 handleToolErrors 设为 true 时，错误会作为 ToolMessages 返回，以便大模型恢复。
```typescript
import { ToolNode } from "@langchain/langgraph/prebuilt";

const toolNode = new ToolNode(tools, { handleToolErrors: true });

workflow.addNode("tools", toolNode);

---

常见修复方法

必须调用 compile() 才能获得可执行图。 ```python # WRONG builder.invoke({"input": "test"}) # AttributeError!

CORRECT

graph = builder.compile() graph.invoke({"input": "test"})

</python>
<typescript>
必须调用 compile() 才能获得可执行图。
```typescript
// WRONG
await builder.invoke({ input: "test" });

// CORRECT
const graph = builder.compile();
await graph.invoke({ input: "test" });

提供指向 END 的条件路径，以避免无限循环。 ```python # WRONG: Loops forever builder.add_edge("node_a", "node_b") builder.add_edge("node_b", "node_a")

CORRECT

def should_continue(state): return END if state["count"] > 10 else "node_b" builder.add_conditional_edges("node_a", should_continue)

</python>
<typescript>
使用以 END 为返回目标的条件边来打破循环。
```typescript
// WRONG: Loops forever
builder.addEdge("node_a", "node_b").addEdge("node_b", "node_a");

// CORRECT
builder.addConditionalEdges("node_a", (state) => state.count > 10 ? END : "node_b");

其他常见错误： ```python # Router must return names of nodes that exist in the graph builder.add_node("my_node", func) # Add node BEFORE referencing in edges builder.add_conditional_edges("node_a", router, ["my_node"])

Command return type needs Literal for routing destinations (Python)

def node_a(state) -> Command[Literal["node_b", "node_c"]]: return Command(goto="node_b")

START is entry-only - cannot route back to it

builder.add_edge("node_a", START) # WRONG! builder.add_edge("node_a", "entry") # Use a named entry node instead

Reducer expects matching types

return {"items": ["item"]} # List for list reducer, not a string

```typescript
// Always await graph.invoke() - it returns a Promise
const result = await graph.invoke({ input: "test" });

// TS Command nodes need { ends } to declare routing destinations
builder.addNode("router", routerFn, { ends: ["node_b", "node_c"] });

### 不应执行的操作

• 直接修改状态 — 始终从节点返回部分更新字典
• 路由回 START — 它仅用于入口；请改用命名节点
• 忘记为列表字段添加 reducer — 没有 reducer 时，最后一次写入会覆盖之前的值
• 未理解静态边与 Command goto 都会执行，就将两者混用