Python Agent 踩坑实录：asyncio.wait_for 超时后，任务真的停了吗？

Python Agent 踩坑实录：asyncio.wait_for 超时后，任务真的停了吗？

摘要：本文从一次"超时了但 CPU 还在飙"的事故出发，分析 Agent 开发中 asyncio.wait_for 超时后任务没有真正停止的四个隐蔽原因：同步阻塞代码无法被取消、CancelledError 被 except Exception 吞掉、finally 里的 await 被取消、子任务未被级联取消。文章给出超时后的正确清理姿势，以及 Agent 工具调用的超时防护方案。附完整可复用代码和排查清单，适合正在学 Python + Agent 的开发者避坑。

文章目录

• • 一、问题背景：超时了，但任务还在跑
  • 二、最小复现代码：超时了但任务还在打印
  • 三、四个隐蔽的"取消失败"场景
  • • 场景一：同步阻塞代码无法被取消
    • 场景二：CancelledError 被 except Exception 吞掉
    • 场景三：finally 里的 await 被取消
    • 场景四：子任务未被级联取消
  • 四、修复方案：正确的超时与取消
  • • 方案一：用 asyncio.timeout 上下文管理器（Python 3.11+）
    • 方案二：确保任务可取消
    • 方案三：超时后显式清理
  • 五、Agent 工具调用的完整超时防护
  • 六、Agent 开发中的特殊场景
  • • 场景一：LLM API 调用超时
    • 场景二：流式输出的超时
  • 七、总结与排查清单
  • • 核心结论
    • 排查清单
    • 推荐实践
  • 八、写在最后

一、问题背景：超时了，但任务还在跑

最近在写一个 Agent 项目，工具调用需要设置超时，避免某个工具卡死拖垮整个 Agent。

我写了这样的代码：

import asyncio

async def call_tool(tool_name: str, timeout: float = 5.0):
try:
return await asyncio.wait_for(
_execute_tool(tool_name),
timeout=timeout,
)
except asyncio.TimeoutError:
return {"error": f"工具 {tool_name} 超时"}

测试的时候发现一个诡异现象：

• 5 秒后确实返回了超时错误；
• 但后台的 _execute_tool 还在跑；
• 日志里能看到工具在超时后仍然打印输出；
• CPU 占用没有下降；
• 如果连续触发 10 次超时，后台就累积了 10 个"僵尸任务"。

更糟的是，某些工具调用了外部 API，超时后请求并没有真正取消，外部 API 侧还在计费。

排查后发现：asyncio.wait_for 超时后，只是从等待者的视角结束了，任务本身不一定真的停了。

二、最小复现代码：超时了但任务还在打印

先看错误版本：

import asyncio

async def long_running_task():
try:
print("任务开始")
for i in range(10):
await asyncio.sleep(1)
print(f"任务运行中… {i}")
print("任务完成")
return "done"
except asyncio.CancelledError:
print("任务被取消")
raise

async def main():
try:
result = await asyncio.wait_for(long_running_task(), timeout=3.0)
print(f"结果: {result}")
except asyncio.TimeoutError:
print("超时了！")

# 等待 5 秒，观察后台是否还有输出
await asyncio.sleep(5)
print("主协程结束")

asyncio.run(main())

输出：

任务开始
任务运行中… 0
任务运行中… 1
任务运行中… 2
超时了！
任务被取消
主协程结束

看起来正常？任务被取消了。但是，如果任务里有同步阻塞代码，或者吞掉了 CancelledError，结果就完全不同。

三、四个隐蔽的"取消失败"场景

场景一：同步阻塞代码无法被取消

import asyncio
import time

async def blocking_task():
print("任务开始")
time.sleep(10) # 同步阻塞，无法被取消
print("任务完成")
return "done"

async def main():
try:
await asyncio.wait_for(blocking_task(), timeout=3.0)
except asyncio.TimeoutError:
print("超时了！")

await asyncio.sleep(1)
print("主协程结束")

asyncio.run(main())

输出：

任务开始
超时了！
主协程结束
任务完成 # 超时后仍然打印了！

原因：time.sleep(10) 是同步阻塞调用，asyncio 的取消机制只能在 await 点生效。当协程卡在同步代码里时，取消信号无法传递，任务会一直跑到同步代码结束。

这就是为什么第一篇讲 time.sleep 会阻塞事件循环——它不仅阻塞并发，还让超时机制失效。

场景二：CancelledError 被 except Exception 吞掉

import asyncio

async def swallow_cancel_task():
try:
for i in range(10):
await asyncio.sleep(1)
print(f"运行中… {i}")
return "done"
except Exception as e: # CancelledError 在 Python 3.8+ 是 BaseException，但…
print(f"捕获到异常: {e}")
return "error"

async def main():
try:
await asyncio.wait_for(swallow_cancel_task(), timeout=3.0)
except asyncio.TimeoutError:
print("超时了！")

await asyncio.sleep(8)
print("主协程结束")

asyncio.run(main())

关键知识点：

• Python 3.8 之前，CancelledError 继承自 Exception，except Exception 会吞掉它；
• Python 3.8+，CancelledError 继承自 BaseException，except Exception 不会捕获它；
• 但如果代码里用了 except BaseException 或裸 except:，仍然会吞掉。

危险的写法：

try:
await something()
except BaseException: # 危险！会吞掉 CancelledError
pass

try:
await something()
except: # 更危险！裸 except 会吞掉一切
pass

正确做法：捕获 CancelledError 后必须 raise 重新抛出：

try:
await something()
except asyncio.CancelledError:
# 做一些清理
print("清理资源")
raise # 必须重新抛出
except Exception as e:
# 处理业务异常
handle_error(e)

场景三：finally 里的 await 被取消

import asyncio

async def task_with_cleanup():
try:
await asyncio.sleep(10)
return "done"
finally:
print("开始清理")
await asyncio.sleep(1) # 这个 await 会被取消
print("清理完成") # 不会执行

async def main():
try:
await asyncio.wait_for(task_with_cleanup(), timeout=2.0)
except asyncio.TimeoutError:
print("超时了！")

asyncio.run(main())

输出：

开始清理
超时了！

清理完成 永远不会打印。原因：任务被取消后，finally 里如果还有 await，会立刻再次触发 CancelledError，导致清理逻辑中断。

修复方案：用 asyncio.shield() 保护清理逻辑：

async def task_with_cleanup():
try:
await asyncio.sleep(10)
return "done"
finally:
# shield 保护清理任务不被取消
await asyncio.shield(_cleanup())

async def _cleanup():
print("开始清理")
await asyncio.sleep(1)
print("清理完成")

注意：shield 只是保护内部任务不被外部取消，但如果外层协程被强制结束，内部任务仍然可能被丢弃。生产环境建议用 asyncio.create_task 把清理任务独立出去：

async def task_with_cleanup():
try:
await asyncio.sleep(10)
finally:
# 清理任务独立运行，不依赖当前协程的生命周期
asyncio.create_task(_cleanup())

场景四：子任务未被级联取消

import asyncio

async def child_task(name: str):
try:
while True:
await asyncio.sleep(1)
print(f"子任务 {name} 运行中")
except asyncio.CancelledError:
print(f"子任务 {name} 被取消")
raise

async def parent_task():
# 启动子任务，但不保存引用
asyncio.create_task(child_task("A"))
asyncio.create_task(child_task("B"))

await asyncio.sleep(100)
return "done"

async def main():
try:
await asyncio.wait_for(parent_task(), timeout=3.0)
except asyncio.TimeoutError:
print("父任务超时！")

await asyncio.sleep(5)
print("主协程结束")

asyncio.run(main())

输出：

子任务 A 运行中
子任务 B 运行中
子任务 A 运行中
子任务 B 运行中
子任务 A 运行中
父任务超时！
子任务 A 运行中 # 父任务超时后，子任务还在跑！
子任务 B 运行中
子任务 A 运行中
子任务 B 运行中
主协程结束

原因：asyncio.wait_for 只取消 parent_task 本身，不会自动取消 parent_task 里创建的子任务。子任务变成"孤儿"，继续运行直到自己结束。

这就是第三篇讲的"任务被 GC 静默取消"的反面——这里任务没被 GC，但也没被正确取消，变成了泄漏。

修复方案：

async def parent_task():
child_a = asyncio.create_task(child_task("A"))
child_b = asyncio.create_task(child_task("B"))
try:
await asyncio.sleep(100)
return "done"
finally:
# 父任务被取消时，级联取消子任务
for child in (child_a, child_b):
if not child.done():
child.cancel()
# 等待子任务真正结束
await asyncio.gather(child_a, child_b, return_exceptions=True)

Python 3.11+ 推荐用 TaskGroup，它会自动处理级联取消：

async def parent_task():
async with asyncio.TaskGroup() as tg:
tg.create_task(child_task("A"))
tg.create_task(child_task("B"))
await asyncio.sleep(100)
return "done"

四、修复方案：正确的超时与取消

方案一：用 asyncio.timeout 上下文管理器（Python 3.11+）

import asyncio

async def call_tool(tool_name: str, timeout: float = 5.0):
try:
async with asyncio.timeout(timeout):
result = await _execute_tool(tool_name)
return result
except TimeoutError:
return {"error": f"工具 {tool_name} 超时"}

async def _execute_tool(tool_name: str):
# 内部必须是可取消的异步代码
await asyncio.sleep(2)
return f"{tool_name} 的结果"

asyncio.timeout 是 Python 3.11 引入的，比 wait_for 更推荐，因为它：

• 语义更清晰，async with 包裹的代码块超时后会被取消；
• 支持嵌套超时，内层超时不会影响外层；
• 与 TaskGroup 配合更好。

方案二：确保任务可取消

核心原则：所有耗时的操作都必须是可取消的异步操作。

注意 asyncio.to_thread：它虽然能让同步代码不阻塞事件循环，但线程一旦启动就无法被取消。asyncio.wait_for 超时后，线程会继续跑到结束：

import asyncio
import time

def _sync_blocking():
time.sleep(10) # 线程会跑完
return "done"

async def task():
return await asyncio.to_thread(_sync_blocking)

async def main():
try:
await asyncio.wait_for(task(), timeout=3.0)
except asyncio.TimeoutError:
print("超时了，但线程还在跑")

await asyncio.sleep(8)
print("线程结束了")

asyncio.run(main())

结论：to_thread 适合"不可取消但不关键"的场景，不适合"必须能取消"的场景。

方案三：超时后显式清理

import asyncio
import httpx

async def call_with_timeout(url: str, timeout: float = 5.0):
client = httpx.AsyncClient()
try:
async with asyncio.timeout(timeout):
response = await client.get(url)
return response.json()
except TimeoutError:
return {"error": "请求超时"}
finally:
# 无论成功失败，都关闭 client
await client.aclose()

关键点：超时后的清理逻辑要放在 finally 里，并且用 asyncio.shield 保护：

async def call_with_timeout(url: str, timeout: float = 5.0):
client = httpx.AsyncClient()
try:
async with asyncio.timeout(timeout):
return await client.get(url)
except TimeoutError:
return {"error": "请求超时"}
finally:
# 保护清理逻辑
await asyncio.shield(client.aclose())

五、Agent 工具调用的完整超时防护

把上面的所有能力整合到一个工具调用封装里：

import asyncio
from typing import Callable, Awaitable, TypeVar, Any

T = TypeVar("T")

class ToolTimeoutError(Exception):
"""工具超时异常"""
pass

async def call_tool_with_timeout(
tool_name: str,
tool_func: Callable[..., Awaitable[T]],
*args,
timeout: float = 10.0,
cleanup: Callable[[], Awaitable[None]] | None = None,
**kwargs,
) –> T:
"""
调用工具，带超时和清理

:param tool_name: 工具名（用于日志）
:param tool_func: 工具函数（必须是可取消的异步函数）
:param timeout: 超时时间
:param cleanup: 超时后的清理函数
"""
task = None
try:
async with asyncio.timeout(timeout):
task = asyncio.create_task(tool_func(*args, **kwargs))
return await task
except TimeoutError:
# 任务可能还在运行，尝试取消
if task and not task.done():
task.cancel()
try:
await task
except (asyncio.CancelledError, Exception):
pass

# 执行清理
if cleanup:
try:
await asyncio.shield(cleanup())
except Exception as e:
print(f"[WARN] 清理失败: {e}")

raise ToolTimeoutError(f"工具 {tool_name} 超时（{timeout}s）")

使用示例：

async def search_web(query: str, client: httpx.AsyncClient):
response = await client.get(f"https://api.example.com/search?q={query}")
return response.json()

async def cleanup_http_client(client: httpx.AsyncClient):
await client.aclose()

async def main():
client = httpx.AsyncClient()
try:
result = await call_tool_with_timeout(
"search_web",
search_web,
"python agent",
client=client,
timeout=5.0,
cleanup=lambda: cleanup_http_client(client),
)
print(result)
except ToolTimeoutError as e:
print(f"工具调用失败: {e}")

六、Agent 开发中的特殊场景

场景一：LLM API 调用超时

LLM API 的超时时间通常比较长（30~60 秒），如果超时后没有正确取消，会导致：

• 请求继续执行，token 照常消耗；
• 连接池被占满，后续请求排队；
• 用户侧看到"卡住"，但服务端还在跑。

修复：

async def call_llm_with_timeout(prompt: str, timeout: float = 60.0):
try:
async with asyncio.timeout(timeout):
return await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
)
except TimeoutError:
raise ToolTimeoutError(f"LLM 调用超时（{timeout}s）")

注意：httpx（OpenAI SDK 底层用的库）在收到取消信号时，会主动关闭连接并中止请求。前提是请求还在"等待响应"阶段，如果已经在流式输出中，取消后已生成的部分 token 仍然会被计费。

场景二：流式输出的超时

流式输出有两个超时：

• 首 token 超时：从发起请求到收到第一个 token 的时间；
• token 间隔超时：两个 token 之间的最大间隔。

async def stream_with_timeout(prompt: str, first_token_timeout: float = 10.0):
stream = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
stream=True,
)

# 首 token 超时
try:
async with asyncio.timeout(first_token_timeout):
first_chunk = await anext(stream)
except TimeoutError:
raise ToolTimeoutError(f"首 token 超时（{first_token_timeout}s）")

# 处理首 token
yield first_chunk

# token 间隔超时
async for chunk in stream:
try:
async with asyncio.timeout(token_interval_timeout):
yield chunk
except TimeoutError:
raise ToolTimeoutError(f"token 间隔超时（{token_interval_timeout}s）")

实现要点：

• 首 token 超时用 asyncio.timeout 包裹第一次 anext(stream)；
• token 间隔超时用 asyncio.timeout 包裹每次迭代；
• 超时后抛出 ToolTimeoutError，由上层统一处理。

七、总结与排查清单

核心结论

asyncio.wait_for 超时后，只是从等待者的视角结束了，任务本身不一定真的停了。要让任务真正停止，必须满足三个条件：

排查清单

遇到"超时了但任务还在跑"，按这个顺序排查：

• 任务里有没有 time.sleep()、同步 IO、CPU 密集计算？
• 有没有 except BaseException 或裸 except: 吞掉 CancelledError？
• finally 里的清理逻辑有没有 await？有没有被取消？
• 有没有 asyncio.create_task 创建的子任务？父任务取消时有没有级联取消？
• 有没有用 asyncio.to_thread 跑同步代码？线程是否无法取消？
• 外部 API 请求（HTTP、数据库）是否真的被中止？还是只是客户端超时？

推荐实践

• Python 3.11+ 优先用 asyncio.timeout + TaskGroup；
• 所有耗时操作都用可取消的异步版本；
• 清理逻辑放 finally，用 asyncio.shield 保护；
• 子任务用 TaskGroup 管理，自动级联取消；
• 工具调用统一封装超时 + 取消 + 清理（见第五节）。

八、写在最后

回到开头的问题：asyncio.wait_for 超时后，任务真的停了吗？

答案是：不一定。它只是让 await 这一侧不再等待了，任务本身是否停止，取决于任务内部代码是否"配合"取消。

这也是 Agent 开发里最容易踩的坑之一——你以为超时了，资源就释放了，实际上后台可能还挂着一堆"僵尸任务"，悄悄消耗 CPU、连接池，甚至外部 API 的计费。

希望这篇文章能帮你避开这些坑。如果你在 Agent 开发中也遇到过类似问题，欢迎在评论区交流。

附：本文所有代码均基于 Python 3.11+ 验证，完整可运行示例已内嵌在各章节中。