--- title: "Token Bucket 算法实现 LLM API 速率限制：工程参数与调优实践" route: "/posts/2026/04/13/token-bucket-rate-limiting-llm-api/" canonical_path: "/posts/2026/04/13/token-bucket-rate-limiting-llm-api/" canonical_url: "https://blog2.hotdry.top/posts/2026/04/13/token-bucket-rate-limiting-llm-api/" markdown_path: "/agent/posts/2026/04/13/token-bucket-rate-limiting-llm-api/index.md" markdown_url: "https://blog2.hotdry.top/agent/posts/2026/04/13/token-bucket-rate-limiting-llm-api/index.md" agent_public_path: "/agent/posts/2026/04/13/token-bucket-rate-limiting-llm-api/" agent_public_url: "https://blog2.hotdry.top/agent/posts/2026/04/13/token-bucket-rate-limiting-llm-api/" kind: "research" generated_at: "2026-04-13T19:18:17.960Z" version: "1" slug: "2026/04/13/token-bucket-rate-limiting-llm-api" date: "2026-04-13T17:09:03+08:00" category: "ai-systems" year: "2026" month: "04" day: "13" --- # Token Bucket 算法实现 LLM API 速率限制：工程参数与调优实践 > 深入解析 Token Bucket 算法的工程实现，涵盖桶容量、填充速率、令牌成本估算与突发流量处理的具体参数配置。 ## 元数据 - Canonical: /posts/2026/04/13/token-bucket-rate-limiting-llm-api/ - Agent Snapshot: /agent/posts/2026/04/13/token-bucket-rate-limiting-llm-api/index.md - 发布时间: 2026-04-13T17:09:03+08:00 - 分类: [ai-systems](/agent/categories/ai-systems/index.md) - 站点: https://blog2.hotdry.top ## 正文 在构建面向 LLM API 的网关或代理层时，速率限制是不可绕过的工程挑战。与传统 REST API 不同，LLM API 的请求成本以 token 计费，且输出 token 数量具有高度不确定性，这使得速率限制策略需要更精细的设计。Token Bucket 算法因其对突发流量的友好支持和对长期速率的精确控制，成为 LLM API 速率限制的首选方案。本文将从工程实现角度，详细阐述 Token Bucket 的核心参数配置、代码级实现细节以及生产环境的调优策略。 ## Token Bucket 算法核心原理 Token Bucket 的工作机制可以类比为一个盛放令牌的容器：系统以固定速率向桶中添加令牌，每个请求需要消耗一定数量的令牌才能放行。当桶满时，新加入的令牌会溢出；当桶空时，请求必须等待令牌积累到足够数量或直接被拒绝。这一算法之所以适用于 LLM 场景，是因为它能够在保证长期平均速率的同时，容忍一定程度的突发流量，而不会像固定窗口算法那样在边界处产生突刺。 在 LLM API 场景中，Token Bucket 需要处理两个维度的限制：请求速率（Requests Per Minute, RPM）和 token 速率（Tokens Per Minute, TPM）。前者控制 API 调用频次，后者控制实际计算资源的消耗。很多成熟的 LLM 提供商（如 Anthropic、OpenAI）都采用这种双限制模式，因此我们的实现也需要支持多桶并行校验。 ## 核心工程参数的设计与取值 ### 桶容量（Bucket Capacity） 桶容量决定了系统能够容忍的最大突发量。在传统 API 场景中，桶容量通常设置为与填充速率相同的量级，但在 LLM 场景中，考虑到请求的输入 token 差异巨大，建议将桶容量设置为预期平均 token 消耗量的 2 到 5 倍。例如，如果平均每次请求消耗 1000 个 token（包括输入和预估输出），则桶容量可设置为 3000 到 5000。容量过小会导致正常突发被误截，容量过大则会稀释速率限制的效果。 需要特别注意的是，桶容量并非越大越好。过大的容量意味着系统可能在长时间累积后一次性释放大量请求，形成"流量洪峰"，对下游 LLM 服务造成压力。建议桶容量不超过每分钟填充量的 2 倍。 ### 填充速率（Refill Rate） 填充速率决定了系统长期允许的吞吐量。计算填充速率时，需要综合考虑两个因素：目标 QPS（每秒查询数）和单次请求的平均 token 消耗。如果目标是每秒处理 10 个请求，每个请求平均消耗 500 token，则 token 填充速率应设置为每秒 5000 token，即每分钟 300000 token。 在工程实现中，填充速率的设置还需要留有安全余量。建议实际填充速率设置为目标吞吐量的 80% 到 90%，预留部分容量应对实际消耗超出预估的情况。对于需要同时限制 RPM 和 TPM 的场景，建议设置两个独立的 Token Bucket 分别进行校验。 ### 令牌消耗成本（Token Cost） Token Bucket 的消耗量并非固定为 1，而是根据请求的实际消耗动态计算。对于 LLM API，每次请求应该消耗「输入 token 数量加上预估的输出 token 数量」。输出 token 的预估通常采用经验公式：以输入 token 数量的 1.2 到 1.5 倍作为缓冲区间。例如，输入 1000 token 时，可按 1200 到 1500 token 进行扣减。 这种预估策略的优势在于：它能在保证不超限的前提下，尽量减少因输出超出预估而导致的限流。同时，在请求完成后，应根据实际输出的 token 数量进行「补扣」或「退还」操作，以保持计费的精确性。现代 LLM API 大多支持流式输出，补扣操作可以在最后一个 chunk 返回后异步执行。 ## 突发流量处理策略 ### 突发容许机制 Token Bucket 天然支持突发流量，因为桶中积累的令牌可以在短时间内集中消耗。关键在于如何设计突发容许阈值。一般建议将突发容量设置为填充速率的 2 到 5 秒累积量。例如，若每秒填充 100 token，则突发容量可设置为 200 到 500 token。这样可以在不影响长期速率的前提下，容忍用户短时间内的高频调用。 在实际生产环境中，突发流量往往来自于批量任务或定时同步场景。针对这类场景，建议在应用层实现任务队列和错峰调度，而非单纯依赖增大桶容量。通过将大任务拆分为小批次均匀分布，可以在不触发限流的前提下完成大规模处理。 ### 多级桶策略 单一桶难以同时满足短期突发和长期限制的需求。推荐采用「双桶策略」：一个用于限制每分钟请求数（RPM 桶），另一个用于限制每分钟 token 数（TPM 桶）。两个桶并行校验，只有同时通过时请求才能放行。这种设计在 LangChain 的 InMemoryRateLimiter 中有成熟实现，可作为工程参考。 此外，还可以引入「用户级桶」和「全局桶」的二级结构。用户级桶负责保护单个租户的配额不被其他租户侵蚀，全局桶则用于保护后端 LLM 服务的整体容量。当全局桶接近耗尽时，即使用户级桶仍有余额，也需要对请求进行排队或限流。 ## 代码级实现示例 以下是一个支持双桶校验和动态令牌消耗的简化实现，采用 Python 异步风格编写： ```python import asyncio import time from dataclasses import dataclass, field from typing import Dict, Optional import logging logger = logging.getLogger(__name__) @dataclass class TokenBucket: """Token Bucket 实现，支持动态令牌消耗与并发安全""" capacity: int refill_rate: float # 每秒填充令牌数 tokens: float = field(default=0) last_refill: float = field(default_factory=time.time) _lock: asyncio.Lock = field(default_factory=asyncio.Lock) async def consume(self, cost: float, wait: bool = True) -> bool: """尝试消耗令牌，成功返回 True，否则返回 False 或等待""" async with self._lock: self._refill() if self.tokens >= cost: self.tokens -= cost logger.debug(f"消耗 {cost} 令牌，剩余 {self.tokens}/{self.capacity}") return True if not wait: logger.warning(f"令牌不足，需要 {cost}，当前仅 {self.tokens}") return False # 计算等待时间 wait_time = (cost - self.tokens) / self.refill_rate logger.info(f"令牌不足，等待 {wait_time:.2f} 秒") await asyncio.sleep(wait_time) self._refill() self.tokens -= cost return True 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 class LLMRateLimiter: """双桶限流器：RPM + TPM""" def __init__( self, rpm_capacity: int = 60, rpm_refill_rate: float = 1.0, # 每秒 1 个请求 tpm_capacity: int = 300000, tpm_refill_rate: float = 5000.0, # 每秒 5000 token ): self.rpm_bucket = TokenBucket(rpm_capacity, rpm_refill_rate) self.tpm_bucket = TokenBucket(tpm_capacity, tpm_refill_rate) async def acquire( self, input_tokens: int, estimated_output_ratio: float = 1.3, wait: bool = True ) -> bool: """获取请求许可，自动计算 token 消耗""" # 预估输出 token 数并计算总消耗 estimated_output = int(input_tokens * estimated_output_ratio) total_cost = input_tokens + estimated_output # 先检查请求级桶 rpm_ok = await self.rpm_bucket.consume(1, wait) if not rpm_ok: return False # 再检查 token 级桶 tpm_ok = await self.tpm_bucket.consume(total_cost, wait) return tpm_ok async def report_actual_usage(self, actual_output_tokens: int): """根据实际输出补扣/退还令牌""" # 实际消耗与预估的差异，用于调优 pass ``` 上述实现展示了 Token Bucket 的核心逻辑：异步安全的锁保护、基于时间的令牌补充、动态消耗量计算以及可选的等待策略。在生产环境中，还需要增加指标埋点（当前令牌余额、等待时间、限流事件次数）以支持后续调优。 ## 监控指标与调优闭环 速率限制参数并非一成不变，需要通过持续监控进行动态调整。建议采集以下核心指标： **令牌相关指标**：当前桶内令牌余额反映系统当前的余量情况，若持续接近零则说明限制过严或流量超出预期。令牌消耗速率用于验证实际填充速率是否与预期一致。等待时间分布则反映请求被延迟的程度，若 P99 等待时间过长可能需要调整容量或填充速率。 **限流事件指标**：被拒绝的请求数量和占比是评估限流策略有效性的直接依据。需要按用户、端点、错误码等多个维度进行细分，以识别异常流量模式或配置错误。 **业务指标关联**：将限流事件与业务指标（如成功处理的请求数、平均响应延迟）进行关联分析，可以更准确地评估限流策略对用户体验的影响。 ## 回退与容错策略 即使配置了合理的 Token Bucket 参数，仍然可能遇到突发流量超过系统容量的极端情况。此时需要设计优雅的回退策略：立即返回 429 状态码并附带合理的 Retry-After 头；实现指数退避加随机抖动（jitter）避免惊群效应；为高优先级用户提供单独的限流桶以保证关键业务不被影响。 此外，建议在限流器前增加熔断机制：当后端 LLM 服务响应变慢或错误率上升时，主动降低放行速率，防止级联故障扩散。这种「自适应限流」可以在流量洪峰和后端故障双重压力下保护系统的可用性。 ## 小结 Token Bucket 算法为 LLM API 速率限制提供了灵活而强大的基础。关键在于合理配置三个核心参数：桶容量（通常为填充速率的 2 到 5 秒累积量）、填充速率（基于目标吞吐量并预留 10% 到 20% 安全余量）、令牌消耗成本（输入 token 加上预估输出的 1.2 到 1.5 倍）。配合双桶策略（ RPM 加 TPM ）、用户级与全局级二级结构以及完善的监控回退机制，可以构建出既保护后端资源又最大化业务吞吐量的速率限制系统。 在实际落地过程中，建议从小流量场景开始验证参数效果，逐步调整至最优配置，并通过监控数据持续迭代。 ## 资料来源 - [LLM Rate Limiting: Maximizing API Throughput Without Getting Throttled](https://www.nitrix-reloaded.com/2025/04/13/llm-rate-limiting-maximizing-api-throughput-without-getting-throttled/) - [InMemoryRateLimiter — LangChain Documentation](https://python.langchain.com/api_reference/core/rate_limiters/langchain_core.rate_limiters.InMemoryRateLimiter.html) ## 同分类近期文章 ### [Polymarket单边卖No策略的库存风险管理与做市商返利优化](/agent/posts/2026/04/14/polymarket-one-sided-no-position-inventory-risk-management/index.md) - 日期: 2026-04-14T02:53:43+08:00 - 分类: [ai-systems](/agent/categories/ai-systems/index.md) - 摘要: 聚焦持续卖出No头的单边做市策略，从金融工程角度分析寸头管理、对手方风险暴露、对冲成本计算与做市商返利优化路径。 ### [构建 Polymarket 自动化机器人：过滤非体育市场与持续买入 No 合约的工程实现](/agent/posts/2026/04/14/polymarket-bot-filter-non-sports-buy-no-contracts/index.md) - 日期: 2026-04-14T02:02:40+08:00 - 分类: [ai-systems](/agent/categories/ai-systems/index.md) - 摘要: 详解如何通过 Polymarket CLOB API 构建自动化交易机器人，实现非体育市场过滤与 No 合约持续买入的完整工程方案。 ### [多代理量化交易系统架构：角色分工、数据流编排与策略执行](/agent/posts/2026/04/14/multi-agent-quantitative-trading-architecture/index.md) - 日期: 2026-04-14T01:50:30+08:00 - 分类: [ai-systems](/agent/categories/ai-systems/index.md) - 摘要: 深入解析开源 AI 对冲基金项目的多代理系统架构设计，涵盖 19 个专业化代理的角色分工、集中式状态管理与串并联混合的数据流编排模式。 ### [Claude-Mem 深度解析：会话级自动记忆压缩与上下文注入机制](/agent/posts/2026/04/14/claude-mem-automatic-context-compression/index.md) - 日期: 2026-04-14T00:26:31+08:00 - 分类: [ai-systems](/agent/categories/ai-systems/index.md) - 摘要: 剖析 Claude Code 插件如何通过 5 个生命周期钩子实现会话上下文自动捕获，利用 AI 压缩后注入未来会话，突破上下文窗口限制。 ### [构建 AI Agent 基准污染检测流水线：自动化架构与工程参数](/agent/posts/2026/04/13/building-ai-agent-benchmark-contamination-detection-pipeline/index.md) - 日期: 2026-04-13T21:50:56+08:00 - 分类: [ai-systems](/agent/categories/ai-systems/index.md) - 摘要: 围绕 AI Agent 基准污染检测流水线，详述数据泄露与基准腐化的自动化识别架构、工程实现参数及持续监控策略。