LangGraph--Graph API

1. 核心概念

LangGraph 的设计理念是 “Nodes 做工作，Edges 决定流程”。

1.1 Graph

• 图是整个工作流的抽象。执行图意味着节点间消息的传递和状态更新。

• 图采用 message passing 的方式，每个节点处理完状态后，将更新沿边发送到下一个节点。

• 图的执行分为 super-step（超步）：

  • 同一超步中的节点可并行执行。
  • 不同超步顺序执行。
  • 节点在收到消息前是inactive，收到消息后变为active。
  • 图执行结束条件：所有节点inactive且无消息在传输。

1.2 State（状态）

• 状态是图的共享数据结构，记录当前工作流的快照。

• 由 schema 和 reducer 组成：

  • schema：定义状态字段，可以使用TypedDict、dataclass或Pydantic BaseModel。
  • reducer：定义状态更新逻辑。默认覆盖，可自定义函数（如operator.add叠加列表）。

1.2.1 多 schema 支持

• 图的内部状态可以有多个 schema：

  • OverallState：完整状态
  • InputState/OutputState：输入输出子集
  • PrivateState：内部节点通信状态

• 节点可以读写整个OverallState，也可以写入子状态PrivateState。

1.2.2 Reducers 示例

fromtypingimportAnnotatedfromtyping_extensionsimportTypedDictfromoperatorimportaddclassState(TypedDict):foo:intbar:Annotated[list[str],add]# 使用 reducer 叠加

• 默认 reducer 是 覆盖（overwrite）
• 可用Overwrite强制覆盖，不使用 reducer。
• add_messages是处理消息列表状态的专用 reducer（处理更新和追加）。

1.3 Nodes（节点）

• 节点是执行逻辑的函数，接收state、config、runtime三个参数。

• 可以是同步或异步函数。

• 节点内部可调用 LLM 或普通函数逻辑。

• 节点函数可以返回：

  • 更新状态字典{key: value}
  • Command对象（同时更新状态和控制流）

• 内置节点：

  • START：图的入口节点
  • END：图的终止节点

节点示例

fromlanggraph.graphimportStateGraph,START,ENDdefnode_a(state):return{"foo":"bar"}graph=StateGraph(dict)graph.add_node("node_a",node_a)graph.add_edge(START,"node_a")graph.add_edge("node_a",END)

节点高级特性

• 缓存（Caching）：基于输入缓存节点结果。
• runtime context：传递非状态信息（如数据库连接或模型配置）。
• 递归控制：通过config["metadata"]["langgraph_step"]获取当前超步计数，可主动或被动处理递归限制。

1.4 Edges（边）

• 边控制图的流向和条件。

• 边类型：

  • Normal Edge：固定连接节点
  • Conditional Edge：根据routing_function(state)动态选择下一节点
  • Entry Point：用户输入进入的第一个节点
  • Conditional Entry Point：动态选择入口节点

Conditional Edge 示例

defroute(state):return"node_b"ifstate["foo"]=="bar"else"node_c"graph.add_conditional_edges("node_a",route)

Command 对象

• 将状态更新与控制流合并。

fromlanggraph.typesimportCommanddefmy_node(state):returnCommand(update={"foo":"bar"},goto="node_b")

• 可用于：

  • 子图跳转
  • 人工干预（Human-in-the-loop）
  • 动态状态更新与路由

1.5 Send

• 用于动态生成节点和状态，支持 Map-Reduce 模式。

fromlanggraph.typesimportSenddefcontinue_to_jokes(state):return[Send("generate_joke",{"subject":s})forsinstate["subjects"]]

2. 消息（Messages）

• 常用于聊天应用，将消息列表存入状态。
• 推荐使用add_messagesreducer 处理消息列表。
• MessagesState提供了预定义状态结构{"messages": list[AnyMessage]}。

消息更新示例

fromlanggraph.graphimportMessagesStatefromlanggraph.graph.messageimportadd_messagesclassState(MessagesState):documents:list[str]

3. 运行时（Runtime）

• Runtime对象在节点中传递上下文信息。

• 可传递：

  • 用户信息
  • 模型选择
  • 数据库连接等

示例

@dataclassclassContext:user_id:strdefnode_with_runtime(state,runtime:Runtime[Context]):print(runtime.context.user_id)

4. 图的构建与执行

4.1 StateGraph

• 核心类，负责图的管理。

• 支持：

  • 添加节点add_node
  • 添加边add_edge,add_conditional_edges
  • 设置入口set_entry_point
  • 设置终止节点set_finish_point
  • 编译.compile()
  • 执行.invoke(inputs)

4.2 Graph Compilation

• 必须在执行前调用.compile()

• 编译会：

  • 检查节点完整性（无孤立节点）
  • 设置缓存、断点
  • 准备状态通道和 reducer

4.3 Recursion / 超步限制

• 默认递归限制为 25 步
• 可通过invoke(config={"recursion_limit": N})修改
• 可以通过config["metadata"]["langgraph_step"]访问当前步数，支持 主动处理 递归限制。

5. 图迁移（Graph Migrations）

• 支持修改节点、边和状态字段

• 对于中断的线程：

  • 允许添加新节点和边
  • 不允许删除或重命名节点（可能引发错误）

• 状态兼容：

  • 新增或删除键向前向后兼容
  • 修改类型不兼容可能导致旧线程出错

6. 可视化（Visualization）

• 支持图可视化，便于理解复杂工作流
• 官方文档提供内置方法和指南

7. 总结

LangGraph Graph API 提供了：