05 用 LangGraph 构建可控的多步 Agent
结论先说:多步 Agent 的难点不是“步骤多”,而是“循环分几层” #
很多人第一次写 Agent,会把所有失败处理都塞进一个大循环里。但 SecurityClaw 里最有价值的设计,是把循环拆成了两层:
- 内层局部重试:处理一次技能执行里的参数解析、超时、局部失败
- 外层图式编排:处理整个 Agent 是否需要重新规划、恢复、停止
如果不把这两层分开,Agent 很快就会陷入一种很糟糕的状态:你不知道自己是在“重试一个工具”,还是在“重跑整个计划”。
一、先区分两种循环 #
1.1 内层循环:一次执行里的局部重试 #
MAX_RETRIES = 3
for attempt in range(MAX_RETRIES):
try:
result = skill.run(parsed_params)
break
except (ParseError, SkillTimeout) as e:
if attempt == MAX_RETRIES - 1:
return {"execution_error": str(e)}
# 只重试这一步,不重新规划整个任务
这类循环解决的是:
- 参数格式不对
- 单次技能调用超时
- 某个局部解析失败
它的单位是一次动作。
1.2 外层循环:整个 Agent 的规划-执行-评估 #
# 伪代码:图式编排的主循环
while not done(state):
plan = think(state)
step_result = execute(plan)
state = evaluate_and_update(state, step_result)
这层循环解决的是:
- 当前计划是否还成立
- 结果是否足够回答问题
- 是否要进入恢复路径
- 是否应该直接停止
它的单位是整个任务推进。
二、为什么 LangGraph 适合做“外层循环” #
LangGraph 在这里并不是为了画图,而是为了把“状态转移条件”显式化。
flowchart TD
Input[输入问题] --> Decide[Decide / Planning]
Decide --> Execute[Execute Skill]
Execute --> Evaluate{结果是否可接受?}
Evaluate -->|是| End[Format Response]
Evaluate -->|否| Recover{是否需要恢复?}
Recover -->|是| Decide
Recover -->|否| Stop[强制停止]
这个图真正带来的好处有三点:
2.1 状态转移可审计 #
你可以明确知道:
- 是哪一个节点决定进入恢复
- 是哪一个条件触发停止
- 是哪一个字段影响了下一步路由
2.2 每个节点职责更纯 #
例如:
decide_node只负责产出 planexecute_node只负责执行 skillevaluate_node只负责判断证据是否足够
2.3 中断点天然存在 #
安全场景常常需要:
- 高风险动作前暂停
- 人工确认后继续
- checkpoint 后恢复
如果是普通 while,这些点都要你自己插;LangGraph 天然把这些边界暴露出来。
三、这篇最重要的模型:多步 Agent 其实是“局部循环 + 全局循环”的组合 #
把这件事抽象成更准确的伪代码:
# 伪代码:多步 Agent 的两层控制
for graph_round in range(MAX_GRAPH_ROUNDS):
plan = planner.make_plan(state)
for action in plan.actions:
result = execute_with_local_retry(action, max_retries=3)
state = record_result(state, result)
if result.is_fatal:
break
decision = evaluator.decide(state)
if decision == "stop":
return final_response(state)
if decision == "recover":
state = recover(state)
continue
你会发现,这才是“真正可运行”的 Agent 模型。
不是一个循环,而是:
- 动作级重试
- 任务级再规划
- 系统级停止条件
四、为什么不把重试都放在外层 #
这是一个很容易做错的设计。
方案 A:任何失败都直接回到规划节点 #
优点:逻辑统一 缺点:
- 一个小的格式错误也要重新调用 LLM
- token 浪费大
- 容易把短暂故障放大成全局重规划
方案 B:局部问题局部重试,结构性问题再回到外层图 #
优点:
- 便宜
- 失败隔离更清楚
- 不会把所有问题都推给 LLM 缺点:
- 系统设计更复杂
- 需要明确定义“什么是局部失败,什么是结构性失败”
SecurityClaw 选的是 B。这是对的。因为在安全场景里,很多失败根本不是“重新规划”能解决的,而只是:
- 参数少一个字段
- API 临时超时
- 单个技能返回格式不合法
这种时候立刻整轮回退,只会放大问题。
五、恢复路径为什么必须单独建模 #
恢复不是“再试一次”。恢复是:
- 判断为什么失败
- 选择新的策略
- 把失败作为证据输入下一轮
# 伪代码:恢复节点
if evaluation.reason == "empty_result":
next_action = "expand_search_scope"
elif evaluation.reason == "permission_denied":
next_action = "request_human_confirmation"
elif evaluation.reason == "timeout":
next_action = "fallback_tool"
如果没有恢复节点,系统会出现两种糟糕行为:
- 盲目重试:同一个技能执行三次,结果毫无变化
- 过早失败:其实换一个工具就行,却直接告诉用户“做不到”
恢复节点本质上是把“失败”从异常变成结构化输入。
六、边界条件:什么时候必须强制停止 #
任何多步 Agent 都要有硬边界。不然系统迟早进入无穷重试。
至少要有这三类停止条件 #
6.1 轮数上限 #
if state["graph_round"] >= max_rounds:
return "stop"
6.2 预算上限 #
if budget.remaining <= 0:
return "stop"
6.3 证据充分 #
if evaluation.evidence_sufficient:
return "stop"
这三类条件覆盖了:
- 防无限循环
- 防成本失控
- 防已经足够却继续过度执行
七、测试应该怎么证明这套循环是可靠的 #
真正该测的不是“最终答案对不对”,而是:
- 局部重试是否只影响当前 action
- 外层恢复是否真的改变了下一轮路径
- 达到最大轮数时是否会停止
- 超时、解析失败、空结果这三类失败是否被区分
def test_execute_timeout_triggers_local_retry():
...
def test_empty_result_enters_recover_path():
...
def test_max_rounds_forces_stop():
...
这些测试的价值比展示一个“答对了的问题”更高,因为它们证明了这套图在坏路径下仍然可控。
八、这一篇真正要你记住的框架 #
多步 Agent 的本质不是“让模型多想几轮”,而是把失败分层处理:局部失败局部重试,结构性失败进入恢复,系统级边界负责停止。
如果你只记住一个图,就记这个:
动作级重试
↓
任务级再规划
↓
系统级停止条件
这就是“能跑通”和“能长期运行”的区别。
九、验证命令 #
pytest tests/test_runner.py tests/test_scheduler.py -v
如果项目有独立的 evaluate / recover 测试,再补:
pytest tests/ -k "retry or recover or max_rounds" -v