第 05 章《事件协议:InboundMessage + OutboundEvent》· AI Agent 事件协议怎么做？10 种 OutboundEvent 联合类型实战（nanobot）

H咖肥

318人浏览 · 2026-08-12 10:53:03

H咖肥 · 2026-08-12 10:53:03 发布

本文回答什么问题:MessageBus 上跑的"消息体"长什么样?InboundMessage 有哪些字段?OutboundEvent 10 种事件怎么用?

预计阅读时间:12 分钟
源码版本:GitHub HKUDS/nanobot main 分支主线代码(仓库相对路径)

第 04 章讲了 MessageBus 的"骨架",本章讲"血肉"——总线上跑的 InboundMessage / OutboundMessage / OutboundEvent 三类消息体长什么样。

1. 整体定位:为什么需要统一的事件协议

如果每个通道自定义消息体,AgentLoop 要写 17 个 if-else 来适配。统一协议让 AgentLoop 只关心"消息体长这样"——通道层做协议转换,业务层做事件分发。

核心要点速查(建议收藏)

3 个核心 dataclass:InboundMessage(nanobot/bus/events.py L23-L39)、OutboundMessage(L42-L58)、OutboundEvent 联合(L58)
10 种 OutboundEvent 子类(nanobot/bus/outbound_events.py L17-L89):Progress / StreamDelta / StreamEnd / RetryWait / TurnEnd / GoalStatus / GoalStateSync / SessionUpdated / RuntimeModelUpdated / TurnModelUpdated
session_key 默认格式:f"{channel}:{chat_id}",可 session_key_override 覆盖
runtime_control 元数据:metadata["_runtime_control"] 让内部指令走通道(如 mcp_reload)
2 个工厂函数:outbound_message_for_event() / outbound_event_from_message()(L92-L108 / L110-L130)

2. InboundMessage:用户发来的消息

# nanobot/bus/events.py L23-L39
@dataclass
class InboundMessage:
    channel: str                                    # 通道名 "telegram" / "discord" / ...
    sender_id: str                                  # 用户 ID
    chat_id: str                                    # 会话 ID(DM / 群 ID)
    content: str                                    # 文本内容
    timestamp: datetime = field(default_factory=datetime.now)
    media: list[str] = field(default_factory=list)  # 媒体 URL 列表
    metadata: dict[str, Any] = field(default_factory=dict)
    session_key_override: str | None = None        # 可选 session 覆盖

    @property
    def session_key(self) -> str:
        return self.session_key_override or f"{self.channel}:{self.chat_id}"

关键字段:

channel / sender_id / chat_id:三元组定位"哪个用户在哪个通道的哪个会话"
media:图片 / 文件 / 音频 URL 列表(经通道 SDK 转换)
metadata["_runtime_control"]:内部指令(详见 §5.3)
session_key:默认拼接,可覆盖用于 thread-scoped session

3. OutboundMessage:Agent 回复的消息

# nanobot/bus/events.py L42-L58
@dataclass
class OutboundMessage:
    channel: str
    chat_id: str
    content: str
    event: OutboundEvent | None = None              # 类型化事件(可选)
    metadata: dict[str, Any] = field(default_factory=dict)

event 字段是关键:10 种 OutboundEvent 子类(详见 §4),通道实现根据 event.type 决定如何渲染——流式打字 / 显示进度 / 重试等待等。

4. OutboundEvent:10 种类型化事件

# nanobot/bus/outbound_events.py L17-L89
class ProgressEvent(OutboundEvent):           # 进度更新(进度条)
class StreamDeltaEvent(OutboundEvent):        # 流式文本片段
class StreamEndEvent(OutboundEvent):          # 流式结束
class RetryWaitEvent(OutboundEvent):          # 重试等待(显示 spinner)
class TurnEndEvent(OutboundEvent):            # 回合结束(含 latency_ms)
class GoalStatusEvent(OutboundEvent):         # 长期目标状态变化
class GoalStateSyncEvent(OutboundEvent):      # 目标状态同步
class SessionUpdatedEvent(OutboundEvent):     # session 元信息更新(标题 / 标签)
class RuntimeModelUpdatedEvent(OutboundEvent):# 模型热更新
class TurnModelUpdatedEvent(OutboundEvent):   # 单回合模型切换

典型用法(以 StreamDelta 为例):

# AgentRunner 流式输出
async for delta in provider.stream(messages):
    await self.bus.publish_outbound(OutboundMessage(
        channel="telegram",
        chat_id=chat_id,
        content="",          # delta 在 event.content
        event=StreamDeltaEvent(content=delta.text, delta_id=delta.id),
    ))

通道收到 event.type == "stream_delta" 后,会编辑原消息显示增量文本(而不是发新消息)。

5. 关键模式与决策

5.1 工厂函数:`outbound_message_for_event()`(L92-L108)

def outbound_message_for_event(
    *,
    channel: str,
    chat_id: str,
    event: OutboundEvent,
    content: str | None = None,
    metadata: Mapping[str, Any] | None = None,
) -> OutboundMessage:
    """Build an :class:`OutboundMessage` for a typed event."""
    return OutboundMessage(
        channel=channel,
        chat_id=chat_id,
        content=_event_content(event) if content is None else content,
        event=event,
        metadata=dict(metadata or {}),
    )

为什么用工厂:让 event + content 自动绑定,避免调用方忘记同步字段。

5.2 反向提取:`outbound_event_from_message()`(L110-L130)

def outbound_event_from_message(msg: OutboundMessage) -> OutboundEvent | None:
    """Extract typed event from message (or None if plain text)."""
    return msg.event

作用:中间件(如 hook)想看消息携带什么事件类型时,直接调这个函数即可。

5.3 runtime_control 内部指令

# nanobot/bus/events.py L17-L20
metadata: dict[str, Any]
# 当 metadata["_runtime_control"] 为真时,AgentLoop 走"内部指令"分支
# 支持的指令:mcp_reload / image_generation_reload / provider_reload

用法:内部通道(如 WebUI 配置面板)发 InboundMessage(metadata={"_runtime_control": True}, content="mcp_reload"),AgentLoop 重新加载 MCP 服务而非当作用户输入。

5.4 为什么用 dataclass 联合?

类型安全:OutboundEvent 联合保证 10 种之一,Python 类型检查可发现遗漏
自带数据:每个事件 dataclass 自带字段(如 StreamDeltaEvent.content),不需要 metadata 字典字符串传
演进友好:新增 TurnEndEvent(latency_ms=...) 不破坏老通道

6. 常见问题 / 避坑

Q:OutboundEvent 有 10 种,我通道必须实现全部?

A:。通道只关心自己关心的类型——event is None 表示纯文本消息,直接发 content 即可。要做流式就只处理 StreamDeltaEvent / StreamEndEvent;要做 spinner 就只处理 RetryWaitEvent。

Q:`session_key_override` 怎么用?

A:用于线程级 session(同一 chat_id 不同 topic)。WebUI 支持 thread,可以在 metadata["thread_id"] 设置后,AgentLoop 自动用 f"{channel}:{chat_id}:{thread_id}" 作为 session_key。

Q:`media` 字段是 URL 还是字节?

A:。通道收到 URL 后自行下载并 send(如 Telegram 调用 sendPhoto(photo=url))。音频转录用 transcription.py 处理。

7. 小结

InboundMessage:channel/sender_id/chat_id/content/media/metadata/session_key_override
OutboundMessage:channel/chat_id/content/event/metadata
10 种 OutboundEvent:Progress / StreamDelta / StreamEnd / RetryWait / TurnEnd / GoalStatus / GoalStateSync / SessionUpdated / RuntimeModelUpdated / TurnModelUpdated
2 个工厂函数:outbound_message_for_event() / outbound_event_from_message()
runtime_control 让内部指令走通道而非用户输入

本文要点速查

3 个核心 dataclass:InboundMessage / OutboundMessage / OutboundEvent 联合
10 种 OutboundEvent 让通道按类型渲染(流式 / 进度 / 重试 / …)
runtime_control 让 MCP / provider reload 等内部指令走通道
下一步:第 06 章《配置子系统 Config》—— 这些 dataclass 的字段约束在哪里

按角色推荐

系统架构师:必读(§5 的 dataclass 联合 + runtime_control 是核心契约)
LLM Agent 开发者:必读(所有 Agent 行为都基于这 3 个 dataclass)
LLM Provider 适配者:选读(知道 OutboundEvent 是 stream / progress 载体即可)
聊天通道开发者:必读(必须实现 §4 的 10 种事件分发)
Tool / MCP 工具开发者:选读(知道 metadata[“_runtime_control”] 用于 reload)

下一步

第 06 章《配置子系统 Config》 —— Pydantic 配置 + 5 个常见误区(主题群"架构与基础设施",第 1 周)
第 10 章《AgentLoop 编排核心》 —— InboundMessage / OutboundMessage 在 AgentLoop 怎么流转(主题群"Agent 核心",第 3 周)
第 17 章《LLMProvider 抽象》 —— Provider 调用产出 LLMResponse(主题群"LLM Provider",第 4 周)

tags:#nanobot #AI Agent #LLM #Python #源码解析 #事件协议 #dataclass