应于协程内部用 time.perf_counter() 手动埋点测实际运行耗时，避免外层计时失效；任务取消时需 try/finally 保证统计；查运行位置可用 asyncio.current_task() 结合 inspect.getframeinfo() 安全获取堆栈。

如何在python中实时监控asyncio任务的执行耗时与堆栈？

用 asyncio.create_task() 包裹任务时如何打点耗时

直接在 asyncio.create_task() 外层加计时会失效，因为任务提交后立即返回，不等执行完成。真正要测的是协程体实际运行时间，不是调度延迟。

推荐在协程函数内部手动埋点，用 time.time() 或更精确的 time.perf_counter()：

import time
import asyncio
<p>async def fetch_data():
start = time.perf_counter()
await asyncio.sleep(1)  # 模拟IO
duration = time.perf_counter() - start
print(f"fetch_data took {duration:.3f}s")
return "done"</p>

别用 time.time() 测短任务——系统时钟可能被 NTP 调整，perf_counter() 是单调递增的
如果任务被取消，await 抛出 CancelledError，需用 try/finally 确保耗时统计不漏
避免在 hot path（如每秒数千次调用）里打印，改用 logging.debug() 并控制日志级别

任务运行中获取当前堆栈（非 traceback）

想看某个正在跑的 task 此刻卡在哪一行，不能等它挂掉再看 traceback——得主动抓快照。Python 3.11+ 可用 task.get_coro().__code__.co_filename 和 task.get_coro().cr_frame.f_lineno，但更通用且安全的方式是调用 asyncio.current_task() + inspect.getframeinfo()：

Python Sdk

inference.sh 的 Python SDK：运行 AI 应用、构建智能体，并集成 150 多个模型。包名：inferencesh (pip install inferencesh)。支持同步/异步……

下载

import inspect
import asyncio
<p>async def long_running():
while True:
task = asyncio.current_task()
frame = task.get_coro().cr_frame
info = inspect.getframeinfo(frame)
print(f"at {info.filename}:{info.lineno}")
await asyncio.sleep(0.5)</p>

注意：cr_frame 在协程暂停时可能为 None，访问前要判空
Python cr_frame，得降级用 task.get_stack()（返回 list[FrameSummary]），但只在 task 被挂起时才有效
task.get_stack() 返回的是挂起点堆栈，不是当前执行点；真要查“正在哪”，必须用 cr_frame 或 sys._current_frames() 配合 task ID 查

全局监控所有活跃 task 的耗时与位置

靠每个协程自己打点太琐碎。可以用后台 task 定期扫描 asyncio.all_tasks()，对每个未完成的 task 计算「已存活时间」并尝试提取位置信息：

import asyncio
import time
from datetime import timedelta
<p>async def monitor_tasks():
while True:
for task in asyncio.all_tasks():
if task.done() or task.cancelled():
continue</p><h1>粗略存活时间（从创建到现在）</h1><pre class="brush:python;toolbar:false;">        age = time.time() - getattr(task, '_created_at', time.time())
        try:
            frame = task.get_coro().cr_frame
            if frame:
                loc = f"{frame.f_code.co_filename}:{frame.f_lineno}"
                print(f"[{timedelta(seconds=age)}] {task.get_name()} @ {loc}")
        except Exception:
            pass  # 忽略无法读取帧的 task（如刚创建、已销毁）
    await asyncio.sleep(1)

启动前给 task 打上创建时间戳

def create_tracked_task(coro, *, name=None): task = asyncio.create_task(coro, name=name) task._created_at = time.time() return task

asyncio.all_tasks() 返回的是当前 event loop 的所有 task，跨 loop 不可见
不要在监控循环里做重操作（如 deep copy 堆栈），否则会影响被监控任务的响应性
某些 task（如由 asyncio.to_thread() 创建的）其 cr_frame 不可用，需 fallback 到 repr(task) 看大致状态

为什么不能依赖 sys.settrace 或 asyncio hooks？

有人想用 sys.settrace() 全局钩子来捕获 async 函数进入/退出，但这在 asyncio 下基本不可行：

sys.settrace() 对协程对象本身不触发，只对普通函数生效；await 行不会触发 line trace
asyncio.add_debugging_hook()（3.12+）仅支持 task creation/cancellation，不提供执行中 hook
第三方库如 aiodebug 内部也是靠定期轮询 + cr_frame 实现，不是真正的实时 hook
真正低开销的方案只有两种：协程内手动埋点，或用 tracemalloc + asyncio.run() 的 debug mode 查内存分配热点，而非执行路径