asyncio 编排模式

DEV-02: asyncio 编排模式

学习目标

• 掌握工具并行执行与并发上限控制
• 理解故障仲裁逻辑：失败 → 兄弟取消 → drain 优雅关闭
• 理解并行 guardrails 与 asyncio.gather 模型取消
• 掌握流式事件队列设计（asyncio.Queue + sentinel）

项目实践

工具并行执行

_execute_tool_plan() 默认使用 asyncio.gather() 并行执行所有工具类别：

# 工具执行并发模型
results = await asyncio.gather(
    execute_function_tools(plan.function_runs),
    execute_computer_actions(plan.computer_actions),
    execute_shell_calls(plan.shell_calls),
    execute_mcp_approval(plan.pending_mcp_approvals),
)

FunctionTool 批次内部的并发通过 max_function_tool_concurrency 控制。

故障仲裁逻辑

_FunctionToolBatchExecutor 的故障处理：

为什么需要 drain：直接丢弃 cancelled tasks 可能导致资源泄漏（如文件句柄未关闭、连接未释放）。Drain 机制等待被取消的任务完成清理。

并行 Guardrails 执行

第一轮输入 guardrails 的并行执行：

# 模型任务与 guardrails 任务并发
all_tasks = [model_task] + guardrail_tasks
done, pending = await asyncio.gather(*all_tasks, return_exceptions=True)

# tripwire 触发 → 取消模型任务
if guardrail_tripwire_triggered:
    model_task.cancel()

关键细节：asyncio.gather 等待所有任务完成。如果 guardrail tripwire 触发，模型任务被取消，但其请求可能已发出（token 已消耗）。

流式事件队列

from asyncio import Queue

class RunResultStreaming:
    _event_queue: Queue[StreamEvent | QueueCompleteSentinel]
    run_loop_task: asyncio.Task

    async def stream_events(self):
        while True:
            event = await self._event_queue.get()
            if isinstance(event, QueueCompleteSentinel):
                break
            yield event

设计要点：

• QueueCompleteSentinel 标记流结束
• 后台 asyncio.Task 持续推入事件
• 消费者通过 stream_events() 异步消费

会话级并发

场景	机制
SQLiteSession 多实例写同文件	文件级 RLock（引用计数）
WebSocket model 并发请求	请求级锁（per-request lock）
WebSocket 连接复用	per-event-loop WeakKeyDictionary 缓存

问题与规避

问题	规避方案
工具并发耗尽 API rate limit	设置 max_function_tool_concurrency 限制并发数
流式模式后台任务泄漏（进程退出后 task 仍在运行）	始终消费 stream_events() 到结束，或显式 cancel()
Drain 耗时过长拖慢整体执行	工具内部实现合理的 CancelledError 处理（如 try/except asyncio.CancelledError + 快速清理）

设计取舍

为什么使用 asyncio.gather 而非 asyncio.TaskGroup

SDK 使用 asyncio.gather(..., return_exceptions=True) 而非 Python 3.11+ 的 asyncio.TaskGroup，主要原因：SDK 需要支持 Python 3.10+，而 TaskGroup 是 3.11 引入的。

代价：asyncio.gather 的异常处理比 TaskGroup 更繁琐——需要手动检查 return_exceptions 结果中的异常实例。

为什么故障仲裁区分 Exception / CancelledError / BaseException

CancelledError 是 Python asyncio 的正常取消信号（非错误），不应覆盖真正的 Exception。而 BaseException（如 KeyboardInterrupt）的优先级最低——不应让它覆盖 Exception 或 CancelledError。

参考来源

• Python asyncio 文档
• OpenAI Agents SDK: tool_execution.py