状态机架构
状态机架构（State Machine） 是一种基于有限状态自动机的 Agent 架构模式。通过定义明确的状态集合、转换条件和动作，实现 Agent 行为的可预测控制和流程自动化，特别适合对话系统、业务流程自动化等固定流程场景。
一、核心原理
1.1 设计哲学
状态机架构的核心思想是状态驱动的行为控制：
┌─────────────────────────────────────────────────────────────┐
│ 状态机设计哲学 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 传统 Agent（状态不可控）： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ 输入 → LLM → 输出 │ │
│ │ │ │
│ │ 每次调用状态不确定，难以预测和控制 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 状态机 Agent（状态可控）： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ ┌──────────────────────────────┐ │ │
│ │ │ │ │ │
│ │ ↓ │ │ │
│ │ ┌──────────┐ 条件A ┌──────────┐ │ │
│ │ │ 状态S1 │ ────────────→ │ 状态S2 │ │ │
│ │ └──────────┘ └────┬─────┘ │ │
│ │ ↑ │ │ │
│ │ │ │ 条件B │ │
│ │ │ ↓ │ │
│ │ ┌────┴─────┐ ┌──────────┐ │ │
│ │ │ 状态S4 │ ←──────────── │ 状态S3 │ │ │
│ │ └──────────┘ └──────────┘ │ │
│ │ │ │ │
│ │ │ 条件C │ │
│ │ ↓ │ │
│ │ ┌──────────┐ │ │
│ │ │ 结束 │ │ │
│ │ └──────────┘ │ │
│ │ │ │
│ │ 状态明确、转换可控、行为可预测 │ │
│ │ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1.2 核心概念
| 概念 | 英文 | 说明 | 示例 |
|---|---|---|---|
| 状态 | State | Agent 当前的模式或阶段 | 等待输入、处理中、完成 |
| 转换 | Transition | 从一个状态到另一个状态的改变 | 输入有效时转换 |
| 事件 | Event | 触发状态转换的外部刺激 | 用户消息、超时 |
| 条件 | Guard | 决定转换是否发生的条件 | 输入格式正确 |
| 动作 | Action | 状态转换时执行的操作 | 发送回复、调用 API |
1.3 状态机类型
┌─────────────────────────────────────────────────────────────┐
│ 状态机类型 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 有限状态机（FSM） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 最简单的状态机，只有状态和转换 │ │
│ │ │ │
│ │ [空闲] ──消息──→ [处理] ──完成──→ [空闲] │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 2. Mealy 状态机 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 输出依赖于当前状态和输入 │ │
│ │ │ │
│ │ [状态A] ──输入X/输出Y──→ [状态B] │ │
│ │ 动作绑定在转换上 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 3. Moore 状态机 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 输出只依赖于当前状态 │ │
│ │ │ │
│ │ [状态A/输出X] ──输入──→ [状态B/输出Y] │ │
│ │ 动作绑定在状态上 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 4. 层次状态机（HSM） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 状态可以嵌套，形成层次结构 │ │
│ │ │ │
│ │ ┌─────────────────────────┐ │ │
│ │ │ [主状态] │ │ │
│ │ │ ┌─────┐ ┌─────┐ │ │ │
│ │ │ │子A │ │子B │ │ │ │
│ │ │ └─────┘ └─────┘ │ │ │
│ │ └─────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1.4 与其他架构对比
| 对比维度 | 状态机 | ReAct | DAG |
|---|---|---|---|
| 确定性 | 最高 | 低 | 高 |
| 可控性 | 最高 | 低 | 中 |
| 灵活性 | 低 | 最高 | 中 |
| 并行能力 | 低 | 无 | 高 |
| 适用场景 | 固定流程 | 探索任务 | 数据管道 |
二、工作流程
2.1 完整工作流程图
┌─────────────────────────────────────────────────────────────────────┐
│ 状态机完整工作流程 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ 初始化 │ │
│ │ (Init) │ │
│ └──────┬──────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────────┐ │
│ │ 状态机主循环 │ │
│ │ │ │
│ │ ┌──────────────────────────────────────────────────────┐ │ │
│ │ │ │ │ │
│ │ │ ┌─────────────────────────────────────────────┐ │ │ │
│ │ │ │ 1. 获取当前状态 │ │ │ │
│ │ │ │ current_state = state_machine.current │ │ │ │
│ │ │ └─────────────────────────────────────────────┘ │ │ │
│ │ │ │ │ │ │
│ │ │ ↓ │ │ │
│ │ │ ┌─────────────────────────────────────────────┐ │ │ │
│ │ │ │ 2. 等待事件 │ │ │ │
│ │ │ │ event = wait_for_event() │ │ │ │
│ │ │ └─────────────────────────────────────────────┘ │ │ │
│ │ │ │ │ │ │
│ │ │ ↓ │ │ │
│ │ │ ┌─────────────────────────────────────────────┐ │ │ │
│ │ │ │ 3. 检查转换条件 │ │ │ │
│ │ │ │ for transition in current_state.transitions:│ │ │
│ │ │ │ if transition.guard(event): │ │ │ │
│ │ │ │ next_state = transition.target │ │ │ │
│ │ │ │ break │ │ │ │
│ │ │ └─────────────────────────────────────────────┘ │ │ │
│ │ │ │ │ │ │
│ │ │ ↓ │ │ │
│ │ │ ┌─────────────────────────────────────────────┐ │ │ │
│ │ │ │ 4. 执行转换动作 │ │ │ │
│ │ │ │ transition.action(event) │ │ │ │
│ │ │ └─────────────────────────────────────────────┘ │ │ │
│ │ │ │ │ │ │
│ │ │ ↓ │ │ │
│ │ │ ┌─────────────────────────────────────────────┐ │ │ │
│ │ │ │ 5. 更新状态 │ │ │ │
│ │ │ │ current_state.exit() │ │ │ │
│ │ │ │ state_machine.current = next_state │ │ │ │
│ │ │ │ next_state.enter() │ │ │ │
│ │ │ └─────────────────────────────────────────────┘ │ │ │
│ │ │ │ │ │ │
│ │ │ ↓ │ │ │
│ │ │ ┌─────────────────────────────────────────────┐ │ │ │
│ │ │ │ 6. 检查是否终态 │ │ │ │
│ │ │ │ if current_state.is_final: │ │ │ │
│ │ │ │ break │ │ │ │
│ │ │ └─────────────────────────────────────────────┘ │ │ │
│ │ │ │ │ │ │
│ │ └────────────────────────┼─────────────────────────────┘ │ │
│ │ │ │ │
│ └────────────────────────────┼─────────────────────────────────┘ │
│ │ │
│ ↓ │
│ ┌─────────────────────┐ │
│ │ 完成/终止 │ │
│ └─────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────┘2.2 订单处理状态机示例
┌─────────────────────────────────────────────────────────────┐
│ 订单处理状态机 │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ │
│ │ 创建 │ │
│ │ (CREATED)│ │
│ └────┬─────┘ │
│ │ │
│ 支付成功 │ 支付失败 │
│ ┌─────────┴─────────┐ │
│ ↓ ↓ │
│ ┌──────────┐ ┌──────────┐ │
│ │ 已支付 │ │ 取消 │ │
│ │ (PAID) │ │(CANCELLED)│ │
│ └────┬─────┘ └──────────┘ │
│ │ │
│ ┌────────┴────────┐ │
│ ↓ ↓ │
│ ┌──────────┐ ┌──────────┐ │
│ │ 发货 │ │ 退款 │ │
│ │(SHIPPED) │ │(REFUNDED)│ │
│ └────┬─────┘ └──────────┘ │
│ │ │
│ ↓ │
│ ┌──────────┐ │
│ │ 完成 │ │
│ │(COMPLETED)│ │
│ └──────────┘ │
│ │
│ 状态转换规则： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ CREATED ──支付成功──→ PAID │ │
│ │ CREATED ──支付失败──→ CANCELLED │ │
│ │ PAID ──发货─────→ SHIPPED │ │
│ │ PAID ──退款─────→ REFUNDED │ │
│ │ SHIPPED ──确认收货──→ COMPLETED │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘2.3 对话状态机示例
"""
对话系统状态机
"""
from enum import Enum
from dataclasses import dataclass
from typing import Callable, Optional
class DialogState(Enum):
"""对话状态"""
GREETING = "greeting" # 问候
COLLECT_INFO = "collect_info" # 收集信息
CONFIRM = "confirm" # 确认
PROCESS = "process" # 处理
COMPLETE = "complete" # 完成
ERROR = "error" # 错误
@dataclass
class Transition:
"""状态转换"""
from_state: DialogState
to_state: DialogState
event: str # 触发事件
guard: Callable # 条件函数
action: Callable # 动作函数
class DialogStateMachine:
"""对话状态机"""
def __init__(self):
self.state = DialogState.GREETING
self.context = {} # 对话上下文
self.transitions = self._build_transitions()
def _build_transitions(self) -> dict:
"""构建状态转换表"""
return {
(DialogState.GREETING, "user_input"): [
{
"target": DialogState.COLLECT_INFO,
"guard": lambda e: "预订" in e.get("text", ""),
"action": self._start_booking
},
{
"target": DialogState.GREETING,
"guard": lambda e: True, # 默认转换
"action": self._greeting
}
],
(DialogState.COLLECT_INFO, "user_input"): [
{
"target": DialogState.CONFIRM,
"guard": lambda e: self._has_all_info(),
"action": self._show_confirmation
},
{
"target": DialogState.COLLECT_INFO,
"guard": lambda e: True,
"action": self._collect_more_info
}
],
(DialogState.CONFIRM, "user_input"): [
{
"target": DialogState.PROCESS,
"guard": lambda e: "确认" in e.get("text", ""),
"action": self._process_booking
},
{
"target": DialogState.COLLECT_INFO,
"guard": lambda e: "修改" in e.get("text", ""),
"action": self._modify_info
}
],
(DialogState.PROCESS, "complete"): [
{
"target": DialogState.COMPLETE,
"guard": lambda e: e.get("success", False),
"action": self._complete_success
},
{
"target": DialogState.ERROR,
"guard": lambda e: True,
"action": self._handle_error
}
]
}
def process_event(self, event: str, data: dict = None):
"""处理事件"""
key = (self.state, event)
transitions = self.transitions.get(key, [])
for t in transitions:
if t["guard"](data or {}):
# 执行动作
t["action"](data or {})
# 转换状态
self.state = t["target"]
return
# 没有匹配的转换
print(f"警告：状态 {self.state} 无法处理事件 {event}")
def _start_booking(self, data):
self.context = {"type": "booking"}
print("开始预订流程...")
def _greeting(self, data):
print("您好！请问有什么可以帮助您的？")
def _has_all_info(self) -> bool:
return all(k in self.context for k in ["date", "time", "name"])
def _show_confirmation(self, data):
print(f"请确认：{self.context}")
def _collect_more_info(self, data):
text = data.get("text", "")
# 解析用户输入，更新上下文
print("请提供更多信息...")
def _process_booking(self, data):
print("正在处理预订...")
# 模拟处理
self.process_event("complete", {"success": True})
def _modify_info(self, data):
print("请提供修改信息...")
def _complete_success(self, data):
print("预订成功！")
def _handle_error(self, data):
print("处理失败，请稍后重试。")
# 使用示例
if __name__ == "__main__":
sm = DialogStateMachine()
# 模拟对话
sm.process_event("user_input", {"text": "我想预订房间"})
sm.process_event("user_input", {"text": "明天下午3点"})
sm.process_event("user_input", {"text": "我叫张三"})
sm.process_event("user_input", {"text": "确认预订"})三、代码实现
3.1 基础框架实现
"""
状态机基础框架
"""
from typing import Dict, List, Callable, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import inspect
@dataclass
class State:
"""状态定义"""
name: str
is_final: bool = False
on_enter: Callable = None
on_exit: Callable = None
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class Transition:
"""状态转换定义"""
from_state: str
to_state: str
event: str
guard: Callable[[Any], bool] = None
action: Callable[[Any], Any] = None
class StateMachine:
"""状态机"""
def __init__(self, name: str = "StateMachine"):
self.name = name
self.states: Dict[str, State] = {}
self.transitions: Dict[str, List[Transition]] = {}
self.current_state: Optional[str] = None
self.history: List[str] = []
self.context: Dict[str, Any] = {}
def add_state(self, state: State):
"""添加状态"""
self.states[state.name] = state
if state.name not in self.transitions:
self.transitions[state.name] = []
def add_transition(self, transition: Transition):
"""添加转换"""
if transition.from_state not in self.transitions:
self.transitions[transition.from_state] = []
self.transitions[transition.from_state].append(transition)
def set_initial_state(self, state_name: str):
"""设置初始状态"""
if state_name not in self.states:
raise ValueError(f"状态 {state_name} 不存在")
self.current_state = state_name
self._enter_state(state_name)
def process_event(self, event: str, data: Any = None):
"""处理事件"""
if not self.current_state:
raise RuntimeError("状态机未初始化")
# 查找匹配的转换
transitions = self.transitions.get(self.current_state, [])
for transition in transitions:
if transition.event != event:
continue
# 检查条件
if transition.guard and not transition.guard(data):
continue
# 执行动作
if transition.action:
result = transition.action(data)
if result is not None:
self.context.update(result) if isinstance(result, dict) else None
# 执行转换
self._transition_to(transition.to_state)
return True
return False # 没有匹配的转换
def _transition_to(self, new_state: str):
"""执行状态转换"""
old_state = self.current_state
# 退出旧状态
self._exit_state(old_state)
# 进入新状态
self._enter_state(new_state)
# 更新当前状态
self.current_state = new_state
self.history.append(f"{old_state} -> {new_state}")
print(f"[{self.name}] {old_state} -> {new_state}")
def _enter_state(self, state_name: str):
"""进入状态"""
state = self.states.get(state_name)
if state and state.on_enter:
state.on_enter(self.context)
def _exit_state(self, state_name: str):
"""退出状态"""
state = self.states.get(state_name)
if state and state.on_exit:
state.on_exit(self.context)
def is_finished(self) -> bool:
"""检查是否到达终态"""
if not self.current_state:
return False
state = self.states.get(self.current_state)
return state.is_final if state else False
def get_state_info(self) -> dict:
"""获取状态信息"""
return {
"current_state": self.current_state,
"is_finished": self.is_finished(),
"history": self.history[-10:], # 最近10次转换
"context": self.context
}
# 使用示例：构建订单状态机
def create_order_state_machine():
"""创建订单状态机"""
sm = StateMachine("OrderStateMachine")
# 添加状态
sm.add_state(State("created", on_enter=lambda c: print("订单已创建")))
sm.add_state(State("paid", on_enter=lambda c: print("订单已支付")))
sm.add_state(State("shipped", on_enter=lambda c: print("订单已发货")))
sm.add_state(State("completed", is_final=True, on_enter=lambda c: print("订单已完成")))
sm.add_state(State("cancelled", is_final=True, on_enter=lambda c: print("订单已取消")))
# 添加转换
sm.add_transition(Transition(
from_state="created",
to_state="paid",
event="pay_success",
action=lambda d: {"paid_at": "now"}
))
sm.add_transition(Transition(
from_state="created",
to_state="cancelled",
event="pay_failed"
))
sm.add_transition(Transition(
from_state="paid",
to_state="shipped",
event="ship",
action=lambda d: {"shipped_at": "now"}
))
sm.add_transition(Transition(
from_state="paid",
to_state="cancelled",
event="refund"
))
sm.add_transition(Transition(
from_state="shipped",
to_state="completed",
event="confirm"
))
# 设置初始状态
sm.set_initial_state("created")
return sm
if __name__ == "__main__":
# 创建并运行
sm = create_order_state_machine()
# 模拟订单流程
sm.process_event("pay_success")
sm.process_event("ship")
sm.process_event("confirm")
print(f"\n最终状态: {sm.current_state}")
print(f"是否完成: {sm.is_finished()}")3.2 使用 LangGraph 实现
"""
使用 LangGraph 实现状态机
"""
from typing import TypedDict
from langgraph.graph import StateGraph, END
class OrderState(TypedDict):
"""订单状态"""
order_id: str
status: str
history: list
def create_order_graph():
"""创建订单状态机图"""
# 定义状态处理函数
def handle_created(state: OrderState):
return {"status": "created"}
def handle_paid(state: OrderState):
return {"status": "paid", "history": state["history"] + ["paid"]}
def handle_shipped(state: OrderState):
return {"status": "shipped", "history": state["history"] + ["shipped"]}
def handle_completed(state: OrderState):
return {"status": "completed", "history": state["history"] + ["completed"]}
def handle_cancelled(state: OrderState):
return {"status": "cancelled", "history": state["history"] + ["cancelled"]}
# 定义条件判断
def should_pay(state: OrderState):
return state.get("event") == "pay"
def should_ship(state: OrderState):
return state.get("event") == "ship"
def should_complete(state: OrderState):
return state.get("event") == "complete"
def should_cancel(state: OrderState):
return state.get("event") == "cancel"
# 创建图
workflow = StateGraph(OrderState)
# 添加节点
workflow.add_node("created", handle_created)
workflow.add_node("paid", handle_paid)
workflow.add_node("shipped", handle_shipped)
workflow.add_node("completed", handle_completed)
workflow.add_node("cancelled", handle_cancelled)
# 设置入口
workflow.set_entry_point("created")
# 添加边
workflow.add_edge("created", "paid")
workflow.add_edge("created", "cancelled")
workflow.add_edge("paid", "shipped")
workflow.add_edge("paid", "cancelled")
workflow.add_edge("shipped", "completed")
workflow.add_edge("completed", END)
workflow.add_edge("cancelled", END)
return workflow.compile()3.3 层次状态机实现
"""
层次状态机实现
"""
from typing import Dict, List, Optional, Set
from dataclasses import dataclass
@dataclass
class HierarchicalState:
"""层次状态"""
name: str
parent: Optional['HierarchicalState'] = None
children: List['HierarchicalState'] = None
on_enter: callable = None
on_exit: callable = None
def __post_init__(self):
if self.children is None:
self.children = []
def add_child(self, child: 'HierarchicalState'):
child.parent = self
self.children.append(child)
def is_ancestor_of(self, state: 'HierarchicalState') -> bool:
"""检查是否是某个状态的祖先"""
current = state.parent
while current:
if current == self:
return True
current = current.parent
return False
def get_path(self) -> List[str]:
"""获取从根到当前状态的路径"""
path = [self.name]
current = self.parent
while current:
path.insert(0, current.name)
current = current.parent
return path
class HierarchicalStateMachine:
"""层次状态机"""
def __init__(self):
self.states: Dict[str, HierarchicalState] = {}
self.root_states: List[HierarchicalState] = []
self.current_state: Optional[HierarchicalState] = None
def add_state(self, state: HierarchicalState, parent: str = None):
"""添加状态"""
self.states[state.name] = state
if parent:
parent_state = self.states.get(parent)
if parent_state:
parent_state.add_child(state)
else:
self.root_states.append(state)
else:
self.root_states.append(state)
def transition_to(self, state_name: str):
"""转换到指定状态"""
new_state = self.states.get(state_name)
if not new_state:
raise ValueError(f"状态 {state_name} 不存在")
# 找到 LCA（最近公共祖先）
lca = self._find_lca(self.current_state, new_state)
# 退出当前状态到 LCA
self._exit_to_lca(self.current_state, lca)
# 从 LCA 进入新状态
self._enter_from_lca(new_state, lca)
self.current_state = new_state
def _find_lca(self, state1: HierarchicalState, state2: HierarchicalState):
"""找到两个状态的最近公共祖先"""
if not state1 or not state2:
return None
ancestors = set()
current = state1
while current:
ancestors.add(current)
current = current.parent
current = state2
while current:
if current in ancestors:
return current
current = current.parent
return None
def _exit_to_lca(self, state: HierarchicalState, lca: HierarchicalState):
"""退出状态到 LCA"""
current = state
while current and current != lca:
if current.on_exit:
current.on_exit()
current = current.parent
def _enter_from_lca(self, state: HierarchicalState, lca: HierarchicalState):
"""从 LCA 进入状态"""
path = []
current = state
while current and current != lca:
path.insert(0, current)
current = current.parent
for s in path:
if s.on_enter:
s.on_enter()
# 使用示例
if __name__ == "__main__":
hsm = HierarchicalStateMachine()
# 添加层次状态
hsm.add_state(HierarchicalState("online", on_enter=lambda: print("上线")))
hsm.add_state(HierarchicalState("idle", on_enter=lambda: print("空闲")), parent="online")
hsm.add_state(HierarchicalState("busy", on_enter=lambda: print("忙碌")), parent="online")
hsm.add_state(HierarchicalState("call", on_enter=lambda: print("通话中")), parent="busy")
# 设置初始状态
hsm.current_state = hsm.states["idle"]
# 转换状态
hsm.transition_to("call")四、适用场景
4.1 最佳适用场景
| 场景类型 | 具体示例 | 状态机优势 |
|---|---|---|
| 对话系统 | 客服机器人、预订系统 | 流程可控、状态清晰 |
| 订单处理 | 电商订单、工单系统 | 状态明确、可审计 |
| 游戏 AI | NPC 行为、关卡流程 | 行为可预测 |
| 业务流程 | 审批流程、工作流 | 流程标准化 |
4.2 场景详解
┌─────────────────────────────────────────────────────────────┐
│ 状态机适用场景详解 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 客服对话系统 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ [欢迎] ──产品咨询──→ [产品介绍] │ │
│ │ │ │ │ │
│ │ │ 购买意向 │ │
│ │ │ ↓ │ │
│ │ │ [收集信息] ──确认──→ [下单] │ │
│ │ │ │ │ │
│ │ │ 取消 │ │
│ │ │ ↓ │ │
│ │ └───────────────→ [结束] │ │
│ │ │ │
│ │ 优势：对话流程可控，每个状态职责明确 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 2. 游戏 NPC AI │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ ┌─── 发现敌人 ──→ [追击] │ │
│ │ │ │ │ │
│ │ [巡逻] ─────┤ 进入攻击范围 │ │
│ │ │ ↓ │ │
│ │ └─── 被攻击 ──→ [战斗] │ │
│ │ │ │ │
│ │ 血量低 │ │
│ │ ↓ │ │
│ │ [逃跑] │ │
│ │ │ │ │
│ │ 安全 │ │
│ │ ↓ │ │
│ │ [巡逻] │ │
│ │ │ │
│ │ 优势：行为可预测，状态转换逻辑清晰 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 3. 审批工作流 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ │ │
│ │ [提交] ──→ [主管审批] ──通过──→ [经理审批] │ │
│ │ │ │ │ │
│ │ 驳回 通过 │ │
│ │ │ │ │ │
│ │ ↓ ↓ │ │
│ │ [退回] [完成] │ │
│ │ │ │
│ │ 优势：流程标准化，审批记录可追溯 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘4.3 不适用场景
| 场景 | 原因 | 推荐替代方案 |
|---|---|---|
| 探索性任务 | 无法预先定义状态 | ReAct 模式 |
| 复杂决策 | 状态爆炸问题 | 规则引擎 |
| 高度动态 | 状态频繁变化 | 行为树 |
| 并行任务 | 状态机串行特性 | DAG 工作流 |
五、局限性与优化
5.1 主要局限性
| 局限性 | 具体表现 | 影响 |
|---|---|---|
| 状态爆炸 | 状态数量指数增长 | 维护困难 |
| 灵活性差 | 新增状态需要修改代码 | 扩展困难 |
| 并行限制 | 难以处理并行事件 | 效率受限 |
| 复杂度 | 大规模状态机难以理解 | 可读性差 |
5.2 优化策略
┌─────────────────────────────────────────────────────────────┐
│ 状态机优化策略 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 层次化设计 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 问题：状态过多导致难以管理 │ │
│ │ │ │
│ │ 优化方案：使用层次状态机 │ │
│ │ │ │
│ │ ┌──────────────────────────┐ │ │
│ │ │ [会话] │ │ │
│ │ │ ┌─────┐ ┌─────┐ │ │ │
│ │ │ │认证 │ │服务 │ │ │ │
│ │ │ └──┬──┘ └──┬──┘ │ │ │
│ │ │ │ │ │ │ │
│ │ │ ┌──┴──┐ ┌──┴──┐ │ │ │
│ │ │ │登录 │ │查询 │ │ │ │
│ │ │ │注册 │ │下单 │ │ │ │
│ │ │ └─────┘ └─────┘ │ │ │
│ │ └──────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 2. 状态模式 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 问题：状态逻辑复杂，难以维护 │ │
│ │ │ │
│ │ 优化方案：将每个状态封装为类 │ │
│ │ │ │
│ │ class IdleState: │ │
│ │ def handle_event(self, event): ... │ │
│ │ │ │
│ │ class BusyState: │ │
│ │ def handle_event(self, event): ... │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 3. 配置驱动 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 问题：硬编码状态难以扩展 │ │
│ │ │ │
│ │ 优化方案：使用配置文件定义状态和转换 │ │
│ │ │ │
│ │ states: │ │
│ │ - name: created │ │
│ │ transitions: │ │
│ │ - event: pay │ │
│ │ target: paid │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 4. 可视化工具 │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 问题：复杂状态机难以理解 │ │
│ │ │ │
│ │ 优化方案：使用可视化工具展示状态图 │ │
│ │ • Mermaid 图表 │ │
│ │ • StateMachine Cat │ │
│ │ • PlantUML │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘六、面试常见问题
Q1: 状态机与行为树有什么区别？
A:
| 对比维度 | 状态机 | 行为树 |
|---|---|---|
| 结构 | 状态 + 转换 | 树形节点 |
| 执行 | 状态保持 | 每帧评估 |
| 灵活性 | 低 | 高 |
| 可读性 | 高（状态图） | 中（树结构） |
| 适用场景 | 固定流程 | 复杂行为 |
选择建议：
- 流程固定、状态明确 → 状态机
- 行为复杂、需要动态组合 → 行为树
Q2: 如何避免状态爆炸问题？
A:
解决策略：
- 层次化：将相关状态组合为父状态
- 正交分解：将独立的状态维度分离
- 参数化状态：用参数区分相似状态
- 状态模式：每个状态独立封装
Q3: 状态机如何处理并发事件？
A:
处理方案：
# 方案1：事件队列
class ConcurrentStateMachine:
def __init__(self):
self.event_queue = []
def push_event(self, event):
self.event_queue.append(event)
def process_events(self):
while self.event_queue:
event = self.event_queue.pop(0)
self.process_event(event)
# 方案2：并行状态机
class ParallelStateMachine:
def __init__(self):
self.regions = [] # 多个独立状态机
def process_event(self, event):
for region in self.regions:
region.process_event(event)Q4: 状态机在 Agent 系统中如何应用？
A:
典型应用：
| 应用场景 | 状态机作用 | 示例状态 |
|---|---|---|
| 对话管理 | 控制对话流程 | 问候、收集信息、确认、完成 |
| 任务执行 | 管理任务生命周期 | 待执行、执行中、暂停、完成 |
| 错误处理 | 错误恢复流程 | 正常、警告、错误、恢复 |
| 用户交互 | 管理交互状态 | 等待输入、处理中、响应 |
Q5: 如何测试状态机？
A:
测试策略：
- 状态覆盖测试：确保每个状态都被访问
- 转换覆盖测试：确保每个转换都被执行
- 路径测试：测试关键路径
- 边界测试：测试状态边界条件
def test_state_machine():
sm = create_order_state_machine()
# 测试正常流程
sm.process_event("pay_success")
assert sm.current_state == "paid"
sm.process_event("ship")
assert sm.current_state == "shipped"
sm.process_event("confirm")
assert sm.current_state == "completed"
assert sm.is_finished()Q6: 状态机与其他架构如何结合？
A:
混合架构示例：
class HybridAgent:
"""状态机 + LLM 混合架构"""
def __init__(self):
self.state_machine = DialogStateMachine()
self.llm = ChatOpenAI()
def process_message(self, message):
# 状态机决定当前阶段
state = self.state_machine.current_state
# 根据状态选择 LLM 行为
if state == DialogState.COLLECT_INFO:
prompt = self._build_collect_prompt(message)
elif state == DialogState.CONFIRM:
prompt = self._build_confirm_prompt(message)
else:
prompt = self._build_general_prompt(message)
# LLM 生成回复
response = self.llm(prompt)
# 状态机处理事件
self.state_machine.process_event("user_input", {"text": message})
return response七、总结
| 概念 | 一句话总结 | 面试关键词 |
|---|---|---|
| 状态机 | 状态驱动的行为控制架构 | 有限状态、转换 |
| 状态 | Agent 的当前模式或阶段 | 当前状态、终态 |
| 转换 | 状态之间的变化过程 | 事件触发、条件判断 |
| 层次状态机 | 状态可嵌套的高级状态机 | 状态层次、LCA |
| 核心优势 | 可预测、可控、可测试 | 流程标准化 |
| 主要局限 | 状态爆炸、灵活性差 | 需层次化设计 |
一句话总结：状态机架构通过明确的状态定义和转换规则，实现 Agent 行为的可预测控制，是对话系统和业务流程自动化的首选架构。
最后更新：2026年3月18日