Agent 中的状态机
用状态机管理 Agent 的执行流程：定义状态、转换条件和动作。状态机让 Agent 的行为可预测、可测试、可恢复。
#type / concept
#status / evergreen
#tech / ai
#tech / architecture
[!info] related notes
- 所属 MOC: AI Agent Application MOC
- 相关概念: 工作流编排, 图工作流, Run State
- 实践: 咨询 Agent 工作流
Agent 中的状态机
一句话定义
状态机是用有限个状态和状态之间的转换规则来描述 Agent 行为的模型。它让 Agent 的执行流程从”模型自由发挥”变成”在预定义的状态空间中转换”，从而可预测、可测试、可恢复。
它解决什么问题
完全由 LLM 自由决定执行流程的 Agent 有两个问题：
- 不可预测: 你不知道 Agent 下一步会做什么
- 难以恢复: 中断后不知道应该回到哪个状态
状态机通过预定义状态和转换规则，给 Agent 一个”有限的行为空间”。
核心原理
状态机的基本组成
状态 (State): Agent 当前处于什么阶段
转换 (Transition): 从一个状态到另一个状态的条件
动作 (Action): 进入/退出状态时执行的操作
事件 (Event): 触发转换的外部或内部信号Agent 状态机示例
┌─────────┐ 用户输入 ┌──────────────┐
│ Idle │ ──────────────→ │ Collecting │
│ (空闲) │ │ (信息收集) │
└─────────┘ └──────┬───────┘
↑ │ 信息完整
│ ▼
│ ┌──────────────┐
│ │ Processing │
│ │ (推理处理) │
│ └──────┬───────┘
│ │ 需要工具
│ ▼
│ ┌──────────────┐
│ │ Tool Calling │
│ │ (工具调用) │
│ └──────┬───────┘
│ │ 工具完成
│ ▼
│ ┌──────────────┐
│ 完成 │ Generating │
└──────────────────── │ (生成回复) │
└──────────────┘状态定义
from enum import Enum
class AgentState(Enum):
IDLE = "idle" # 空闲，等待用户输入
COLLECTING = "collecting" # 收集信息
PROCESSING = "processing" # LLM 推理中
TOOL_CALLING = "tool_calling" # 工具执行中
GENERATING = "generating" # 生成回复
WAITING_APPROVAL = "waiting" # 等待人类审批
COMPLETED = "completed" # 完成
ERROR = "error" # 错误转换规则
TRANSITIONS = {
(AgentState.IDLE, "user_message"): AgentState.COLLECTING,
(AgentState.COLLECTING, "info_complete"): AgentState.PROCESSING,
(AgentState.PROCESSING, "need_tool"): AgentState.TOOL_CALLING,
(AgentState.PROCESSING, "no_tool"): AgentState.GENERATING,
(AgentState.TOOL_CALLING, "tool_done"): AgentState.PROCESSING,
(AgentState.TOOL_CALLING, "need_approval"): AgentState.WAITING_APPROVAL,
(AgentState.WAITING_APPROVAL, "approved"): AgentState.TOOL_CALLING,
(AgentState.WAITING_APPROVAL, "rejected"): AgentState.PROCESSING,
(AgentState.GENERATING, "done"): AgentState.COMPLETED,
(AgentState.GENERATING, "error"): AgentState.ERROR,
}典型工程实现
用 LangGraph 实现状态机
LangGraph 的 StateGraph 本质上就是一个状态机：
from langgraph.graph import StateGraph, END
graph = StateGraph(AgentState)
# 节点 = 状态的动作
graph.add_node("collect", collect_info)
graph.add_node("process", process_with_llm)
graph.add_node("call_tool", call_tool)
graph.add_node("generate", generate_response)
# 边 = 转换规则
graph.set_entry_point("collect")
graph.add_conditional_edges("collect", route_after_collect)
graph.add_conditional_edges("process", route_after_process)
graph.add_edge("call_tool", "process")
graph.add_edge("generate", END)状态持久化
from langgraph.checkpoint import SqliteSaver
# 状态持久化到数据库
checkpointer = SqliteSaver.from_conn_string(":memory:")
app = graph.compile(checkpointer=checkpointer)
# 每次转换都自动保存 checkpoint
# 中断后可以从 checkpoint 恢复常见设计模式
1. 分层状态机
顶层状态（如”对话中”、“执行中”）内部嵌套子状态机。
2. 事件驱动状态机
状态转换由事件触发，而不是由代码主动轮询。
3. 有限状态 + LLM 决策
状态是有限的，但每个状态内的具体行为由 LLM 决定。
常见坑
- 状态太多: 超过 10 个状态就难以维护
- 转换条件模糊: 导致状态机”卡住”
- 不做状态持久化: 重启后丢失状态
- 状态和行为混在一起: 状态机只管转换，行为应该在节点函数中
和其他概念的关系
- vs 图工作流: 状态机是图工作流的特例（有状态的有向图）
- vs 工作流编排: 状态机是编排的一种实现模式
- vs Run State: Run State 是状态机的具体应用
- vs Checkpoint: Checkpoint 是状态持久化的机制
总结
状态机让 Agent 的行为从不可预测变成有限可预测。它不是限制 Agent 的能力，而是给 Agent 一个可靠的执行框架。