AI Agent Engineer

AI Agent Engineer

消息、Session Key 与入口

5 min read

消息、Session Key 与入口

对应源码

主题源码
通用消息对象nanobot/bus/events.py::InboundMessage、OutboundMessage
异步消息队列nanobot/bus/queue.py::MessageBus
Session Key 规则nanobot/session/keys.py、nanobot/agent/loop.py::AgentLoop._effective_session_key
bus 入口与并发分派nanobot/agent/loop.py::AgentLoop.run、AgentLoop._dispatch
direct 入口nanobot/agent/loop.py::AgentLoop.process_direct

学习目标

读完后，你应当能够：

  • 解释 InboundMessage 和 OutboundMessage 的职责；
  • 说明默认 session key 如何产生；
  • 区分 MessageBus 路径和 process_direct() 路径；
  • 找到 unified session 和 override 影响 Session 归属的位置。

前置知识

1. InboundMessage 是入口统一格式

定义位置：

  • nanobot/bus/events.py::InboundMessage

核心字段：

字段含义
channel消息来源，如 cli、api、websocket
sender_id发送者标识
chat_id对话或频道标识
content用户文本或系统注入文本
timestamp入站时间
media媒体路径或引用
metadataChannel/运行时附加数据
session_key_override显式覆盖默认 Session 归属

默认属性：

session_key_override or f"{channel}:{chat_id}"

这意味着 Session 默认按“入口类型 + 对话标识”隔离。

sender_id 默认不参与 session key

Session 归属默认使用 channel 和 chat_id，不是发送者 ID。群聊、线程和 Channel adapter 如何选择 chat_id，会直接影响上下文隔离。

2. 有效 Session Key 可能进一步变化

InboundMessage.session_key 是消息对象给出的默认值，但 AgentLoop 还会计算 effective session key，用于：

  • unified session；
  • thread-scoped override；
  • Subagent 或内部 continuation；
  • 某些运行表面的显式 session ID。

因此调试 Session 串线时，要同时观察：

msg.session_key
effective_session_key
SessionManager 实际读取的 key
磁盘文件中的 metadata.key

相关源码：

  • nanobot/session/keys.py
  • nanobot/agent/loop.py::_effective_session_key
  • nanobot/agent/loop.py::_dispatch

3. OutboundMessage 是通用投递格式

定义位置：

  • nanobot/bus/events.py::OutboundMessage

核心字段：

  • channel
  • chat_id
  • content
  • reply_to
  • media
  • metadata
  • buttons

metadata 可能承载：

  • 消息路由信息；
  • stream/progress 标记；
  • latency；
  • Channel 无关的 _agent_ui 结构化数据。

这些内部 metadata 不应默认视为稳定公共 API。

4. MessageBus 只是两个 Queue

nanobot/bus/queue.py::MessageBus 维护：

inbound: Queue[InboundMessage]
outbound: Queue[OutboundMessage]

职责很小：

  • 发布入站；
  • 消费入站；
  • 发布出站；
  • 消费出站；
  • 暴露队列长度。

它不负责：

  • Session 持久化；
  • 消息重放；
  • Runtime Event；
  • Provider 调用；
  • Channel 权限；
  • retry。

这种小职责使 Channel 和 Agent 核心解耦。

5. 两种进入 AgentLoop 的方式

方式 A：MessageBus

flowchart LR
    CH["Channel / 交互式 CLI"] --> IN["bus.publish_inbound"]
    IN --> RUN["AgentLoop.run"]
    RUN --> DIS["_dispatch"]
    DIS --> PM["_process_message"]
    PM --> OUT["bus.publish_outbound"]
    OUT --> CM["ChannelManager / CLI consumer"]

AgentLoop.run()：

  • 持续消费 inbound；
  • 处理 priority command 和 runtime control；
  • 将同一活跃 session 的新消息放入 pending queue；
  • 为其他消息创建 Task；
  • 最终由 _dispatch() 取得 session lock。

方式 B：process_direct()

flowchart LR
    E["CLI -m / API / SDK"] --> PD["process_direct"]
    PD --> LOCK["session lock"]
    LOCK --> PM["_process_message"]
    PM --> RET["直接返回结果"]

它仍然复用同一个 _process_message() 和状态机，只是入口不依赖 Channel 的 inbound/outbound dispatcher。

6. 同 Session 串行

AgentLoop._dispatch() 为 effective session key 取得独立 asyncio.Lock：

同一 key → 共用一把 lock → Turn 串行
不同 key → 不同 lock → 可以并发

此外还有全局并发 gate，限制同时处理的请求数量。

同一 session 在 Turn 进行中收到新消息时，可能进入该 session 的 pending queue，而不是启动竞争 Turn。后续在 AgentRunner 的安全边界处注入。

7. 各入口的 Session 习惯

入口典型 key
CLI 单次显式 --session，默认 cli:direct
Channel通常为 <channel>:<chat_id>
APIapi:<session_id>，默认 api:default
SDK由 SDK 参数或默认值决定
Subagent 回传使用 override 对齐主 Session

不要假设 CLI、API 和 WebUI 会天然共享历史。只有相同 effective key 或启用 unified session 时才可能共享。

8. 观察任务

在 AgentLoop.process_direct() 或 _dispatch() 设置断点，观察：

  • msg.channel
  • msg.chat_id
  • msg.session_key
  • session_key
  • self._session_locks.keys()
  • self._pending_queues.keys()

分别运行：

nanobot agent -c $nanobotConfig -w $nanobotWorkspace -s "cli:lesson-a" -m "记住 A"
nanobot agent -c $nanobotConfig -w $nanobotWorkspace -s "cli:lesson-b" -m "我让你记住了什么？"

预期：不同 session 不应依赖彼此的对话历史。

源码与测试

主题位置
消息 dataclassnanobot/bus/events.py
Queuenanobot/bus/queue.py
Session key helpernanobot/session/keys.py
主消费循环AgentLoop.run
session 串行AgentLoop._dispatch
direct 入口AgentLoop.process_direct
unified session 测试tests/agent/test_unified_session.py
workspace/session scopetests/agent/test_workspace_scope.py

自检

  1. MessageBus 为什么不保存历史？
  2. 为什么 sender_id 和 chat_id 不能混为一谈？
  3. process_direct() 为什么仍能复用 AgentLoop 状态机？
  4. 同一 session 的新消息为什么不应直接启动另一个竞争 Turn？

下一步

阅读 AgentLoop 状态机。

Backlinks