技术笔记 LangGraph 核心与实战 Agent AI 架构工作流 LangGraph 状态机 Python

LangGraph 是什么，为什么不用链式调用：从控制权反转看图状态机演进

Asaakii2026-03-112026-10-08

本文是「LangGraph 核心与实战」系列专栏的第 1 篇。专栏总览参见：《LangGraph 核心架构全景：用状态图构建可控 Agent 工作流》。

构建基于大语言模型（LLM）的应用时，最直观的写法通常是将多个调用像链条一样串联起来：

# 典型的链式调用原型
step1_res = llm.invoke(f"提取用户问题中的核心实体: {user_input}")
step2_res = llm.invoke(f"根据实体检索补充上下文: {step1_res.content}")
final_res = llm.invoke(f"基于检索结果生成最终回复: {step2_res.content}")

这种模式通常被称为链式调用（Chain）。它结构直接，便于在十分钟内搭出一个可运行的原型。但在真实业务场景中，一旦系统需要根据中间结果判断是否走退款分支、需要在模型输出不合规时重新生成、或者需要多个专业节点并行处理数据，纯链式调用的代码就会迅速退化为层层嵌套的条件分支与全局变量传递。

理解链式调用的局限性，是理解为什么 LangGraph 采用“图结构”来建模系统的起点。

链式调用在工程化中的三大瓶颈

当业务复杂度提升后，线性链式调用会面临三项确定性的工程阻碍：

  flowchart TD
    subgraph PainPoint1 ["瓶颈 1: 条件分支膨胀"]
        A1["业务逻辑扩展"] --> B1["多层嵌套 if-else"] --> C1["难以追踪执行路径"]
    end

    subgraph PainPoint2 ["瓶颈 2: 循环与重试缺失"]
        A2["生成结果未通过校验"] --> B2["手写 while True 包装"] --> C2["中间数据容易污染或死锁"]
    end

    subgraph PainPoint3 ["瓶颈 3: 隐式状态管理"]
        A3["多节点跨步传递"] --> B3["散落的局部变量"] --> C3["缺乏统一快照与断点恢复能力"]
    end

1. 条件分支导致代码结构膨胀

现实任务往往不是单一路线。例如在客服工单处理中，需要根据第一步识别出的情绪与问题类型，分流到完全不同的处理管线：

# 链式调用中处理分支的常见形态
intent = classify_intent(user_input)

if intent == "refund":
    auth_status = check_user_auth(user_input)
    if auth_status == "passed":
        result = handle_refund(user_input)
    else:
        result = request_reauth(user_input)
elif intent == "complaint":
    sentiment = analyze_sentiment(user_input)
    if sentiment == "angry":
        result = escalate_to_human(user_input)
    else:
        result = auto_pacify(user_input)
else:
    result = general_faq(user_input)

当分支数量增加到五六个，或者分支之间存在交叉汇聚时，主逻辑被大量的控制流胶水代码淹没。每新增一个业务分支，都需要在复杂的嵌套层级中谨慎寻找修改位置，极易引发回归缺陷。

2. 无法原生表达循环、自纠与重试

智能体系统的核心特征在于自适应反馈（Feedback Loop）：如果生成的 SQL 语法错误，系统需要捕获数据库报错，并带着错误堆栈再次请求模型进行修复。

在链式调用中，循环必须借由外层的 while 循环、计数器和中断标志位来实现。这种命令式的循环控制让状态变异完全散落在各个局部作用域中，很难对每一次重试的输入输出进行独立的审计与链路追踪。

3. 状态传递依赖隐式变量，缺乏统一数据契约

在线性调用中，前几步产生的数据通常保存在局部变量（如 step1_res、auth_status）中。当流程延伸到第 7 步甚至第 10 步时，很难一眼看出“当前步骤到底依赖前序哪些步骤的输出”。一旦发生异常中断，内存中的局部变量随调用栈释放而蒸发，系统无法实现断点续跑。

LangGraph 的核心思想：状态机与图拓扑

针对上述问题，LangGraph 放弃了“流水线链条”模型，转向了经典的**有限状态机（Finite State Machine）与计算图（Computation Graph）**抽象：

  flowchart LR
    StateContainer[("统一状态容器 (State)<br/>声明式 TypedDict")]
    
    NodeA["处理节点 A (Node)<br/>读取 State, 返回增量更新"]
    NodeB["处理节点 B (Node)<br/>读取 State, 返回增量更新"]
    
    RouteEdge{"路由决策边 (Edge)<br/>判断走分支 1 还是分支 2"}
    
    NodeA --> RouteEdge
    RouteEdge -->|"条件匹配"| NodeB
    RouteEdge -->|"需自纠"| NodeA
    
    NodeA <-->|读写更新| StateContainer
    NodeB <-->|读写更新| StateContainer

在 LangGraph 中，业务系统由三个核心元语组成：

State（状态容器）：整张图的共享“单一事实源（Single Source of Truth）”。通常使用 Python 的 TypedDict 或 Pydantic 强类型声明。所有节点都从该状态读取上下文，并在执行完毕后将产出合并回状态中。
Node（节点）：独立的函数单元。每一个节点对应一个具体的原子操作（如调用模型、执行数据库查询、格式化文档）。节点的函数签名统一为 (state: State) -> dict，它只负责计算并返回需要变更的局部键值。
Edge（边）：定义节点之间的转移通道。边分为两类：
- 普通边（Edge）：固定顺序转移（节点 A 结束后无条件进入节点 B）；
- 条件边（Conditional Edge）：由一个专职的路由函数根据当前 State 决定下一步跳向哪个具体节点。

这种设计的本质是控制权解耦：节点只专注做业务数据转换，不需要知道下一个处理者是谁；路由逻辑被单独抽离到条件边中，整张图的执行拓扑可以在代码外层一览无余。

模式对比：链式调用与状态图的架构差异

评估维度	链式调用（Chain）	状态图（LangGraph）
拓扑结构	单向线性流水线	任意有向图（支持多分支、并发扇出、环形回路）
条件分支处理	嵌入在执行函数内部的命令式 `if-else`	声明式的 `add_conditional_edges`，路由规则独立抽离
循环与迭代	依靠外层命令式 `while` 循环硬编码	图中指向前序节点的反向边，自然形成自省回路
状态共享方式	局部变量隐式传递，容易发生依赖丢失	全局强类型 `State` 显式声明，字段契约一目了然
可观测性与调试	依赖控制台打点，无法直观还原拓扑	图结构支持编译为 Mermaid / PNG，状态支持流式（stream）单步捕获
扩展性	业务越复杂，主函数代码越冗长	新增功能只需定义新节点并增加一条边，存量节点无须改动

代码对比：同一业务场景的两种实现

假设实现一个用户工单初筛系统：系统先识别输入情绪，如果是负面情绪则路由给安抚节点，若是正常情绪则路由给常规解答节点。

方式 1：传统链式/过程式写法

# 逻辑与分支严重缠绕在主函数中
def handle_ticket_linear(user_text: str) -> str:
    # 步骤 1: 情绪分类
    sentiment = "negative" if any(w in user_text for w in ["差", "投诉", "慢"]) else "neutral"
    
    # 步骤 2: 条件分支写死在代码内
    if sentiment == "negative":
        return f"【加急通道】：已收到您的反馈，客服专员已介入处理您的问题：{user_text}"
    else:
        return f"【常规解答】：系统正在检索知识库以解答您的咨询：{user_text}"

在这段代码中，如果有第三方想要测试“加急通道生成文本”这单一环节，很难将逻辑解耦出来单独单测；后续如果想在加急处理失败后重试，代码复杂度将大幅上升。

方式 2：LangGraph 状态图写法

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

# 1. 显式定义数据契约
class TicketState(TypedDict):
    user_text: str
    sentiment: str
    final_reply: str

# 2. 定义高内聚节点（纯函数操作）
def classify_node(state: TicketState) -> dict:
    text = state["user_text"]
    sentiment = "negative" if any(w in text for w in ["差", "投诉", "慢"]) else "neutral"
    return {"sentiment": sentiment}

def urgent_node(state: TicketState) -> dict:
    return {"final_reply": f"【加急通道】：客服专员已介入：{state['user_text']}"}

def standard_node(state: TicketState) -> dict:
    return {"final_reply": f"【常规解答】：知识库检索中：{state['user_text']}"}

# 3. 独立路由决策函数
def route_by_sentiment(state: TicketState) -> str:
    return "urgent" if state["sentiment"] == "negative" else "standard"

# 4. 组装图拓扑
builder = StateGraph(TicketState)

builder.add_node("classify", classify_node)
builder.add_node("urgent", urgent_node)
builder.add_node("standard", standard_node)

builder.add_edge(START, "classify")
builder.add_conditional_edges(
    "classify",
    route_by_sentiment,
    {
        "urgent": "urgent",
        "standard": "standard"
    }
)
builder.add_edge("urgent", END)
builder.add_edge("standard", END)

app = builder.compile()

# 5. 执行调用
res = app.invoke({"user_text": "你们这系统响应太慢了！"})
print(res["final_reply"])