状态机与Agent规划
核对日期:2026-05-09。
1. 定义与边界
状态机与 Agent 规划是指用显式状态、转移条件和动作边界约束 Agent 的规划与执行。Agent 可以在某些节点内做语言理解、候选方案生成或工具选择,但节点转移由状态机控制。
它不是把 Agent 变成纯规则系统。正确做法是:确定性规则控制边界,Agent 处理开放子问题。
2. 为什么重要
生产业务常常需要可验证流程:申请、审核、执行、通知、归档。纯 Agent 很难保证每次都按合规流程走。状态机让系统知道“当前处于哪个阶段、允许做什么、下一步条件是什么”。
3. 核心机制
状态对象:
{
"state": "RiskCheck",
"allowed_actions": ["classify_risk", "ask_clarification"],
"facts": {"amount": 120, "order_id": "o1"},
"guards": ["amount <= user_limit", "order_exists"],
"next_candidates": ["Approval", "Execute"]
}
4. 架构模式
| 模式 | Agent 角色 | 适用 |
|---|---|---|
| Agent inside node | 节点内调用 Agent 产出结构化结果 | 字段抽取、摘要、分类 |
| Agent proposes transition | Agent 建议下一状态,代码校验 | 半结构化流程 |
| State machine as guardrail | 每次工具调用前检查状态 | 高风险操作 |
| Graph workflow | 节点和边组成可恢复图 | LangGraph 类应用 |
5. 工程实现
def step_machine(ctx):
state = load_state(ctx.run_id)
if state.name == "Intake":
facts = agent_extract(ctx.input)
if missing_required(facts):
return transition("Clarify", facts=facts)
return transition("RiskCheck", facts=facts)
if state.name == "RiskCheck":
risk = evaluate_rules(state.facts)
return transition("Approval" if risk.high else "Execute", risk=risk)
raise InvalidState(state.name)
转移配置:
states:
Execute:
allowed_tools: [commit_refund, send_notification]
guards:
- approval_exists_if_high_risk
- idempotency_key_present
on_success: Done
on_failure: Failed
6. 生产实践
- 状态机配置版本化,状态迁移写入审计日志。
- Agent 输出只作为转移输入,不直接改状态。
- 每个状态声明 allowed tools、guards、timeouts。
- 对 pending 状态提供可视化:等待谁、等待什么、超时策略。
- 使用 replay 测试验证历史事件在新状态机下的行为。
7. 常见反模式
- 状态只是字符串,缺少 guards 和 allowed actions。
- Agent 可以从任何状态调用任何工具。
- 状态迁移原因不记录,审计时无法解释。
- 业务状态和 Agent 内存不一致。
- 状态机过细,导致每个 token 级决策都进状态机,开发复杂。
8. 评测方法
- Transition Correctness:状态转移是否符合规则。
- Guard Coverage:高风险动作是否被 guard 覆盖。
- Agent Node Accuracy:节点内 Agent 输出是否准确。
- Replay Consistency:历史案例回放是否可复现。
- Process Completion Rate:流程最终完成或合理失败的比例。
9. 安全与治理
- 状态机是权限控制点,不能由外部文本覆盖。
- 高风险状态转移必须绑定审批人和参数快照。
- 状态恢复时重新校验权限和 guard。
- 对状态配置变更执行代码审查或发布审批。
10. 工程手册补充
10.1 状态机数据结构
状态机适合把不可控的 Agent 判断限制在可控转移里。
machine:
state: Executing
run_id: run_789
plan_version: 2
allowed_transitions:
Executing: [Checkpointed, Replanning, Paused, Aborted, Completed]
guards:
Replanning:
- "replan_count < 3"
- "no_unresolved_side_effect"
Completed:
- "all_required_artifacts_verified"
10.2 工程实践
| 设计点 | 推荐做法 | 反模式 |
|---|---|---|
| Agent 决策 | 只返回候选转移和理由 | 让 Agent 直接改数据库状态 |
| Guard | 用代码检查权限、预算、依赖 | 靠提示词要求“不要越权” |
| Checkpoint | 在状态转移后持久化 | 只在最终完成时保存 |
| 回滚 | 对副作用状态定义补偿边 | 失败后从头再跑 |
伪代码:
candidate = agent.propose_transition(state, observation)
if candidate.name not in machine.allowed(state.name):
raise InvalidTransition(candidate.name)
for guard in guards[candidate.name]:
guard.check(state, candidate)
new_state = transition(state, candidate)
store.checkpoint(new_state)
评测与上线清单:
- 用状态覆盖率评测:每个状态、每条转移、每个 guard 都有测试样本。
- 故障注入:在任意状态杀进程,恢复后应停在合法状态。
- 监控非法转移尝试、guard 拒绝率、暂停时长、补偿成功率。
- 安全上,高风险状态转移必须由代码 gate 执行,不由模型最终裁决。
11. 权威资料
- LangGraph graph API: https://docs.langchain.com/oss/python/langgraph/graph-api
- LangGraph durable execution: https://docs.langchain.com/oss/python/langgraph/durable-execution
- AWS Step Functions state machines: https://docs.aws.amazon.com/step-functions/latest/dg/concepts-statemachines.html
- AWS Step Functions service integration patterns: https://docs.aws.amazon.com/step-functions/latest/dg/connect-to-resource.html
- Anthropic, Building effective agents: https://www.anthropic.com/engineering/building-effective-agents