AI Agent 多轮对话状态机编排：从意图追踪到上下文恢复的工程实践

一、多轮对话的"失忆困境"：状态丢失与意图漂移的工程痛点

构建 AI Agent 时，单轮对话的问答模式相对简单——接收输入、调用模型、返回输出。但当 Agent 需要处理跨越多轮的复杂任务时，问题接踵而至。某智能运维 Agent 在执行"排查集群故障并自动修复"任务时，第一轮识别出节点异常，第二轮尝试重启，第三轮发现重启后服务未恢复——但此时 Agent 已经"忘记"第一轮的诊断结论，重新开始排查，陷入循环。这就是典型的状态丢失问题。

更隐蔽的是意图漂移。用户在多轮交互中可能中途修改需求，或者 Agent 在执行过程中发现需要切换策略。如果缺少显式的意图追踪机制，Agent 会继续沿着旧路径执行，产生无效操作。例如用户先要求"部署 v2 版本"，执行到一半又说"先灰度发布"，Agent 如果没有捕获到意图变更，就会直接全量部署。

状态机是解决这类问题的经典方案。将多轮对话建模为有限状态自动机（FSM），每个状态对应一个明确的对话阶段，状态转移由意图识别和条件判断驱动。这种方式不仅解决了"失忆"问题，还让对话流程可审计、可恢复、可回溯。

二、对话状态机架构：从线性流转到条件分支的三层设计

flowchart TB
    subgraph FSM["多轮对话状态机架构"]
        direction TB
        S1["IDLE<br/>空闲态<br/>等待用户输入"]
        S2["INTENT_PARSE<br/>意图解析<br/>识别用户目标"]
        S3["TASK_PLAN<br/>任务规划<br/>分解执行步骤"]
        S4["EXECUTING<br/>执行态<br/>调用工具/模型"]
        S5["CONFIRMING<br/>确认态<br/>关键操作人工确认"]
        S6["RECOVERING<br/>恢复态<br/>异常后上下文重建"]
        S7["COMPLETED<br/>完成态<br/>结果汇总与归档"]
    end

    S1 -->|"用户输入"| S2
    S2 -->|"意图明确"| S3
    S2 -->|"意图模糊"| S1
    S3 -->|"步骤确认"| S4
    S4 -->|"需要确认"| S5
    S5 -->|"用户确认"| S4
    S5 -->|"用户拒绝"| S3
    S4 -->|"执行异常"| S6
    S6 -->|"上下文恢复"| S4
    S4 -->|"任务完成"| S7
    S7 -->|"新任务"| S1

    style S1 fill:#f9f,stroke:#333
    style S4 fill:#9cf,stroke:#333
    style S6 fill:#f96,stroke:#333
    style S7 fill:#9f9,stroke:#333

状态机架构分为三层：

第一层：核心状态定义。每个状态有明确的进入条件、执行逻辑和退出条件。IDLE 态只负责接收输入；INTENT_PARSE 态调用意图分类模型；TASK_PLAN 态根据意图生成执行计划；EXECUTING 态按计划逐步调用工具；CONFIRMING 态在关键操作前暂停等待人工确认；RECOVERING 态在异常发生时重建上下文；COMPLETED 态汇总结果并归档。

第二层：状态转移规则。转移不是硬编码的 if-else，而是由转移条件表驱动。每条转移规则包含：源状态、触发事件、守卫条件、目标状态、转移动作。守卫条件可以是"意图置信度 > 0.8"或"连续失败次数 < 3"等动态判断。这种声明式的转移规则让状态机可配置、可测试。

第三层：上下文快照与恢复。每次状态转移时，将当前对话上下文（包括意图栈、已执行步骤、中间结果）序列化为快照。当 Agent 因异常中断后重新启动时，从最近的快照恢复，避免从头开始。快照采用增量存储策略——只保存状态差异，减少存储开销。

三、对话状态机的代码实现

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Callable, Any
import json
import time

class DialogState(Enum):
    """对话状态枚举"""
    IDLE = "idle"
    INTENT_PARSE = "intent_parse"
    TASK_PLAN = "task_plan"
    EXECUTING = "executing"
    CONFIRMING = "confirming"
    RECOVERING = "recovering"
    COMPLETED = "completed"

@dataclass
class ContextSnapshot:
    """对话上下文快照，用于异常恢复"""
    session_id: str
    state: DialogState
    intent_stack: list = field(default_factory=list)
    executed_steps: list = field(default_factory=list)
    intermediate_results: dict = field(default_factory=dict)
    failure_count: int = 0
    timestamp: float = field(default_factory=time.time)

    def to_dict(self) -> dict:
        return {
            "session_id": self.session_id,
            "state": self.state.value,
            "intent_stack": self.intent_stack,
            "executed_steps": self.executed_steps,
            "intermediate_results": self.intermediate_results,
            "failure_count": self.failure_count,
            "timestamp": self.timestamp,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "ContextSnapshot":
        data["state"] = DialogState(data["state"])
        return cls(**data)

@dataclass
class TransitionRule:
    """状态转移规则"""
    source: DialogState
    event: str
    guard: Callable[[ContextSnapshot], bool]
    target: DialogState
    action: Optional[Callable[[ContextSnapshot], None]] = None

class DialogStateMachine:
    """多轮对话状态机引擎"""

    def __init__(self):
        self.rules: list[TransitionRule] = []
        self.snapshots: dict[str, ContextSnapshot] = {}
        self.snapshot_store: list[dict] = []

    def add_rule(self, rule: TransitionRule):
        self.rules.append(rule)

    def _find_transition(
        self, ctx: ContextSnapshot, event: str
    ) -> Optional[TransitionRule]:
        """查找匹配的转移规则，守卫条件必须通过"""
        for rule in self.rules:
            if rule.source == ctx.state and rule.event == event:
                if rule.guard(ctx):
                    return rule
        return None

    def _save_snapshot(self, ctx: ContextSnapshot):
        """保存上下文快照，采用增量存储"""
        snapshot_data = ctx.to_dict()
        self.snapshot_store.append(snapshot_data)
        self.snapshots[ctx.session_id] = ctx

    def transit(self, session_id: str, event: str) -> DialogState:
        """执行状态转移"""
        ctx = self.snapshots.get(session_id)
        if not ctx:
            ctx = ContextSnapshot(
                session_id=session_id, state=DialogState.IDLE
            )
            self._save_snapshot(ctx)

        rule = self._find_transition(ctx, event)
        if not rule:
            # 无匹配规则，保持当前状态
            return ctx.state

        # 执行转移动作
        if rule.action:
            rule.action(ctx)

        # 更新状态
        old_state = ctx.state
        ctx.state = rule.target
        ctx.timestamp = time.time()

        # 保存快照
        self._save_snapshot(ctx)
        return ctx.state

    def recover(self, session_id: str) -> Optional[ContextSnapshot]:
        """从快照恢复上下文"""
        return self.snapshots.get(session_id)

# ===== 构建状态机实例 =====
def build_dialog_fsm() -> DialogStateMachine:
    """构建对话状态机，注册所有转移规则"""
    fsm = DialogStateMachine()

    # IDLE -> INTENT_PARSE：用户输入触发意图解析
    fsm.add_rule(TransitionRule(
        source=DialogState.IDLE,
        event="user_input",
        guard=lambda ctx: True,
        target=DialogState.INTENT_PARSE,
    ))

    # INTENT_PARSE -> TASK_PLAN：意图置信度足够高
    fsm.add_rule(TransitionRule(
        source=DialogState.INTENT_PARSE,
        event="intent_resolved",
        guard=lambda ctx: len(ctx.intent_stack) > 0,
        target=DialogState.TASK_PLAN,
    ))

    # INTENT_PARSE -> IDLE：意图模糊，回到空闲态
    fsm.add_rule(TransitionRule(
        source=DialogState.INTENT_PARSE,
        event="intent_ambiguous",
        guard=lambda ctx: True,
        target=DialogState.IDLE,
    ))

    # EXECUTING -> CONFIRMING：关键操作需确认
    fsm.add_rule(TransitionRule(
        source=DialogState.EXECUTING,
        event="need_confirmation",
        guard=lambda ctx: True,
        target=DialogState.CONFIRMING,
    ))

    # EXECUTING -> RECOVERING：连续失败超过阈值
    fsm.add_rule(TransitionRule(
        source=DialogState.EXECUTING,
        event="execution_failed",
        guard=lambda ctx: ctx.failure_count >= 3,
        target=DialogState.RECOVERING,
    ))

    # RECOVERING -> EXECUTING：上下文恢复后继续执行
    fsm.add_rule(TransitionRule(
        source=DialogState.RECOVERING,
        event="context_recovered",
        guard=lambda ctx: len(ctx.executed_steps) > 0,
        target=DialogState.EXECUTING,
    ))

    # COMPLETED -> IDLE：任务完成，等待新任务
    fsm.add_rule(TransitionRule(
        source=DialogState.COMPLETED,
        event="new_task",
        guard=lambda ctx: True,
        target=DialogState.IDLE,
    ))

    return fsm

关键设计决策说明：ContextSnapshot 采用 dataclass 序列化方案，而非 pickle，因为 pickle 存在安全风险且跨版本不兼容。TransitionRule 的守卫条件使用回调函数而非字符串表达式，避免 eval() 带来的注入风险。快照存储使用追加模式（append-only），支持按时间回溯到任意历史状态。

四、状态机方案的边界与权衡

优势方面：状态机让对话流程显式化，每个状态和转移都有明确定义，便于团队协作和代码审查。上下文快照机制使得异常恢复成为可能——进程崩溃后可以从最近快照恢复，而非丢失全部进度。声明式转移规则让状态机可配置，新增对话场景只需添加规则，无需修改核心引擎。

劣势方面：状态机的表达能力有限。当对话场景复杂到需要嵌套子状态（如"执行中"包含"等待API响应"和"等待人工确认"两个并行子状态）时，FSM 变得难以维护。此时需要升级为层次状态机（HSM）或行为树。此外，状态爆炸是另一个风险——如果每个意图都对应独立的状态转移路径，状态数量会指数级增长。缓解策略是将通用逻辑抽象为共享状态，仅对差异化路径定义专用状态。

适用边界：状态机适合流程确定、状态数量可控的对话场景（如运维操作、工单处理、部署流程）。不适合开放域闲聊或高度发散的创意生成场景——这类场景的状态空间不可枚举，FSM 无法覆盖。

性能考量：快照的序列化开销随上下文大小线性增长。当对话历史包含大量中间结果时，每次转移的快照保存可能成为瓶颈。实践中可以设置快照间隔——仅在关键状态转移时保存完整快照，其他时候只记录增量日志。

五、总结

多轮对话状态机编排解决了 AI Agent 在长流程任务中的两个核心问题：状态丢失和意图漂移。通过将对话建模为有限状态自动机，每个阶段有明确的进入/退出条件，状态转移由守卫条件驱动，上下文快照保障异常恢复能力。落地时需注意三点：一是控制状态数量，避免状态爆炸；二是快照策略选择增量存储而非全量复制；三是当对话复杂度超过 FSM 表达能力时，及时升级为层次状态机或行为树。工程实践中，建议先用状态转移表梳理所有可能的对话路径，再编码实现，而非边写边加状态——后者几乎必然导致状态爆炸。