构建LLM API限流处理系统：从令牌桶算法到智能负载均衡

1. 项目概述：当免费LLM API遇上限流，我们如何优雅应对？

在LLM应用开发中，尤其是个人项目、初创公司或教育研究场景，使用免费或带有试用额度的LLM API（如OpenRouter、Google AI Studio、Groq等）是控制成本、快速验证想法的绝佳途径。然而，天下没有免费的午餐，这些服务无一例外都伴随着严格的限流策略——无论是每分钟请求数、每日请求配额，还是每分钟令牌数限制。直接调用这些API，稍有不慎就会触发429 Too Many Requests或配额耗尽，导致应用中断，用户体验直线下降。

我最近在开发一个多轮对话的智能客服原型时，就深刻体会到了这一点。项目初期为了快速上线，我整合了多个来自free-llm-api-resources列表的免费API作为后备模型池。结果，在模拟并发测试时，系统频繁报错，不是因为A接口的每分钟请求超限，就是B接口的每日额度用尽。这让我意识到，单纯地调用API是远远不够的，必须有一套系统性的限流处理与容错策略，来保障服务的稳定性和连续性。

本文将基于free-llm-api-resources这个宝藏清单，深入拆解一套完整的LLM API限流处理解决方案。这套方案不仅适用于免费API，其核心思想同样可以迁移到任何需要管理多个、有配额限制的后端服务的场景。我们将从设计思路、核心策略、具体实现到避坑经验，一步步构建一个健壮、高效的API调用中间层。

2. 核心策略设计：从单点容错到智能调度

面对多个有限流规则的API端点，我们的目标是在有限的资源内最大化可用性和成功率。这需要超越简单的“重试”逻辑，构建一个分层、智能的调度系统。核心设计围绕以下几个原则展开：

1. 透明化与可观测性：所有API的调用状态、剩余配额、响应延迟都必须被实时监控和记录。
2. 优先级与降级：为不同API设定优先级，优先使用高质量或高配额的服务，当首选服务不可用时，能无缝切换到备选。
3. 预防而非补救：主动根据限流规则进行流量控制，尽量避免触发服务端的429错误。
4. 成本与性能平衡：在免费额度、响应速度、模型能力之间做出权衡。

基于这些原则，我设计了一个包含四层防御的限流处理架构：

2.1 第一层：客户端限流器（预防性限流）

这是最基础也是最重要的一层。它的作用是在我们的代码内部，模拟服务端的限流规则，主动控制发往每个API的请求速率，从而避免触发服务端的限制。

• 令牌桶算法实现：为每个API端点维护一个“令牌桶”。桶以固定速率（如每秒N个令牌）填充，每个请求消耗一个令牌。如果桶空，则请求必须等待或立即被拒绝。这完美对应了“每秒/每分钟请求数”的限制。
• 配额预算管理：对于“每日/每月请求数”或“令牌数”这类总量限制，我们需要一个“预算管理器”。它记录每个API周期内已使用的量，并在接近上限时发出警告或自动切换。
• 实现要点：可以使用像python的ratelimit库或asyncio的信号量（Semaphore）来简单实现。关键是要根据free-llm-api-resources中列出的具体规则进行精确配置。例如，对于OpenRouter的免费版，我们需要设置一个每分钟20次请求的全局限制器。

2.2 第二层：智能重试与回退（反应性容错）

尽管有第一层的预防，网络波动或意外的突发流量仍可能导致请求失败。这一层负责优雅地处理失败。

• 指数退避重试：对于网络错误或5xx服务器错误，采用指数退避策略进行重试（如等待1秒、2秒、4秒...后重试）。绝对不要对4xx客户端错误（特别是429）进行简单的立即重试，这只会加剧问题。
• 429错误特殊处理：当收到429错误时，解析响应头（如Retry-After）获取服务端建议的重试时间。如果没有该头，则采用一个更长的、随机的退避时间（如30-60秒）。
• 回退策略：当某个API连续失败或配额明确耗尽时，立即将请求路由到预设的备用API上。这要求我们的系统能感知每个端点的“健康状态”。

2.3 第三层：多API负载均衡与路由（资源调度）

这是系统的“大脑”。它根据预设策略和实时状态，决定将当前请求发送给哪个API。

• 策略类型：
  • 优先级路由：定义一个API优先级列表（如 Google AI Studio -> Groq -> OpenRouter）。总是优先使用列表中最靠前的、可用的API。
  • 加权随机路由：根据每个API的剩余配额比例或性能评分，分配请求权重。剩余配额多的API获得更高权重。
  • 最低延迟路由：定期探测各API的响应延迟，将新请求发给当前延迟最低的。
• 状态管理：维护一个全局的“API状态表”，记录每个端点的：是否启用、剩余配额估算、最近错误率、平均响应时间、当前并发数等。这个表是动态更新的。

2.4 第四层：请求队列与异步处理（流量整形）

对于无法立即处理的请求（例如所有API都暂时达到速率上限），引入一个队列机制，让请求排队等待，而不是直接返回失败。

• 异步任务队列：使用Celery、RQ或asyncio.Queue，将LLM调用封装为异步任务。用户请求提交后立即返回一个任务ID，后续通过轮询或WebSocket获取结果。
• 好处：
  1. 平滑流量峰值，避免瞬时冲击。
  2. 为用户提供了更友好的体验（“请求已接收，正在处理”）。
  3. 便于实现请求的优先级调度（VIP用户请求优先）。
• 注意事项：需要额外考虑任务结果的存储、过期清理和用户通知机制。

3. 基于Python的实战实现

下面，我将以一个Python实现为例，展示如何构建这个系统的核心部分。我们将使用aiohttp进行异步HTTP调用，redis作为令牌桶和状态存储，asyncio处理并发。

3.1 环境准备与依赖安装

首先，确保你的环境已准备好。我们主要需要以下库：

pip install aiohttp redis httpx python-dotenv

如果使用令牌桶算法，可以选择ratelimit或pyrate-limiter：

pip install pyrate-limiter

项目目录结构建议如下：

llm_api_gateway/ ├── config/ │ ├── __init__.py │ └── api_configs.py # 存放所有API的配置（密钥、端点、限流规则） ├── core/ │ ├── __init__.py │ ├── rate_limiter.py # 客户端限流器 │ ├── circuit_breaker.py # 熔断器（可选，用于处理连续故障） │ ├── load_balancer.py # 负载均衡与路由 │ └── api_client.py # 封装了重试、回退的API客户端 ├── models/ │ └── api_status.py # API状态数据模型 ├── utils/ │ └── logging_setup.py # 日志配置 ├── .env # 环境变量（API密钥） └── main.py # 主程序入口或FastAPI应用

3.2 核心模块详解与代码实现

3.2.1 API配置管理 (config/api_configs.py)

这是所有策略的基石。我们需要为每个API建立详细的档案。

from dataclasses import dataclass from typing import Optional, Dict, Any from enum import Enum class Provider(Enum): OPENROUTER_FREE = "openrouter_free" GOOGLE_AI_STUDIO = "google_ai_studio" GROQ_FREE = "groq_free" CEREBRAS_FREE = "cerebras_free" @dataclass class APIConfig: """单个API的配置模型""" name: str provider: Provider base_url: str api_key: str # 从环境变量读取 headers: Dict[str, str] # 限流规则 (根据 free-llm-api-resources 列表) requests_per_minute: int # RPM requests_per_day: Optional[int] = None # 每日上限 tokens_per_minute: Optional[int] = None # TPM # 路由权重和优先级 priority: int = 1 # 数字越小优先级越高 weight: float = 1.0 # 模型端点（例如，OpenRouter的模型名） default_model: Optional[str] = None # 是否启用 enabled: bool = True # 配置实例化 API_CONFIGS: Dict[Provider, APIConfig] = { Provider.OPENROUTER_FREE: APIConfig( name="OpenRouter Free Tier", provider=Provider.OPENROUTER_FREE, base_url="https://openrouter.ai/api/v1", api_key=os.getenv("OPENROUTER_API_KEY"), headers={"Authorization": f"Bearer {os.getenv('OPENROUTER_API_KEY')}"}, requests_per_minute=20, # 免费版限制 requests_per_day=50, priority=3, default_model="google/gemma-2b-it:free" ), Provider.GOOGLE_AI_STUDIO: APIConfig( name="Google AI Studio (Gemini Flash)", provider=Provider.GOOGLE_AI_STUDIO, base_url="https://generativelanguage.googleapis.com/v1beta", api_key=os.getenv("GOOGLE_AI_STUDIO_KEY"), headers={"Content-Type": "application/json"}, requests_per_minute=5, # Gemini 3 Flash tokens_per_minute=250000, requests_per_day=20, priority=1, # 高优先级，因为配额相对宽松 weight=2.0, default_model="models/gemini-1.5-flash" ), Provider.GROQ_FREE: APIConfig( name="Groq Cloud (Llama 3.1 8B)", provider=Provider.GROQ_FREE, base_url="https://api.groq.com/openai/v1", api_key=os.getenv("GROQ_API_KEY"), headers={"Authorization": f"Bearer {os.getenv('GROQ_API_KEY')}"}, requests_per_minute=30, # 假设值，需根据实际调整 requests_per_day=14400, priority=2, default_model="llama-3.1-8b-instant" ), }

实操心得：务必定期查阅free-llm-api-resources的GitHub仓库更新，因为免费服务的规则变动频繁。最好将这部分配置外置成JSON或YAML文件，方便动态更新而无需重启服务。

3.2.2 分布式令牌桶限流器 (core/rate_limiter.py)

为了实现多进程/多机器环境下的统一限流，我们使用Redis实现一个分布式的令牌桶。

import asyncio import time import redis.asyncio as redis from typing import Optional class DistributedTokenBucketLimiter: """基于Redis的分布式令牌桶限流器""" def __init__(self, redis_client: redis.Redis, bucket_key: str, capacity: int, refill_rate: float): """ Args: redis_client: Async Redis客户端 bucket_key: 用于存储桶状态的Redis键 capacity: 桶容量（令牌数） refill_rate: 每秒补充的令牌数 """ self.redis = redis_client self.bucket_key = bucket_key self.capacity = capacity self.refill_rate = refill_rate async def acquire(self, tokens: int = 1, timeout: float = 10.0) -> bool: """ 尝试获取指定数量的令牌。 Args: tokens: 需要的令牌数 timeout: 等待超时时间（秒） Returns: bool: 是否成功获取 """ start_time = time.time() while time.time() - start_time < timeout: # 使用Lua脚本保证原子性 lua_script = """ local key = KEYS[1] local capacity = tonumber(ARGV[1]) local refill_rate = tonumber(ARGV[2]) local tokens_needed = tonumber(ARGV[3]) local now = tonumber(ARGV[4]) -- 获取当前桶状态: [last_update_time, current_tokens] local bucket_data = redis.call('hmget', key, 'last_update', 'tokens') local last_update = bucket_data[1] local current_tokens = bucket_data[2] -- 初始化桶 if not last_update then last_update = now current_tokens = capacity redis.call('hmset', key, 'last_update', last_update, 'tokens', current_tokens) else last_update = tonumber(last_update) current_tokens = tonumber(current_tokens) end -- 计算应补充的令牌 local time_passed = now - last_update local tokens_to_add = math.floor(time_passed * refill_rate) if tokens_to_add > 0 then current_tokens = math.min(capacity, current_tokens + tokens_to_add) last_update = now redis.call('hmset', key, 'last_update', last_update, 'tokens', current_tokens) end -- 检查是否有足够令牌 if current_tokens >= tokens_needed then current_tokens = current_tokens - tokens_needed redis.call('hset', key, 'tokens', current_tokens) return 1 -- 成功 else return 0 -- 失败 end """ now = time.time() success = await self.redis.eval( lua_script, 1, self.bucket_key, self.capacity, self.refill_rate, tokens, now ) if success: return True # 等待一小段时间再重试 await asyncio.sleep(0.1) return False # 超时 # 使用示例：为每个API创建限流器 async def create_api_limiters(redis_client): limiters = {} for provider, config in API_CONFIGS.items(): # 将每分钟限制转换为每秒补充率 rpm = config.requests_per_minute capacity = max(5, rpm // 2) # 桶容量设为限流值的一半，允许一定突发 refill_rate = rpm / 60.0 # 每秒补充的令牌数 key = f"rate_limit:{provider.value}" limiters[provider] = DistributedTokenBucketLimiter(redis_client, key, capacity, refill_rate) return limiters

注意事项：Redis的Lua脚本确保了在并发环境下“检查-更新”操作的原子性，这是实现准确限流的关键。桶容量(capacity)不宜设置得过小，否则无法应对正常的请求波动；也不宜过大，否则失去了限流的意义。通常设置为限流值的1/2到1倍之间。

3.2.3 带熔断与重试的智能API客户端 (core/api_client.py)

这个客户端集成了限流检查、指数退避重试和简单的熔断逻辑。

import aiohttp import asyncio import logging from typing import Dict, Any, Optional, Tuple from dataclasses import dataclass from .rate_limiter import DistributedTokenBucketLimiter from config.api_configs import APIConfig, Provider logger = logging.getLogger(__name__) @dataclass class APIResponse: success: bool data: Optional[Dict[str, Any]] = None error: Optional[str] = None provider: Optional[Provider] = None latency: float = 0.0 # 响应延迟 class IntelligentAPIClient: def __init__(self, session: aiohttp.ClientSession, rate_limiters: Dict[Provider, DistributedTokenBucketLimiter]): self.session = session self.rate_limiters = rate_limiters # 熔断器状态：记录连续失败次数 self.circuit_breaker: Dict[Provider, int] = {p: 0 for p in API_CONFIGS.keys()} self.MAX_FAILURES = 5 # 连续失败5次触发熔断 self.RESET_TIMEOUT = 60 # 熔断60秒后尝试恢复 async def make_request( self, config: APIConfig, payload: Dict[str, Any], max_retries: int = 3 ) -> APIResponse: """ 发送请求，包含限流、重试和熔断逻辑。 """ provider = config.provider # 1. 检查熔断器 if self.circuit_breaker.get(provider, 0) >= self.MAX_FAILURES: logger.warning(f"Circuit breaker OPEN for {provider}. Skipping.") return APIResponse(success=False, error=f"Circuit breaker open for {provider}") # 2. 申请令牌（限流） limiter = self.rate_limiters.get(provider) if limiter: acquired = await limiter.acquire(timeout=2.0) if not acquired: logger.warning(f"Rate limit exceeded for {provider}. Request queued or failed.") # 这里可以触发路由到其他API，或进入队列 return APIResponse(success=False, error=f"Rate limit exceeded for {provider}") # 3. 准备请求 url = f"{config.base_url}/chat/completions" # 以OpenAI格式为例 headers = {**config.headers, "Content-Type": "application/json"} if config.default_model and "model" not in payload: payload["model"] = config.default_model # 4. 带指数退避的重试循环 last_exception = None for attempt in range(max_retries + 1): # 0, 1, 2, 3 try: start_time = asyncio.get_event_loop().time() async with self.session.post(url, json=payload, headers=headers, timeout=aiohttp.ClientTimeout(total=30)) as response: latency = asyncio.get_event_loop().time() - start_time if response.status == 200: # 成功，重置该API的失败计数 self.circuit_breaker[provider] = 0 data = await response.json() return APIResponse(success=True, data=data, provider=provider, latency=latency) elif response.status == 429: # Too Many Requests # 即使有客户端限流，也可能因其他原因触发429 retry_after = response.headers.get('Retry-After') wait_time = int(retry_after) if retry_after and retry_after.isdigit() else (2 ** attempt + 5) logger.warning(f"API {provider} rate limited (429). Retry after {wait_time}s.") self.circuit_breaker[provider] += 1 if attempt < max_retries: await asyncio.sleep(wait_time) continue # 重试 else: return APIResponse(success=False, error="Rate limited after max retries", provider=provider) elif 500 <= response.status < 600: # 服务器错误 logger.error(f"API {provider} server error {response.status}.") self.circuit_breaker[provider] += 1 if attempt < max_retries: wait = (2 ** attempt) + 1 # 指数退避：1, 3, 7秒... await asyncio.sleep(wait) continue else: return APIResponse(success=False, error=f"Server error {response.status}", provider=provider) else: # 其他4xx错误，如400 Bad Request, 401 Unauthorized error_text = await response.text() logger.error(f"API {provider} client error {response.status}: {error_text}") # 客户端错误通常不需要重试，但记录一次失败 self.circuit_breaker[provider] += 1 return APIResponse(success=False, error=f"Client error {response.status}: {error_text[:200]}", provider=provider) except (aiohttp.ClientError, asyncio.TimeoutError) as e: last_exception = e logger.error(f"Network/Timeout error for {provider} (attempt {attempt+1}): {e}") self.circuit_breaker[provider] += 1 if attempt < max_retries: await asyncio.sleep(2 ** attempt) continue # 所有重试都失败 return APIResponse(success=False, error=f"All retries failed. Last exception: {last_exception}", provider=provider) async def reset_circuit_breaker(self, provider: Provider): """手动重置某个API的熔断器""" self.circuit_breaker[provider] = 0 logger.info(f"Circuit breaker reset for {provider}.")

避坑技巧：对于429错误，一定要尊重响应头中的Retry-After。如果没有，切勿使用过短的退避时间，否则你的IP或API密钥可能会被临时封禁。指数退避中的随机抖动（jitter）也是一个好实践，可以避免多个客户端同时重试造成的“惊群效应”。

3.2.4 负载均衡与路由管理器 (core/load_balancer.py)

这个模块根据策略和实时状态选择最合适的API。

import random from typing import List, Optional from config.api_configs import API_CONFIGS, APIConfig, Provider from .api_client import IntelligentAPIClient, APIResponse from models.api_status import APIStatus # 假设有一个存储状态的数据模型 class LoadBalancer: def __init__(self, api_client: IntelligentAPIClient): self.api_client = api_client self.api_status: Dict[Provider, APIStatus] = self._init_api_status() def _init_api_status(self): """初始化API状态跟踪""" status = {} for provider, config in API_CONFIGS.items(): if config.enabled: status[provider] = APIStatus( provider=provider, config=config, total_requests=0, successful_requests=0, total_latency=0.0, last_error=None, last_success_time=None ) return status def _calculate_score(self, status: APIStatus) -> float: """计算一个API的当前得分，用于路由决策""" score = 0.0 config = status.config # 基础优先级权重（数字越小，优先级越高，得分越高） score += (10 - config.priority) * 20 # 成功率权重（最近N次请求） if status.total_requests > 0: success_rate = status.successful_requests / status.total_requests score += success_rate * 100 # 延迟权重（平均延迟越低，得分越高） if status.successful_requests > 0: avg_latency = status.total_latency / status.successful_requests # 假设1秒为基准，延迟越低加分越多 latency_score = max(0, 50 - avg_latency * 10) score += latency_score # 配置权重 score += config.weight * 10 # 如果最近有错误，减分 if status.last_error and (time.time() - status.last_error.timestamp) < 300: # 5分钟内 score -= 30 return score async def select_api(self, strategy: str = "priority") -> Optional[APIConfig]: """根据策略选择一个可用的API配置""" enabled_apis = [(p, s) for p, s in self.api_status.items() if s.config.enabled] if not enabled_apis: return None if strategy == "priority": # 按优先级排序，选择第一个可用的（这里简化了“可用”判断） sorted_apis = sorted(enabled_apis, key=lambda x: x[1].config.priority) for provider, status in sorted_apis: # 简单的健康检查：连续失败次数过多则跳过 if status.consecutive_failures < 3: return status.config elif strategy == "weighted_random": # 加权随机 scores = [self._calculate_score(s) for _, s in enabled_apis] # 将得分转换为权重，确保非负 weights = [max(s, 0.1) for s in scores] # 最小权重0.1 total_weight = sum(weights) if total_weight > 0: rand = random.uniform(0, total_weight) cumulative = 0 for (provider, status), weight in zip(enabled_apis, weights): cumulative += weight if rand <= cumulative: return status.config elif strategy == "lowest_latency": # 最低延迟（需要足够的历史数据） valid_apis = [(p, s) for p, s in enabled_apis if s.successful_requests > 5] if valid_apis: selected = min(valid_apis, key=lambda x: x[1].total_latency / x[1].successful_requests) return selected[1].config else: # 回退到优先级 return await self.select_api("priority") # 兜底：返回优先级最高的 return enabled_apis[0][1].config if enabled_apis else None async def route_request(self, payload: Dict[str, Any]) -> APIResponse: """路由请求到选定的API，并更新状态""" selected_config = await self.select_api(strategy="weighted_random") if not selected_config: return APIResponse(success=False, error="No available API endpoint") provider = selected_config.provider status = self.api_status[provider] status.total_requests += 1 # 调用API response = await self.api_client.make_request(selected_config, payload) # 更新状态 if response.success: status.successful_requests += 1 status.total_latency += response.latency status.last_success_time = time.time() status.consecutive_failures = 0 else: status.last_error = response.error status.consecutive_failures += 1 return response

4. 系统集成与高级特性

将上述模块组合起来，我们可以构建一个完整的API网关服务。这里以FastAPI为例，展示一个简单的集成端点。

from fastapi import FastAPI, HTTPException, BackgroundTasks from pydantic import BaseModel import asyncio import redis.asyncio as redis from core.load_balancer import LoadBalancer from core.api_client import IntelligentAPIClient from core.rate_limiter import create_api_limiters import aiohttp app = FastAPI(title="LLM API Gateway with Rate Limiting") # 全局初始化 redis_client = None api_client = None load_balancer = None class ChatRequest(BaseModel): messages: list model: Optional[str] = None temperature: float = 0.7 max_tokens: Optional[int] = None @app.on_event("startup") async def startup_event(): global redis_client, api_client, load_balancer # 初始化Redis连接 redis_client = redis.from_url("redis://localhost:6379", decode_responses=True) # 创建限流器 limiters = await create_api_limiters(redis_client) # 创建aiohttp会话和智能客户端 session = aiohttp.ClientSession() api_client = IntelligentAPIClient(session, limiters) # 创建负载均衡器 load_balancer = LoadBalancer(api_client) @app.on_event("shutdown") async def shutdown_event(): if api_client and api_client.session: await api_client.session.close() if redis_client: await redis_client.close() @app.post("/v1/chat/completions") async def chat_completion(request: ChatRequest, background_tasks: BackgroundTasks): """ 统一的聊天补全端点，内部处理路由、限流和重试。 """ payload = { "messages": request.messages, "temperature": request.temperature, } if request.model: payload["model"] = request.model if request.max_tokens: payload["max_tokens"] = request.max_tokens response = await load_balancer.route_request(payload) if response.success: # 可以在这里添加日志、监控数据上报等后台任务 background_tasks.add_task(log_success, response) return response.data else: # 如果所有策略都失败，返回一个友好的错误 raise HTTPException( status_code=503, detail={ "error": "Service temporarily unavailable", "message": "All LLM providers are currently unreachable or rate-limited. Please try again later.", "provider_error": response.error } ) async def log_success(response): """异步记录成功请求的日志""" # 实现你的日志逻辑，例如写入数据库或发送到监控系统 pass # 添加一个管理端点，查看当前各API状态 @app.get("/admin/status") async def get_api_status(): status_info = [] for provider, status in load_balancer.api_status.items(): success_rate = (status.successful_requests / status.total_requests * 100) if status.total_requests > 0 else 0 avg_latency = status.total_latency / status.successful_requests if status.successful_requests > 0 else 0 status_info.append({ "provider": provider.value, "enabled": status.config.enabled, "total_requests": status.total_requests, "success_rate": f"{success_rate:.1f}%", "avg_latency_ms": f"{avg_latency*1000:.1f}", "consecutive_failures": status.consecutive_failures, "last_error": status.last_error, }) return {"apis": status_info}

4.1 配额预算与每日重置

对于每日限额，我们需要一个独立的预算跟踪和重置机制。

import asyncio from datetime import datetime, timezone, timedelta class DailyQuotaManager: def __init__(self, redis_client: redis.Redis): self.redis = redis_client def _get_today_key(self, provider: Provider) -> str: """生成基于日期的Redis键，例如 quota:openrouter_free:2024-01-01""" today = datetime.now(timezone.utc).date().isoformat() return f"quota:{provider.value}:{today}" async def increment_usage(self, provider: Provider, tokens_used: int = 1) -> bool: """增加使用量，并检查是否超限""" key = self._get_today_key(provider) config = API_CONFIGS[provider] # 使用Redis的INCRBY和EXPIRE命令 # 设置键的过期时间为48小时，确保跨日重置 pipe = self.redis.pipeline() pipe.incrby(key, tokens_used) pipe.expire(key, 172800) # 48小时 current_usage = await pipe.execute()[0] if config.requests_per_day and current_usage > config.requests_per_day: logger.warning(f"Daily quota exceeded for {provider}. Usage: {current_usage}/{config.requests_per_day}") # 可以在这里触发警报或自动禁用该API return False return True async def get_remaining_quota(self, provider: Provider) -> int: """获取今日剩余配额""" key = self._get_today_key(provider) config = API_CONFIGS[provider] current_usage = int(await self.redis.get(key) or 0) if config.requests_per_day: return max(0, config.requests_per_day - current_usage) return float('inf') # 无限制

在IntelligentAPIClient.make_request的成功分支中，调用quota_manager.increment_usage(provider)来记录使用量。负载均衡器在select_api时，可以查询get_remaining_quota，将配额即将耗尽的API权重降为0。

4.2 请求队列实现（异步处理）

对于需要排队处理的场景，可以集成一个简单的内存队列或使用更专业的任务队列如Celery。

import asyncio from asyncio import Queue from typing import Callable, Any class AsyncRequestQueue: def __init__(self, maxsize: int = 1000): self.queue = Queue(maxsize=maxsize) self.processing_tasks = set() async def enqueue(self, request_id: str, payload: Dict[str, Any], callback: Callable): """将请求放入队列""" await self.queue.put((request_id, payload, callback)) async def start_worker(self, num_workers: int = 3): """启动工作协程处理队列中的请求""" for i in range(num_workers): task = asyncio.create_task(self._worker(f"worker-{i}")) self.processing_tasks.add(task) task.add_done_callback(self.processing_tasks.discard) async def _worker(self, name: str): """工作协程：从队列取任务，通过负载均衡器处理""" logger.info(f"Starting queue worker {name}") while True: try: request_id, payload, callback = await self.queue.get() logger.debug(f"Worker {name} processing request {request_id}") # 使用我们之前定义的负载均衡器处理请求 response = await load_balancer.route_request(payload) # 调用回调函数（例如，通过WebSocket通知前端，或更新数据库） await callback(request_id, response) self.queue.task_done() except asyncio.CancelledError: break except Exception as e: logger.error(f"Worker {name} error: {e}") # 可选：将失败的任务重新放回队列或记录到死信队列 await asyncio.sleep(1)

在FastAPI端点中，可以将耗时请求放入队列，立即返回一个任务ID。

task_results = {} # 临时存储结果，生产环境应用数据库 @app.post("/v1/chat/completions/async") async def create_async_chat_task(request: ChatRequest): request_id = str(uuid.uuid4()) payload = {**request.dict()} # 存储一个初始状态 task_results[request_id] = {"status": "pending", "result": None} # 定义回调：当任务完成时更新结果 async def on_task_done(req_id: str, response: APIResponse): task_results[req_id] = { "status": "completed" if response.success else "failed", "result": response.data if response.success else {"error": response.error}, "provider": response.provider.value if response.provider else None } # 放入队列 await request_queue.enqueue(request_id, payload, on_task_done) return {"task_id": request_id, "status": "queued", "message": "Request is being processed."} @app.get("/v1/chat/completions/async/{task_id}") async def get_async_task_result(task_id: str): result = task_results.get(task_id) if not result: raise HTTPException(status_code=404, detail="Task not found") return result

5. 监控、告警与运维建议

一个健壮的系统离不开监控。以下是一些关键的监控点：

1. API健康度仪表盘：利用之前实现的/admin/status端点，构建一个简单的仪表盘，实时展示各API的成功率、延迟、剩余配额估算和熔断状态。
2. 错误日志聚合：将所有API调用错误（429, 5xx, 网络超时）集中日志，并设置告警。例如，某个API在5分钟内连续失败10次，应触发告警。
3. 配额消耗预警：每日配额使用量达到80%时，发送通知（邮件、Slack等），提醒开发者可能需要切换API或升级计划。
4. 性能指标：记录每个请求的端到端延迟、令牌使用量，用于分析成本效益和优化模型选择。

运维建议：

• 密钥轮换与管理：将API密钥存储在环境变量或专业的密钥管理服务中，切勿硬编码。定期检查free-llm-api-resources列表，注册新的免费服务以扩充你的“模型池”。
• 配置热更新：实现一个机制，在不重启服务的情况下更新API配置（如限流参数、优先级），以快速响应服务商的政策变化。
• 降级与兜底：当所有外部API都不可用时，应有一个兜底策略。例如，返回一个缓存的通用回复，或者切换到一个极其轻量级的本地模型（如TinyLLama），哪怕效果差一些，也比直接报错好。
• 测试与演练：定期进行故障演练，模拟某个API完全失效或配额突然耗尽的情况，确保你的负载均衡和降级逻辑能按预期工作。

6. 常见问题与排查技巧实录

在实际部署和运行这套系统时，我遇到了不少坑，这里总结一下最常见的几个问题及其解决方法。

问题一：明明配置了每分钟20次的限流，为什么还是频繁收到429错误？

• 可能原因1：令牌桶容量设置过小。如果桶容量是1，即使补充速率是20/分钟，也无法应对任何瞬时并发。解决：将桶容量设置为限流值的1/2到1倍，例如对于20 RPM，容量设置为10-20。
• 可能原因2：分布式环境下的时钟同步问题。如果限流器依赖服务器时间，而多台机器时间不同步，会导致限流不准。解决：使用Redis等中心化存储来管理令牌桶状态，确保全局一致性。
• 可能原因3：服务端的限流规则更复杂。有些服务可能除了全局RPM，还有针对IP、账户、模型等多个维度的限流。解决：仔细阅读API文档，如果可能，在请求头中添加唯一标识（如X-Request-ID），并在收到429错误时分析返回的错误信息。

问题二：负载均衡器总是选择同一个API，导致其配额快速耗尽。

• 可能原因：如果使用“优先级”策略，且高优先级的API一直健康，那么流量永远不会落到低优先级API上。解决：
  1. 引入配额感知路由：在select_api逻辑中，查询DailyQuotaManager获取剩余配额，当剩余配额低于一定阈值（如10%）时，大幅降低其权重或临时禁用。
  2. 使用加权随机+动态权重：根据实时成功率和剩余配额动态计算权重，而不是固定值。
  3. 实现手动流量切换：通过管理接口，可以临时调整某个API的enabled状态或优先级。

问题三：异步队列中的任务堆积，响应延迟非常高。

• 可能原因：任务生产速度持续高于消费速度。可能是工作协程数量不足，或者所有API都达到了限流上限，导致每个任务处理极慢。解决：
  1. 监控队列长度：暴露queue.qsize()作为监控指标，设置告警。
  2. 动态扩缩容：根据队列长度动态增加或减少工作协程的数量。
  3. 设置任务超时与丢弃：为每个队列任务设置最大等待时间，超时后直接向用户返回“系统繁忙”错误，避免无限等待。对于非关键任务，可以实施有选择的丢弃。
  4. 分析瓶颈：如果是因为所有API都达到限流，说明免费资源已无法满足当前需求，需要考虑接入付费API或进行应用层面的优化（如缓存常见回答、合并请求等）。

问题四：如何测试整个限流和容错系统？

• 单元测试：对DistributedTokenBucketLimiter、IntelligentAPIClient中的重试逻辑等进行单元测试，模拟网络超时和不同HTTP状态码。
• 集成测试：使用像pytest-asyncio这样的工具，编写测试用例，模拟多个API端点不同行为（如一个快速成功，一个慢速，一个返回429）。
• 混沌工程测试：在测试环境中，使用工具临时阻断对某个API的网络访问，或使用Mock服务器模拟其返回429/500错误，观察系统是否能自动切换到备用API，以及熔断器是否正确触发和恢复。

这套基于free-llm-api-resources的限流处理策略，本质上是一个资源管理和调度系统。它的价值在于将不稳定的、有限制的免费资源，通过技术手段整合成一个相对稳定、可用的服务池。在实际项目中，我从一个经常被限流报警吵醒的状态，到系统能够平稳运行数周无需人工干预，这套方案起到了决定性的作用。它的核心思想——预防限流、优雅降级、智能调度——在任何依赖第三方服务的微服务架构中都具有普适性。