事件驱动 Agent 状态机

事件驱动 Agent 状态机

学习目标

理解 Nanobot 如何将一个完整的 Agent Turn 拆分为 8 个离散状态，通过事件驱动的状态机实现可追踪、可中断的执行流程。

前置知识

• Agent 核心循环设计 — Agent Turn 的基本概念与执行模型
• 事件驱动 Agent 架构 — 事件驱动模式的基础原理

下文假设你已理解上述概念，直接聚焦 Nanobot 的具体实现。

项目实践

8 状态 Turn 循环

Nanobot 的核心创新之一是将传统的”一气呵成”式 Agent Turn 拆分为 8 个明确定义的状态：

每个状态对应 Turn 生命周期中的一个阶段：

状态	职责	事件	下一阶段
RESTORE	恢复运行时检查点、提取文档、处理附件	ok	COMPACT
COMPACT	会话压缩准备，检查是否需要归档旧消息	ok	COMMAND
COMMAND	斜杠命令分发（如 /stop、/new）	dispatch / shortcut	BUILD / DONE
BUILD	构建 LLM 上下文：加载历史、记忆、技能	ok	RUN
RUN	调用 AgentRunner 执行 LLM 循环	ok	SAVE
SAVE	持久化新消息到会话，清理检查点	ok	RESPOND
RESPOND	组装输出消息	ok	DONE

状态转换表

Nanobot 使用一个明确的转换字典来驱动状态迁移：

_TRANSITIONS = {
    (TurnState.RESTORE, "ok"): TurnState.COMPACT,
    (TurnState.COMPACT, "ok"): TurnState.COMMAND,
    (TurnState.COMMAND, "dispatch"): TurnState.BUILD,
    (TurnState.COMMAND, "shortcut"): TurnState.DONE,
    (TurnState.BUILD, "ok"): TurnState.RUN,
    (TurnState.RUN, "ok"): TurnState.SAVE,
    (TurnState.SAVE, "ok"): TurnState.RESPOND,
    (TurnState.RESPOND, "ok"): TurnState.DONE,
}

驱动循环通过状态名动态查找对应的处理器：

while ctx.state is not TurnState.DONE:
    handler_name = f"_state_{ctx.state.name.lower()}"
    handler = getattr(self, handler_name, None)
    event = await handler(ctx)
    next_state = self._TRANSITIONS.get((ctx.state, event))
    ctx.state = next_state

TurnContext 数据携带

每个 Turn 通过 TurnContext 数据类在状态之间传递中间结果。它既是状态机的上下文，也是 Turn 的运行记录：

• msg: 原始入站消息
• session: 当前会话
• history: 会话历史
• initial_messages: 发送给 LLM 的消息列表
• final_content: LLM 的最终回复
• tools_used: 本轮使用的工具
• trace: 每个状态的执行时间和错误记录

性能追踪

每个状态执行后会记录 StateTraceEntry，包含起始时间、耗时和事件类型：

ctx.trace.append(StateTraceEntry(
    state=ctx.state,
    started_at=t0,
    duration_ms=duration,
    event=event,
))

这使得调试时可以精确到每个状态的性能瓶颈。

问题与规避

状态爆炸风险

问题：随着功能增加，状态数量可能膨胀，转换表变得难以维护。

Nanobot 的规避：保持 8 个核心状态不变，将复杂逻辑封装在单个状态处理器内部（如 BUILD 状态中加载历史、记忆、技能等全部上下文），而非创建更多状态。

命令短路

问题：命令执行后仍需继续 Turn，但某些命令应该直接结束。

规避：COMMAND 状态是唯一可以跳转到 DONE 的状态。短路命令（如 /help）直接返回 shortcut 事件，跳过 BUILD、RUN、SAVE 阶段。注意短路命令需要自行持久化消息到会话。

并发会话锁

问题：同一会话的多个消息可能并发处理，导致状态混乱。

规避：每个会话有独立的 asyncio.Lock，确保同一会话的消息串行处理，跨会话则通过 Semaphore 控制并发上限。

设计取舍

为什么用状态机而非单一函数

优势：

• 每个状态职责单一，易于测试和调试
• 转换表是显式的，新增状态不需要修改其他状态的逻辑
• 可以精确追踪每个阶段的耗时

代价：

• 状态之间需要通过 TurnContext 共享数据，增加了上下文携带成本
• 相比单一函数，代码行数更多

替代方案：直接写一个大的 async def process_message() 函数，按顺序执行各个步骤。Nanobot 选择状态机的原因是为了支持 /stop 命令在任意状态中断当前 Turn。

参考来源

• Nanobot AgentLoop 源码
• Agent 核心循环设计
• 事件驱动 Agent 架构