日志、RuntimeEvent 与测试策略
对应源码
| 主题 | 源码 |
|---|---|
| Turn 日志和 state trace | nanobot/agent/loop.py::AgentLoop._process_message |
| Hook、progress 和可观测回调 | nanobot/agent/hook.py、nanobot/agent/progress_hook.py |
| 用户消息与运行事件 | nanobot/bus/events.py、nanobot/bus/runtime_events.py |
| WebSocket 事件适配 | nanobot/channels/websocket.py |
| CLI/Gateway 日志入口 | nanobot/cli/commands.py::agent、nanobot/cli/gateway.py |
| 分层测试 | tests/agent/、tests/providers/、tests/channels/、tests/tools/、tests/security/、tests/session/ |
学习目标
读完后,你应当能够:
- 区分日志、State trace、Runtime Event、MessageBus 消息和持久化历史;
- 为一次 Turn 选择最小观察组合;
- 用相邻测试核对源码行为,而不是只读实现;
- 为新功能选择合适的测试层级。
前置知识
- 完成 保存、响应与 RuntimeEvent;
- 能运行单个
pytest文件或测试函数; - 知道 mock/fake Provider 的用途:控制模型响应,避免真实网络和不稳定输出。
1. 五类证据不要混用
| 证据 | 回答的问题 | 主要位置 |
|---|---|---|
| 日志 | 运行时发生了什么异常或关键动作? | loguru.logger 调用 |
StateTraceEntry | Turn 在哪个状态停留多久、是否出错? | nanobot/agent/loop.py |
RuntimeEvent | UI/订阅者应看到什么高层运行状态? | nanobot/bus/runtime_events.py |
Inbound/OutboundMessage | 业务消息如何在入口与核心之间流动? | nanobot/bus/events.py |
| Session/Memory 文件 | 哪些状态在重启后仍然存在? | workspace 下持久化文件 |
例如,Tool 正在运行可以同时出现在日志、Agent progress hook 和 checkpoint 中,但三者的消费者和寿命不同。当前 RuntimeEventBus 本身不定义 Tool 进度事件。
2. 日志:给开发者看的叙事
nanobot 使用 Loguru。CLI 初始化时移除默认 handler,再添加带时间、级别和来源的输出。agent 命令提供:
--logs / --no-logs对应位置:
nanobot/cli/commands.py顶部 logger 配置;nanobot/cli/commands.py::agent;nanobot/cli/gateway.py的文件/服务日志设置。
调试一次请求时,优先搜索这些关键句:
Processing message fromRegistered ... toolsLLM transient errorTask cancelled for sessionError processing message for sessionShutdown: flushed ... session(s)
日志关闭不等于事件关闭
--no-logs控制 nanobot logger 是否输出,不会移除 MessageBus、RuntimeEventBus 或 Session 保存逻辑。
3. State trace:Turn 状态机的局部剖面
TurnContext.trace 保存 StateTraceEntry,字段包括:
stateduration_mseventerror
AgentLoop._process_message() 在每个 state handler 前后计时并追加条目。它最适合回答:
- 卡在 BUILD、RUN 还是 SAVE?
- 某个状态耗时是否异常?
- 状态 handler 是否抛错?
它不是全局分布式 tracing,也不会自动串联外部 Provider 和 Tool 服务的内部 span。
4. Runtime Event:给运行表面看的结构化事件
RuntimeEventBus 发布结构化 dataclass 事件,WebSocket coordinator 等订阅者再把它转换成对应协议。当前精确事件集合是:
SessionTurnStarted;TurnRunStatusChanged;TurnCompleted;GoalStateChanged;RuntimeModelChanged。
文本 delta、reasoning delta 和 Tool start/end 不属于 RuntimeEventBus dataclass。它们通过 AgentHook、ProgressHook 和调用方传入的 callback 传播,再由 SDK、Channel 或 WebSocket adapter 转成各自事件。
核心原则是:
Agent core 的高层状态留在 RuntimeEventBus;细粒度流式进度留在 hook/callback;WebUI 特有的 envelope、sequence、replay 和 JSON 字段留在 WebSocket 边界。
这使 SDK、WebUI 或以后新增的运行表面可以复用同一核心事件,而不把前端协议塞进 AgentRunner。
关键位置:
nanobot/bus/runtime_events.pynanobot/agent/hook.pynanobot/agent/progress_hook.pynanobot/channels/websocket.py
5. usage 与 latency
Runner 累积各轮 Provider usage。若 Provider 没有返回 usage,当前实现会估算并标记 estimated_tokens。AgentLoop 保存 _last_usage,进度 hook 会记录 prompt/completion/cached token。
Turn latency 会被:
- 由
RuntimeEventPublisher暂存,并随TurnCompleted发布; - 在保存阶段写入最后一条 assistant message;
- 供 WebUI transcript 展示。
调试成本问题时,要先区分 Provider 原始 token 与 nanobot 估算值,不能把 estimated_tokens 当作账单事实。
6. 一次 Turn 的最小观察组合
初学阶段建议只开四个观察点:
flowchart LR A["入口消息<br/>channel / session_key"] --> B["Turn trace<br/>state / duration"] B --> C["Runner 结果<br/>stop_reason / tools / usage"] C --> D["Session JSONL<br/>最终持久化"]
只有当问题涉及 WebUI 进度或断线恢复时,再加入 Runtime Event 和 WebSocket frame;涉及 Provider 暂时性错误时,再加入 retry metadata。
这样能避免在几十种事件中迷失。
7. 测试目录就是可执行说明书
nanobot 的测试大体按源代码职责镜像:
| 主题 | 优先读的测试 |
|---|---|
| Turn 状态与保存 | tests/agent/test_loop_runner_integration.py、test_loop_save_turn.py |
| Runner 核心 | tests/agent/test_runner_core.py |
| 工具调用 | tests/agent/test_runner_tool_execution.py |
| 注入 | tests/agent/test_runner_injections.py |
| Runtime Event | tests/agent/test_loop_progress.py |
| Session | tests/session/ |
| Provider | tests/providers/ |
| WebSocket | tests/channels/test_websocket_channel.py |
| Tool | tests/tools/ |
| 安全 | tests/security/ 与安全相关 tool tests |
源码告诉你“如何实现”;测试更直接地告诉你维护者承诺“哪些输入应产生哪些结果”。
8. 推荐的测试阅读法
对一个行为,按以下顺序:
- 找一个名字最接近问题的测试;
- 识别 fake Provider、fixture 和初始 Session;
- 写下触发动作;
- 只看断言,先描述可观察契约;
- 再回源码解释断言为何成立;
- 修改实验时,先把所需的最小测试与依赖代码复制到
learning-CN/examples或独立临时工作区;不要直接修改code-EN,也不必一开始就运行整套测试。
示例:想理解并行工具,不应从 asyncio.gather 开始猜,而应先读:
tests/agent/test_loop_tool_context.pytests/agent/test_runner_tool_execution.py
然后核对 Tool.concurrency_safe 和 _partition_tool_batches()。
9. 分层验证
| 改动类型 | 最小验证 |
|---|---|
| 纯 Tool 逻辑 | 对应 tests/tools/test_*.py |
| Runner 行为 | fake Provider + tests/agent/test_runner_*.py |
| Loop 状态/Session | tests/agent/test_loop_*.py + 相邻 session tests |
| Channel 协议 | 对应 Channel test,不调用真实平台 |
| WebUI 数据契约 | Python WebSocket/API tests + webui 前端 tests |
| 打包 | Python package build + WebUI build/output 检查 |
真实 DeepSeek 请求适合做最终 smoke test,不适合替代确定性的单元测试。
10. 边界
测试通过不证明生产环境安全
单元测试通常控制 DNS、文件系统、Provider 和时间。真实系统还需要配置审查、权限边界、密钥保护和部署层 sandbox。
Runtime Event 不是审计日志
它服务于运行进度和 UI 状态,可能是短生命周期、内存内、可重放窗口有限的事件流。需要长期审计时不能只依赖它。
自检
- “模型调用很慢”首先看日志还是 State trace?为什么?
- Provider 不返回 token usage 时,文档应怎样描述该值?
- 为什么 WebSocket envelope 不应该由 AgentRunner 直接构造?
- 新增一个 Tool 时,为什么真实 DeepSeek 请求不是最小测试?