LangGraph

1. 概念

LangGraph 是用于构建、管理和部署长时间运行、有状态智能体与工作流的底层编排框架。 它把应用建模为图：

• State：保存当前应用状态
• Node：读取状态并执行逻辑
• Edge：决定下一个节点 可以把核心关系概括为：

$$\text{Graph} = \text{State} + \text{Nodes} + \text{Edges}$$

节点可以包含大语言模型，也可以只是普通 Python 代码。因此，LangGraph 能够在同一流程中混合确定性步骤和模型驱动步骤。

2. LangGraph 的特点

LangGraph 主要解决复杂智能体的流程编排问题，重点能力包括：

• 持久化执行
• 检查点与断点恢复
• 短期和长期记忆
• 流式输出
• 人机协同
• 条件分支与循环
• 子图
• 错误处理与重试
• 工作流可视化和追踪 LangGraph 是较底层的工具，不负责替开发者决定提示词、工具策略或智能体架构。

3. 安装

pip install -U langgraph

如果图中的节点需要使用 LangChain 的模型和消息组件，还需要安装相应的 LangChain 及模型集成包。

4. 三个核心概念

4.1 State

State 是图在某一时刻的共享数据结构。节点从 State 中读取数据，并返回需要更新的字段。 State 可以使用 TypedDict、dataclass 或 Pydantic 模型定义。每个字段还可以配置 reducer，用于决定新旧值如何合并。 默认情况下，节点返回的新值会覆盖旧值：

\text{new_state[key]}=\text{node_update[key]}

如果字段使用自定义 reducer，则可以追加或合并数据：

\text{new_value}=\text{reducer}(\text{old_value},\text{update})

4.2 Node

Node 是执行具体逻辑的函数。它接收当前 State，并返回部分状态更新。

def greet_node(state: dict):
    return {"message": f"你好，{state['name']}"}

节点不必返回完整 State，只需要返回自己要更新的字段。

4.3 Edge

Edge 决定节点执行完成后下一步去哪里：

• 普通边：固定从节点 A 走到节点 B
• 条件边：根据当前 State 动态选择节点
• START：图的起点
• END：图的终点

builder.add_edge(START, "greet")
builder.add_edge("greet", END)

5. 创建第一个图

下面的例子创建一个从输入到输出的最简单图。

from typing_extensions import TypedDict
from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    message: str


def hello_node(state: State):
    return {"message": state["message"] + "，你好"}


builder = StateGraph(State)
builder.add_node("hello", hello_node)
builder.add_edge(START, "hello")
builder.add_edge("hello", END)

graph = builder.compile()
result = graph.invoke({"message": "小明"})
print(result)

构建图的一般步骤是：

1. 定义 State
2. 编写节点函数
3. 添加节点
4. 添加普通边或条件边
5. 编译图
6. 使用 invoke 或 stream 执行 图必须先调用 compile()，才能执行。

6. 条件路由

条件边使用路由函数根据 State 决定下一节点。

from typing import Literal


def route(state: State) -> Literal["success", "failure"]:
    if state["message"]:
        return "success"
    return "failure"


builder.add_conditional_edges(
    "check",
    route,
    {"success": "success", "failure": "failure"},
)

条件路由适合实现：

• 判断是否需要继续处理
• 根据分类结果进入不同流程
• 失败重试或转入备用流程
• 根据模型输出选择工具或子任务

7. 循环工作流

LangGraph 支持边指向已经执行过的节点，因此可以构建循环。 一个常见的智能体循环是：

$$\text{Model}\rightarrow\text{Tool}\rightarrow\text{Model}\rightarrow\cdots$$

循环必须有明确的终止条件，例如模型不再请求工具、任务状态变为完成或达到最大执行次数，否则可能产生无限循环。

8. 消息状态

对话型应用通常需要在 State 中保存消息列表。LangGraph 提供了 MessagesState，并使用 add_messages reducer 处理新消息追加和已有消息更新。

from langgraph.graph import MessagesState


class State(MessagesState):
    documents: list[str]

使用消息状态时，节点通常读取最后一条消息：

def read_last_message(state: MessagesState):
    message = state["messages"][-1]
    return {"messages": [{"role": "ai", "content": message.content}]}

9. Reducer

Reducer 决定节点更新如何应用到 State。

9.1 默认覆盖

没有指定 reducer 时，新值覆盖旧值：

class State(TypedDict):
    count: int

如果当前状态为 {"count": 1}，节点返回 {"count": 2}，新状态就是 {"count": 2}。

9.2 自定义合并

对于需要累积的列表，可以使用 reducer：

from operator import add
from typing import Annotated


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

如果旧值是 ["开始"]，更新是 ["完成"]，合并后是 ["开始", "完成"]。 消息列表更推荐使用 add_messages，因为它能够根据消息 ID 正确处理追加和更新。

10. Command

Command 用于在一个节点函数中同时完成状态更新和路由。

from typing import Literal
from langgraph.types import Command


def decide(state: State) -> Command[Literal["next", "end"]]:
    if state["approved"]:
        return Command(update={"status": "继续"}, goto="next")
    return Command(update={"status": "结束"}, goto="end")

Command 的常见用途包括：

• update：更新 State
• goto：跳转到指定节点
• graph=Command.PARENT：从子图跳转到父图节点
• resume：在人工中断后恢复执行 如果只需要根据状态路由而不需要同时更新状态，使用条件边通常更清晰。

11. Send 与 Map-Reduce

当需要对数量未知的一组对象分别执行同一个节点时，可以使用 Send 动态创建任务。 例如：

1. 第一个节点生成多个主题
2. 对每个主题并行调用生成节点
3. 使用 reducer 汇总结果

from langgraph.types import Send


def fan_out(state: dict):
    return [
        Send("generate", {"subject": subject})
        for subject in state["subjects"]
    ]


builder.add_conditional_edges("split", fan_out)

这种模式适合批量分析、并行检索和 Map-Reduce 工作流。

12. 持久化与 Checkpointer

Checkpointer 在图的执行过程中保存检查点，使应用可以：

• 保存会话历史
• 跨调用恢复状态
• 从中断位置继续执行
• 在失败后恢复
• 查看不同时间点的状态

from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "thread-1"}}
result = graph.invoke(input_data, config=config)

生产环境应根据持久化需求选择合适的数据库或存储实现。InMemorySaver 适合学习和临时测试，不适合进程重启后仍需保留数据的生产场景。

13. 人机协同与 Interrupt

对于发送邮件、修改数据库、删除文件或执行高风险操作等场景，可以让图暂停并等待人工确认。

from langgraph.types import interrupt


def human_review(state: dict):
    answer = interrupt("是否批准这项操作？")
    return {"approved": answer}

图暂停后，可以使用 Command(resume=...) 传入人工决定并继续执行。 人工介入的关键点是：

• 中断前的副作用应当可重复或具有幂等性
• 使用 checkpointer 保存中断时的状态
• 恢复时使用同一个 thread_id
• 对高风险动作设置明确的审批边界

14. Runtime Context

Context 用于传递本次运行的外部依赖或配置，例如用户 ID、模型提供商、数据库连接等。它不一定需要写入图的 State。

from dataclasses import dataclass

from langgraph.runtime import Runtime


@dataclass
class Context:
    user_id: str


def user_node(state: dict, runtime: Runtime[Context]):
    return {"message": f"用户：{runtime.context.user_id}"}

State 适合保存会影响流程的业务数据；Context 适合保存本次运行的配置和依赖。

15. 节点的幂等性

启用 checkpointer 后，节点可能因为中断、重试或恢复而再次执行。因此节点中的外部副作用需要设计成幂等的。 例如，写入数据库时可以使用：

• 幂等键
• Upsert
• 写入前检查
• 可识别的事务 ID 如果一个节点每次重试都会重复创建订单或重复发送邮件，就可能造成严重问题。

16. 流式执行

invoke 等待图执行完成并返回最终状态；stream 可以在图执行过程中逐步输出状态更新。

for update in graph.stream(
    {"message": "开始"},
    stream_mode="updates",
):
    print(update)

流式输出适合展示：

• 模型生成过程
• 当前执行节点
• 工具调用和工具结果
• 长任务的阶段性进度

17. Graph API 与 Functional API

LangGraph 提供两种主要的建模方式：

• Graph API：显式定义 State、Nodes 和 Edges，适合分支多、需要可视化的工作流
• Functional API：使用更接近普通 Python 控制流的函数和任务，适合线性流程或希望少写图结构代码的场景 选择依据不是功能强弱，而是流程表达方式和可维护性。

18. LangGraph 的执行模型

LangGraph 可以看作一个由状态驱动的执行引擎：

$$\text{Input State}\rightarrow\text{Node}\rightarrow\text{State Update}\rightarrow\text{Edge Routing}\rightarrow\text{Next Node}$$

一次执行通常包含以下过程：

1. 图接收初始输入并创建本次运行的状态
2. 从 START 节点进入一个或多个节点
3. 节点读取当前状态并执行普通代码或模型调用
4. 节点返回部分状态更新
5. reducer 将更新合并到图状态
6. 普通边或条件边决定下一批节点
7. 图不断执行，直到所有路径到达 END 或被中断
8. 如果配置了 checkpointer，在关键步骤保存检查点 LangGraph 使用离散的 super-step 组织执行。一个 super-step 可以包含并行运行的一组节点；节点完成后，图再根据新状态进入下一步。

19. Graph API 的完整结构

使用 Graph API 时，通常需要明确四类内容：

19.1 State Schema

定义状态字段、类型和 reducer。

from typing_extensions import TypedDict


class State(TypedDict):
    question: str
    answer: str
    approved: bool

19.2 Nodes

节点是实际执行逻辑的函数，可以是同步函数或异步函数。

def answer_node(state: State):
    return {"answer": f"问题是：{state['question']}"}

19.3 Edges

边负责连接节点，表达固定或动态的控制流。

builder.add_edge(START, "answer")
builder.add_edge("answer", END)

19.4 Compile

编译阶段检查图的结构，并配置 checkpointer、断点和其他运行参数。

graph = builder.compile()

compile() 不是模型训练，而是把节点、边和运行配置组装成可以执行的图对象。

20. LangGraph 与 LangChain 的详细区别

20.1 一句话区别

LangChain 主要解决“如何方便地使用模型、工具和智能体”；LangGraph 主要解决“如何精确地编排一个有状态、可恢复的复杂执行流程”。

20.2 定位和抽象层级

对比项	LangChain	LangGraph
定位	LLM 应用和智能体开发框架	工作流与智能体编排运行时
抽象层级	较高层	较低层、更接近执行引擎
核心入口	create_agent、模型、工具、消息	StateGraph、State、Node、Edge
默认思路	组装一个可配置的通用智能体循环	自己定义完整的状态图
学习成本	较低，适合快速开始	较高，需要理解状态和图执行
控制程度	通过参数、中间件和工具配置	可以控制每个节点、边和状态更新

20.3 核心单元不同

LangChain 的核心单元是：

• Model：调用大语言模型
• Message：表示对话内容
• Tool：提供外部能力
• Agent：让模型循环调用工具
• Middleware：扩展智能体行为 开发者通常告诉 LangChain“使用哪个模型、有哪些工具、遵循什么提示词”，然后由通用 Agent Loop 运行。 LangGraph 的核心单元是：
• State：保存整个流程的共享数据
• Node：执行一个明确的步骤
• Edge：定义节点之间的转移
• Reducer：定义状态更新的合并方式
• Checkpointer：保存和恢复执行状态 开发者需要明确整个流程如何走，而不仅仅是提供工具列表。

20.4 控制流不同

LangChain 的 Agent Loop 通常是：

$$\text{Model}\rightarrow\text{Tool}\rightarrow\text{Model}\rightarrow\text{Final Answer}$$

模型根据上下文决定是否调用工具以及下一步动作。 LangGraph 可以表达更丰富的控制流：

$$\text{Input}\rightarrow\begin{cases}\text{检索}\rightarrow\text{生成}\text{人工审批}\rightarrow\text{执行}\text{失败}\rightarrow\text{重试或降级}\end{cases}$$

流程的关键分支由图的边和路由函数决定，模型只负责被安排到某些节点中。

20.5 状态管理不同

LangChain 的简单 Agent 通常围绕消息列表工作，开发者主要关心用户消息、模型消息和工具消息。 LangGraph 把状态提升为一等公民。状态可以同时保存：

• 用户问题
• 当前任务阶段
• 消息历史
• 检索文档
• 工具调用结果
• 审批状态
• 错误信息
• 重试次数
• 最终结构化输出 每个字段可以独立配置覆盖或追加等更新规则，因此适合复杂业务流程。

20.6 执行可预测性不同

LangChain Agent 的下一步可能由模型动态决定，因此灵活，但可能出现：

• 不必要的工具调用
• 工具调用顺序变化
• 多余的思考轮次
• 由于模型输出差异导致流程变化 LangGraph 可以把关键业务步骤写成确定性节点和边，只把需要推理的部分交给模型。这样更容易进行测试、审计、成本控制和安全限制。

20.7 持久化和恢复不同

LangChain 可以通过 checkpointer 保存 Agent 会话，但简单应用通常只需要短期对话记忆。 LangGraph 从设计上就围绕“可暂停、可保存、可恢复”组织执行，适合：

• 运行时间较长的研究任务
• 需要跨进程继续的任务
• 等待人工审批的流程
• 失败后从检查点恢复
• 需要查看历史状态的应用

20.8 人机协同不同

LangChain 可以通过中间件加入人工审批逻辑。 LangGraph 通过 interrupt() 明确暂停图，并使用 Command(resume=...) 恢复。暂停前的状态由 checkpointer 保存，恢复时仍使用原来的 thread_id。 这种方式适合：

• 发送邮件前确认
• 删除文件前确认
• 执行付款前审批
• 修改数据库前审核
• 模型无法决定时请求用户补充信息

20.9 错误处理不同

在 LangChain 中，错误通常放在工具函数、重试中间件、回退模型或 Agent 配置中处理。 在 LangGraph 中，可以把错误转成状态并路由到专门节点：

$$\text{Node}\rightarrow\begin{cases}\text{Retry Node}\text{Fallback Node}\text{Human Review Node}\text{Failure Node}\end{cases}$$

错误处理本身成为流程的一部分，而不是只在流程外部捕获异常。

20.10 多智能体不同

LangChain 可以创建多个 Agent，但多个 Agent 之间的交接、共享上下文和终止条件需要额外设计。 LangGraph 可以将智能体表示为节点或子图，并明确：

• 哪个智能体先执行
• 哪些状态可以共享
• 什么时候交接给另一个智能体
• 交接后如何返回
• 哪些情况转人工 因此 LangGraph 更适合研究员、规划器、执行器、审核员等角色协作。

20.11 代码组织方式不同

LangChain 的典型代码重点是配置模型和工具：

agent = create_agent(
    model=model,
    tools=[search, calculator],
    system_prompt="你是一个研究助手。",
)

LangGraph 的典型代码重点是定义流程：

builder = StateGraph(State)
builder.add_node("research", research_node)
builder.add_node("review", review_node)
builder.add_edge(START, "research")
builder.add_conditional_edges("research", route_after_research)
builder.add_edge("review", END)
graph = builder.compile()

前者的重点是“给智能体什么能力”；后者的重点是“整个任务如何执行”。

20.12 适用场景对照

场景	更适合的工具	原因
单轮模型问答	LangChain	调用简单，代码少
一个模型加几个工具	LangChain	create_agent 可以快速完成
简单 RAG	LangChain	文档、检索和生成组件易于组合
固定的短流程	LangChain Chain 或 LangGraph	流程简单时 Chain 足够，需状态时用 LangGraph
多步审批流程	LangGraph	需要状态、分支和人工中断
长时间运行任务	LangGraph	支持检查点和恢复
多智能体协作	LangGraph	方便编排角色交接和共享状态
高风险工具操作	LangGraph	可插入明确的审批节点和恢复路径
精确审计的业务流程	LangGraph	节点和边构成可检查的执行路径

20.13 二者的关系

LangChain 和 LangGraph 不是互相排斥的两个框架：

• LangChain 提供模型、消息、工具和智能体组件
• LangGraph 提供状态管理和流程编排
• LangChain 的 Agent 可以建立在 LangGraph 之上
• LangGraph 可以使用 LangChain，也可以只使用普通 Python 函数 常见架构是：

$$\text{LangChain Components}+\text{LangGraph Runtime}=\text{可控的 LLM Workflow}$$

因此，“使用 LangGraph”不代表不能使用 LangChain；很多项目正是用 LangChain 负责组件，用 LangGraph 负责流程。

20.14 迁移路径

1. 使用 LangChain 直接调用模型
2. 使用 LangChain 工具和 create_agent 快速验证需求
3. 当出现多个阶段、分支、循环或审批时，定义明确的 State
4. 把原有模型和工具调用封装为 LangGraph 节点
5. 使用边和条件路由重建流程
6. 加入 checkpointer、interrupt、重试和 LangSmith 追踪 迁移时通常不需要重写模型和工具，只需要改变它们被调用和编排的方式。

21. 典型工作流结构

21.1 顺序工作流

$$A\rightarrow B\rightarrow C$$

适合固定的文档处理、提取、审核流程。

21.2 条件分支

$$A\rightarrow\begin{cases}B\C\end{cases}$$

适合分类、审批和错误处理。

21.3 循环智能体

$$\text{思考}\rightarrow\text{调用工具}\rightarrow\text{观察结果}\rightarrow\text{思考}$$

适合需要多轮工具调用的任务。

21.4 并行与汇总

$$A\rightarrow B_1,B_2,\ldots,B_n\rightarrow C$$

适合并行检索、批量分析和多文档处理。

22. 设计 LangGraph 工作流的步骤

1. 先写出业务流程图，不要一开始就写代码
2. 区分确定性逻辑和需要模型判断的逻辑
3. 为会影响后续路由的数据设计 State 字段
4. 为每个明确职责设计一个 Node
5. 为固定转移添加普通边
6. 为需要判断的转移添加条件边
7. 为列表累积、消息追加等字段配置 reducer
8. 为长任务和人工审批配置 checkpointer
9. 为外部副作用设计幂等方案
10. 为循环设计终止条件、最大步数和失败路径

23. 调试重点

• 检查每个节点的输入和输出 State
• 检查 reducer 是否会意外覆盖或重复追加数据
• 检查条件边是否覆盖所有可能结果
• 为循环设置最大步数或明确终止条件
• 对外部副作用设计幂等性
• 对中断和恢复使用稳定的 thread_id
• 检查恢复时节点是否会重新执行
• 使用 LangSmith 检查节点执行路径、状态变化和工具调用

24. 学习顺序

1. 学会定义 State、Node 和 Edge
2. 编写顺序工作流
3. 学习条件边和循环
4. 学习消息状态和 reducer
5. 学习 checkpointer 与会话记忆
6. 学习 interrupt 与人工审批
7. 学习 Command、Send 和子图
8. 结合 LangChain 模型和工具构建智能体

25. 关键术语

• State：图运行期间共享的数据
• Node：执行逻辑并返回状态更新的函数
• Edge：决定下一步执行位置的连接
• Reducer：合并旧状态和节点更新的函数
• Checkpointer：保存图状态的组件
• Thread：由 thread_id 标识的一条执行或会话轨迹
• Interrupt：暂停图并等待外部输入
• Command：同时更新状态并控制路由
• Send：动态发送不同状态到多个节点
• Subgraph：嵌套在其他图中的子图
• Super-step：图执行中的一个离散迭代步骤