1. SecurityClaw 学习笔记/

05 用 LangGraph 构建可控的多步 Agent

结论先说:多步 Agent 的难点不是“步骤多”,而是“循环分几层” #

很多人第一次写 Agent,会把所有失败处理都塞进一个大循环里。但 SecurityClaw 里最有价值的设计,是把循环拆成了两层:

  1. 内层局部重试:处理一次技能执行里的参数解析、超时、局部失败
  2. 外层图式编排:处理整个 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 只负责产出 plan
  • execute_node 只负责执行 skill
  • evaluate_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"

如果没有恢复节点,系统会出现两种糟糕行为:

  1. 盲目重试:同一个技能执行三次,结果毫无变化
  2. 过早失败:其实换一个工具就行,却直接告诉用户“做不到”

恢复节点本质上是把“失败”从异常变成结构化输入。


六、边界条件:什么时候必须强制停止 #

任何多步 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"

这三类条件覆盖了:

  • 防无限循环
  • 防成本失控
  • 防已经足够却继续过度执行

七、测试应该怎么证明这套循环是可靠的 #

真正该测的不是“最终答案对不对”,而是:

  1. 局部重试是否只影响当前 action
  2. 外层恢复是否真的改变了下一轮路径
  3. 达到最大轮数时是否会停止
  4. 超时、解析失败、空结果这三类失败是否被区分
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