Agent 并发调度：多个 Agent 同时跑不是简单的多线程

Agent 并发调度多个 Agent 同时跑不是简单的多线程一、见证奇迹的时刻当 10 个 Agent 同时启动系统瘫了在设计多 Agent 协作系统时一个直观的思路是让每个 Agent 独立运行互不干扰直接用 asyncio 或 ThreadPoolExecutor 并发执行。测试时跑 3 个 Agent 没问题推到生产环境 10 个 Agent 一起跑——结果是 GPU OOM、上下文混乱、部分 Agent 超时后静默失败。这不是 Agent 的设计问题是并发调度策略的缺失。传统 Web 服务的并发模型请求独立、无共享状态、无状态服务在多 Agent 场景下会大面积失效。原因在于Agent 的运行不是无状态的——它需要维护对话上下文、工具调用历史、中间推理步骤。而且多个 Agent 共享同一个 LLM 推理服务本质上是共享一个有容量限制的资源池。更棘手的是Agent 之间的执行时间高度不确定。同样的任务Agent A 可能 3 秒完成也可能因为需要 5 次工具调用而花费 30 秒。固定时间片或简单 FIFO 调度会导致不确定性进一步放大。二、共享资源池与有状态任务为什么 Agent 调度不是 Web 调度flowchart TD A[用户任务请求] -- B{调度器 Dispatcher} B -- C[任务队列 PriorityQueue] C -- D{资源感知调度} D -- E{LLM API 容量检查} E --|有剩余配额| F[分配 Agent 执行] E --|配额耗尽| G[加入等待队列] D -- H{显存容量检查} H --|显存充足| F H --|显存不足| I[触发显存回收或排队] D -- J{上下文窗口检查} J --|未超限| F J --|超限| K[上下文压缩或分片] F -- L[Agent 执行循环] L -- M{执行结果} M --|需要工具调用| N[工具调用 结果注入] N -- L M --|完成| O[结果收集] subgraph 并发控制 P[信号量 Semaphore] Q[速率限制 RateLimiter] R[超时控制器 TimeoutManager] end L -.- P L -.- Q L -.- R style E fill:#e3f2fd,color:#1565c0 style H fill:#fff3e0,color:#e65100 style J fill:#fce4ec,color:#c62828Agent 并发调度需要同时管理三类资源LLM API 容量API 的并发请求限制RPM/TPM超出限制的请求会被拒绝或排队GPU 显存如果使用本地部署模型多个 Agent 同时推理可能撑爆显存上下文窗口每个 Agent 的对话历史不断增长多个 Agent 同时运行会浪费大量 Token。三、资源感知调度器设计不仅管理任务顺序更管理资源配额import asyncio from dataclasses import dataclass, field from typing import List, Dict, Optional import time dataclass class ResourceQuota: 资源配额定义 max_concurrent_requests: int 5 # LLM API 最大并发请求数 max_gpu_memory_gb: float 40.0 # 最大可用 GPU 显存 max_context_tokens: int 32000 # 单 Agent 最大上下文 Token rpm_limit: int 500 # API 每分钟请求限制 dataclass class AgentTask: Agent 任务定义 task_id: str agent_type: str prompt: str priority: int 0 # 优先级数值越小优先级越高 context_tokens: int 0 # 当前上下文 Token 数 estimated_gpu_memory: float 0.0 # 预估 GPU 显存占用 max_tool_calls: int 10 # 最大工具调用次数 timeout_seconds: int 300 # 超时时间 class ResourceAwareScheduler: 资源感知的 Agent 调度器 def __init__(self, quota: ResourceQuota): self.quota quota # 设计原因PriorityQueue 而非普通 Queue # 紧急任务如用户交互需要优先执行而非排在批量任务后面 self.task_queue: List[AgentTask] [] self.active_tasks: Dict[str, AgentTask] {} # 设计原因信号量控制 LLM API 并发数 # 超过限制的调用排队等待而非失败避免上游重试风暴 self.api_semaphore asyncio.Semaphore(quota.max_concurrent_requests) # 设计原因滑动窗口速率限制器 # 记录过去60秒内的请求时间戳确保不超过 RPM 限制 self._request_timestamps: List[float] [] self._rpm_window 60.0 async def _check_rpm_limit(self): 检查 RPM 限制 now time.time() # 清理窗口外的记录 self._request_timestamps [ ts for ts in self._request_timestamps if now - ts self._rpm_window ] if len(self._request_timestamps) self.quota.rpm_limit: # 设计原因计算需要等待的时间而非固定 sleep # 动态等待最小化延迟窗口边界过了立即放行 wait_time self._request_timestamps[0] self._rpm_window - now 0.01 await asyncio.sleep(max(0, wait_time)) self._request_timestamps.append(now) async def submit_task(self, task: AgentTask): 提交任务到调度器 # 设计原因优先检查资源后确认能否立即执行 # 避免任务已入队但资源不足导致的虚假等待 if len(self.active_tasks) self.quota.max_concurrent_requests: asyncio.create_task(self._execute_agent(task)) else: self.task_queue.append(task) # 按优先级排序数值小的优先 self.task_queue.sort(keylambda t: t.priority) async def _execute_agent(self, task: AgentTask): 执行单个 Agent 的核心逻辑 self.active_tasks[task.task_id] task try: # 设计原因使用 wait_for 包装执行防止单个 Agent 无限等待 # 超时后优雅终止而非强制 kill给 Agent 保存中间状态的机会 await asyncio.wait_for( self._agent_loop(task), timeouttask.timeout_seconds ) except asyncio.TimeoutError: print(fAgent {task.task_id} 执行超时触发降级逻辑) await self._handle_timeout(task) finally: del self.active_tasks[task.task_id] await self._dispatch_next()四、调度策略的权衡公平性与效率的永恒矛盾不同调度策略的选择直接影响系统的行为特征。FIFO 调度实现最简单按到达顺序执行。问题是长任务会阻塞后续所有短任务Head-of-Line Blocking导致平均响应时间被少数长任务拉高。优先级调度紧急任务优先用户体验好。代价是需要定义优先级标准且低优先级任务可能永远得不到执行Starvation。需要引入 Aging 机制任务等待时间越长优先级越高。资源感知调度综合考虑 API 配额、显存、上下文窗口后再决定是否执行。最大化资源利用率但调度器复杂度最高且资源预估不准时调度效果反而变差。Token 预算调度为每个 Agent 分配 Token 预算超过预算后触发上下文压缩或任务终止。控制成本效果好但可能提前终止有价值的探索路径。实际系统的建议优先级调度 资源感知 Token 预算的三层混合策略。第一层用优先级排序任务队列第二层用资源感知决定何时执行第三层用 Token 预算做安全兜底。五、总结多 Agent 并发调度的核心挑战在于管理共享的 LLM 推理资源。Agent 任务是有状态的、执行时间不确定的不能简单套用无状态 Web 服务的并发模型。资源感知调度器需要同时监控 API 并发限制、GPU 显存使用和上下文窗口长度。调度策略的选择涉及公平性与效率的权衡优先级调度可能导致低优先级任务饥饿FIFO 调度可能导致头阻塞。实际系统推荐采用优先级调度、资源感知和 Token 预算的三层混合策略。