用 LangGraph 重写状态机并恢复中断

这是一次重写，不是新功能

先把定位说清楚，因为这一点很容易含糊过去：这一篇要用 LangGraph 替换掉第 3 篇的手写工具循环和第 7 篇的审批挂起逻辑。写完之后，agent/loop.py 那个函数会被删掉，最终版以图为准。

不说清楚的话，读者跟到这里会有一个合理的困惑：我该用哪份代码？两份都留着吗？答案是只留图。第 3 篇那个循环的价值在于它让你看清了 tool calling 的机制——那是理解成本最低的形式。但它不是最终形态。

手写循环到这里为止

第 3 篇的循环只有工具调用时非常清楚：

for step in range(MAX_STEPS):
    reply = await provider.chat(messages, tools=schemas)
    if not reply.tool_calls:
        return reply.content
    for call in reply.tool_calls:
        result = await execute(call, user)
        messages.append(tool_message(call.id, result))

十几行，一眼看得懂。但从第 11 篇到第 12 篇，这个循环里陆续要塞进：检索、证据充分性判断、拒答分支、字段收集的多轮状态、写工具的审批挂起、重试与降级。真写下来会变成这样：

for step in range(MAX_STEPS):
    if handoff_memory and handoff_memory['missing']:
        ...   # 字段收集分支
    elif not evidence and needs_retrieval(messages):
        ...   # 检索分支
    elif pending_approval:
        ...   # 从审批恢复的分支，但这里根本恢复不了
    reply = await provider.chat(...)
    for call in reply.tool_calls:
        if is_write(call):
            ...   # 挂起，然后呢？

问题不在于代码变长了。三件事同时坏掉了：

状态散落在局部变量里。 handoff_memory、evidence、pending_approval、step 全是这个协程栈上的变量。协程一结束，它们就没了。

没人能回答「这个会话现在在哪」。 排查一个卡住的会话，只能看日志猜它执行到了哪一行。

审批挂起没法真正实现。 这是最硬的那道墙。interrupt 在手写版本里意味着 return 一个 approval_required 响应——协程结束，栈销毁，messages 数组、已收集的字段、检索到的证据全部消失。用户点确认时，服务端只能从数据库里的审批记录重新拼出上下文，而重新拼出来的东西和原来那个不一定等价。

第 7 篇那一篇里，我把审批记录的字段列得很细（arguments、conversation_id、user_id），就是在用数据库手动补偿这件事。它能跑，但补偿的只有那次工具调用本身。整个会话的中间状态没有被保存。

LangGraph 解决的是这三件事。它不帮你写 prompt，不提升回答质量，不让检索变准——把它当成一个「带持久化的状态机运行时」，期待就对了。

State Schema：状态从变量变成数据

class AgentState(TypedDict, total=False):
    messages: Annotated[list[BaseMessage], add_messages]
    user_id: str
    evidence: list[Chunk]
    approval_id: str | None
    handoff: HandoffMemory | None
    step: int
    error: str | None

Annotated[..., add_messages] 是这里唯一需要额外理解的东西。它是一个 reducer：节点返回 {'messages': [新消息]} 时，框架把新消息追加到已有列表上，而不是替换。其他字段没有 reducer，默认是覆盖语义——evidence 每次检索都整个换掉，这是想要的行为。

搞混这两种语义会得到很隐蔽的 bug：给 messages 用覆盖语义，历史对话会在某个节点执行后突然只剩一条。

节点是纯函数，只读状态、返回增量：

async def retrieve(state: AgentState) -> dict:
    query = state['messages'][-1].content
    chunks = await retriever.hybrid(query, user_id=state['user_id'])
    return {'evidence': chunks}

async def decide(state: AgentState) -> dict:
    step = state.get('step', 0) + 1
    if not evidence_sufficient(state.get('evidence', [])):
        return {'handoff': start_or_continue_handoff(state), 'step': step}
    reply = await model_with_tools(state)
    return {'messages': [reply], 'step': step}

不要在节点里改全局变量或直接写数据库以外的状态。 节点返回什么，框架就把什么存进 checkpoint；偷偷改的东西不在 checkpoint 里，恢复之后就不见了。这条规则很容易在赶时间的时候破掉，代价是一个只在重启后才复现的 bug。

图：条件路由集中在一个节点

graph = StateGraph(AgentState)
graph.add_node('retrieve', retrieve)
graph.add_node('decide', decide)
graph.add_node('tool', run_read_tool)
graph.add_node('write_tool', run_write_tool)
graph.add_node('handoff', collect_fields)
graph.add_node('answer', compose_answer)

graph.add_edge(START, 'retrieve')
graph.add_edge('retrieve', 'decide')
graph.add_conditional_edges('decide', route, {
    'tool': 'tool', 'write': 'write_tool',
    'handoff': 'handoff', 'answer': 'answer',
})
graph.add_edge('tool', 'decide')
graph.add_edge('handoff', 'decide')
graph.add_edge('write_tool', 'decide')
graph.add_edge('answer', END)

def route(state: AgentState) -> str:
    if state.get('step', 0) >= MAX_STEPS:
        return 'answer'                       # 步数上限仍然是硬约束
    if state.get('handoff', {}).get('missing'):
        return 'handoff'
    last = state['messages'][-1]
    if not getattr(last, 'tool_calls', None):
        return 'answer'
    return 'write' if is_write(last.tool_calls[0]) else 'tool'

Preview

节点、条件边与中断点

图里三条回边都指向 decide。这是刻意的：所有路由决策集中在一个函数里，图就还能读；把条件散到各个节点上（tool 自己决定下一步去哪），很快就没人能画出这张图了。

第 3 篇那张工具循环图和这张图是同一套逻辑。区别在于，那张图画的是代码的形状——你得对着代码想象它；这张图就是代码，graph.add_conditional_edges 那几行和图上的箭头一一对应。

MAX_STEPS 的检查放在 route 里而不是某个节点里。它是路由决策，不是某一步的业务逻辑。放错位置的话，超步数时会先执行一次工具调用才发现该停了。

interrupt：中断即落盘

from langgraph.types import interrupt

async def run_write_tool(state: AgentState) -> dict:
    call = state['messages'][-1].tool_calls[0]
    approval = await approvals.create_from_call(call, state['user_id'])
    decision = interrupt({
        'approval_id': str(approval.id),
        'summary': summarize(call),
        'arguments': call.args,
    })
    if decision != 'approved':
        return {'messages': [tool_message(call.id, {'ok': False, 'error': 'rejected'})]}
    result = await execute_write(call, state['user_id'])   # 幂等键仍在这层
    return {'messages': [tool_message(call.id, result)]}

interrupt() 做两件事：把当前 state 写进 checkpoint，然后抛出一个可序列化的中断信息。API 层把它转成第 7 篇那个 approval_required 的 SSE 事件——响应格式不变，前端不需要因为后端换了实现而改代码。

interrupt 的返回值是恢复时传进来的那个值。这个写法读起来像同步阻塞，实际上函数在这里退出了，恢复时会从头重新执行这个节点，直到再次遇到 interrupt 时拿到 resume 值继续往下。

这个「重新执行」的语义非常重要，下一节专门说它。

恢复接口：

@router.post('/approve/{approval_id}')
async def approve(approval_id: UUID, user: User = Depends(current_user)):
    approval = await approvals.lock_for_update(approval_id)
    if not approval or approval.user_id != user.id:
        raise HTTPException(404, 'approval_not_found')
    if approval.status != 'pending' or approval.expires_at < now():
        raise HTTPException(409, 'approval_expired')
    await approvals.mark(approval.id, 'approved')
    config = {'configurable': {'thread_id': approval.conversation_id}}
    return await app.ainvoke(Command(resume='approved'), config)

第 7 篇的所有校验一条都不能省：重新读用户身份、比较 approval.user_id、检查状态和过期、行锁防并发双击。框架管的是图状态，不管你的业务权限——thread_id 只要猜对就能 resume，权限校验是你自己的事。

checkpointer：用官方的，不自建表

from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver

async with AsyncPostgresSaver.from_conn_string(settings.database_url) as checkpointer:
    await checkpointer.setup()
    app = graph.compile(checkpointer=checkpointer)

thread_id 直接用 conversation_id，一个会话一条时间线。

这里有个坑必须提前处理。checkpointer.setup() 会自建几张表（checkpoints、checkpoint_writes、checkpoint_blobs 等），而这些表不在你的 Alembic 模型里。下次 alembic revision --autogenerate 时，Alembic 发现数据库里有模型里没有的表，会生成 op.drop_table('checkpoints')。有人不看 diff 直接 apply，所有挂起中的会话就没了。

在 env.py 里排除它们：

CHECKPOINT_TABLES = {'checkpoints', 'checkpoint_writes',
                     'checkpoint_blobs', 'checkpoint_migrations'}

def include_object(obj, name, type_, reflected, compare_to):
    if type_ == 'table' and name in CHECKPOINT_TABLES:
        return False
    return True

一句话原则：checkpoint 表的 schema 归框架管，业务表归 Alembic 管，两边不要交叉。框架升级时它自己迁移这些表，你只要别去动它。

备份策略也要分开写进文档。业务表需要 PITR；checkpoint 表体积会长得很快（每个节点执行都写一次），而且过期会话的检查点没有保留价值，应该配一个定期清理——保留最近 7 天，或者只保留 approvals 表里还有 pending 记录的那些 thread_id。

重启恢复实验

这是这一篇唯一需要亲手跑一遍的验证。步骤：

# 1. 发起一个需要审批的请求，记下 approval_id 和 conversation_id
curl -N -X POST localhost:8000/chat -d '{"message":"帮我建个工单：VPN 连不上"}'
# -> approval_required, approval_id=a_456, conversation_id=c_123

# 2. 确认 checkpoint 落库了
psql -c "select thread_id, checkpoint_ns from checkpoints where thread_id='c_123'"

# 3. 杀掉进程，重新起一个
docker compose restart api

# 4. 在新进程上确认
curl -X POST localhost:8000/approve/a_456 -H "Authorization: Bearer $TOKEN"

Preview

审批等待期跨过一次滚动发布

预期结果：工单只创建一张，回答里保留了重启前的完整对话，conversation_id 不变。

如果恢复后模型被重复调用了，说明你踩到了上一节说的「节点会重新执行」。interrupt 之前的代码会跑第二遍——上面那段 run_write_tool 里，approvals.create_from_call 就会执行两次，产生一条多余的审批记录。

处理方式有两种。一是把副作用挪到 interrupt 之后，节点前半段保持纯粹；二是让那个副作用本身幂等。第二种更稳，而且工具已经这样做了——第 7 篇的幂等键在这里第二次发挥作用：即使 execute_write 因为某种原因跑了两遍，唯一索引保证只成功一次。

这就是为什么第 7 篇要先建幂等机制再引入框架。反过来的顺序会让你在这里遇到一个很难归因的重复写入。

什么时候不该上 LangGraph

只有一个模型调用的问答。 用第 2 篇的 Provider 直接调，不需要图。为一次 chat() 建一个状态机是纯粹的开销。
没有需要跨请求存活的状态。 图的核心价值是 checkpoint。全部逻辑都在一个请求内跑完的话，你付了框架的成本却没拿到它的收益。
调试成本敏感的场景。 手写循环出错时堆栈直接指向那一行；图出错时你要先搞清它停在哪个节点、state 里有什么、哪条条件边被选中了。这个成本是真实的，团队里第一次接触 LangGraph 的人要花几天才能顺畅排查。

迁移完成后，保留第 3 篇和第 7 篇的测试。它们现在测的是节点级行为（工具选择、schema 校验、幂等、审批权限），而不是那个已经被删掉的循环。这批用例是判断「重写有没有改坏东西」的唯一依据——重写最常见的失败模式不是新代码写错了，而是旧代码里某个不起眼的校验被漏掉了。

本篇验收标准

graph.get_graph().draw_mermaid() 能输出图结构，节点和条件边与文中一致。
审批挂起期间重启服务，resume 后不丢消息、不重复写入、conversation_id 不变。
checkpoint 用官方 Postgres saver；alembic revision --autogenerate 生成的迁移里没有任何 checkpoint 表。
第 3、7、10 篇的关键评测用例在新编排下全部通过，分数不低于重写前的基线。
节点重新执行时不产生多余的审批记录或重复工单。
有一条清理任务处理过期会话的检查点，且不会误删仍在 pending 的会话。

后端到这里就完整了：能查、能写、能检索、能引用、能拒答、能审批、能恢复。下一篇进浏览器，把这些能力做成一个用户真的能用的界面——难点集中在 SSE 的增量解析和断线重连。

#用 LangGraph 重写状态机并恢复中断

#这是一次重写，不是新功能

#手写循环到这里为止

#State Schema：状态从变量变成数据

#图：条件路由集中在一个节点

#interrupt：中断即落盘

#checkpointer：用官方的，不自建表

#重启恢复实验

#什么时候不该上 LangGraph

#本篇验收标准