free-llm-api-resources：LLM API并发控制实战指南——高效掌控速率限制与请求调度

2026-04-12 09:59:45作者：谭伦延

在基于免费LLM API构建应用时，开发者常面临请求被拒、服务封禁等问题，核心症结在于缺乏有效的并发控制策略。free-llm-api-resources作为免费LLM推理资源的集合，其API调用涉及多种速率限制机制（如请求/分钟、令牌/天等），需要针对性的并发管理方案。本文将围绕LLM API并发控制的核心痛点，系统介绍从问题诊断到工具优化的全流程实践，帮助开发者在合规前提下最大化API利用率。

问题诊断：免费LLM API的并发挑战

速率限制触发机制解析

免费LLM API通常通过多层级限制保护服务稳定性，主要包括：

请求频率限制：如OpenRouter的20次/分钟、50次/天限制
令牌消耗限制：如Groq的tokens/minute限制
IP级别的并发限制：部分API会限制单IP的同时连接数

这些限制通过响应头字段传递，如x-ratelimit-limit（总配额）、x-ratelimit-remaining（剩余配额）和x-ratelimit-reset（重置时间）。项目中src/pull_available_models.py模块已实现对这些头部信息的解析，为并发控制提供数据基础。

典型并发问题场景

突发流量超限：短时间集中请求导致分钟级限制触发
资源竞争冲突：多线程同时调用同一API导致整体超限
配额分配不均：未根据不同API的限制特性差异化控制

核心方案：三级并发控制体系

1. 固定延迟基础控制

适用场景：限制宽松的API（如Mistral）、简单脚本调用
实现方案：在请求间添加固定等待时间，确保不超过基础频率限制。

import time

class FixedDelayController:
    def __init__(self, min_interval=1):
        self.min_interval = min_interval  # 最小请求间隔(秒)
        self.last_request_time = 0
        
    def acquire(self):
        current_time = time.time()
        elapsed = current_time - self.last_request_time
        if elapsed < self.min_interval:
            time.sleep(self.min_interval - elapsed)  # 确保最小间隔
        self.last_request_time = time.time()

# 使用示例：创建1秒间隔的控制器
controller = FixedDelayController(1)
for prompt in prompts:
    controller.acquire()
    response = requests.post(api_url, json={"prompt": prompt})

优缺点分析：实现简单但效率低，无法充分利用动态配额。

2. 线程池并发调度

适用场景：多模型并行获取、批量处理任务
实现方案：通过限制线程池大小控制并发数量，结合队列管理请求顺序。

from concurrent.futures import ThreadPoolExecutor, as_completed

def process_model(model):
    # 模型处理逻辑，包含基础延迟控制
    controller.acquire()
    return fetch_model_data(model)

# 控制并发数为5，避免触发IP级并发限制
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(process_model, m): m for m in models}
    for future in as_completed(futures):
        model = futures[future]
        try:
            result = future.result()
        except Exception as e:
            logger.error(f"处理模型{model}失败: {str(e)}")

优缺点分析：平衡效率与稳定性，但需手动调整线程数适配不同API限制。

3. 动态限流算法实现

适用场景：严格限制的API（如Groq、OpenRouter）、生产环境应用
实现方案：基于API返回的实时配额数据，动态调整请求频率和并发数。

class DynamicRateLimiter:
    def __init__(self):
        self.limits = {}  # 存储各API的限制信息
        self.last_reset = {}  # 配额重置时间
        
    def update_limits(self, api_name, headers):
        # 解析响应头更新限制信息
        self.limits[api_name] = {
            "requests": int(headers["x-ratelimit-limit-requests"]),
            "tokens": int(headers["x-ratelimit-limit-tokens"]),
            "reset": int(headers["x-ratelimit-reset"])
        }
        self.last_reset[api_name] = time.time()
        
    def get_safe_interval(self, api_name, current_requests):
        # 计算安全请求间隔
        limits = self.limits[api_name]
        remaining_time = limits["reset"] - (time.time() - self.last_reset[api_name])
        remaining_requests = limits["requests"] - current_requests
        return remaining_time / max(remaining_requests, 1)  # 避免除零

# 使用示例：结合动态限流调用Groq API
limiter = DynamicRateLimiter()
response = requests.post(groq_url)
limiter.update_limits("groq", response.headers)
# 根据当前请求数计算下一次请求的安全间隔
interval = limiter.get_safe_interval("groq", current_request_count)
time.sleep(interval)

优缺点分析：最大化资源利用率，但实现复杂，需持续监控配额变化。

实战策略：多API协同与场景适配

OpenRouter API：令牌桶流量整形

针对OpenRouter的20次/分钟限制，采用令牌桶算法平滑请求流量：

from token_bucket import TokenBucket

# 初始化令牌桶：容量20，每分钟补充20个令牌
bucket = TokenBucket(20, 20/60)

def openrouter_request(prompt):
    while not bucket.consume(1):  # 获取1个令牌
        time.sleep(0.1)  # 令牌不足时等待
    return requests.post(openrouter_url, json={"prompt": prompt})

Groq API：响应头驱动的动态调整

利用Groq返回的详细限制头实现精细化控制：

def get_groq_limits():
    # 发送测试请求获取限制信息
    response = requests.post(groq_test_url)
    return {
        "rpd": int(response.headers["x-ratelimit-limit-requests"]),
        "tpm": int(response.headers["x-ratelimit-limit-tokens"])
    }

# 根据每日请求限制计算安全并发数
limits = get_groq_limits()
max_concurrent = max(1, limits["rpd"] // (24 * 60))  # 平均到每分钟的请求数

多API协同策略：优先级队列调度

当同时调用多个API时，通过优先级队列实现差异化调度：

import queue

# 创建优先级队列，高优先级API先执行
q = queue.PriorityQueue()
# 添加任务：(优先级, API名称, 请求参数)
q.put((1, "groq", {"prompt": "紧急任务"}))
q.put((2, "openrouter", {"prompt": "常规任务"}))

# 处理队列任务
while not q.empty():
    priority, api, params = q.get()
    if api == "groq":
        groq_request(params)
    else:
        openrouter_request(params)