第 3 课：Agent 多步循环 + 审批 —— 把"对话"升级成"行动者"

本节目标：理解 Agent 为什么不是 ChatGPT、ReAct 循环怎么运作、HIGH 风险工具的审批暂停-恢复机制、以及"穷人版 Checkpointer"如何让 Agent 状态跨请求持久化。

这一课你会真正理解 Agent 和普通 ChatBot 的本质区别。

1. 单轮 Tool Calling 的天花板

第1课讲了单轮工具调用——LLM 决策一次、工具执行一次、LLM 总结一次。两行核心代码就能跑。

但 3 个真实需求会立刻撞墙：

需求	为什么单轮做不到
复合查询："查张三余额，不到100就发充值提醒"	第2步判断依赖第1步结果，需要多步串联
自我纠错："查不到时换个工具再试"	单轮看到 ERROR 只能告诉用户"失败了"
HIGH 风险审批："删文档前等人批准"	单轮要么直接删（危险）要么直接拒绝（傻）

核心认知：单轮 Tool Calling 是一次性决策，Agent 是持续循环 + 状态演进。

2. ReAct：让 LLM "边走边想"

ReAct = Reason + Act。模式很简单：

code

Loop:
  1. Thought: 我现在该做什么？
  2. Action: 调用 some_tool(args)
  3. Observation: 工具返回了什么？
  4. 回到 1，直到 LLM 说"我能回答了"

每一步 LLM 都看到全部历史，所以能：基于上一步结果决定下一步、看到失败后换工具重试、多步组合形成复合能力。

核心实现——agent_node：

python

async def agent_node(state: AgentGraphState) -> dict[str, Any]:
    sp = list(state.get("scratchpad") or [])   # 读草稿纸
    step = int(state.get("step_count", 0)) + 1

    # LLM 看到全部历史，输出下一步决策
    messages = prompt_manager.get("agent_react").render(
        tools_desc=_tools_desc(),
        scratchpad_text=_scratchpad_text(sp),   # 关键：每一轮都注入全部历史
        question=state.get("question", ""),
    )
    result = await llm_service.chat_json(messages)

关键设计：scratchpad 是 ReAct 的"草稿纸"——记录每轮的 thought / action / observation。LLM 每次只输出下一步。要不要继续循环？那是路由的事，不是节点的事。

3. 课程版的 Agent 架构：4 个文件

Code

agents/
├── state.py          ← 草稿纸结构（AgentGraphState）
├── nodes.py          ← agent_node + halt_node + 路由
├── agent_graph.py    ← LangGraph StateGraph 构图
└── loop_detection.py ← 死循环检测

拓扑只有 2 个节点：

Architecture Flow

START → agent ── route_after_agent ──┬─ continue → agent（循环！）
                                      ├─ done → END
                                      ├─ max_steps → halt → END
                                      ├─ loop → halt → END
                                      ├─ approval_required → halt → END
                                      └─ error → halt → END

构图代码 18 行：

python

def build_agent_graph():
    builder = StateGraph(AgentGraphState)
    builder.add_node("agent", agent_node)
    builder.add_node("halt", halt_node)

    builder.add_edge(START, "agent")
    builder.add_conditional_edges(
        "agent", route_after_agent,
        {
            "continue": "agent",        # ← 自反边：循环的核心！
            "done": END,
            "max_steps": "halt",
            "loop": "halt",
            "approval_required": "halt",
            "error": "halt",
        },
    )
    builder.add_edge("halt", END)
    return builder.compile()

"continue": "agent" 是自反边——LangGraph 看到这条会反复执行 agent_node，直到路由返回其他值。

4. agent_node 的 6 条分支（核心逻辑）

每轮决策要处理 6 种情况：

分支	触发条件	写什么到 scratchpad	是否暂停
LLM 调用失败	网络、超时	error observation	halt: error
输出格式错	JSON 解析失败	parse error	halt: error
LLM 宣告 done	`is_done=true`	thought + final	halt: done
工具不存在/RBAC 拒	未注册/无权限	observation 提示	禁止：继续（让 LLM 换）
HIGH 工具	`risk == HIGH`	写审批 + 标记	halt: approval_required
LOW/MEDIUM 工具	普通工具	observation = 结果	禁止：继续

HIGH 工具分支是最关键的：

python

if risk == RiskLevel.HIGH:
    # 1. 写审批表
    approval = approvals_db.create_pending(
        tool_name=tool_name, tool_args=tool_args,
        risk=risk, requested_by=state.get("user_id"),
    )
    # 2. scratchpad 留标记（还没真执行）
    sp.append({
        "action": {"name": tool_name, "args": tool_args},
        "observation": f"[待审批] approval_id={approval.id}",
        "approval_id": approval.id,
    })
    # 3. 声明 halt，路由导向暂停
    return {"scratchpad": sp, "halt_reason": "approval_required",
            "pending_approval_id": approval.id}

注意：注意：HIGH 检测在 agent_node，不在 tool_registry.execute()。这正是第2课的原则——ToolRegistry 只记录风险等级，Agent 层决定策略。

5. 路由：循环的"指挥棒"

Code

def route_after_agent(state) -> Literal[...]:
    reason = state.get("halt_reason")
    if reason in ("error", "approval_required"):
        return reason              # 优先级最高：节点显式声明终止
    if state.get("is_done"):
        return "done"              # 成功优先于防御性中止
    if state.get("step_count") >= state.get("max_steps", 6):
        return "max_steps"         # 防失控
    if detect_loop(state.get("scratchpad")):
        return "loop"              # 防死循环
    return "continue"              # 以上都不是 → 再来一轮

优先级为什么这样排：is_done 必须在 max_steps 之前——否则 Agent 刚好在第6步算出答案，却被判"超步"。成功路径永远优先于防御性中止。

6. "穷人版 Checkpointer"：状态跨请求存活

这是 Ch14 最妙的设计。

问题：审批可能等 5 分钟到 5 天，不能让 Agent 占着协程等。必须把状态序列化到磁盘。

方案：整个 run 的状态存 SQLite 一行，3 段代码搞定。

第 1 段：start —— 创建 run + 跑图到暂停

python

async def start(self, question, *, user_id, tenant_id, roles):
    run_id = runs_db.create(question, max_steps, user_id, tenant_id, roles)
    return await self._run_until_halt(run_id)

第 2 段：_run_until_halt —— 跑图 + 回写状态

python

async def _run_until_halt(self, run_id):
    run = runs_db.get(run_id)           # 从 DB 读
    init = {
        "scratchpad": [s.model_dump() for s in run.scratchpad],  # 反序列化
        "question": run.question, ...
    }
    final = await graph.ainvoke(init)   # 跑图

    # 根据结果决定状态
    if final.get("is_done"):
        run.status = "done"
    elif reason == "approval_required":
        run.status = "paused"           # ← 等审批
    elif reason in ("max_steps", "loop"):
        run.status = "done"             # 兜底 answer
    else:
        run.status = "failed"

    runs_db.save(run)                   # 落盘！

scratchpad 是状态本体。JSON 存 DB，每次跑图时反序列化重新加载——这就是 Checkpointer 的本质。

第 3 段：resume —— 审批后从断点续跑

python

async def resume(self, run_id):
    run = runs_db.get(run_id)
    approval = approval_service.get(run.pending_approval_id)

    last = run.scratchpad[-1]
    tool_name = last.action.get("name")
    orig_args = last.action.get("args")

    # 根据审批结论生成 observation
    if approval.status == APPROVED:
        obs = self._execute_tool_safe(tool_name, orig_args)
    elif approval.status == EDITED:
        obs = self._execute_tool_safe(tool_name, approval.edited_args)
        last.action = {"name": tool_name, "args": approval.edited_args}  # 审计一致性！
    elif approval.status == REJECTED:
        obs = f"[审批被拒绝] reason={approval.reason}"
    elif approval.status == EXPIRED:
        obs = "[审批超时，视为拒绝]"

    last.observation = obs       # 把结果补回 scratchpad
    return await self._run_until_halt(run_id)  # 接着跑！

精髓：Agent 不"等"审批。暂停时彻底释放协程，状态全落 DB。审批回来后从 DB 完整重建状态继续。这跟 Web 应用的无状态思想一样——可 horizontal scale。

7. 完整请求的"长征"：一次需要审批的删除操作

Code

POST /agent/ask  { "question": "删除文档 doc_42" }
  ↓
agent_node Round 1: LLM 决策调 delete_document → 发现 risk=HIGH
  → 写审批表 → scratchpad 加"待审批" → halt_reason=approval_required
  ↓
_run_until_halt: run.status = "paused", runs_db.save(run)
  ↓
返回 { "status": "paused", "pending_approval_id": "..." }
  ↓
 Agent 彻底睡着——零协程占用，全部状态在 SQLite

—————— 5 分钟后，管理员审批 ——————

POST /approvals/{id}/approve
  ↓ 批准
POST /agent/runs/{run_id}/resume
  ↓
resume(): 读审批结论 → APPROVED → 执行 delete_document → observation 回填 scratchpad
  → _run_until_halt(run_id) 接着跑
  ↓
agent_node Round 2: LLM 看到 scratchpad 里上一步已成功
  → is_done=true, final_answer="文档 doc_42 已删除"
  ↓
返回 { "status": "done", "answer": "文档 doc_42 已删除" }

跨 3 个 HTTP 请求、至少 2 个人、可能跨几天，状态始终完整存活。

8. 5 条工程心法

节点只走一步，路由决定循环——agent_node 不写 while True，循环靠图的自反边。每一步可独立测试
scratchpad 是状态的唯一来源——能被序列化、反序列化、跨机器复制。Agent 的"记忆"就是这段 JSON
所有终止收敛到 halt 节点——max_steps/loop/approval/error/done 都走 halt，加监控改文案只改一处
HIGH 工具是执行前阻塞，不是执行后审计——agent_node 只写审批表，resume 才真正 execute
持久化全部状态，永远别假设服务不重启——哪怕审批 5 秒回来也走 DB。明天重启/扩容/换 Python 版本都不影响

9. 自测：5 个问题

问题 1：agent_node 在 LLM JSON 解析失败时为什么不让它重试，而是直接走 halt:error？

问题 2：scratchpad 只记 observation 不记 thought，会出现什么问题？（4 个连锁后果）

问题 3：EDITED 审批时为什么必须把 scratchpad 的 action.args 改成批准后的值？

问题 4：route_after_agent 里 is_done 优先级为什么高于 max_steps？反过来会怎样？

问题 5（最重要）：审批超时 1 小时自动作废，在哪一层加？为什么不能在 agent_node 里做？

答案与解析

问题 1：JSON 解析失败为什么不让 LLM 重试

3 层原因：

节点承诺：agent_node 只走一步。节点内重试破坏原子语义 + 时长不可控
重试已在更合适的层：llm_service.chat_json 内部已重试过 JSON 格式错误（把错误回注 prompt 让 LLM 自纠）。agent_node 拿到 ok=false 意味着重试已失败
失败必须可见：悄悄重试 → 监控看不到、审计追不到。fail-fast → halt_reason 立刻记录，报警能触发

重试在"恰当的层"做：越底层越自动、越上层越克制。LLM 失败不可怕，失败时沉默不可观察才可怕。

问题 2：不记 thought 的 4 个连锁后果

后果	说明
LLM 重复犯错	看不到上轮"为什么调这个工具"，可能做一模一样的无用决策
调试不可能	只知道调了什么工具，不知道为什么调。线上出问题永远查不出根因
审计黑洞	监管问"为什么删了用户文档"，没 thought = 只能回答"LLM 决定的"
循环检测失灵	两次语义不同的 `search_knowledge_base` 调用，没 thought 会被误判成循环

深层认知：Agent 可观测性 ≠ "它做了什么"，而是"它为什么这么做"。thought 是 LLM 的动机日志。

问题 3：EDITED 为什么必须改 args

python

# LLM 原话：delete_document(doc_id="doc_42")
# 审批人改成：delete_document(doc_id="doc_43") 后批准
last.action = {"name": tool_name, "args": new_args}  # 关键行

不改的 3 个后果：

scratchpad 自相矛盾：action 记录删 doc_42，observation 记录删了 doc_43——下一轮 LLM 困惑
LLM 推理基于错误前提：以为 doc_42 已删（实际没删），后续整个推理链全错
审计追责死循环："谁下令删了 doc_43？"——LLM 没说过、审批表也没明确记录真实参数

通用原则：任何"提议→修改→执行"流程里，落盘状态必须反映真实执行。原始提议另存，主字段必须是真相。

问题 4：is_done 优先级为什么高于 max_steps

假设 max_steps=6，LLM 刚好在第 6 轮给出 final_answer。如果先判 max_steps：

Agent 明明算出了答案，却被判"已达最大步数"——用户体验灾难（"算出来了但显示超时"）
花了 6 轮 token 结果被丢弃，用户只能重问，又烧 6 轮
max_steps 监控被夸大，失败率虚高

通用原则："成功路径"永远优先于"防御性中止"。先确认成果，再判断兜底。

问题 5：审批超时在哪一层加

答案：不在 agent_node，在后台定时任务。

因为 Agent 暂停后零协程运行——agent_node 根本没有机会执行。超时靠外部触发器：

Code

Cron (每5分钟) → approval_service.expire_overdue()
  → PENDING 超时审批 → EXPIRED
  → 下次 resume() 看到 EXPIRED → obs="[审批超时，视为拒绝]"
  → LLM 基于这条 observation 决定换路/报错

进阶：定时任务还可以自动调 resume() 让 paused 的 run 走完（但要限流防打爆 LLM API）。

通用原则：不要让"睡着"的进程自己醒来。睡着的状态必须靠外部触发器唤醒——这是分布式系统的"事件外部性"原则。

本节要点

Agent = LLM + 工具 + 循环 + 状态。缺一环都不叫 Agent
循环不靠 while True，靠图的自反边 + 路由——每一步可测、可观察、可中断
scratchpad 是状态本体，JSON 存 DB，跨请求完整存活
HIGH 工具暂停-审批-恢复是 Agent 区别于 ChatBot 的灵魂——把 LLM 从"建议者"变成"可控的行动者"
"成功路径"优先于"防御性中止"；外部触发器唤醒"睡着"的状态——两条铁律适用所有分布式系统