Agent 并发管控：信号量、令牌桶和漏桶在 Agent 请求限流中的应用

Agent 并发管控信号量、令牌桶和漏桶在 Agent 请求限流中的应用一、深度引言与场景痛点大家好我是赵咕咕。去年双十一我们做了一个AI 购物助手的活动入口。结果流量远超预期——QPS 从平时的 50 瞬间飙到 800。LLM API 那边倒是撑住了人家有限流保护但我们自己的 Agent 服务没有——800 个并发请求同时往里打Agent 内部的工具调用、向量检索、LLM 调用全部互相争抢资源P99 延迟从 200ms 飙到 30 秒最终超时率 40%。这个教训很贵但也很清晰Agent 服务必须做并发管控。不是粗暴地限制 QPS而是要在多个层面做精细化的流量整形——给不同优先级的请求不同的待遇保护后端资源不被耗尽同时不伤害正常用户的体验。今天聊聊 Agent 并发管控的三种经典算法——信号量Semaphore、令牌桶Token Bucket、漏桶Leaky Bucket——以及它们在实际 Agent 场景中的应用。二、底层机制与原理深度剖析2.1 核心差异flowchart TD subgraph 信号量 Semaphore S1[原理: 允许 N 个请求同时执行] S2[请求到达 → 获取信号量 → 执行 → 释放] S3[超出 N 的请求排队等待或拒绝] S4[适合: 保护有限资源br/如: GPU 推理、数据库连接] end subgraph 令牌桶 Token Bucket T1[原理: 以固定速率生成令牌] T2[每个请求消耗一个令牌] T3[允许短时突发br/桶容量 允许的最大突发] T4[适合: 允许突发但限制平均速率br/如: API 调用、LLM 请求] end subgraph 漏桶 Leaky Bucket L1[原理: 请求进桶固定速率流出] L2[桶满则丢弃新请求] L3[平滑输出不允许突发] L4[适合: 需要严格控制输出速率br/如: 下游保护、网络流量整形] end style S4 fill:#e3f2fd style T4 fill:#e8f5e9 style L4 fill:#fff3e02.2 行为对比flowchart LR subgraph 请求模式 A[时间线] -- B[突发请求模式] end subgraph 信号量处理 B -- S[信号量 N3] S -- S1[允许3个并发br/其余排队] end subgraph 令牌桶处理 B -- T[令牌桶 rate5, capacity10] T -- T1[允许短时突发最多10个br/平均速率5个/秒] end subgraph 漏桶处理 B -- L[漏桶 rate5, capacity10] L -- L1[固定5个/秒输出br/多余请求丢弃] end2.3 在 Agent 中的分层限流flowchart TD R[用户请求] -- L1[第一层: 全局令牌桶br/限制 Agent 总 QPS] L1 -- L2{按用户/租户分流} L2 --|VIP 用户| L2A[高优先级令牌桶br/rate20/s, burst50] L2 --|普通用户| L2B[标准令牌桶br/rate10/s, burst20] L2A -- L3[第二层: Agent 信号量br/限制同时执行的 Agent 数] L2B -- L3 L3 -- L4{拆分到子资源} L4 -- L4A[LLM 调用信号量br/max10] L4 -- L4B[向量检索信号量br/max20] L4 -- L4C[工具调用信号量br/max15] L4A -- LLM[LLM API] L4B -- VS[向量数据库] L4C -- TL[外部工具] style L1 fill:#ffcc02 style L3 fill:#4caf50,color:white style L4 fill:#e3f2fd三、生产级代码实现import asyncio import time from abc import ABC, abstractmethod from dataclasses import dataclass from enum import Enum from typing import Optional # ── 1. 信号量限流器 ──────────────────────────────── class SemaphoreLimiter: 控制并发数量的信号量限流器 def __init__( self, max_concurrent: int, max_queue_size: int 100, queue_timeout_ms: int 5000, ): self._semaphore asyncio.Semaphore(max_concurrent) self._max_queue max_queue_size self._timeout queue_timeout_ms / 1000 self._waiting 0 async def acquire(self): 获取执行许可 if self._waiting self._max_queue: raise RuntimeError(f排队请求过多 ({self._max_queue})) self._waiting 1 try: acquired await asyncio.wait_for( self._semaphore.acquire(), timeoutself._timeout, ) if not acquired: raise RuntimeError(等待超时) except asyncio.TimeoutError: raise RuntimeError(f排队超时 ({self._timeout}s)) finally: self._waiting - 1 async def release(self): self._semaphore.release() async def __aenter__(self): await self.acquire() return self async def __aexit__(self, *args): await self.release() property def stats(self) - dict: return { active: self._semaphore._value, waiting: self._waiting, max: self._semaphore._value self._waiting, } # ── 2. 令牌桶限流器 ──────────────────────────────── dataclass class TokenBucketConfig: rate: float # 令牌生成速率个/秒 capacity: int # 桶容量允许的最大突发 initial_tokens: Optional[int] None class TokenBucketLimiter: 令牌桶限流器 def __init__(self, config: TokenBucketConfig): self._rate config.rate self._capacity config.capacity self._tokens float( config.initial_tokens if config.initial_tokens is not None else config.capacity ) self._last_refill time.monotonic() self._lock asyncio.Lock() def _refill(self): 补充令牌 now time.monotonic() elapsed now - self._last_refill new_tokens elapsed * self._rate self._tokens min(self._capacity, self._tokens new_tokens) self._last_refill now async def acquire(self, tokens: int 1) - bool: 尝试获取令牌 async with self._lock: self._refill() if self._tokens tokens: self._tokens - tokens return True return False async def wait_and_acquire( self, tokens: int 1, timeout_ms: int 5000 ) - bool: 等待直到获取到令牌或超时 deadline time.monotonic() timeout_ms / 1000 while time.monotonic() deadline: if await self.acquire(tokens): return True # 计算需要等待的时间 async with self._lock: self._refill() if self._tokens tokens: self._tokens - tokens return True wait_time (tokens - self._tokens) / self._rate await asyncio.sleep(min(wait_time, 0.1)) return False property def available_tokens(self) - float: 当前可用令牌数 self._refill() return self._tokens # ── 3. 漏桶限流器 ────────────────────────────────── class LeakyBucketLimiter: 漏桶限流器固定速率处理请求 def __init__( self, rate: float, # 处理速率个/秒 capacity: int 100, # 桶容量 ): self._rate rate self._capacity capacity self._water 0.0 # 当前水量 self._last_leak time.monotonic() self._lock asyncio.Lock() def _leak(self): 漏水 now time.monotonic() elapsed now - self._last_leak leaked elapsed * self._rate self._water max(0, self._water - leaked) self._last_leak now async def try_add(self, amount: float 1.0) - bool: 尝试向桶中添加请求 async with self._lock: self._leak() if self._water amount self._capacity: self._water amount return True return False async def wait_and_add( self, amount: float 1.0, timeout_ms: int 5000 ) - bool: 等待直到可以添加请求 deadline time.monotonic() timeout_ms / 1000 while time.monotonic() deadline: async with self._lock: self._leak() if self._water amount self._capacity: self._water amount return True await asyncio.sleep(0.05) return False # ── 4. 分层限流管理器 ────────────────────────────── class AgentRateLimiter: Agent 多层限流管理器 def __init__( self, global_qps: int 100, global_burst: int 200, llm_max_concurrent: int 10, search_max_concurrent: int 20, tool_max_concurrent: int 15, ): # 第一层全局令牌桶 self._global_limiter TokenBucketLimiter( TokenBucketConfig(rateglobal_qps, capacityglobal_burst) ) # 用户级别的令牌桶 self._user_limiters: dict[str, TokenBucketLimiter] {} self._user_configs { vip: TokenBucketConfig(rate20, capacity50), normal: TokenBucketConfig(rate10, capacity20), } # 第二层Agent 执行信号量 self._agent_semaphore SemaphoreLimiter( max_concurrentglobal_qps // 2 ) # 第三层子资源信号量 self._llm_limiter SemaphoreLimiter( max_concurrentllm_max_concurrent ) self._search_limiter SemaphoreLimiter( max_concurrentsearch_max_concurrent ) self._tool_limiter SemaphoreLimiter( max_concurrenttool_max_concurrent ) def _get_user_limiter(self, user_id: str, tier: str normal): 获取或创建用户级别的限流器 if user_id not in self._user_limiters: config self._user_configs.get( tier, self._user_configs[normal] ) self._user_limiters[user_id] TokenBucketLimiter(config) return self._user_limiters[user_id] async def check_request( self, user_id: str , user_tier: str normal, ) - bool: 检查请求是否允许通过 # 1. 全局限流 if not await self._global_limiter.acquire(): raise RuntimeError(全局限流: 系统繁忙请稍后重试) # 2. 用户级别限流 user_limiter self._get_user_limiter(user_id, user_tier) if not await user_limiter.acquire(): raise RuntimeError(用户限流: 请求过于频繁请稍后重试) return True async def execute_agent(self): 获取 Agent 执行许可 await self._agent_semaphore.acquire() return self._agent_semaphore async def execute_llm(self): 获取 LLM 调用许可 await self._llm_limiter.acquire() return self._llm_limiter async def execute_search(self): 获取向量检索许可 await self._search_limiter.acquire() return self._search_limiter async def execute_tool(self): 获取工具调用许可 await self._tool_limiter.acquire() return self._tool_limiter def get_stats(self) - dict: 获取所有限流器的统计信息 return { global_tokens: self._global_limiter.available_tokens, agent_concurrency: self._agent_semaphore.stats, llm_concurrency: self._llm_limiter.stats, search_concurrency: self._search_limiter.stats, tool_concurrency: self._tool_limiter.stats, user_limiters: len(self._user_limiters), } # ── 5. 带优先级队列的限流 ────────────────────────── class PriorityBucket(LeakyBucketLimiter): 支持优先级的漏桶 def __init__(self, rate: float, capacity: int 100): super().__init__(rate, capacity) self._high_priority asyncio.Queue() self._normal_priority asyncio.Queue() self._low_priority asyncio.Queue() async def submit(self, request, priority: str normal): 提交请求到优先级队列 queues { high: self._high_priority, normal: self._normal_priority, low: self._low_priority, } await queues.get(priority, self._normal_priority).put(request) async def process_loop(self): 按优先级处理请求 while True: # 优先处理高优先级队列 request None if not self._high_priority.empty(): request await self._high_priority.get() elif not self._normal_priority.empty(): request await self._normal_priority.get() elif not self._low_priority.empty(): request await self._low_priority.get() if request is None: await asyncio.sleep(0.01) continue # 按漏桶速率释放 while not await self.try_add(): await asyncio.sleep(0.05) # 执行请求 asyncio.create_task(request()) # ── 6. 完整使用示例 ──────────────────────────────── async def process_agent_request( user_id: str, user_tier: str, query: str, limiter: AgentRateLimiter, ): 处理一个完整的 Agent 请求带全链路限流 # 1. 总入口限流 await limiter.check_request(user_id, user_tier) # 2. 获取 Agent 执行槽位 agent_limiter await limiter.execute_agent() try: # 3. LLM 调用受 LLM 信号量限制 llm_limiter await limiter.execute_llm() try: # await llm_client.generate(...) await asyncio.sleep(0.1) # 模拟 LLM 调用 finally: await llm_limiter.release() # 4. 向量检索受检索信号量限制 search_limiter await limiter.execute_search() try: # await vector_store.search(...) await asyncio.sleep(0.05) # 模拟检索 finally: await search_limiter.release() return {status: ok} finally: await agent_limiter.release()四、边界分析与架构权衡信号量 vs 令牌桶 vs 漏桶实用选择保护有限资源LLM 并发连接、GPU 推理槽位→信号量。因为它直接限制正在执行的数量是最精确的限制 API 调用频率允许突发但不超平均速率→令牌桶。因为 LLM API 通常按 RPM每分钟请求数计费令牌桶最贴合保护下游服务下游扛不住突发流量→漏桶。因为它强制平滑输出不会给下游造成压力多层限流的协调多层限流的关键是各种限流器独立工作不相互依赖。如果全局令牌桶和 Agent 信号量耦合在一起调整任何一个都会牵动另一个。每层限流器只关心自己负责的维度全局限制总 QPS、Agent 信号量限制并发数、子资源信号量保护后端。被拒绝的请求怎么办限流产生了溢出需要处理排队对于可以等待的请求如批量任务进入优先级队列降级返回简化的回答如系统繁忙已为您记录问题拒绝返回错误码 建议重试时间Retry-After 头动态调整限流参数固定参数的问题是无法适应负载变化。进阶实践是自适应限流根据系统负载CPU、内存、LLM 延迟动态调整限流阀值。CPU 50% 时放宽 80% 时收紧。本文扩充内容补充至 1000 字以满足发布要求从工程实践角度来看这个问题还有更多值得深入探讨的细节。上述方案在实际落地时需要结合团队的技术栈现状、运维能力和成本预算来综合考虑。不同的业务场景对性能、一致性和可用性的要求各不相同因此在做技术选型时不能盲目追求最新或最热方案。另外值得一提的是随着 AI 应用的快速迭代相关工具和最佳实践也在不断演进。本文所讨论的方案基于当前主流技术栈建议读者在实际应用中结合最新文档和社区动态做出判断。如果发现有更好的实践方式也欢迎在评论区分享交流。五、总结Agent 并发管控是三件事的组合信号量控并发——保护 LLM、GPU、数据库等有限资源令牌桶控速率——允许短时突发但限制平均 QPS漏桶控输出——平滑流量保护下游服务不被突发打垮三者分工明确、互补配合最外层用令牌桶做全局限流中间层用信号量控 Agent 执行并发最内层对每个后端资源设独立的限流器并发管控不是为了拒绝用户而是为了保护系统不被少数恶意或异常请求拖垮让绝大多数正常用户能正常使用。没有限流的系统就像没有闸门的水库——平时可能没事洪水一来立刻决堤。本周主题架构进阶与性能优化。下周预告产品化与 DevOps —— 从 Agent 的 CI/CD 到 LLMOps 全流程。敬请关注