AI Agent Engineer

AI Agent Engineer

asyncio 并发、取消与资源清理

8 min read

asyncio 并发、取消与资源清理

对应源码

主题源码
总队列消费和任务创建nanobot/agent/loop.py::AgentLoop.run、_dispatch
Session 串行与任务取消nanobot/agent/loop.py::_cancel_active_tasks
Tool 批次并发nanobot/agent/runner.py::_partition_tool_batches、_execute_tools
外部进程和 Shellnanobot/agent/tools/shell.py::ExecTool
MCP 资源清理nanobot/agent/loop.py::close_mcp
Gateway 关闭nanobot/cli/commands.py::_run_gateway

学习目标

读完后，你应当能够：

  • 解释“同一 Session 串行、不同 Session 并行”的实现；
  • 说明同一 Turn 内多个 Tool Call 何时可以并行；
  • 跟踪 /stop 到 Task.cancel() 和 checkpoint 恢复；
  • 找出 Gateway 正常关闭时的清理顺序。

前置知识

  • 知道协程只有在 await 等让出控制权时才会切换；
  • 知道 asyncio.Lock 保护临界区，Semaphore 限制并发数；
  • 知道取消会在被取消的协程中抛出 asyncio.CancelledError，因此清理后通常还要继续抛出。

1. 先区分四种并发

nanobot 中至少有四个不同层面的并发：

层面是否并行控制位置
不同 Session 的 Turn可以AgentLoop.run() 与 _dispatch()
同一 Session 的多个 Turn串行；新消息还可能作为 mid-turn injection_session_locks、_pending_queues
同一次模型响应中的多个 Tool Call仅安全工具可组成并发批次AgentRunner._partition_tool_batches()
Gateway 的 Agent、Channel、WebUI、健康检查并发运行nanobot/cli/commands.py::_run_gateway() 内的 tasks

把这四层混成一句“nanobot 是异步的”，会让调试几乎失去方向。

2. 从总队列到每个 Session

AgentLoop.run() 持续从 MessageBus.consume_inbound() 取消息。普通消息会创建一个 _dispatch(msg) task，因此主循环还能继续接收 /stop 或其他会话的新消息。

sequenceDiagram
    participant Bus as MessageBus
    participant AL as AgentLoop.run
    participant A as Session A Task
    participant B as Session B Task

    Bus->>AL: A 的消息
    AL->>A: create_task(_dispatch)
    Bus->>AL: B 的消息
    AL->>B: create_task(_dispatch)
    Note over A,B: 两个 Session 可并行推进

_dispatch() 随后取得两个限制器：

  1. _session_locks[session_key]：保证同一 Session 只有一个 owner；
  2. _concurrency_gate：限制整个 AgentLoop 同时处理的请求数。

全局并发上限来自环境变量 NANOBOT_MAX_CONCURRENT_REQUESTS，源码默认值是 3；设置为 0 或负数时不创建 Semaphore，即不做这层限制。

两个限制器解决不同问题

Session lock 保护会话顺序与状态；全局 semaphore 保护 Provider、CPU、内存和下游工具不被无限并发压垮。

3. 同一 Session 的后续消息

当 Session 正在运行时，AgentLoop 会公布一个最大长度为 20 的 _pending_queues[session_key]。后续普通消息优先进入该队列，供 AgentRunner 在迭代边界注入，而不是启动一个竞争 Turn。

特殊情况：

  • priority command 会内联处理；
  • 可分派命令不会作为自然语言注入；
  • 队列满时，源码回退为排队等待 Session lock 的新 task；
  • Turn 结束后，尚未消费的消息会重新发布到 inbound bus，避免静默丢失。

这说明 _pending_queues 不是普通消息积压队列，而是“正在运行 Turn 的可注入邮箱”。

4. Tool Call 并发不是默认全开

主循环创建 AgentRunSpec 时启用 concurrent_tools=True，但 AgentRunner 不会无条件 gather() 所有工具。

_partition_tool_batches() 检查每个 Tool 的 concurrency_safe：

  • 可安全并发的相邻调用可以进入同一批次，由 asyncio.gather() 执行；
  • 非并发安全的 Tool 单独形成批次；
  • 各批次仍按顺序运行。

Tool.concurrency_safe 的基类默认值位于：

  • nanobot/agent/tools/base.py::Tool.concurrency_safe

一个典型理由是：读操作可能并行安全，而会修改共享文件、进程或会话状态的操作通常需要串行。新增 Tool 时不能因为它是 async def 就宣称并发安全。

5. /stop 如何变成取消

控制命令最终调用 AgentLoop._cancel_active_tasks(key)，从 _active_tasks 取出该 Session 尚未完成的 task 并调用 cancel()。

取消随后沿调用栈传播：

flowchart TD
    A["/stop"] --> B["_cancel_active_tasks"]
    B --> C["Task.cancel()"]
    C --> D["AgentRunner / Tool / Provider 收到 CancelledError"]
    D --> E["ExecTool 杀死并回收子进程"]
    D --> F["AgentLoop 尝试恢复 runtime checkpoint"]
    E --> G["重新抛出 CancelledError"]
    F --> G

关键实现：

  • AgentLoop._cancel_active_tasks()
  • AgentLoop._dispatch() 的 except asyncio.CancelledError
  • AgentRunner.run() 与 _run_tool() 的取消分支
  • ExecTool.execute() 与 _kill_process()

ExecTool 在超时或取消时都会终止 subprocess，并等待进程退出，避免遗留僵尸进程。取消分支完成清理后再次 raise，让上层知道这次 Turn 并未正常结束。

6. 为什么取消后还能看到部分过程

工具循环期间，Runner 会把 checkpoint 写入 Session metadata。AgentLoop._dispatch() 捕获取消后，调用 _restore_runtime_checkpoint()，尝试把已经形成的 assistant/tool 消息恢复到 Session 历史，再保存 Session。

因此，取消的语义不是“回滚整个 Turn”，而更接近：

停止后续执行，同时尽力保留已经发生且有教学价值的中间结果。

这也是 Agent 系统与数据库事务的重要区别。真实世界的工具副作用通常无法自动撤销。

7. Gateway 正常关闭

Gateway 启动时并发运行 AgentLoop、ChannelManager，以及按配置启用的 WebUI/健康服务。收到关闭信号后，nanobot/cli/commands.py 的 finally 路径会执行清理，核心顺序包括：

  1. 取消仍运行的 runtime tasks；
  2. gather(..., return_exceptions=True) 等待它们收尾；
  3. 停止 Channel；
  4. 关闭 MCP 连接和后台任务；
  5. SessionManager.flush_all() 以 fsync=True 重写缓存中的 Session；
  6. 恢复信号处理器。

AgentLoop.close_mcp() 特别说明：stdio MCP transport 使用 AnyIO cancel scope，应该在打开它的 task 中关闭。

8. 调试方法

观察跨 Session 并发

用两个不同 Session 同时发送一个会等待数秒的请求，给以下位置打断点：

  • AgentLoop.run() 创建 task 的位置；
  • AgentLoop._dispatch() 的 async with lock, gate；
  • AgentRunner._request_model()。

记录每个 task 的 session_key 和进入时间。预期：不同 key 可以重叠，同一 key 不会同时拥有 lock。

观察取消

给一个会执行长命令的 Turn 发送 /stop，观察：

  • _cancel_active_tasks() 返回的取消数；
  • ExecTool._kill_process()；
  • _restore_runtime_checkpoint()；
  • Session JSONL 是否保存了取消前的 tool result。

不要用具有破坏性副作用的命令做取消实验

取消只能停止尚未完成的进程，不能可靠撤销已经写入外部系统的数据。

9. 源码与测试

源码：

  • nanobot/agent/loop.py::run
  • nanobot/agent/loop.py::_dispatch
  • nanobot/agent/loop.py::_cancel_active_tasks
  • nanobot/agent/loop.py::close_mcp
  • nanobot/agent/runner.py::_execute_tools
  • nanobot/agent/runner.py::_partition_tool_batches
  • nanobot/agent/tools/shell.py::ExecTool
  • nanobot/cli/commands.py::_run_gateway

测试：

  • tests/agent/test_unified_session.py
  • tests/agent/test_runner_injections.py
  • tests/agent/test_stop_preserves_context.py
  • tests/agent/test_runner_tool_execution.py
  • tests/agent/test_loop_tool_context.py
  • tests/session/test_session_fsync.py

自检

  1. 为什么只有 _session_locks 还不够？
  2. 一个 Tool 是只读操作，是否就一定能把 concurrency_safe 设为 True？
  3. 取消 Turn 为什么不能等同于回滚？
  4. _pending_queues 中未消费的消息为什么要重新发布？

下一篇：错误、重试与数据一致性

Backlinks