Python Agent 踩坑实录：一个 time.sleep 让我的并发工具链变成了“串行机器人”

一、问题背景：不是并发吗，怎么变成排队了？

最近在学 Python + Agent，写了一个多工具调用的 Demo：

• Agent 可以同时调度 3 个工具；
• 每个工具模拟耗时 2 秒；
• 我用了 asyncio.gather()，预期总耗时应该是 2 秒左右。

结果跑出来：

A start
A end
B start
B end
C start
C end
耗时: 6.00s

并发呢？

这个问题在生产环境中的表现比 Demo 严重得多：FastAPI 的 SSE 流式输出卡住、WebSocket 心跳超时断开、Agent 调用外部 API 时前端像死了一样没有任何 token 返回——很多时候，Agent “变慢”“卡死”“不流式输出”，不是模型的问题，而是事件循环被一行同步代码堵住了。

本文从现象出发，逐步定位根因，给出修复方案，最后还有一个 Agent 开发中的坑：异步子 Agent 的假同步等待。

二、最小复现代码： time.sleep 毁掉所有并发

先看错误版本：

import asyncio
import time

async def tool(name: str, seconds: int):
    print(f"{name} start")
    time.sleep(seconds)  # 问题在这里
    print(f"{name} end")
    return name

async def main():
    start = time.perf_counter()

    await asyncio.gather(
        tool("A", 2),
        tool("B", 2),
        tool("C", 2),
    )

    print(f"耗时: {time.perf_counter() - start:.2f}s")


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

输出：

A start
A end
B start
B end
C start
C end
耗时: 6.00s

三个工具本应同时运行，结果挨个排队执行。问题就出在这一行：

time.sleep(seconds)

三、原因分析：async def 不等于异步

asyncio 是单线程事件循环模型。当你写：

async def tool(...):
    time.sleep(2)

虽然函数是 async def，但 time.sleep(2) 是同步阻塞调用。它会卡住当前线程，事件循环在这 2 秒内完全无法工作，所有协程都被“饿死”。

更准确地说，事件循环的工作方式是：协程执行到 await 时，把控制权交还给事件循环，循环再去调度其他就绪的协程。而 time.sleep() 没有任何 await 操作，它只是让当前线程硬等 2 秒——这 2 秒内，事件循环完全没有机会切换到其他任务。

一句话总结：

async def 只是“可以异步”，不代表里面每一行代码都会自动异步。任何同步阻塞调用，都会让所有并发任务一起陪葬。

常见的同步阻塞调用

危险写法	所在库/场景	阻塞原因
time.sleep()	标准库	阻塞当前线程
requests.get()	requests	同步 HTTP 请求，阻塞整个事件循环
同步数据库驱动	pymysql / psycopg2	同步 I/O 等待
大文件同步读写	open().read()	阻塞磁盘 I/O
CPU 密集计算	纯 Python 计算	单线程模型下直接阻塞整个事件循环

四、修复方案：三种方式让事件循环“活”过来

方案一：换成 asyncio.sleep（最简单）

import asyncio
import time


async def tool(name: str, seconds: int):
    print(f"{name} start")
    await asyncio.sleep(seconds)  # 正确：把控制权交还事件循环
    print(f"{name} end")
    return name


async def main():
    start = time.perf_counter()

    await asyncio.gather(
        tool("A", 2),
        tool("B", 2),
        tool("C", 2),
    )

    print(f"耗时: {time.perf_counter() - start:.2f}s")


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

输出：

A start
B start
C start
A end
B end
C end
耗时: 2.00s

三个任务几乎同时开始，总耗时约 2 秒。

方案二：换异步 HTTP 库（Agent 工具调用推荐）

真实 Agent 项目里，工具函数经常要调外部 API。如果用 requests，同样会阻塞事件循环：

import requests

async def search_web(query: str):
    resp = requests.get(f"https://api.example.com/search?q={query}")  # 阻塞！
    return resp.json()

正确做法是换成 httpx：

import httpx

async def search_web(query: str):
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://api.example.com/search",
            params={"q": query},
            timeout=10,
        )
        return resp.json()

方案三：asyncio.to_thread 包装同步库（兼容旧 SDK）

如果底层 SDK 只有同步版本，可以用 asyncio.to_thread() 把同步函数丢到线程池执行，不阻塞事件循环：

import asyncio
import requests

def _sync_search(query: str):
    resp = requests.get(
        "https://api.example.com/search",
        params={"q": query},
        timeout=10,
    )
    return resp.json()

async def search_web(query: str):
    return await asyncio.to_thread(_sync_search, query)

注意：线程池默认只有 min(32, os.cpu_count() + 4) 个线程，并发量大时需要调整。CPU 密集任务仍然要考虑 ProcessPoolExecutor。

修复效果对比

场景	错误写法	耗时	正确写法	耗时
3 个工具，各等 2 秒	time.sleep(2)	6.00s	asyncio.sleep(2)	2.00s
10 个 HTTP 请求	requests.get()	20s+	httpx 异步	~2s

五、工程实践延伸：异步子 Agent 的“假同步等待”陷阱

time.sleep 的坑在单函数层面很容易排查，但在 Agent 多工具/多子 Agent 架构中，阻塞会以更隐蔽的方式出现。

同步子 Agent 模式

很多 Agent 框架支持“子 Agent”调度：主 Agent 把任务分发给子 Agent，子 Agent 执行完返回结果。如果子 Agent 是同步阻塞的：

# 错误：主 Agent 必须等子 Agent 全部完成才能继续
async def main_agent():
    result_a = await sub_agent_research()   # 等 10 秒
    result_b = await sub_agent_weather()    # 再等 5 秒
    return combine(result_a, result_b)      # 总耗时 15 秒

即使每个 await 本身不阻塞事件循环，但因为业务逻辑上必须顺序等待，整体耗时仍然是叠加的。

异步子 Agent 模式

正确的做法是让子 Agent 独立执行，主 Agent 不阻塞：

# 正确：子 Agent 异步调度，主 Agent 立即返回任务 ID
async def main_agent():
    task_a = asyncio.create_task(sub_agent_research())
    task_b = asyncio.create_task(sub_agent_weather())
    
    # 主 Agent 可以先做其他事，或者流式输出进度
    await asyncio.sleep(0)  # 让出控制权
    
    result_a = await task_a
    result_b = await task_b
    return combine(result_a, result_b)  # 总耗时 ~10 秒（取最长）

这种模式的核心区别在于：子 Agent 可以并行执行，主 Agent 只负责调度和结果聚合，不阻塞推理流程。在现代 Agent 框架中，异步子 Agent 会立即返回任务 ID，在远程服务器上独立执行，主 Agent 不会因为等待子任务而失去响应能力。

踩坑检查清单

检查项	危险信号	修复建议
工具函数内部	有 time.sleep()	换 await asyncio.sleep()
HTTP 调用	用了 requests	换 httpx.AsyncClient
数据库操作	同步驱动	异步驱动 / asyncio.to_thread
文件读写	同步 open()	aiofiles / asyncio.to_thread
子 Agent 调度	顺序 await 多个子任务	asyncio.create_task() 并行
CPU 密集计算	纯 Python 循环	ProcessPoolExecutor

六、排查技巧：如何确认事件循环被阻塞了

方法一：检查线程名

import threading

async def tool():
    print(threading.current_thread().name)

如果打印出来是 MainThread，并且里面执行了耗时同步操作，那基本就是它了。

方法二：开启 asyncio debug 模式

asyncio.run(main(), debug=True)

debug 模式会检测慢回调、未 await 的协程等问题。

方法三：在流式输出中插入计时点

如果是 FastAPI + SSE 场景，可以在生成器中插入计时：

async def event_stream():
    while True:
        start = time.perf_counter()
        chunk = await get_next_chunk()
        elapsed = time.perf_counter() - start
        if elapsed > 1.0:
            print(f"[WARN] chunk generation blocked for {elapsed:.2f}s")
        yield chunk

当 elapsed 异常大时，说明事件循环在等待期间被其他同步操作占用了。

七、总结

这次踩坑让我记住了一句话：

在 asyncio 里，任何同步阻塞调用，都会让所有并发任务一起陪葬。

写 Python Agent 时，工具函数不是加个 async 就完事了。真正要关注的是：里面有没有阻塞事件循环的同步调用。

核心要点回顾：

1. async def 不等于异步，time.sleep() 会阻塞整个事件循环；
2. 三个修复方案：asyncio.sleep → httpx → asyncio.to_thread；
3. Agent 多工具场景下，子 Agent 调度也要避免“假同步等待”；
4. 排查手段：线程名检查、asyncio debug 模式、流式计时点。

把 time.sleep 换成 asyncio.sleep，把 requests 换成 httpx，或者用 asyncio.to_thread 包装同步代码——你的 Agent 才会真正并发起来。

如果你也在写 Agent，建议先从这个小坑开始排查。很多时候，“Agent 变慢”“卡死”“不流式输出”，不是模型的问题，而是事件循环被一行同步代码堵住了。