03 LangGraph 工作流核心机制
结论先说:LangGraph 在这里解决的不是“多轮对话”,而是“可恢复的状态转移” #
如果你只把 LangGraph 理解成“把节点连起来的图工具”,那你会严重低估它在 Agent 系统里的作用。
在 SecurityClaw 里,LangGraph 真正解决的是:
- 如何把规划、执行、评估拆成不同职责节点
- 如何限制每个节点只能更新自己负责的 state 字段
- 如何把失败变成显式状态,而不是 try/except 后的一段模糊文本
- 如何让恢复、停止和 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"
})
这段代码说明,模型真正能决定的不是“下一步具体怎么执行”,而只是:
executeanswerretry
这其实就是把模型的自由度收紧到了动作意图层。
这比让模型直接路由整个系统好在哪 #
如果让模型随便输出“跳转到哪个节点”,那它就拥有了控制流权限。
而 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
这样做不是洁癖,而是为了避免三件坏事:
- 多个节点抢着写同一字段
- checkpoint 恢复时搞不清哪些字段才可信
- 测试里无法定位哪一步污染了状态
这一点可以压成一句话 #
图式编排的本质,不是多节点,而是把状态写权限切开。
五、失败为什么不能只靠异常处理 #
如果你把 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"
这样做有两个关键收益:
- LLM 下一轮能看到失败证据
- 系统可以根据失败类型决定恢复而不是盲目重试
也就是说,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 系统里的角色压成这四句话:
- 把控制流从 prompt 里拉回代码
- 把失败从异常字符串提升为显式状态
- 把节点职责按状态写权限切开
- 把恢复和停止做成可审计的边
这才是它真正比裸 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