本文详解如何在 asyncio 异步框架中合理整合 CPU 密集型操作（如本地代码执行、模型推理预处理等），通过 loop.run_in_executor 结合线程/进程池，突破单线程事件循环瓶颈，真正实现 I/O 与 CPU 的并发协同。
本文详解如何在 asyncio 异步框架中合理整合 cpu 密集型操作（如本地代码执行、模型推理预处理等），通过 `loop.run_in_executor` 结合线程/进程池，突破单线程事件循环瓶颈，真正实现 i/o 与 cpu 的并发协同。
在典型的 LLM 应用流水线中（如你描述的 run_agent），一个任务往往混合了三类操作：
- ✅ 高延迟 I/O 操作：调用 OpenAI API、本地 HTTP 模型服务（天然适合
async/await）； - ⚠️ 轻量同步计算：JSON 解析、字符串拼接等——通常可接受在 event loop 线程中执行；
- ❌ CPU/GPU 密集型操作：本地代码执行（
exec()）、向量化后处理、大张量变换、规则引擎匹配等——会阻塞 event loop，导致所有协程停滞。
你的 htop 显示单核 100% 占用，正是典型信号：run_agent 内部存在未剥离的同步计算逻辑，使 asyncio 失去并发优势。
✅ 正确解法：职责分离 + 执行器卸载
核心原则是 “async only for I/O, sync + executor for CPU”。你需要将 run_agent 中的 CPU 工作抽离为纯同步函数，并交由 ThreadPoolExecutor（I/O-bound 或短时 CPU）或 ProcessPoolExecutor（长时 CPU/GIL 敏感）执行：
import asyncio
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
# ✅ 提取纯同步的 CPU 密集型函数（无 await、无 async）
def cpu_heavy_preprocess(sample_data: dict) -> dict:
# 例如：执行本地 Python 代码、复杂正则、NumPy 数值计算
import numpy as np
result = np.fft.fft(sample_data["signal"]) # 真实 CPU 负载
return {"processed": result.tolist()}
def run_local_code_safely(code_str: str, context: dict) -> dict:
# 注意：生产环境需沙箱化！此处仅为示意
local_ns = {"__builtins__": {}}
exec(code_str, local_ns, context)
return context.get("output", {})
# ✅ 在 async 函数中安全调用
async def run_single_sample(task_sample: TaskSample):
loop = asyncio.get_running_loop()
# Step 1: I/O-heavy — 原生 async（OpenAI 调用等）
llm_response = await call_openai_api(task_sample.prompt)
# Step 2: CPU-heavy — 卸载到线程池（适合 I/O 等待+短计算）
with ThreadPoolExecutor(max_workers=4) as pool:
processed = await loop.run_in_executor(
pool,
cpu_heavy_preprocess,
{"signal": llm_response["raw_audio"]}
)
# Step 3: 若存在极重计算（如 PyTorch 模型推理），改用进程池避 GIL
# with ProcessPoolExecutor(max_workers=2) as pool:
# result = await loop.run_in_executor(pool, run_torch_model, processed)
# Step 4: 最终 async 后续（保存、日志等）
await save_result(task_sample.id, {**llm_response, "postprocessed": processed})
⚠️ 关键注意事项
-
不要在 executor 中调用
async函数：run_in_executor只接受同步可调用对象。若需在子进程中启动新 event loop，请显式使用asyncio.run()（仅限ProcessPoolExecutor场景，且需确保模块可序列化）； -
线程池 vs 进程池选择：
-
ThreadPoolExecutor：适用于含 I/O 等待的混合负载（如数据库查询 + 轻量计算），开销小； -
ProcessPoolExecutor：适用于纯 CPU-bound、GIL 敏感操作（如scipy.optimize、pandas.groupby.agg大数据集），但进程启动/通信成本高，慎用于高频小任务；
-
- 资源竞争防护：多个 executor 任务可能并发访问共享资源（如文件、全局变量），务必加锁或使用线程/进程安全结构；
-
错误传播：
run_in_executor抛出的异常会原样被await捕获，无需额外包装； -
配置建议：
max_workers不宜盲目设高——线程池推荐min(32, os.cpu_count() + 4)，进程池通常os.cpu_count() // 2起步，结合压测调整。
? 总结
你的用例完全合理，问题不在于设计 flawed，而在于 async 函数边界未对齐执行模型。真正的高性能 ML 编排，必然是分层的：
? 顶层 asyncio：调度、I/O 并发、状态协调；
? 中层 Executor：隔离 CPU/GPU 工作，释放 event loop；
? 底层同步函数：专注计算逻辑，保持无状态、易测试、可复用。
重构后，你将观察到 htop 中多核利用率显著提升，整体吞吐量（samples/sec）随并发度线性增长——这才是异步 + 并行的正确打开方式。
