LangGraph 的状态驱动编排机制

前言

在简单的问答场景中，调用一次大模型就能得到结果；但当任务需要多轮推理、工具调用、条件分支、失败重试、人工审核甚至长期运行时，单纯依赖一条线性调用链很快就会变得难以维护。

LangGraph 的价值并不是“再封装一个大模型调用接口”，而是提供了一种显式、可恢复、可观测的智能体编排方式：开发者用图描述流程，用状态承载上下文，用节点执行任务，再由运行时根据边和状态决定下一步动作。

一、为什么线性工作流不够用了

传统的 LLM 应用通常可以表示为：用户问题 → Prompt → 大模型 → 输出解析 → 最终答案

这种结构适合固定流程，但真实的 Agent 往往更接近下面的过程：

判断用户问题属于哪一类；
决定是否需要检索资料；
调用搜索、数据库或业务 API；
检查工具返回结果是否足够；
如果结果不可靠，则重新检索或改写问题；
生成答案；
必要时等待人工审核；
审核通过后继续执行。

这里至少包含了顺序、分支、循环、并行和暂停恢复等控制逻辑。如果把这些逻辑全部写在一个 while 循环和大量 if...else 中，代码虽然能够运行，但执行路径、上下文变化和异常恢复都会逐渐变得不可控。

LangGraph 的核心思路就是不要把 Agent 写成一个不可见的黑盒循环，而要把它拆解成一张可以观察和控制的状态图。

二、图、节点与状态

LangGraph 的基础抽象可以概括为：Graph = Nodes + Edges + State

1. Graph：工作流的整体容器

Graph 描述整个应用的结构，包括：

有哪些执行节点；
节点之间如何连接；
哪些路径是固定的；
哪些路径需要根据运行结果动态选择；
何时开始；
何时结束。

在 Python API 中，常见的入口是 StateGraph。开发者定义状态结构、注册节点和边，最后通过 compile() 将图编译成可执行对象。

2. Node：执行一个具体步骤

节点通常是一个函数，也可以封装模型调用、工具调用、数据库查询或业务逻辑。它的基本模式是：读取当前状态 → 执行任务 → 返回状态更新

例如，一个检索节点可以读取 question，调用向量数据库，然后只返回新增的 documents：

def retrieve_node(state):
    documents = retriever.invoke(state["question"])
    return {
        "documents": documents
    }

节点并不需要返回完整状态，而是返回对状态的增量更新，即为读取当前状态并返回更新内容的函数。

3. Edge：决定下一步去哪

边负责表达控制流，常见类型包括：

普通边：节点 A 执行后必然进入节点 B；
条件边：根据状态或节点输出选择不同路径；
循环边：节点执行后重新回到之前的节点；
动态跳转：由运行过程决定下一个目标节点。

例如：

1 2	问题分类 → 需要检索？ → 检索 → 生成 └── 不需要 → 直接生成

4. State：整个图共享的“工作记忆”

State 是 LangGraph 编排机制的核心。它不是简单的参数对象，而是贯穿整个执行过程的共享数据结构。

一个典型的状态可以这样定义：

from typing import Annotated
from typing_extensions import TypedDict
import operator

class AgentState(TypedDict):
    question: str
    documents: list
    answer: str
    attempts: int
    messages: Annotated[list, operator.add]

其中：

question 保存用户原始问题；
documents 保存检索结果；
answer 保存当前答案；
attempts 记录重试次数；
messages 可以通过 Reducer 追加消息，而不是覆盖原值。

设计 State 时，一个重要原则是：尽量保存结构化、原始的数据，而不是提前拼装好的 Prompt 文本。需要调用模型时，再由节点根据当前状态构造 Prompt，这样更便于调试、复用和状态检查。

三、一次执行是如何被编排的

LangGraph 的运行机制可以用“状态更新驱动节点激活”来理解。

假设有这样一个结构：START → classify → retrieve → generate → END

一次执行大致会经历以下过程：

用户输入被写入初始状态；
classify 节点被激活；
节点读取状态并返回分类结果；
运行时合并状态更新；
根据边的定义激活下一个节点；
新节点读取更新后的状态并继续执行；
直到抵达 END 或触发中断。

LangGraph 的底层执行模型受到 Pregel/Bulk Synchronous Parallel 思想影响。一个超步骤通常包含三个阶段：Plan → Execution → Update

Plan：确定本轮需要执行哪些节点；
Execution：执行这些节点，必要时可以并行执行；
Update：统一提交节点产生的状态更新。

在当前超步骤中产生的更新，通常会在下一轮调度时对其他节点可见，而不是让并行节点在执行中途互相读取到不完整的数据。

状态驱动的执行循环

四、条件路由与循环：Agent 灵活性的来源

1. 条件边实现动态决策

条件边并不是让大模型直接接管整个程序，而是让模型或业务代码输出一个有限、明确的路由结果。

例如：

def route_after_classify(state):
    if state["question_type"] == "knowledge":
        return "retrieve"
    if state["question_type"] == "smalltalk":
        return "generate"
    return "human_review"

然后将路由结果连接到不同节点：

builder.add_conditional_edges(
    "classify",
    route_after_classify,
    {
        "retrieve": "retrieve",
        "generate": "generate",
        "human_review": "human_review",
    },
)

这种方式的关键优势是：**让模型负责判断，让程序负责约束。**模型可以决定“应该查资料还是直接回答”，但真正可执行的目标节点仍然由开发者预先注册。这样既保留了 Agent 的灵活性，也避免让模型生成任意代码或任意执行路径。

2. 循环实现反思和重试

很多 Agent 任务不是一次执行就能完成。例如：

1
2
3

检索 → 生成 → 评估
             ├── 合格 → 结束
             └── 不合格 → 改写问题 → 再检索

这类流程可以通过循环边表达：

def route_after_evaluate(state):
    if state["quality"] == "good":
        return "end"
    if state["attempts"] >= 3:
        return "human_review"
    return "retry"

循环必须设置明确的退出条件，例如：

最大尝试次数；
质量分数达到阈值；
工具调用失败次数超过限制；
已经获得足够证据；
进入人工审核。

否则 Agent 可能陷入“检索—生成—评估—再检索”的无限循环。

带条件分支和循环的RAGAgent

五、状态合并：并行执行为什么不会互相覆盖

当多个节点并行运行时，它们可能同时向同一个状态字段写入数据。例如，一个节点负责搜索网页，另一个节点负责查询内部知识库：

1
2
3

          ┌── 网页搜索 ──┐
输入问题  ─┤             ├─→ 汇总节点
          └── 内部检索 ──┘

如果两个节点都向 documents 写入结果，就需要规定这些更新如何合并。否则，后提交的结果可能覆盖先提交的结果。这就是 Reducer 的作用。它描述的是：旧值 + 本轮多个更新 → 新值

例如，可以让多个节点返回的文档列表进行追加：

from typing import Annotated
import operator

class SearchState(TypedDict):
    question: str
    documents: Annotated[list, operator.add]

这样网页搜索和内部检索产生的结果就可以合并，而不是互相覆盖。底层通道也可以理解为节点之间传递数据的通信机制：通道拥有值类型、更新类型以及更新函数，用于决定状态如何变化。但是要注意并行并不等于无条件地同时执行所有任务，只有在图结构和调度条件允许时，多个节点才会被安排到同一超步骤中。开发者还需要考虑：

外部 API 是否支持并发；
是否存在速率限制；
多个结果是否具备确定的排序；
合并操作是否满足幂等性；
某个并行节点失败后，整体流程如何处理。

六、持久化、中断与恢复

LangGraph 的生产级能力，主要体现在它不仅能“跑完一次”，还能够保存执行过程。

1. Checkpoint：保存每个阶段的状态

Checkpointer 会在执行过程中保存状态快照，检查点通常以线程为组织单位，并通过 thread_id 区分不同会话或任务。

这意味着系统可以具有恢复中断的对话、继续执行长期任务、在故障后从最近状态恢复、查看某一次运行的历史状态、对历史状态进行回溯和分支执行等能力。

可以把 Checkpoint 理解为游戏中的存档点：

1
2
3

节点 A 完成 → 保存状态
节点 B 完成 → 保存状态
节点 C 失败 → 从节点 B 的状态恢复

2. Interrupt：在人与 Agent 之间建立暂停点

当 Agent 需要进行高风险操作时，不应该直接执行。例如：

删除数据；
发送正式邮件；
提交付款；
修改生产环境配置；
发布面向用户的内容。

这时可以在图中设置中断点：生成操作计划 → 暂停 → 人工确认 → 继续执行

LangGraph 的 interrupt() 可以暂停图执行，将状态保存下来，等待外部输入后再通过 Command 恢复。要可靠恢复，通常需要配置 Checkpointer，并为执行指定 thread_id。

可恢复的人工介入流程

七、LangGraph 与普通 Chain 的区别

对比维度	线性 Chain	LangGraph
流程结构	主要是顺序执行	支持顺序、分支、循环和并行
状态管理	常由调用者手动传递	通过共享 State 统一管理
控制流	通常由代码外层控制	由图中的边和路由函数表达
失败恢复	往往需要自行实现	可结合 Checkpoint 恢复执行
人工介入	需要额外设计暂停机制	可使用 Interrupt 形成暂停点
可观测性	主要依赖日志和回调	可以查看节点、状态和执行路径
适用场景	固定的输入输出链路	复杂 Agent、RAG、多智能体和长期任务