小节三：用 asyncio 并发异步 Tool——当 Tool 本身就是 async 时，事件循环更自然

ThreadPoolExecutor 很适合：

已有同步函数

但很多现代 Tool 原本就是异步的。

例如：

async def fetch_api(...):
    ...

或者：

异步 HTTP 客户端
异步数据库
异步消息队列

这时用：

asyncio

更自然。

什么是 asyncio？

可以先用一句很直观的话理解，一个线程里，任务遇到等待时主动让出执行权，让其他任务继续。

比如：

请求 API A
等待网络
  ↓
先去请求 API B
等待网络
  ↓
请求 API C

不需要为每个 I/O 都开线程。

最小例子

import asyncio


async def task(
    name: str,
    delay: float,
) -> str:
    await asyncio.sleep(
        delay
    )

    return (
        f"{name} done"
    )


async def main():
    results = await asyncio.gather(
        task("A", 1.0),
        task("B", 1.0),
        task("C", 1.0),
    )

    print(results)


if __name__ == "__main__":
    asyncio.run(
        main()
    )

三个任务各 1 秒。

串行大约 3 秒。

并发接近 1 秒。

用 asyncio.gather() 执行 Tool

如果 Tool 都是：

async def

可以：

results = await asyncio.gather(
    *coroutines
)

但有一个注意点。

默认情况下，如果某个 coroutine 抛异常，可能影响整体。

所以 Agent Harness 更适合：

return_exceptions=True

然后逐个转换成 ToolResult。

完整异步并行 Tool 示例

这个完整版本已经把并发上限放进批量执行器，而不是只在旁边讲一个 Semaphore。AsyncToolRegistry.execute() 会把普通工具异常转换成 ToolResult，所以某个调用失败不会直接打断 gather()；如果你的 Registry 允许异常继续向外抛，则要使用 return_exceptions=True 并逐项归一化，或在 run_one() 内完成捕获。取消信号不要当成普通 Tool Error 吞掉，它应该继续向上传播，让整批任务及时停止。

异步 Tool 的结构和线程池版本非常像，只是调度单位从 Future 换成 coroutine。真正异步的 HTTP、数据库和 SDK 调用，可以直接交给事件循环推进，再用 Semaphore 控制最大并发量。

这张图也能看出为什么不要把阻塞同步函数直接塞进事件循环：asyncio 的优势来自协程在等待 I/O 时主动让出执行权，如果函数本身一直阻塞，整个事件循环仍然会被卡住。

ThreadPoolExecutor 和 asyncio 怎么选？

可以先用一个很实用的判断：

已有同步 Tool
→ ThreadPoolExecutor

Tool 本身就是 async
→ asyncio

不要为了“看起来高级”强行把所有同步函数改成 async。

如果 Engine 已经是异步的，但某个旧 Tool 仍是同步 I/O，可以在 Python 3.9 中使用：

result = await asyncio.to_thread(
    sync_tool,
    **arguments,
)

它会把阻塞函数交给线程执行，避免直接卡住事件循环。需要注意，取消等待同样不保证线程里的函数立刻停止，因此 Tool 自身的网络、文件或子进程超时仍然要保留。

CPU 密集型任务呢？

例如：

巨大矩阵计算
压缩
视频编码
模型推理

线程不一定能带来理想提升。

这种任务可能更适合：

ProcessPoolExecutor

或者独立任务服务。

但这不是本节重点。

Agent Tool 最常见的并发收益主要来自，I/O。

所以先把 I/O 并发掌握。

asyncio 也要控制并发上限

假设一次：

1000 个 HTTP Tool Call

asyncio.gather() 真的可以瞬间发很多请求。

这也可能把外部服务打爆。

所以需要：

Semaphore

例如：

semaphore = asyncio.Semaphore(4)

每次最多 4 个并发。

带并发上限的例子

这就是并发控制。

Agent 可以提出 10 个 Action。

但 Harness 只允许：

同时 3 个

这仍然体现，模型提出动作，系统决定执行策略。