1. SecurityClaw 学习笔记/

03 LangGraph 工作流核心机制

结论先说:LangGraph 在这里解决的不是“多轮对话”,而是“可恢复的状态转移” #

如果你只把 LangGraph 理解成“把节点连起来的图工具”,那你会严重低估它在 Agent 系统里的作用。

在 SecurityClaw 里,LangGraph 真正解决的是:

  1. 如何把规划、执行、评估拆成不同职责节点
  2. 如何限制每个节点只能更新自己负责的 state 字段
  3. 如何把失败变成显式状态,而不是 try/except 后的一段模糊文本
  4. 如何让恢复、停止和 checkpoint 变成可审计逻辑

一、先看最小骨架:节点不是步骤,而是职责边界 #

workflow = StateGraph(State)
workflow.add_node("skills_check", check_skills)
workflow.add_node("think", think_node)
workflow.add_node("reflect", reflect_node)
workflow.add_node("direct_answer", direct_answer_node)
workflow.add_node("response_final", final_output_node)

把这段代码翻译成人话,就是:

  • skills_check:当前能力够不够
  • think:下一步计划是什么
  • reflect:执行结果说明了什么
  • direct_answer:是否可以直接回答
  • response_final:如何生成最终输出

这已经不是“流程顺序”了,而是“状态职责分工”。


二、显式状态机为什么比 while True 更适合 Agent #

很多最小 demo 的写法其实是:

while True:
    plan = llm(messages)
    result = call_tool(plan)
    if done(result):
        break
    messages.append(result)

这类写法的问题不是“不能跑”,而是状态和控制流全混在一起

  • 谁改了 state?
  • 失败后应该回到哪一步?
  • 哪个条件决定停止?
  • 恢复逻辑怎么插?

SecurityClaw 用图把这些隐式逻辑展开成显式转移:

flowchart TD
    start([start]) --> skills_check
    skills_check --> think
    think --> decide_next{decide_next}
    decide_next -- execute --> reflect
    decide_next -- answer --> direct_answer
    decide_next -- retry --> skills_check
    reflect --> evaluate{last step success?}
    evaluate -- no --> think
    evaluate -- yes --> response_final
    direct_answer --> response_final
    response_final --> end([end])

这张图真正值钱的地方 #

不是它好看,而是它明确回答了:

  • 失败后回哪
  • 回退是重新规划,还是直接停止
  • 哪个节点有权决定下一条边

三、decide_next 的意义:模型只能提议动作类型 #

workflow.add_conditional_edges("think", decide_next, {
    "execute": "reflect",
    "answer": "direct_answer",
    "retry": "skills_check"
})

这段代码说明,模型真正能决定的不是“下一步具体怎么执行”,而只是:

  • execute
  • answer
  • retry

这其实就是把模型的自由度收紧到了动作意图层

这比让模型直接路由整个系统好在哪 #

如果让模型随便输出“跳转到哪个节点”,那它就拥有了控制流权限。

而 SecurityClaw 的做法是:

模型只输出抽象动作,代码把抽象动作映射成真实边。

这样做的好处是:

  • 控制流不被 prompt 漂移带走
  • 恢复路径仍然是代码定义的
  • 可以独立测试路由逻辑

四、为什么每个节点只更新自己负责的 state 字段 #

这是 LangGraph 在 Agent 场景里的最大价值之一。

class AgentState(TypedDict):
    messages: list[Message]
    task_plan: Optional[Plan]
    execution_history: list[StepResult]
    skill_usage: dict[str, int]
    error_count: int
    final_output: Optional[str]

节点职责应该像这样分 #

think_node #

  • 更新 task_plan
  • 可能追加 messages

reflect_node #

  • 更新 execution_history
  • 更新 error_count

response_final #

  • 更新 final_output

这样做不是洁癖,而是为了避免三件坏事:

  1. 多个节点抢着写同一字段
  2. checkpoint 恢复时搞不清哪些字段才可信
  3. 测试里无法定位哪一步污染了状态

这一点可以压成一句话 #

图式编排的本质,不是多节点,而是把状态写权限切开。


五、失败为什么不能只靠异常处理 #

如果你把 Agent 的失败处理写成:

try:
    result = call_tool(...)
except Exception:
    result = "tool failed"

那你只是把失败藏进了一段字符串里。

SecurityClaw 的思路是:失败本身就是状态。

# 伪代码:失败状态显式回流
if not last.success:
    state["messages"].append(last.error)
    state["error_count"] += 1
    return "think"

这样做有两个关键收益:

  1. LLM 下一轮能看到失败证据
  2. 系统可以根据失败类型决定恢复而不是盲目重试

也就是说,LangGraph 在这里承载的不只是 happy path,而是失败路径的结构化表达


六、你提到的 max_retries,正好暴露了状态机设计里的 tradeoff #

这是你刚点出来的那个地方:

if task_plan.action == "retry" and error_count >= 3:
    return "answer"

这个判断的意义不是“模型错了”,而是“系统必须在这里收回停止权”。

为什么它是合理的 #

  • 防止无限 retry
  • 防止 token 和时间无限消耗
  • 防止系统在失败状态里自我催眠

为什么它又值得改进 #

你说得对,它完全可以配置化:

max_retries = config.get("max_retries", 3)
if task_plan.action == "retry" and error_count >= max_retries:
    return "answer"

这样更好的原因 #

  • 不同任务容错阈值不同
  • 查询类和写入类任务的恢复策略不同
  • 本地调试和线上运行的预算敏感度不同

所以这里的 tradeoff 很典型:

  • 写死阈值:稳定、简单、好测
  • 配置化阈值:灵活、真实、但复杂度更高

在 v1 阶段,SecurityClaw 先选了前者。我能理解。


七、为什么 checkpoint 和图拓扑必须分开理解 #

app = workflow.compile(checkpointer=SqliteSaver(conn))

这句很容易被误解成:“图结构和状态都保存到 SQLite 里了。”

其实不是。

Checkpoint 保存的是 #

  • 某一轮运行到哪里
  • 当前状态长什么样

图拓扑决定的是 #

  • 还能往哪走
  • 哪些节点存在
  • 哪些边有效

所以如果你改了:

  • add_node
  • 条件边
  • state 字段结构

旧 checkpoint 就可能恢复失败。

这个问题在图式编排里比 while 循环更明显,因为图拓扑本身就是运行时协议的一部分。


八、这一篇真正要带走的框架 #

你可以把 LangGraph 在 Agent 系统里的角色压成这四句话:

  1. 把控制流从 prompt 里拉回代码
  2. 把失败从异常字符串提升为显式状态
  3. 把节点职责按状态写权限切开
  4. 把恢复和停止做成可审计的边

这才是它真正比裸 while True 更有价值的地方。


九、验证命令 #

pytest tests/test_loop.py -v --log-cli-level=INFO

如果你要验证 checkpoint 相关行为:

sqlite3 data/conversations.db "SELECT * FROM checkpoints ORDER BY created_at DESC LIMIT 5;"
rm -f data/conversations.db