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 示例

from typing import Annotated
from typing_extensions import TypedDict
from operator import add
class State(TypedDict):
    foo: int
    bar: Annotated[list[str], add]  # 使用 reducer 叠加

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

1.3 Nodes（节点）

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

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

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

• 节点函数可以返回：

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

• 内置节点：

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

节点示例

from langgraph.graph import StateGraph, START, END

def node_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 示例

def route(state):
    return "node_b" if state["foo"] == "bar" else "node_c"

graph.add_conditional_edges("node_a", route)

Command 对象

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

from langgraph.types import Command
def my_node(state):
    return Command(update={"foo":"bar"}, goto="node_b")

• 可用于：

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

1.5 Send

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

from langgraph.types import Send
def continue_to_jokes(state):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

2. 消息（Messages）

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

消息更新示例

from langgraph.graph import MessagesState
from langgraph.graph.message import add_messages
class State(MessagesState):
    documents: list[str]

3. 运行时（Runtime）

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

• 可传递：

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

示例

@dataclass
class Context:
    user_id: str

def node_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 提供了：

模块	功能
State	管理图的共享数据，支持多 schema 和 reducer
Node	执行逻辑，支持状态更新、Command、缓存、runtime context
Edge	控制流，可固定、条件、动态或入口
Messages	用于存储聊天记录和其他序列化消息
Send / Command	支持 Map-Reduce 或动态控制流
Runtime	提供节点执行上下文，不属于状态
Recursion / Metadata	超步控制与调试信息
Graph Compilation	检查图结构，准备执行，支持缓存
Migration / Visualization	动态修改图结构与状态，图形化展示