LangGraph 状态机 Agent 实战 2026：构建可控的复杂 AI 工作流

把 Agent 写成"循环里调模型"很快会失控：没法暂停等人审批、崩了从头再来、出问题不知道卡在哪一步。LangGraph 的答案是把 Agent 显式建成状态机——节点是步骤、边是转移规则、状态可持久化。本文用一个完整的中文实战例子（带人工审批的报销处理 Agent）走通核心 API。框架选型与概念对比见 LangChain vs LangGraph，英文完整指南见 LangGraph Complete Guide。

场景：报销审核 Agent

流程：收到报销申请 → AI 提取并校验 → 金额小于阈值自动通过 / 超过则暂停等人工审批 → 出结果。这个"中途等人"的需求，恰好覆盖 LangGraph 三大核心：状态、条件边、interrupt。

第一步：定义状态

python
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
class ExpenseState(TypedDict):
    raw_text: str            # 原始报销描述
    amount: float            # AI 提取的金额
    category: str            # 类目
    decision: Literal['approved', 'rejected', 'pending']
    reason: str

状态就是一个 TypedDict——图里每个节点读它、返回要更新的字段。所有节点共享这一份状态，这是和"链式传参"的本质区别。

第二步：写节点（普通函数）

python
import json
from openai import OpenAI
client = OpenAI()
def extract(state: ExpenseState):
    resp = client.chat.completions.create(
        model='gpt-4o-mini',
        response_format={'type': 'json_object'},
        messages=[{'role': 'user', 'content':
            f'从报销描述提取 JSON {{"amount": 数字, "category": "交通|餐饮|办公|其他"}}：{state["raw_text"]}'}],
    )
    data = json.loads(resp.choices[0].message.content)
    return {'amount': data['amount'], 'category': data['category']}
def auto_approve(state: ExpenseState):
    return {'decision': 'approved', 'reason': f'金额 {state["amount"]} 低于阈值，自动通过'}
def human_review(state: ExpenseState):
    from langgraph.types import interrupt
    # 图在这里暂停，把上下文抛给外部；恢复时拿到人的决定
    verdict = interrupt({'amount': state['amount'], 'category': state['category']})
    if verdict == 'approve':
        return {'decision': 'approved', 'reason': '人工审批通过'}
    return {'decision': 'rejected', 'reason': '人工审批拒绝'}

第三步：条件边 + 组图

python
def route(state: ExpenseState):
    return 'auto_approve' if state['amount'] < 500 else 'human_review'
builder = StateGraph(ExpenseState)
builder.add_node('extract', extract)
builder.add_node('auto_approve', auto_approve)
builder.add_node('human_review', human_review)
builder.add_edge(START, 'extract')
builder.add_conditional_edges('extract', route)   # 按金额分流
builder.add_edge('auto_approve', END)
builder.add_edge('human_review', END)

条件边就是普通 Python 函数返回下一个节点名——所有业务规则（阈值、分支、重试上限）都在可测试的代码里，而不是埋在提示词里求模型自觉。

第四步：持久化 + 运行 + 恢复

python
from langgraph.checkpoint.memory import MemorySaver   # 生产换 PostgresSaver
from langgraph.types import Command
graph = builder.compile(checkpointer=MemorySaver())
cfg = {'configurable': {'thread_id': 'expense-1024'}}
提交一笔大额报销 → 图跑到 human_review 处暂停
result = graph.invoke({'raw_text': '上周出差高铁加酒店共 2380 元'}, cfg)
result 里带 __interrupt__，包含抛出的上下文 → 推给审批人（IM/邮件/后台）
—— 可能是几小时后，另一个进程 ——
final = graph.invoke(Command(resume='approve'), cfg)   # 凭 thread_id 恢复
print(final['decision'], final['reason'])              # approved 人工审批通过

这一段是 LangGraph 的核心价值：暂停可以跨进程、跨天——状态在 checkpointer 里，恢复请求从任何地方来都行。自己用 while 循环写 Agent，这个能力要造一周轮子。

生产清单

checkpointer 用 PostgresSaver（内存版重启即失忆）；每个业务会话独立 thread_id

AI 提取的结构化输出要校验（金额为负？类目越界？）——schema 校验方案见 Zod vs Pydantic

节点级超时与重试：网络/模型抖动不该让整张图挂掉，失败节点单独重跑（状态还在）

接入追踪看每个节点的耗时与输入输出（LangSmith vs Langfuse），没有追踪的状态机出了问题没法定位

对外服务用 FastAPI 包一层 SSE 流式（实现配方）

FAQ

Q：什么时候不该用？ 流程是一条直线（提取→转换→输出）且不需要暂停/恢复时——普通函数顺序调用更简单。判断标准：运行时需不需要根据结果改路径或等外部输入。

Q：多 Agent 怎么做？ supervisor 也是一张图：路由节点决定派活给哪个子图,子图各自维护状态。和 CrewAI 等框架的取舍见多 Agent 框架对比。

Q：模型怎么选？ 提取/分类节点用 mini 档足够，涉及推理决策的节点再上旗舰——按节点配模型是状态机架构的隐藏福利（模型库）。

---

*最后更新：2026 年 6 月。API 以 LangGraph 官方文档为准。*