企业官网建设流程全解析

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:#c62828

Agent 并发调度需要同时管理三类资源：

1. LLM API 容量：API 的并发请求限制（RPM/TPM），超出限制的请求会被拒绝或排队；
2. GPU 显存：如果使用本地部署模型，多个 Agent 同时推理可能撑爆显存；
3. 上下文窗口：每个 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(key=lambda 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), timeout=task.timeout_seconds ) except asyncio.TimeoutError: print(f"Agent {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 预算的三层混合策略。