图 API
原文：Graph API
一句话
LangGraph 将代理工作流建模为图，通过定义状态（State）、节点（Nodes）和边（Edges）来构建复杂、循环的工作流。
什么时候翻这页
当你需要了解 LangGraph 的核心 API 结构，学习如何定义状态、创建节点和连接它们时，可以参考本页内容。
核心概念
图（Graphs）
LangGraph 将代理工作流建模为图，使用三个关键组件定义代理行为：
- 状态（State）：表示应用程序当前快照的共享数据结构，可以是任何数据类型，但通常使用共享状态模式定义。
- 节点（Nodes）：编码代理逻辑的函数，接收当前状态作为输入，执行一些计算或副作用，并返回更新后的状态。
- 边（Edges）：根据当前状态确定下一个要执行节点的函数，可以是条件分支或固定转换。
通过组合节点和边，可以创建随时间演化的复杂循环工作流。
StateGraph
StateGraph 类是使用的主要图类，由用户定义的 State 对象参数化。
编译图
构建图时，首先定义状态，然后添加节点和边，最后编译它。编译是一个简单的步骤，它提供对图结构的基本检查（没有孤立的节点等），也是你可以指定运行时参数（如 checkpointers 和断点）的地方。
怎么做
定义状态
定义图时首先要定义图的 State。State 包括图的 schema 和指定如何应用更新的 reducer 函数。schema 将是图中所有节点和边的输入模式，可以是 TypedDict 或 Pydantic 模型。
Schema
指定图模式的主要方法是使用 TypedDict。如果要在状态中提供默认值，请使用 dataclass。如果需要递归数据验证，也可以使用 Pydantic BaseModel 作为图状态（注意 Pydantic 的性能低于 TypedDict 或 dataclass）。
Reducers
Reducers 是理解节点更新如何应用于 State 的关键。State 中的每个键都有自己的独立归约函数。如果没有明确指定归约函数，则假定对该键的所有更新都应该覆盖它。
在图状态中使用消息
在许多情况下，将先前的对话历史存储为图状态中的消息列表很有帮助。为此，我们可以向图状态添加一个存储 Message 对象列表的键（通道），并使用归约函数（如 operator.add 或 add_messages）来更新它。
创建节点
节点是接受以下参数的 Python 函数（同步或异步）：
state—图的状态config—包含配置信息（如thread_id）和跟踪信息（如tags）的RunnableConfig对象runtime—包含运行时上下文和其他信息的Runtime对象
使用 add_node 方法将节点添加到图中。
连接节点
边定义了逻辑如何路由以及图如何决定停止。主要有几种类型的边：
- 正常边：直接从一个节点到下一个节点。
- 条件边：调用函数确定下一个要去的节点。
- 入口点：用户输入到达时首先调用的节点。
- 条件入口点：调用函数确定用户输入到达时首先调用的节点。
使用 add_edge 方法添加正常边，使用 add_conditional_edges 方法添加条件边。
使用 Send 和 Command
- Send：用于支持 map-reduce 等设计模式，允许在运行时动态创建边。
- Command：用于控制图执行的多功能原语，可以结合状态更新和路由。
处理递归限制
递归限制设置图在单次执行期间可以执行的最大超级步骤数。可以通过 config 字典中的 recursion_limit 键在运行时设置递归限制。
命令 / API 速查
创建和编译图
from langgraph.graph import StateGraph, START, END
# 定义状态
class State(TypedDict):
messages: list[dict]
# 创建图构建器
builder = StateGraph(State)
# 添加节点
builder.add_node("node_name", node_function)
# 添加边
builder.add_edge(START, "node_name")
builder.add_edge("node_name", END)
# 编译图
graph = builder.compile(checkpointer=checkpointer)
处理状态
# 使用 TypedDict 定义状态
class State(TypedDict):
messages: list[dict]
counter: int
# 使用带归约器的状态
from typing import Annotated, List
from operator import add
class State(TypedDict):
messages: Annotated[List[dict], add]
counter: int
创建节点
# 基本节点
def my_node(state: State) -> dict:
# 处理逻辑
return {"key": "value"}
# 使用运行时上下文的节点
def my_node_with_runtime(state: State, runtime: Runtime) -> dict:
# 访问运行时信息
print(f"Thread ID: {runtime.execution_info.thread_id}")
return {"key": "value"}
# 添加节点到图
builder.add_node("my_node", my_node)
添加边
# 添加正常边
builder.add_edge("node_a", "node_b")
# 添加条件边
def routing_function(state: State) -> str:
# 返回下一个节点名称
return "node_b" if state["condition"] else "node_c"
builder.add_conditional_edges("node_a", routing_function)
# 添加条件边带映射
builder.add_conditional_edges(
"node_a",
routing_function,
{"condition_a": "node_b", "condition_b": "node_c"}
)
使用 Send
from langgraph.types import Send
def continue_to_jokes(state: OverallState):
return [Send("generate_joke", {"subject": s}) for s in state['subjects']]
builder.add_conditional_edges("node_a", continue_to_jokes)
使用 Command
from langgraph.types import Command
# 从节点返回 Command
def my_node(state: State) -> Command[Literal["other_node"]]:
return Command(
update={"key": "value"},
goto="other_node"
)
# 使用 resume 继续
graph.invoke(Command(resume="value"), config)
处理消息
from langgraph.graph import MessagesState
from langgraph.graph.message import add_messages
# 使用 MessagesState
class State(MessagesState):
additional_field: str
# 或自定义消息处理
class State(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
与 Hello-Agents / Claude Code 的联系
在 Hello-Agents 中，我们学习了 LangGraph 的基本概念和简单用法。本页内容提供了更深入的 API 细节，特别是关于状态管理、节点和边的更高级用法。Hello-Agents 中的示例使用了 StateGraph 和基本节点/边概念，而本页内容扩展了这些概念，包括归约器、消息处理、Send 和 Command 等高级特性。
初学者易错点
- 忘记编译图：在使用图之前必须调用
.compile()方法。 - 混合路由机制：不要在同一节点混合使用正常边和动态路由（Command），因为两条路径都可能执行，使图行为难以理解。
- 幂等性设计：当使用 checkpointer 时，节点可能会重新执行，确保节点逻辑是幂等的。
- 状态更新误解：节点返回的是状态更新，而不是整个状态对象。
- 递归限制处理：不正确处理递归限制可能导致图执行失败，应考虑使用
RemainingSteps进行主动处理。