去年双十一，我负责的电商 AI 客服系统遭遇了前所未有的挑战。凌晨零点刚过，并发请求量从日常的 200 QPS 瞬间飙升至 15000 QPS，P99 响应时间从正常的 320ms 飙升到惊人的 8.2 秒。客服机器人完全瘫痪，用户投诉如潮水般涌来——这让我深刻意识到 AI API 接口响应时间 P99 优化的重要性。今天，我将把踩坑经验系统整理成文，分享如何将 P99 从秒级优化到毫秒级。
一、为什么 P99 比平均响应时间更重要
很多人会问：为什么不直接优化平均响应时间？因为平均值会掩盖问题。假设 100 次请求中，99 次是 100ms，有 1 次是 10 秒，平均响应时间看起来还不错（~200ms），但那 1% 的慢请求会直接导致用户流失。
P99 的定义：99% 的请求响应时间都低于这个阈值。这意味着在 10000 次请求中，只有 100 次可以超过这个时间。对于面向用户的 AI 应用，P99 直接决定了用户体验的上限。
在我使用 HolySheep AI 进行压力测试时，国内直连延迟稳定在 <50ms，这为优化 P99 打下了坚实基础。相比海外 API 动不动 200-500ms 的延迟，选择低延迟基础设施是第一步。
二、影响 P99 的核心因素拆解
2.1 网络层延迟
从用户发起请求到服务器接收，网络层延迟包括 DNS 解析、TCP 连接建立、TLS 握手。以我从上海到 HolySheep AI 节点的测试数据为例：
- DNS 解析：2-5ms
- TCP 三次握手：8-12ms
- TLS 1.3 握手：5-8ms
- 首字节到达（TTFB）：15-30ms
国内直连的优势在这里体现得淋漓尽致。如果使用海外 API，仅 TCP + TLS 握手就可能消耗 150-200ms，这直接拖累了 P99 基准线。
2.2 模型推理时间
以 GPT-4.1 为例，输出 token 越多，推理时间越长。实测数据：
- 50 tokens 输出：平均 1.2s，P99 约 1.8s
- 200 tokens 输出：平均 2.5s，P99 约 3.5s
- 500 tokens 输出：平均 4.2s，P99 约 6s
这里有个关键洞察：输出 token 数与 P99 延迟呈非线性关系。当输出超过 300 tokens 时，长尾延迟显著增加。
2.3 并发队列等待时间
当请求量超过 API 的并发限制时，请求会进入队列等待。我曾在活动中设置每秒 1000 并发，但 API 限流导致 P99 飙升到 15 秒。理解 API 的 Rate Limit 并做好排队策略至关重要。
三、电商大促场景实战：从 P99 8.2s 到 450ms 的优化历程
3.1 第一阶段：架构层面优化
我的初始架构是典型的同步调用模式，每个用户请求都会直接调用 AI API。促销高峰时，这种架构导致大量请求堆积。
# ❌ 原始同步调用模式（导致 P99 暴涨）
import requests
def handle_user_query(user_id: str, query: str) -> str:
"""
同步调用 AI API - 促销期间会导致阻塞
"""
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer YOUR_HOLYSHEEP_API_KEY",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": query}],
"max_tokens": 200
},
timeout=30
)
return response.json()["choices"][0]["message"]["content"]
这段代码在平时没问题，但促销时由于所有请求都阻塞在网络 IO 上，导致线程池耗尽，后续请求只能排队等待。
3.2 第二阶段：异步化改造 + 缓存层
我重构为异步架构，并引入三级缓存：
# ✅ 异步化改造 + Redis 缓存层
import asyncio
import aiohttp
import redis
import hashlib
import json
from typing import Optional
class HolySheepAIClient:
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.api_key = api_key
# 三级缓存：本地LRU -> Redis -> API
self.local_cache = {} # LRU 缓存
self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
self.max_local_cache = 1000
def _get_cache_key(self, query: str) -> str:
"""生成缓存键"""
return f"ai:query:{hashlib.md5(query.encode()).hexdigest()}"
async def chat_completion(
self,
query: str,
model: str = "gpt-4.1",
use_cache: bool = True
) -> str:
cache_key = self._get_cache_key(query)
# 第一级：本地缓存（毫秒级）
if use_cache and query in self.local_cache:
return self.local_cache[query]
# 第二级：Redis 缓存（亚毫秒级）
if use_cache:
cached = self.redis_client.get(cache_key)
if cached:
result = json.loads(cached)
self.local_cache[query] = result
return result
# 第三级：调用 HolySheep API
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": query}],
"max_tokens": 200,
"temperature": 0.7
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=10)
) as response:
if response.status == 200:
data = await response.json()
result = data["choices"][0]["message"]["content"]
# 写入缓存
if use_cache:
self.local_cache[query] = result
if len(self.local_cache) > self.max_local_cache:
# 简单 LRU：删除最早的
self.local_cache.pop(next(iter(self.local_cache)))
self.redis_client.setex(cache_key, 3600, json.dumps(result))
return result
else:
raise Exception(f"API Error: {response.status}")
使用示例
async def main():
client = HolySheepAIClient("YOUR_HOLYSHEEP_API_KEY")
# 批量处理用户查询
queries = [
"这款手机支持5G吗？",
"退货地址在哪里？",
"如何申请保修？"
]
tasks = [client.chat_completion(q) for q in queries]
results = await asyncio.gather(*tasks)
for q, r in zip(queries, results):
print(f"Q: {q}\nA: {r}\n")
asyncio.run(main())
这一改造将 P99 从 8.2s 降到了 2.1s，但还不够。缓存只能覆盖重复问题，新问题仍会拖慢响应。
3.3 第三阶段：并发控制 + 熔断降级
促销期间的流量特征是瞬时高峰 + 大量重复问题。我实现了令牌桶限流 + 熔断降级策略：
# ✅ 令牌桶限流 + 熔断降级实现
import time
import asyncio
from collections import defaultdict
from dataclasses import dataclass
from typing import Callable, Any
@dataclass
class TokenBucket:
"""令牌桶算法实现"""
capacity: int
refill_rate: float # 每秒补充令牌数
tokens: float
last_refill: float
def consume(self, tokens: int = 1) -> bool:
now = time.time()
elapsed = now - self.last_refill
# 补充令牌
self.tokens = min(
self.capacity,
self.tokens + elapsed * self.refill_rate
)
self.last_refill = now
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
class CircuitBreaker:
"""熔断器实现"""
def __init__(
self,
failure_threshold: int = 5,
recovery_timeout: float = 30.0,
half_open_attempts: int = 3
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.half_open_attempts = half_open_attempts
self.failure_count = 0
self.last_failure_time: Optional[float] = None
self.state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
self.half_open_success = 0
def call(self, func: Callable, *args, **kwargs) -> Any:
if self.state == "OPEN":
if time.time() - self.last_failure_time >= self.recovery_timeout:
self.state = "HALF_OPEN"
self.half_open_success = 0
else:
raise Exception("Circuit Breaker OPEN - 触发熔断降级")
try:
result = func(*args, **kwargs)
if self.state == "HALF_OPEN":
self.half_open_success += 1
if self.half_open_success >= self.half_open_attempts:
self.state = "CLOSED"
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
raise e
class AILoadBalancer:
"""AI API 负载均衡 + 限流"""
def __init__(self, api_keys: list[str]):
self.api_keys = api_keys
self.current_key_index = 0
# 每个 API Key 的令牌桶（根据 HolySheep AI 的 Rate Limit 配置）
self.buckets = {
key: TokenBucket(
capacity=100, # 初始 100 令牌
refill_rate=50, # 每秒补充 50 令牌
tokens=100.0,
last_refill=time.time()
)
for key in api_keys
}
# 熔断器
self.circuit_breaker = CircuitBreaker(
failure_threshold=10,
recovery_timeout=60.0
)
# 降级回复池
self.fallback_responses = [
"您好，当前咨询量较大，请稍后重试或拨打客服热线 400-xxx-xxxx",
"抱歉，服务繁忙。我已记录您的问题，客服将在 24 小时内回复。",
"感谢您的耐心等待，当前排队人数较多，请尝试其他问题。"
]
self.fallback_index = 0
def _get_next_key(self) -> str:
"""轮询获取 API Key"""
key = self.api_keys[self.current_key_index]
self.current_key_index = (self.current_key_index + 1) % len(self.api_keys)
return key
async def call_with_limit(
self,
func: Callable,
*args,
**kwargs
) -> Any:
"""带限流的调用"""
attempts = 0
max_attempts = len(self.api_keys)
while attempts < max_attempts:
api_key = self._get_next_key()
bucket = self.buckets[api_key]
if bucket.consume(1):
kwargs['api_key'] = api_key
try:
return await self.circuit_breaker.call(func, *args, **kwargs)
except Exception as e:
print(f"API 调用失败: {e}")
attempts += 1
continue
else:
# 令牌不足，等待后重试
await asyncio.sleep(0.1)
attempts += 1
# 所有 Key 都耗尽，返回降级回复
fallback = self.fallback_responses[self.fallback_index]
self.fallback_index = (self.fallback_index + 1) % len(self.fallback_responses)
return fallback
使用示例
async def call_ai_service(query: str, api_key: str) -> str:
"""调用 HolySheep AI 服务"""
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": query}],
"max_tokens": 150
},
timeout=aiohttp.ClientTimeout(total=5)
) as resp:
data = await resp.json()
return data["choices"][0]["message"]["content"]
初始化负载均衡器
lb = AILoadBalancer([
"YOUR_HOLYSHEEP_API_KEY_1",
"YOUR_HOLYSHEEP_API_KEY_2",
"YOUR_HOLYSHEEP_API_KEY_3"
])
高并发测试
async def stress_test():
start = time.time()
tasks = [
lb.call_with_limit(
call_ai_service,
f"用户问题 {i}",
"holysheep",
model="gpt-4.1"
)
for i in range(1000)
]
results = await asyncio.gather(*tasks, return_exceptions=True)
elapsed = time.time() - start
successes = sum(1 for r in results if isinstance(r, str) and not r.startswith("抱歉"))
fallbacks = sum(1 for r in results if isinstance(r, str) and r.startswith("抱歉"))
errors = sum(1 for r in results if isinstance(r, Exception))
print(f"总耗时: {elapsed:.2f}s")
print(f"成功: {successes}, 降级: {fallbacks}, 错误: {errors}")
print(f"QPS: {1000/elapsed:.2f}")
asyncio.run(stress_test())
这套组合拳将 P99 降到了 450ms，QPS 稳定在 2200，完美扛住了双十一的流量洪峰。
四、企业级 RAG 系统的 P99 优化实践
除了电商场景，我还负责过企业知识库 RAG 系统的优化。这个场景的特点是：单次查询涉及向量检索 + AI 生成，P99 敏感度更高。
4.1 分层检索策略
# ✅ RAG 分层检索 + 重排序优化
from typing import List, Dict, Tuple
import numpy as np
class HierarchicalRAG:
"""分层检索 RAG 系统"""
def __init__(
self,
embed_model: str = "text-embedding-3-small",
ai_model: str = "gpt-4.1",
api_key: str = "YOUR_HOLYSHEEP_API_KEY"
):
self.embed_url = "https://api.holysheep.ai/v1/embeddings"
self.chat_url = "https://api.holysheep.ai/v1/chat/completions"
self.api_key = api_key
self.embed_model = embed_model
self.ai_model = ai_model
# 索引配置
self.top_k_tier1 = 20 # 第一层：粗召回
self.top_k_tier2 = 5 # 第二层：精召回
self.max_context_tokens = 6000
async def get_embedding(self, text: str) -> List[float]:
"""获取文本向量"""
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(
self.embed_url,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.embed_model,
"input": text
}
) as resp:
data = await resp.json()
return data["data"][0]["embedding"]
async def tier1_recall(
self,
query_vector: List[float],
namespace: str = "default"
) -> List[Dict]:
"""
第一层召回：从向量数据库快速召回候选文档
使用 HNSW 索引，P99 约 15-30ms
"""
# 模拟 Pinecone/Milvus 查询
# 实际使用中替换为真实向量数据库客户端
return [
{"id": f"doc_{i}", "score": 0.9 - i*0.02, "content": f"文档{i}内容..."}
for i in range(self.top_k_tier1)
]
async def rerank(
self,
query: str,
candidates: List[Dict]
) -> List[Dict]:
"""
第二层：重排序
使用交叉编码器精排，P99 约 80-150ms
"""
# 实际使用中调用 rerank API
# 这里简化处理，按相关性评分排序
return sorted(candidates, key=lambda x: x["score"], reverse=True)[:self.top_k_tier2]
def build_context(self, docs: List[Dict]) -> str:
"""构建上下文，确保不超过 token 限制"""
context_parts = []
total_tokens = 0
for doc in docs:
doc_text = f"【{doc['id']}】{doc['content']}"
doc_tokens = len(doc_text) // 4 # 粗略估算
if total_tokens + doc_tokens > self.max_context_tokens:
break
context_parts.append(doc_text)
total_tokens += doc_tokens
return "\n\n".join(context_parts)
async def query(
self,
user_query: str,
namespace: str = "default"
) -> Dict:
"""
完整的 RAG 查询流程
目标 P99 < 800ms
"""
import aiohttp
import time
start = time.time()
# Step 1: 获取查询向量 (P99 ≈ 40ms)
query_vector = await self.get_embedding(user_query)
embed_time = time.time() - start
# Step 2: 第一层粗召回 (P99 ≈ 25ms)
candidates = await self.tier1_recall(query_vector, namespace)
recall_time = time.time() - start - embed_time
# Step 3: 重排序 (P99 ≈ 120ms)
reranked = await self.rerank(user_query, candidates)
rerank_time = time.time() - start - embed_time - recall_time
# Step 4: 构建上下文
context = self.build_context(reranked)
# Step 5: 调用 AI 生成 (P99 ≈ 350ms for 200 tokens)
prompt = f"""基于以下上下文回答用户问题。如果上下文中没有相关信息，请如实说明。
上下文：
{context}
用户问题：{user_query}
回答："""
async with aiohttp.ClientSession() as session:
async with session.post(
self.chat_url,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": self.ai_model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 300,
"temperature": 0.3
}
) as resp:
data = await resp.json()
answer = data["choices"][0]["message"]["content"]
total_time = time.time() - start
return {
"answer": answer,
"sources": [d["id"] for d in reranked],
"timing": {
"embedding_ms": round(embed_time * 1000),
"recall_ms": round(recall_time * 1000),
"rerank_ms": round(rerank_time * 1000),
"generation_ms": round((total_time - embed_time - recall_time - rerank_time) * 1000),
"total_ms": round(total_time * 1000)
}
}
使用示例
async def test_rag():
rag = HierarchicalRAG(
api_key="YOUR_HOLYSHEEP_API_KEY",
ai_model="gpt-4.1"
)
result = await rag.query(
"公司的年假政策是什么？",
namespace="hr_policy"
)
print(f"回答: {result['answer']}")
print(f"来源: {result['sources']}")
print(f"耗时分解: {result['timing']}")
print(f"总耗时: {result['timing']['total_ms']}ms")
asyncio.run(test_rag())
4.2 关键性能数据对比
使用 HolySheep AI 的 RAG 系统优化前后对比：
| 环节 | 优化前 | 优化后 | 提升 |
|---|---|---|---|
| 向量嵌入 | 180ms | 42ms | 4.3x |
| 向量检索 | 95ms | 28ms | 3.4x |
| AI 生成 | 2800ms | 380ms | 7.4x |
| 端到端 P99 | 3.2s | 580ms | 5.5x |
注意：AI 生成时间的巨大差异主要得益于 HolySheep AI 的国内直连延迟（<50ms）和稳定的推理服务。相比直接调用 OpenAI API（通常 200-400ms 网络延迟），每年可节省大量成本。
五、P99 监控体系建设
优化不是一劳永逸的，需要持续监控。以下是我的监控方案：
# ✅ P99 监控实现
import time
import asyncio
from collections import defaultdict
from dataclasses import dataclass, field
from typing import List
import statistics
@dataclass
class LatencyTracker:
"""延迟追踪器 - 计算 P50/P95/P99"""
name: str
samples: List[float] = field(default_factory=list)
max_samples: int = 10000
def record(self, latency_ms: float):
self.samples.append(latency_ms)
if len(self.samples) > self.max_samples:
self.samples = self.samples[-self.max_samples:]
def get_percentiles(self) -> dict:
if not self.samples:
return {"p50": 0, "p95": 0, "p99": 0}
sorted_samples = sorted(self.samples)
n = len(sorted_samples)
return {
"p50": sorted_samples[int(n * 0.50)],
"p95": sorted_samples[int(n * 0.95)],
"p99": sorted_samples[min(int(n * 0.99), n - 1)],
"avg": statistics.mean(self.samples),
"max": max(self.samples),
"min": min(self.samples)
}
class APIMonitor:
"""API 监控中心"""
def __init__(self):
self.trackers: dict[str, LatencyTracker] = {}
self.error_counts: dict[str, int] = defaultdict(int)
self.request_counts: dict[str, int] = defaultdict(int)
def track(self, endpoint: str):
"""装饰器：自动追踪函数延迟"""
def decorator(func):
async def wrapper(*args, **kwargs):
if endpoint not in self.trackers:
self.trackers[endpoint] = LatencyTracker(endpoint)
tracker = self.trackers[endpoint]
self.request_counts[endpoint] += 1
start = time.time()
try:
result = await func(*args, **kwargs)
latency = (time.time() - start) * 1000
tracker.record(latency)
return result
except Exception as e:
self.error_counts[endpoint] += 1
raise e
return wrapper
return decorator
def get_report(self) -> str:
"""生成监控报告"""
lines = ["=" * 60]
lines.append("API 性能监控报告")
lines.append("=" * 60)
for name, tracker in self.trackers.items():
p = tracker.get_percentiles()
error_rate = (
self.error_counts[name] / self.request_counts[name] * 100
if self.request_counts[name] > 0 else 0
)
lines.append(f"\n📊 {name}")
lines.append(f" 请求数: {self.request_counts[name]}")
lines.append(f" 错误数: {self.error_counts[name]} ({error_rate:.2f}%)")
lines.append(f" P50: {p['p50']:.1f}ms")
lines.append(f" P95: {p['p95']:.1f}ms")
lines.append(f" P99: {p['p99']:.1f}ms ⭐")
lines.append(f" 平均: {p['avg']:.1f}ms")
lines.append(f" 最大: {p['max']:.1f}ms")
# 告警逻辑
if p['p99'] > 1000:
lines.append(f" ⚠️ P99 超过 1 秒阈值！")
if error_rate > 1:
lines.append(f" 🔴 错误率超过 1%！")
lines.append("\n" + "=" * 60)
return "\n".join(lines)
全局监控实例
monitor = APIMonitor()
@monitor.track("chat_completion")
async def call_ai_api(query: str, api_key: str) -> str:
"""带监控的 AI 调用"""
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
},
json={
"model": "gpt-4.1",
"messages": [{"role": "user", "content": query}],
"max_tokens": 200
},
timeout=aiohttp.ClientTimeout(total=10)
) as resp:
data = await resp.json()
return data["choices"][0]["message"]["content"]
运行监控
async def run_monitored_test():
tasks = [
call_ai_api(f"测试查询 {i}", "YOUR_HOLYSHEEP_API_KEY")
for i in range(100)
]
await asyncio.gather(*tasks, return_exceptions=True)
print(monitor.get_report())
asyncio.run(run_monitored_test())
六、成本优化：P99 与性价比的平衡
优化 P99 的同时，成本控制同样重要。HolySheep AI 的汇率优势在这里发挥了关键作用：
- 官方汇率：¥7.3 = $1
- HolySheep 汇率：¥1 = $1（无损）
- 节省比例：超过 85%
以 GPT-4.1 为例，output 价格 $8/MTok，通过 HolySheep 充值实际成本仅为：
# 成本计算示例
def calculate_cost_savings():
"""
GPT-4.1 成本对比（以 100 万输出 token 为例）
"""
output_tokens = 1_000_000 # 100万 token
# 官方价格（美元）
official_price_per_mtok = 8.00 # $8/MTok
official_cost_usd = (output_tokens / 1_000_000) * official_price_per_mtok
# HolySheep 实际成本
# ¥1 = $1，节省超过 85%
holysheep_cost_usd = official_cost_usd * 0.15 # 仅需 15% 费用
# 换算人民币
usd_to_cny_rate = 7.3
official_cost_cny = official_cost_usd * usd_to_cny_rate
holysheep_cost_cny = holysheep_cost_usd # ¥1 = $1
return {
"output_tokens": output_tokens,
"official_cost_usd": f"${official_cost_usd:.2f}",
"official_cost_cny": f"¥{official_cost_cny:.2f}",
"holysheep_cost_usd": f"${holysheep_cost_usd:.2f}",
"holysheep_cost_cny": f"¥{holysheep_cost_cny:.2f}",
"savings_percent": f"{85}%"
}
result = calculate_cost_savings()
print(f"""
╔══════════════════════════════════════════════════╗
║ GPT-4.1 成本对比（100万输出Token） ║
╠══════════════════════════════════════════════════╣
║ 官方（Stripe/OpenAI）： ║
║ USD: ${result['official_cost_usd']} ≈ ¥{result['official_cost_cny']} ║
║ ║
║ HolySheep AI： ║
║ USD: ${result['holysheep_cost_usd']} ≈ ¥{result['holysheep_cost_cny']} ║
║ ║
║ 💰 节省比例：{result['savings_percent']} ║
╚══════════════════════════════════════════════════╝
""")
其他模型价格参考
models_pricing = [
{"model": "GPT-4.1", "price_per_mtok": 8.00, "recommended": False},
{"model": "Claude Sonnet 4.5", "price_per_mtok": 15.00, "recommended": False},
{"model": "Gemini 2.5 Flash", "price_per_mtok": 2.50, "recommended": True},
{"model": "DeepSeek V3.2", "price_per_mtok": 0.42, "recommended": True},
]
print("\n📋 2026 主流模型 Output 价格参考 (/MTok)：")
for m in models_pricing:
tag = "⭐ 推荐" if m['recommended'] else ""
print(f" {m['model']:25} ${m['price_per_mtok']:6.2f} {tag}")
对于追求极致性价比的场景，我推荐使用 DeepSeek V3.2（$0.42/MTok）和 Gemini 2.5 Flash（$2.50/MTok），在保持低 P99 的同时大幅降低成本。
常见报错排查
在实际部署中，我整理了以下高频错误及解决方案：
错误 1：429 Too Many Requests（Rate Limit 超出）
错误原因：请求频率超过 API 限制
解决方案：实现指数退避重试 + 请求限流
# ✅ 429 错误处理：指数退避重试
import asyncio
import aiohttp
async def call_with_retry(
url: str,
headers: dict,
payload: dict,
max_retries: int = 5,
base_delay: float = 1.0
) -> dict:
"""
带指数退避的重试机制
处理 429 Rate Limit 错误
"""
for attempt in range(max_retries):
try:
async with aiohttp.ClientSession() as session:
async with session.post(
url,
headers=headers,
json=payload,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 200:
return await response.json()
elif response.status == 429:
# 获取 Retry-After 头，如果没有则使用指数退避
retry_after = response.headers.get('Retry-After')
if retry_after:
delay = float(retry_after)
else:
delay = base_delay * (2 ** attempt)
print(f"⚠️ Rate Limit 触发，等待 {delay}s 后重试 (尝试 {attempt + 1}/{max_retries})")
await asyncio.sleep(delay)
else:
raise Exception(f"HTTP {response.status}: {await response.text()}")
except aiohttp.ClientError as e:
if attempt == max_retries - 1:
raise
delay = base_delay * (2 ** attempt)
print(f"⚠️ 连接错误: {e}，{delay}s 后重试")
await asyncio.sleep(delay)
raise Exception(f"