并发上限应设为rpm÷60×0.8；用asyncio.semaphore控制异步请求；redis list作跨进程队列；429时按retry-after休眠；合并请求前用tokenizer预估token并分块。
☞☞☞AI 智能聊天, 问答助手, AI 智能搜索, 多模态理解力帮你轻松跨越从0到1的创作门槛☜☜☜
用 asyncio.Semaphore 控制 Python 并发请求数
直接硬编码 max_concurrent = 10 是最常见也最容易出错的做法。关键不是设多少，而是这个值必须小于你账户在控制台看到的 RPM（每分钟请求数）除以 60 后再打八折——比如 RPM=60，那并发上限应设为 8，而非 10。
Python 异步场景下，asyncio.Semaphore 是轻量且线程安全的选择：
import asyncio sem = asyncio.Semaphore(8) <p>async def call_deepseek(prompt): async with sem: # 阻塞直到有许可 return await _real_api_call(prompt)</p>
注意：不要在 async with 外部做耗时操作（如大文件读取、复杂预处理），否则会虚占信号量；也不要在循环里反复创建新 Semaphore 实例。
用 Redis List 做跨进程任务队列
单机 asyncio 满足不了多服务实例或突发流量场景。此时必须把队列提到中间件层，Redis 是最快落地的选择——不用部署 Kafka，LPUSH + BRPOP 就能撑住每秒数百任务。
典型结构：
- 生产者用
redis.lpush("deepseek:queue", json.dumps(task)) - 消费者用
redis.brpop("deepseek:queue", timeout=5)阻塞拉取 - 每个 worker 进程独立维护自己的
API_KEY，避免密钥复用触发风控 - 失败任务写入
deepseek:failed列表，不直接丢弃
别忽略 BRPOP 的 timeout 参数：设为 0 会导致 CPU 空转，设为 5 是平衡延迟与资源的合理值。
统一LLM网关 - 一个API对接70+AI模型，使用单一API密钥即可调用GPT、Claude、Gemini、Qwen、Deepseek、Grok等主流模型。
重试时必须检查 Retry-After 响应头
收到 429 错误后立刻重试，等于主动撞墙。DeepSeek API 在返回 429 时，响应头中大概率带 Retry-After 字段，单位是秒。
实操要点：
- 优先读取
response.headers.get("Retry-After")，存在就按该值休眠 - 若为空，才 fallback 到指数退避：
1s → 2s → 4s，最多重试3次 - 不要用
time.sleep()在异步函数里阻塞，改用await asyncio.sleep(delay) - 记录每次重试的
retry_count和原始prompt，方便定位高频失败请求模式
批量请求合并需警惕 token 超限
把 10 条短 query 合成一条长 prompt 看似省调用次数，但 DeepSeek 免费版单次请求最大 token 数是 2048，商用版也常设 8192 上限。一旦超限，直接返回 400 错误，比 429 更难排查。
安全做法：
- 用 tokenizer（如
transformers.AutoTokenizer.from_pretrained("deepseek-ai/deepseek-coder-1.3b-base")）提前估算总 token 数 - 对长文本做分块：按语义切分（如按段落/句号），而非简单按字符数切
- 合并时加明确分隔符，例如
"---QUERY---\n{q1}\n---QUERY---\n{q2}"，避免模型混淆上下文 - 永远在合并前检查
len(prompt.encode("utf-8"))，防止非 ASCII 字符导致 token 计数偏差
真正容易被忽略的是：token 计数工具和 API 实际使用的 tokenizer 不一致时，预估结果会系统性偏高或偏低。上线前务必用真实响应里的 usage.total_tokens 反向校准。
大量免费API接口：立即使用
涵盖生活服务API、金融科技API、企业工商API、等相关的API接口服务。免费API接口可安全、合规地连接上下游，为数据API应用能力赋能！
