限流

引言 Agent系统面临独特的流量挑战：LLM推理的QPS受GPU数量硬性限制、工具调用可能触发外部API限流、突发的复杂查询可能导致单请求消耗数十倍于平均值的资源。传统的固定阈值限流无法应对这些挑战，Agent系统需要更智能的限流和熔断机制。 2026年，自适应限流已成为Agent系统的标配——系统根据实时负载和资源利用率动态调整限流策略，在保护系统的同时最大化吞吐量。 Agent系统的流量特征 传统Web应用流量 Agent系统流量 │ │ │ ╱╲ │ ╱╲ │ ╱ ╲ ╱╲ │ ╱ ╲╱╲╱╲ │ ╱ ╲__╱ ╲___ │ ╱ ╲___ │________________ │________________ 时间 时间 相对均匀的请求量 突发性强、长尾明显 每请求资源消耗相近 单请求资源消耗差异巨大（10-100x） 令牌桶限流 基础令牌桶 import asyncio import time from dataclasses import dataclass @dataclass class TokenBucket: """令牌桶限流器""" capacity: float # 桶容量 refill_rate: float # 每秒补充令牌数 tokens: float = 0 # 当前令牌数 last_refill: float = 0 # 上次补充时间 def __post_init__(self): self.tokens = self.capacity self.last_refill = time.monotonic() async def acquire(self, tokens: int = 1) -> bool: """获取令牌""" while True: self._refill() if self.tokens >= tokens: self.tokens -= tokens return True # 令牌不足，等待补充 wait_time = (tokens - self.tokens) / self.refill_rate await asyncio.sleep(min(wait_time, 0.1)) def _refill(self): """补充令牌""" now = time.monotonic() elapsed = now - self.last_refill self.tokens = min( self.capacity, self.tokens + elapsed * self.refill_rate ) self.last_refill = now def try_acquire(self, tokens: int = 1) -> bool: """非阻塞获取令牌""" self._refill() if self.tokens >= tokens: self.tokens -= tokens return True return False 多维度限流 class AgentRateLimiter: """Agent多维度限流器""" def __init__(self): # 不同维度的限流器 self.limiters = { "requests_per_minute": TokenBucket(capacity=100, refill_rate=100/60), "tokens_per_minute": TokenBucket(capacity=50000, refill_rate=50000/60), "tool_calls_per_minute": TokenBucket(capacity=50, refill_rate=50/60), "concurrent_sessions": SemaphoreLimiter(max_concurrent=20), "gpu_concurrent": SemaphoreLimiter(max_concurrent=4), } async def check_request(self, request: dict) -> dict: """检查请求是否被允许""" required = { "requests_per_minute": 1, "concurrent_sessions": 1, } # 根据请求类型添加额外限制 if request.get("estimated_tokens"): required["tokens_per_minute"] = request["estimated_tokens"] if request.get("tool_calls"): required["tool_calls_per_minute"] = len(request["tool_calls"]) if request.get("requires_gpu"): required["gpu_concurrent"] = 1 # 检查所有维度 for dim, amount in required.items(): limiter = self.limiters[dim] if not limiter.try_acquire(amount): return { "allowed": False, "limited_dimension": dim, "retry_after_ms": self._get_retry_after(dim, amount), "message": f"Rate limit exceeded: {dim}" } return {"allowed": True} def _get_retry_after(self, dimension: str, amount: int) -> int: """计算重试等待时间""" limiter = self.limiters[dimension] if hasattr(limiter, 'refill_rate'): return int((amount - limiter.tokens) / limiter.refill_rate * 1000) return 1000 # 默认1秒 自适应限流 class AdaptiveRateLimiter: """自适应限流器——基于系统负载动态调整""" def __init__(self, config: dict): self.min_limit = config.get("min_limit", 10) self.max_limit = config.get("max_limit", 1000) self.current_limit = config.get("initial_limit", 100) # AIMD参数 self.additive_increase = config.get("additive_increase", 1) self.multiplicative_decrease = config.get("multiplicative_decrease", 0.5) # 系统指标 self.metrics = { "cpu_utilization": 0, "memory_utilization": 0, "gpu_utilization": 0, "p99_latency_ms": 0, "error_rate": 0, } # 调整周期 self.adjustment_interval = 10 # 秒 self.window_requests = 0 self.window_errors = 0 async def should_allow(self, request: dict) -> bool: """判断是否允许请求""" current_rps = self.window_requests / self.adjustment_interval if current_rps >= self.current_limit: return False self.window_requests += 1 return True async def adjust_limits(self): """定期调整限流阈值""" while True: await asyncio.sleep(self.adjustment_interval) # 计算系统健康度 health_score = self._calculate_health() if health_score > 0.8: # 系统健康，增加限流（AIMD的AI） self.current_limit = min( self.current_limit + self.additive_increase, self.max_limit ) logger.info( f"Increasing rate limit to {self.current_limit} " f"(health: {health_score:.2f})" ) elif health_score < 0.5: # 系统不健康，减少限流（AIMD的MD） self.current_limit = max( int(self.current_limit * self.multiplicative_decrease), self.min_limit ) logger.warning( f"Decreasing rate limit to {self.current_limit} " f"(health: {health_score:.2f})" ) # 重置窗口 self.window_requests = 0 self.window_errors = 0 def _calculate_health(self) -> float: """计算系统健康度（0-1）""" weights = { "cpu_utilization": 0.2, "memory_utilization": 0.15, "gpu_utilization": 0.25, "p99_latency_ms": 0.25, "error_rate": 0.15, } thresholds = { "cpu_utilization": 0.8, "memory_utilization": 0.85, "gpu_utilization": 0.9, "p99_latency_ms": 2000, "error_rate": 0.05, } score = 1.0 for metric, weight in weights.items(): value = self.metrics[metric] threshold = thresholds[metric] ratio = value / threshold if threshold > 0 else 0 if ratio > 1: # 超过阈值，扣分 score -= weight * min(ratio - 1, 1) return max(0, score) 熔断器设计 class CircuitBreaker: """熔断器""" class State: CLOSED = "closed" # 正常工作 OPEN = "open" # 熔断，拒绝请求 HALF_OPEN = "half_open" # 半开，试探恢复 def __init__( self, failure_threshold: int = 10, failure_rate_threshold: float = 0.5, recovery_timeout: float = 30.0, half_open_max_calls: int = 3 ): self.state = self.State.CLOSED self.failure_threshold = failure_threshold self.failure_rate_threshold = failure_rate_threshold self.recovery_timeout = recovery_timeout self.half_open_max_calls = half_open_max_calls self.failure_count = 0 self.success_count = 0 self.total_count = 0 self.last_failure_time = None self.half_open_calls = 0 async def call(self, func, *args, **kwargs): """通过熔断器调用函数""" if self.state == self.State.OPEN: if self._should_attempt_recovery(): self.state = self.State.HALF_OPEN self.half_open_calls = 0 logger.info("Circuit breaker entering half-open state") else: raise CircuitBreakerOpenError( f"Circuit breaker is open. " f"Retry after {self._recovery_remaining():.0f}s" ) if self.state == self.State.HALF_OPEN: if self.half_open_calls >= self.half_open_max_calls: raise CircuitBreakerOpenError( "Half-open: max probe calls reached" ) self.half_open_calls += 1 try: result = await func(*args, **kwargs) await self._on_success() return result except Exception as e: await self._on_failure() raise async def _on_success(self): self.success_count += 1 self.total_count += 1 if self.state == self.State.HALF_OPEN: if self.half_open_calls >= self.half_open_max_calls: self.state = self.State.CLOSED self.failure_count = 0 logger.info("Circuit breaker recovered, closing") async def _on_failure(self): self.failure_count += 1 self.total_count += 1 self.last_failure_time = time.monotonic() if self.state == self.State.HALF_OPEN: self.state = self.State.OPEN logger.warning("Circuit breaker re-opened from half-open") elif self.state == self.State.CLOSED: failure_rate = self.failure_count / max(self.total_count, 1) if (self.failure_count >= self.failure_threshold or failure_rate >= self.failure_rate_threshold): self.state = self.State.OPEN logger.error( f"Circuit breaker opened: " f"{self.failure_count}/{self.total_count} failures " f"({failure_rate:.1%})" ) 多级熔断 class MultiLevelCircuitBreaker: """多级熔断器""" def __init__(self): self.breakers = { "llm_inference": CircuitBreaker( failure_threshold=5, failure_rate_threshold=0.3, recovery_timeout=30 ), "tool_search": CircuitBreaker( failure_threshold=10, failure_rate_threshold=0.5, recovery_timeout=15 ), "vector_db": CircuitBreaker( failure_threshold=8, failure_rate_threshold=0.4, recovery_timeout=10 ), "external_api": CircuitBreaker( failure_threshold=3, failure_rate_threshold=0.2, recovery_timeout=60 ), } async def call_with_fallback( self, primary: callable, fallbacks: list, circuit_key: str ): """带降级的熔断调用""" breaker = self.breakers.get(circuit_key) try: if breaker: return await breaker.call(primary) return await primary() except CircuitBreakerOpenError: # 主路径熔断，尝试降级 for i, fallback in enumerate(fallbacks): try: logger.info(f"Trying fallback {i+1}/{len(fallbacks)}") return await fallback() except Exception as e: logger.warning(f"Fallback {i+1} failed: {e}") continue # 所有降级都失败 raise AllFallbacksFailedError( f"All fallbacks failed for {circuit_key}" ) 降级策略 class DegradationStrategy: """Agent降级策略""" LEVELS = { "normal": { "model": "gpt-4o", "max_tools": 10, "max_context": 128000, "streaming": True, }, "degraded_1": { "model": "gpt-4o-mini", # 降级到小模型 "max_tools": 5, "max_context": 32000, "streaming": True, }, "degraded_2": { "model": "gpt-4o-mini", "max_tools": 2, # 仅保留核心工具 "max_context": 8000, "streaming": False, }, "emergency": { "model": "cached-response", # 使用缓存或模板回复 "max_tools": 0, "max_context": 1000, "streaming": False, } } def get_current_level(self, system_load: float) -> str: """根据系统负载获取降级级别""" if system_load < 0.7: return "normal" elif system_load < 0.85: return "degraded_1" elif system_load < 0.95: return "degraded_2" else: return "emergency" 总结 Agent系统的限流与熔断需要从三个层面协同工作：令牌桶实现精确的多维度限流，自适应算法根据系统健康度动态调整阈值，熔断器在故障时快速切断流量并支持降级恢复。关键是在系统保护和用户体验之间找到平衡——过度保护会浪费资源，保护不足会导致雪崩。 ...

引言 Agent 不是孤立的——多个用户同时请求、多个工具并行执行、多个 Agent 协作完成任务。并发控制确保这些并行活动不互相干扰、不超出资源限制、不产生数据竞争。2026年，随着 Agent 集群规模扩大到数百实例，并发控制从单机问题升级为分布式问题。 一、Agent 并发的四个层次 ┌──────────────────────────────────────────────────┐ │ Agent 并发控制层次 │ ├──────────────────┬───────────────────────────────┤ │ 请求级 │ 多用户并发请求 │ │ (Request) │ → 限流 + 队列 + 优先级 │ ├──────────────────┼───────────────────────────────┤ │ Agent级 │ 单用户多Agent实例 │ │ (Agent) │ → 信号量 + 状态隔离 │ ├──────────────────┼───────────────────────────────┤ │ 工具级 │ 多工具并行调用 │ │ (Tool) │ → 并发限制 + 超时控制 │ ├──────────────────┼───────────────────────────────┤ │ 资源级 │ 共享资源访问 │ │ (Resource) │ → 分布式锁 + 乐观锁 │ └──────────────────┴───────────────────────────────┘ 二、请求级并发控制 2.1 多维限流器 import asyncio import time from collections import defaultdict class MultiDimensionRateLimiter: """多维度限流器""" def __init__(self, redis_client): self.redis = redis_client async def check( self, user_id: str, agent_name: str, limits: dict ) -> bool: """多维度限流检查 limits = { "qps": 10, # 每秒请求数 "concurrent": 5, # 最大并发 "daily_tokens": 500000, # 日Token上限 "monthly_cost": 100, # 月成本上限 } """ checks = [] # QPS 限流（滑动窗口） if "qps" in limits: checks.append(self._check_sliding_window( f"qps:{user_id}:{agent_name}", limits["qps"], window=1 )) # 并发限流 if "concurrent" in limits: checks.append(self._check_concurrent( f"conc:{user_id}:{agent_name}", limits["concurrent"] )) # Token 配额 if "daily_tokens" in limits: checks.append(self._check_quota( f"tokens:{user_id}:{date.today()}", limits["daily_tokens"] )) results = await asyncio.gather(*checks) return all(results) async def _check_sliding_window( self, key: str, limit: int, window: int ) -> bool: """滑动窗口限流""" now = time.time() pipe = self.redis.pipeline() # 移除窗口外的记录 pipe.zremrangebyscore(key, 0, now - window) # 添加当前请求 pipe.zadd(key, {str(uuid.uuid4()): now}) # 统计窗口内请求数 pipe.zcard(key) # 设置过期时间 pipe.expire(key, window * 2) results = await pipe.execute() count = results[2] if count > limit: # 移除刚添加的记录 pipe.zrem(key, str(uuid.uuid4())) return False return True async def _check_concurrent(self, key: str, limit: int) -> bool: """并发数限流""" current = await self.redis.incr(key) if current == 1: await self.redis.expire(key, 300) # 5分钟自动过期 if current > limit: await self.redis.decr(key) return False return True async def release_concurrent(self, key: str): """释放并发槽""" await self.redis.decr(key) class PriorityRequestQueue: """优先级请求队列""" def __init__(self, max_concurrent: int = 100): self.semaphore = asyncio.Semaphore(max_concurrent) self.queues = { "high": asyncio.Queue(maxsize=50), "normal": asyncio.Queue(maxsize=500), "low": asyncio.Queue(maxsize=1000), } self._running = True async def submit( self, request: Request, priority: str = "normal" ) -> Response: """提交请求""" queue = self.queues.get(priority, self.queues["normal"]) if queue.full(): raise QueueFullError(f"{priority} queue is full") future = asyncio.Future() await queue.put((request, future)) # 启动消费者（如果未启动） asyncio.create_task(self._consume()) return await future async def _consume(self): """消费请求：按优先级调度""" async with self.semaphore: # 按优先级获取请求 for priority in ["high", "normal", "low"]: queue = self.queues[priority] if not queue.empty(): request, future = await queue.get() try: result = await self._process(request) future.set_result(result) except Exception as e: future.set_exception(e) return # 所有队列为空 await asyncio.sleep(0.01) 三、工具级并发控制 3.1 并行工具调用管理 class ParallelToolExecutor: """并行工具调用管理器""" def __init__(self): self.tool_limits = {} # {tool_name: Semaphore} self.global_limit = asyncio.Semaphore(10) def register_tool( self, name: str, max_concurrent: int = 5, timeout: float = 30.0 ): """注册工具并发限制""" self.tool_limits[name] = { "semaphore": asyncio.Semaphore(max_concurrent), "timeout": timeout } async def execute_parallel( self, tool_calls: list[ToolCall] ) -> list[ToolResult]: """并行执行多个工具调用""" tasks = [] for call in tool_calls: task = asyncio.create_task( self._execute_single(call) ) tasks.append(task) # 等待所有完成，设置全局超时 results = await asyncio.gather(*tasks, return_exceptions=True) return [ result if not isinstance(result, Exception) else ToolResult(success=False, error=str(result)) for result in results ] async def _execute_single(self, call: ToolCall) -> ToolResult: """执行单个工具调用（带并发限制和超时）""" tool_limit = self.tool_limits.get(call.tool) if not tool_limit: raise UnknownToolError(call.tool) # 双重信号量：全局 + 工具级 async with self.global_limit: async with tool_limit["semaphore"]: try: result = await asyncio.wait_for( self._call_tool(call), timeout=tool_limit["timeout"] ) return ToolResult(success=True, data=result) except asyncio.TimeoutError: return ToolResult( success=False, error=f"Tool {call.tool} timed out" ) # 注册工具并发限制 executor = ParallelToolExecutor() executor.register_tool("web_search", max_concurrent=5, timeout=10) executor.register_tool("database_query", max_concurrent=3, timeout=15) executor.register_tool("file_read", max_concurrent=20, timeout=5) executor.register_tool("file_write", max_concurrent=2, timeout=10) executor.register_tool("send_email", max_concurrent=1, timeout=30) 四、资源级并发控制 4.1 分布式锁 class DistributedLock: """基于 Redis 的分布式锁""" def __init__(self, redis_client): self.redis = redis_client self._local_locks = {} # 本地锁缓存 async def acquire( self, resource: str, holder: str, ttl: int = 30, timeout: float = 10.0 ) -> bool: """获取分布式锁 Args: resource: 锁定的资源名 holder: 持有者标识（如 agent_id + session_id） ttl: 锁的生存时间（秒） timeout: 等待获取锁的超时时间 """ lock_key = f"lock:{resource}" start = time.time() while time.time() - start < timeout: # 尝试获取锁（原子操作） acquired = await self.redis.set( lock_key, holder, nx=True, ex=ttl ) if acquired: # 启动看门狗续期 asyncio.create_task(self._watchdog(resource, holder, ttl)) return True # 检查持有者是否是自己（重入） current = await self.redis.get(lock_key) if current == holder: await self.redis.expire(lock_key, ttl) return True # 等待重试 await asyncio.sleep(0.1) return False async def release(self, resource: str, holder: str) -> bool: """释放锁（使用 Lua 脚本保证原子性）""" lua_script = """ if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("DEL", KEYS[1]) else return 0 end """ result = await self.redis.eval( lua_script, 1, f"lock:{resource}", holder ) return bool(result) async def _watchdog(self, resource: str, holder: str, ttl: int): """看门狗：自动续期锁""" lock_key = f"lock:{resource}" renew_interval = ttl * 0.3 # 在30% TTL时续期 while True: await asyncio.sleep(renew_interval) # 检查是否仍持有锁 current = await self.redis.get(lock_key) if current != holder: break # 锁已被释放或被抢 # 续期 await self.redis.expire(lock_key, ttl) class AgentResourceLock: """Agent 资源锁管理""" def __init__(self, lock_manager: DistributedLock): self.locks = lock_manager async def with_resource_lock( self, resource: str, agent_id: str, action: callable, ttl: int = 30 ): """在锁保护下执行操作""" acquired = await self.locks.acquire( resource, agent_id, ttl=ttl ) if not acquired: raise LockAcquisitionError( f"Could not acquire lock for {resource}" ) try: return await action() finally: await self.locks.release(resource, agent_id) 4.2 乐观锁 class OptimisticLock: """乐观锁：适用于低冲突场景""" async def update_with_version( self, resource_id: str, update_fn: callable, max_retries: int = 3 ) -> any: """带版本控制的更新""" for attempt in range(max_retries): # 1. 读取当前版本 current = await self.db.get(resource_id) if not current: raise NotFoundError(resource_id) version = current.version # 2. 计算更新 updated = await update_fn(current.data) # 3. 乐观写入（带版本检查） result = await self.db.update_with_version( resource_id, data=updated, expected_version=version ) if result.success: return updated # 版本冲突，重试 logger.warning( f"Optimistic lock conflict on {resource_id}, " f"attempt {attempt+1}" ) await asyncio.sleep(0.05 * attempt) # 短暂退避 raise ConcurrencyError( f"Failed after {max_retries} retries due to conflicts" ) 五、Agent 级并发控制 class AgentInstanceManager: """Agent 实例管理——控制单用户的 Agent 并发""" def __init__(self, redis_client): self.redis = redis_client self.limits = { "free": 1, # 免费用户：1个并发Agent "pro": 5, # Pro用户：5个 "enterprise": 20, # 企业用户：20个 } async def acquire_slot( self, user_id: str, tier: str, agent_type: str ) -> AgentSlot: """获取 Agent 执行槽""" limit = self.limits.get(tier, 1) key = f"agent_slots:{user_id}" # 检查当前活跃数 active = await self.redis.scard(key) if active >= limit: raise ConcurrencyLimitError( f"User {user_id} has {active}/{limit} active agents" ) # 分配槽位 slot_id = str(uuid.uuid4()) await self.redis.sadd(key, slot_id) await self.redis.expire(key, 3600) # 1小时过期 return AgentSlot( slot_id=slot_id, user_id=user_id, agent_type=agent_type, acquired_at=time.time() ) async def release_slot(self, slot: AgentSlot): """释放 Agent 执行槽""" key = f"agent_slots:{slot.user_id}" await self.redis.srem(key, slot.slot_id) class AgentCoordinator: """Agent 协调器——管理多Agent并发执行""" async def run_agents_concurrent( self, agents: list[Agent], max_parallel: int = 5 ) -> list[AgentResult]: """并发运行多个 Agent""" semaphore = asyncio.Semaphore(max_parallel) async def run_one(agent: Agent) -> AgentResult: async with semaphore: try: result = await agent.run() return AgentResult( agent_id=agent.id, success=True, result=result ) except Exception as e: return AgentResult( agent_id=agent.id, success=False, error=str(e) ) return await asyncio.gather(*[run_one(a) for a in agents]) async def run_agents_pipeline( self, agents: list[Agent], dependencies: dict # {agent_id: [depends_on_ids]} ) -> dict[str, AgentResult]: """按依赖关系运行 Agent（拓扑排序）""" results = {} completed = set() while len(completed) < len(agents): # 找到可执行的 Agent（依赖已完成） ready = [ agent for agent in agents if agent.id not in completed and all( dep in completed for dep in dependencies.get(agent.id, []) ) ] if not ready: # 检测死锁 raise DeadlockError("Dependency cycle detected") # 并发执行就绪的 Agent batch_results = await self.run_agents_concurrent(ready) for result in batch_results: results[result.agent_id] = result completed.add(result.agent_id) return results 六、并发控制监控 class ConcurrencyMonitor: """并发控制监控""" METRICS = { "active_requests": "当前活跃请求数", "queue_depth": "各优先级队列深度", "concurrent_agents": "各用户活跃Agent数", "tool_concurrent_usage": "各工具并发使用数", "lock_wait_time": "锁等待时间分布", "lock_contention_rate": "锁竞争率", "rate_limit_hits": "限流触发次数", } async def health_check(self) -> dict: return { "healthy": await self._check_health(), "active_connections": await self._count_active(), "queue_health": await self._check_queues(), "lock_health": await self._check_locks(), "alerts": await self._get_alerts(), } async def detect_issues(self) -> list[Issue]: issues = [] # 检测队列堆积 for priority, queue in self.queues.items(): if queue.qsize() > queue.maxsize * 0.8: issues.append(Issue( severity="warning", type="queue_backlog", detail=f"{priority} queue at {queue.qsize()}/{queue.maxsize}" )) # 检测锁饥饿 lock_waits = await self._get_lock_wait_times() for resource, wait_time in lock_waits.items(): if wait_time > 5.0: issues.append(Issue( severity="high", type="lock_starvation", detail=f"{resource} lock wait: {wait_time:.1f}s" )) # 检测限流频繁触发 rate_limit_hits = await self._get_rate_limit_hits() for user_id, hits in rate_limit_hits.items(): if hits > 100: # 每分钟超过100次 issues.append(Issue( severity="medium", type="excessive_rate_limiting", detail=f"User {user_id} hit rate limit {hits} times" )) return issues 七、并发控制 Checklist □ 多维限流（QPS + 并发数 + Token + 成本） □ 优先级队列（高/中/低优先级请求隔离） □ 工具级并发限制（不同工具有不同并发上限） □ 全局并发限制（总并发不超过系统容量） □ 分布式锁保护共享资源（带看门狗续期） □ 乐观锁用于低冲突场景 □ Agent 实例数按用户等级限制 □ 依赖感知的 Agent 调度（拓扑排序） □ 死锁检测和预防 □ 并发监控和告警 □ 队列堆积自动扩容 □ 压测验证并发上限 结语 并发控制是 Agent 系统从"能跑"到"能扛"的关键。单机时代，一个信号量就够了；分布式时代，你需要考虑锁的公平性、可重入性、死锁预防和故障恢复。最好的并发控制是"恰到好处"——限制太松会导致雪崩，限制太紧会浪费资源。通过压测找到你的系统边界，然后在那里设置护栏。记住：并发控制不是阻碍性能，而是保护性能。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

为什么智能体系统需要特殊的负载均衡？ 传统的负载均衡主要处理无状态的 HTTP 请求，关注的是 QPS、延迟和可用性。但 AI 智能体系统的负载特征与传统 Web 服务有本质区别：单个请求的计算开销极大（一次 LLM 推理可能耗时数秒到数十秒）、请求之间可能存在状态依赖（多轮对话上下文）、且不同模型的资源需求差异巨大。 一个典型的 AGI 智能体系统可能同时包含：轻量级的意图分类（小模型，毫秒级响应）、重量级的推理任务（大模型，数十秒）、长周期的 Agent 工作流（多步骤，分钟级甚至小时级）、以及实时向量检索（RAG pipeline）。这些异构负载如果混在一起不加区分地调度，必然导致资源浪费和体验劣化。 智能体负载均衡的核心维度 1. 请求分类与路由 第一步是对进入系统的请求进行智能分类。不同于传统负载均衡器只看 URL 路径，智能体系统需要根据请求的语义特征进行路由： class RequestRouter: def __init__(self): self.routes = { "classification": {"model": "claude-haiku", "timeout": 2000, "pool": "fast"}, "reasoning": {"model": "gpt-4o", "timeout": 30000, "pool": "deep"}, "agent_workflow": {"model": "agent-orchestrator", "timeout": 300000, "pool": "long"}, "embedding": {"model": "text-embedding-3", "timeout": 1000, "pool": "vector"}, } def route(self, request): req_type = self.classify(request) route_config = self.routes[req_type] # 检查对应池的可用容量 if self.pool_available(route_config["pool"]): return self.dispatch(request, route_config) else: # 降级策略 return self.fallback(request, req_type) 关键设计点在于：不同类型的请求进入不同的处理池，彼此隔离，避免慢请求阻塞快请求。这就是所谓的"泳道隔离"模式。 2. 连接池与令牌桶 对于 LLM API 调用，两个核心限制是并发连接数和 token 速率限制。有效的管理需要同时处理这两个维度： 连接池管理：每个模型端点维护独立的连接池，池大小根据模型的并发能力和成本预算动态调整。例如，对于按量计费的 API，可以设置较高的并发上限；对于有 RPM 限制的 API，需要严格控制在限额以内。 令牌桶限流：不仅仅限制请求数量，更要限制 token 消耗速率。一个请求可能消耗 100 token，另一个可能消耗 10000 token。按 token 而非请求数来限流，才能真正做到公平调度： class TokenBucket: def __init__(self, capacity, refill_rate): self.capacity = capacity # 桶容量（最大突发） self.refill_rate = refill_rate # 每秒补充的 token 数 self.tokens = capacity self.last_refill = time.time() def consume(self, tokens_needed): self._refill() if self.tokens >= tokens_needed: self.tokens -= tokens_needed return True return False def _refill(self): now = time.time() elapsed = now - self.last_refill self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate) self.last_refill = now 3. 优先级队列 并非所有请求都同等重要。用户的实时对话请求应该优先于后台的批量处理任务。一个成熟的优先级策略通常包含 3-4 个级别： ...