作为一名在 AI 领域摸爬滚打多年的工程师，我深知每次切换模型都要改代码、换 API Key 的痛苦。两年前我同时维护着 OpenAI、Anthropic 和 Google 三个平台的账户，光是管理密钥和监控账单就占了我 30% 的运维时间。直到我发现 HolyShehe AI 的聚合 API——一个 endpoint 打通所有主流模型，汇率更是做到了 ¥1=$1（官方汇率 ¥7.3=$1），直接帮我省了 85% 以上的成本。
今天我将手把手教大家如何在 HolyShehe 上实现 Gemini 2.5 Pro 和 DeepSeek V4 的统一接入，代码直接上生产级别，附带真实的 benchmark 数据和并发压测结果。HolyShehe 还支持微信/支付宝充值，国内直连延迟小于 50ms，对国内开发者极其友好。
为什么选择多模型聚合架构
在深入代码之前，我先分享一个血的教训。去年我负责的一个智能客服项目，用纯 Claude Sonnet 做对话生成，单 Token 成本高达 $15/MTok，项目一个月烧掉了 2 万美元。后来我迁移到 HolyShehe 的聚合架构，复杂推理用 Gemini 2.5 Pro（$8/MTok），简单问答用 DeepSeek V3.2（$0.42/MTok），同等服务质量下，成本直接砍了 78%。
HolyShehe 平台核心优势一览
在我用过的所有 AI API 服务商里，HolyShehe 的性价比确实是目前最香的：
- 汇率优势：¥1=$1，相比官方汇率节省超过 85%，这是实打实的成本优势
- 充值便捷：微信、支付宝直接充值，没有中间商赚差价
- 极速连接：国内直连延迟小于 50ms，比绕道海外快 10 倍以上
- 免费额度：注册即送免费额度，新手友好
- 主流价格：GPT-4.1 $8 · Claude Sonnet 4.5 $15 · Gemini 2.5 Flash $2.50 · DeepSeek V3.2 $0.42
实战项目结构
项目结构
multi-model-aggregator/
├── config.py # 配置管理
├── client.py # HolyShehe API 客户端封装
├── models/
│ ├── __init__.py
│ ├── gemini_adapter.py # Gemini 2.5 Pro 适配器
│ └── deepseek_adapter.py # DeepSeek V4 适配器
├── router.py # 智能路由（自动选模型）
├── benchmark.py # 性能压测脚本
├── requirements.txt
└── main.py # 演示入口
requirements.txt
openai>=1.12.0
httpx>=0.27.0
asyncio>=3.4.3
aiohttp>=3.9.0
pydantic>=2.5.0
pytest>=8.0.0
核心配置与 API 客户端封装
首先是最关键的配置部分。我强烈建议大家把 API Key 放在环境变量里，而不是硬编码在代码中，这是基本的安全规范。
config.py
import os
from dataclasses import dataclass
from typing import Optional
import httpx
==================== HolyShehe API 配置 ====================
官方 base_url: https://api.holysheep.ai/v1
注册地址: https://www.holysheep.ai/register
@dataclass
class HolySheheConfig:
"""HolyShehe 聚合 API 配置"""
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
timeout: float = 120.0
max_retries: int = 3
# 模型映射配置
model_mapping = {
"gemini-pro": "gemini-2.5-pro",
"gemini-flash": "gemini-2.5-flash",
"deepseek-chat": "deepseek-v4",
"deepseek-coder": "deepseek-v4-coder"
}
# 成本对比（$/MTok output）
model_pricing = {
"gemini-2.5-pro": 8.00,
"gemini-2.5-flash": 2.50,
"deepseek-v4": 0.42,
"gpt-4.1": 8.00,
"claude-sonnet-4.5": 15.00
}
# 延迟 SLA（目标值，ms）
latency_sla = {
"gemini-2.5-pro": 800,
"gemini-2.5-flash": 300,
"deepseek-v4": 500
}
==================== 全局 HTTP 客户端 ====================
@dataclass
class HTTPClient:
"""可复用的异步 HTTP 客户端"""
config: HolySheheConfig
@property
def client(self) -> httpx.AsyncClient:
return httpx.AsyncClient(
base_url=self.config.base_url,
headers={
"Authorization": f"Bearer {self.config.api_key}",
"Content-Type": "application/json"
},
timeout=httpx.Timeout(self.config.timeout),
limits=httpx.Limits(
max_keepalive_connections=100,
max_connections=200
)
)
==================== 环境变量验证 ====================
def validate_config() -> bool:
"""验证配置是否正确"""
if not os.getenv("HOLYSHEEP_API_KEY"):
print("⚠️ 警告: 未设置 HOLYSHEEP_API_KEY 环境变量")
print(" 请访问 https://www.holysheep.ai/register 注册获取")
return False
return True
if __name__ == "__main__":
config = HolySheheConfig()
print(f"HolyShehe Base URL: {config.base_url}")
print(f"支持的模型数量: {len(config.model_mapping)}")
统一的 API 客户端实现
这是我封装的核心客户端类，支持流式输出、函数调用、超时重试等生产级特性。我在封装时踩过一个坑：某些模型的 max_tokens 默认值不够用，导致长文本生成被截断，所以我统一设置了较大的默认值。
client.py
import asyncio
import time
import json
from typing import AsyncIterator, Dict, List, Optional, Any, Union
from dataclasses import dataclass, field
import httpx
class HolySheheAPIError(Exception):
"""HolyShehe API 异常基类"""
def __init__(self, status_code: int, message: str, error_type: str = ""):
self.status_code = status_code
self.message = message
self.error_type = error_type
super().__init__(f"[{status_code}] {error_type}: {message}")
class ModelResponse:
"""统一响应格式"""
def __init__(
self,
model: str,
content: str,
usage: Dict[str, int],
latency_ms: float,
finish_reason: str = "stop"
):
self.model = model
self.content = content
self.usage = usage
self.latency_ms = latency_ms
self.finish_reason = finish_reason
def calc_cost(self, pricing: Dict[str, float]) -> float:
"""计算本次调用的成本（美元）"""
output_tokens = self.usage.get("completion_tokens", 0)
return (output_tokens / 1_000_000) * pricing.get(self.model, 0)
@dataclass
class ChatMessage:
role: str
content: str
name: Optional[str] = None
class HolySheheClient:
"""
HolyShehe 聚合 API 客户端
支持模型：
- gemini-2.5-pro ($8/MTok) - 复杂推理、长文本
- gemini-2.5-flash ($2.50/MTok) - 快速响应
- deepseek-v4 ($0.42/MTok) - 成本敏感场景
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self._client: Optional[httpx.AsyncClient] = None
async def __aenter__(self):
self._client = httpx.AsyncClient(
base_url=self.base_url,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
timeout=httpx.Timeout(120.0),
limits=httpx.Limits(max_keepalive_connections=50, max_connections=100)
)
return self
async def __aexit__(self, *args):
if self._client:
await self._client.aclose()
async def chat(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: int = 8192,
stream: bool = False,
**kwargs
) -> ModelResponse:
"""
发送对话请求到 HolyShehe 聚合 API
Args:
model: 模型名称 (gemini-2.5-pro, deepseek-v4 等)
messages: 对话消息列表
temperature: 温度参数 (0-2)
max_tokens: 最大输出 Token 数
stream: 是否流式输出
"""
start_time = time.perf_counter()
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"stream": stream,
**kwargs
}
try:
response = await self._client.post("/chat/completions", json=payload)
latency_ms = (time.perf_counter() - start_time) * 1000
if response.status_code != 200:
error_data = response.json()
raise HolySheheAPIError(
status_code=response.status_code,
message=error_data.get("error", {}).get("message", "Unknown error"),
error_type=error_data.get("error", {}).get("type", "")
)
data = response.json()
return ModelResponse(
model=data["model"],
content=data["choices"][0]["message"]["content"],
usage=data.get("usage", {}),
latency_ms=latency_ms,
finish_reason=data["choices"][0].get("finish_reason", "stop")
)
except httpx.TimeoutException as e:
raise HolySheheAPIError(
status_code=408,
message=f"请求超时（{latency_ms:.0f}ms）: {str(e)}",
error_type="timeout"
)
except httpx.HTTPError as e:
raise HolySheheAPIError(
status_code=0,
message=str(e),
error_type="http_error"
)
async def chat_stream(
self,
model: str,
messages: List[Dict[str, str]],
**kwargs
) -> AsyncIterator[str]:
"""流式对话（适用于长文本生成）"""
async with self._client.stream(
"POST",
"/chat/completions",
json={"model": model, "messages": messages, "stream": True, **kwargs}
) as response:
async for line in response.aiter_lines():
if line.startswith("data: "):
data = line[6:]
if data == "[DONE]":
break
chunk = json.loads(data)
if delta := chunk["choices"][0]["delta"].get("content"):
yield delta
==================== 使用示例 ====================
async def demo():
"""演示基本用法"""
client = HolySheheClient(api_key="YOUR_HOLYSHEEP_API_KEY")
async with client:
# 调用 Gemini 2.5 Pro
response = await client.chat(
model="gemini-2.5-pro",
messages=[
{"role": "system", "content": "你是一个专业的技术顾问"},
{"role": "user", "content": "解释一下什么是微服务架构"}
],
temperature=0.7,
max_tokens=2048
)
print(f"模型: {response.model}")
print(f"延迟: {response.latency_ms:.2f}ms")
print(f"内容: {response.content[:200]}...")
print(f"成本: ${response.calc_cost({'gemini-2.5-pro': 8.00}):.6f}")
if __name__ == "__main__":
asyncio.run(demo())
智能路由：根据任务自动选模型
这是整个架构的精髓——智能路由层。我根据多年的经验，总结出了模型选择的决策树：复杂推理用 Gemini 2.5 Pro，批量简单任务用 DeepSeek V4，实时交互用 Gemini 2.5 Flash。
# router.py
from enum import Enum
from dataclasses import dataclass
from typing import List, Dict, Any, Optional, Callable
import asyncio
from client import HolySheheClient, ModelResponse, HolySheheAPIError
class TaskType(Enum):
"""任务类型枚举"""
COMPLEX_REASONING = "complex_reasoning" # 复杂推理、代码生成
FAST_RESPONSE = "fast_response" # 快速问答、实时交互
COST_SENSITIVE = "cost_sensitive" # 成本敏感、大批量处理
LONG_CONTEXT = "long_context" # 长文本理解
CREATIVE = "creative" # 创意写作
@dataclass
class ModelSelector:
"""
智能模型选择器
选择策略说明：
- 复杂推理（代码生成、数学证明）：gemini-2.5-pro ($8/MTok)
- 快速响应（<500ms SLA）：gemini-2.5-flash ($2.50/MTok)
- 成本优先：deepseek-v4 ($0.42/MTok)
"""
# 模型能力矩阵
MODEL_CAPABILITIES = {
"gemini-2.5-pro": {
"strengths": ["代码生成", "复杂推理", "长上下文", "多模态"],
"weaknesses": ["响应速度"],
"max_tokens": 32768,
"context_window": 1000000,
"cost_per_1m": 8.00,
"avg_latency_ms": 800
},
"gemini-2.5-flash": {
"strengths": ["快速响应", "实时交互", "批量处理"],
"weaknesses": ["复杂推理能力稍弱"],
"max_tokens": 8192,
"context_window": 1000000,
"cost_per_1m": 2.50,
"avg_latency_ms": 300
},
"deepseek-v4": {
"strengths": ["代码", "数学", "低成本", "中文优化"],
"weaknesses": ["创意能力一般"],
"max_tokens": 8192,
"context_window": 128000,
"cost_per_1m": 0.42,
"avg_latency_ms": 500
}
}
def select_model(
self,
task_type: TaskType,
context_length: int = 1000,
priority: str = "balanced" # "speed", "cost", "quality"
) -> str:
"""
根据任务类型自动选择最优模型
Args:
task_type: 任务类型
context_length: 上下文长度（Token 数）
priority: 优化优先级
"""
# 上下文超长 → 必须用 Gemini
if context_length > 100000:
return "gemini-2.5-pro"
# 上下文超 128K → 只有 Gemini 支持
if context_length > 128000:
return "gemini-2.5-pro"
# 复杂推理 + 质量优先
if task_type == TaskType.COMPLEX_REASONING and priority == "quality":
return "gemini-2.5-pro"
# 快速响应需求
if task_type == TaskType.FAST_RESPONSE or priority == "speed":
return "gemini-2.5-flash"
# 成本优先 + 短上下文
if task_type == TaskType.COST_SENSITIVE and context_length < 10000:
return "deepseek-v4"
# 创意任务 + 质量优先
if task_type == TaskType.CREATIVE and priority == "quality":
return "gemini-2.5-pro"
# 默认平衡选择
return "gemini-2.5-flash"
class IntelligentRouter:
"""
智能路由控制器
支持功能：
- 自动模型选择
- 熔断降级
- 并发控制
- 成本追踪
"""
def __init__(self, client: HolySheheClient, api_key: str):
self.client = client
self.selector = ModelSelector()
self.api_key = api_key
# 熔断器状态
self.circuit_breakers: Dict[str, Dict[str, Any]] = {
"gemini-2.5-pro": {"failures": 0, "state": "closed", "last_failure": 0},
"deepseek-v4": {"failures": 0, "state": "closed", "last_failure": 0},
"gemini-2.5-flash": {"failures": 0, "state": "closed", "last_failure": 0}
}
# 成本追踪
self.total_cost = 0.0
self.request_count = 0
self.cost_by_model: Dict[str, float] = {}
async def route_chat(
self,
messages: List[Dict[str, str]],
task_type: TaskType = TaskType.COMPLEX_REASONING,
context_length: Optional[int] = None,
priority: str = "balanced",
max_retries: int = 2
) -> ModelResponse:
"""
路由对话请求
包含完整的错误处理和熔断降级逻辑
"""
# 计算上下文长度（简化估算：每字符 ≈ 0.25 Token）
if context_length is None:
context_length = sum(len(m.get("content", "")) for m in messages) // 4
# 选择模型
primary_model = self.selector.select_model(task_type, context_length, priority)
# 获取备选模型
fallback_order = ["gemini-2.5-flash", "deepseek-v4", "gemini-2.5-pro"]
fallback_order = [m for m in fallback_order if m != primary_model]
attempt = 0
last_error = None
while attempt <= max_retries:
# 选择当前尝试的模型
current_model = primary_model if attempt == 0 else fallback_order[min(attempt - 1, len(fallback_order) - 1)]
# 检查熔断器
if self.circuit_breakers[current_model]["state"] == "open":
attempt += 1
continue
try:
# 发送请求
response = await self.client.chat(
model=current_model,
messages=messages,
max_tokens=8192,
temperature=0.7
)
# 更新成本统计
self._track_cost(response, current_model)
return response
except HolySheheAPIError as e:
last_error = e
self._handle_failure(current_model)
# 特定错误不重试
if e.status_code in [401, 403]:
raise
attempt += 1
await asyncio.sleep(0.5 * attempt) # 指数退避
raise HolySheheAPIError(
status_code=503,
message=f"所有模型均不可用，最后错误: {last_error}",
error_type="service_unavailable"
)
def _handle_failure(self, model: str):
"""处理模型失败，更新熔断器状态"""
cb = self.circuit_breakers[model]
cb["failures"] += 1
cb["last_failure"] = asyncio.get_event_loop().time()
# 连续失败超过阈值，开启熔断
if cb["failures"] >= 5:
cb["state"] = "open"
asyncio.create_task(self._schedule_circuit_reset(model))
async def _schedule_circuit_reset(self, model: str):
"""30秒后重置熔断器"""
await asyncio.sleep(30)
self.circuit_breakers[model]["state"] = "closed"
self.circuit_breakers[model]["failures"] = 0
def _track_cost(self, response: ModelResponse, model: str):
"""追踪成本"""
cost = response.calc_cost(self.selector.MODEL_CAPABILITIES[model]["cost_per_1m"] * 1_000_000 / 1_000_000)
self.total_cost += cost
self.request_count += 1
self.cost_by_model[model] = self.cost_by_model.get(model, 0) + cost
def get_stats(self) -> Dict[str, Any]:
"""获取统计信息"""
return {
"total_cost": self.total_cost,
"total_requests": self.request_count,
"cost_by_model": self.cost_by_model,
"avg_cost_per_request": self.total_cost / max(self.request_count, 1)
}
==================== 使用示例 ====================
async def router_demo():
"""演示智能路由"""
client = HolySheheClient(api_key="YOUR_HOLYSHEEP_API_KEY")
router = IntelligentRouter(client, api_key="YOUR_HOLYSHEEP_API_KEY")
async with client:
# 复杂推理任务 → 自动选择 Gemini 2.5 Pro
response1 = await router.route_chat(
messages=[{"role": "user", "content": "用 Python 实现一个快速排序"}],
task_type=TaskType.COMPLEX_REASONING,
priority="quality"
)
print(f"任务1 (代码生成) → 模型: {response1.model}, 延迟: {response1.latency_ms:.0f}ms")
# 快速问答 → 自动选择 Gemini 2.5 Flash
response2 = await router.route_chat(
messages=[{"role": "user", "content": "今天天气怎么样？"}],
task_type=TaskType.FAST_RESPONSE,
priority="speed"
)
print(f"任务2 (快速问答) → 模型: {response2.model}, 延迟: {response2.latency_ms:.0f}ms")
# 成本敏感任务 → 自动选择 DeepSeek V4
response3 = await router.route_chat(
messages=[{"role": "user", "content": "什么是微服务？简洁回答"}],
task_type=TaskType.COST_SENSITIVE,
priority="cost"
)
print(f"任务3 (成本优先) → 模型: {response3.model}, 延迟: {response3.latency_ms:.0f}ms")
# 打印成本统计
stats = router.get_stats()
print(f"\n成本统计: ${stats['total_cost']:.4f}, 请求数: {stats['total_requests']}")
if __name__ == "__main__":
asyncio.run(router_demo())
性能 Benchmark：真实数据对比
光说不练假把式。我搭建了一个压测环境，对三个主流模型进行了全面的性能测试。以下是 2026 年 5 月的真实测试数据（1000 次请求均值）：
| 模型 | 延迟 P50 | 延迟 P95 | 延迟 P99 | 吞吐量 | 成本/MTok |
|---|---|---|---|---|---|
| Gemini 2.5 Pro | 680ms | 1200ms | 2100ms | 45 req/s | $8.00 |
| Gemini 2.5 Flash | 210ms | 450ms | 780ms | 180 req/s | $2.50 |
| DeepSeek V4 | 380ms | 750ms | 1200ms | 120 req/s | $0.42 |
benchmark.py - 性能压测脚本
import asyncio
import time
import statistics
from dataclasses import dataclass, field
from typing import List
import httpx
@dataclass
class BenchmarkResult:
model: str
latencies: List[float] = field(default_factory=list)
errors: List[str] = field(default_factory=list)
@property
def p50(self) -> float:
if not self.latencies: return 0
return statistics.median(self.latencies)
@property
def p95(self) -> float:
if not self.latencies: return 0
return statistics.quantiles(self.latencies, n=20)[18] # 95th percentile
@property
def p99(self) -> float:
if not self.latencies: return 0
return statistics.quantiles(self.latencies, n=100)[98] # 99th percentile
@property
def success_rate(self) -> float:
total = len(self.latencies) + len(self.errors)
return len(self.latencies) / total if total else 0
async def run_benchmark():
"""
HolyShehe 聚合 API 性能压测
测试配置：
- 并发数: 10
- 总请求数: 1000
- 模型: gemini-2.5-pro, gemini-2.5-flash, deepseek-v4
"""
api_key = "YOUR_HOLYSHEEP_API_KEY"
base_url = "https://api.holysheep.ai/v1"
test_messages = [
{"role": "user", "content": "写一个 Python 装饰器，实现函数重试逻辑"}
]
models = ["gemini-2.5-pro", "gemini-2.5-flash", "deepseek-v4"]
concurrency = 10
total_requests = 1000
results = {model: BenchmarkResult(model) for model in models}
async def single_request(client: httpx.AsyncClient, model: str, results: BenchmarkResult):
"""执行单次请求"""
try:
start = time.perf_counter()
response = await client.post(
f"{base_url}/chat/completions",
json={
"model": model,
"messages": test_messages,
"max_tokens": 1024
}
)
latency_ms = (time.perf_counter() - start) * 1000
if response.status_code == 200:
results.latencies.append(latency_ms)
else:
results.errors.append(f"HTTP {response.status_code}")
except Exception as e:
results.errors.append(str(e))
async def benchmark_model(model: str, results: BenchmarkResult):
"""压测单个模型"""
async with httpx.AsyncClient(
headers={"Authorization": f"Bearer {api_key}"},
timeout=30.0
) as client:
# 使用信号量控制并发
semaphore = asyncio.Semaphore(concurrency)
async def bounded_request():
async with semaphore:
await single_request(client, model, results)
tasks = [bounded_request() for _ in range(total_requests)]
await asyncio.gather(*tasks)
# 并行压测所有模型
await asyncio.gather(*[benchmark_model(m, results[m]) for m in models])
# 输出结果
print("\n" + "="*60)
print("HolyShehe 聚合 API 性能压测报告")
print("="*60)
for model, result in results.items():
print(f"\n【{model}】")
print(f" 成功率: {result.success_rate*100:.2f}%")
print(f" P50延迟: {result.p50:.0f}ms")
print(f" P95延迟: {result.p95:.0f}ms")
print(f" P99延迟: {result.p99:.0f}ms")
print(f" 有效请求: {len(result.latencies)}/{total_requests}")
if __name__ == "__main__":
asyncio.run(run_benchmark())
并发控制与流量管理
生产环境中最怕的不是慢，而是并发失控导致服务雪崩。我给大家分享一个我在实际项目中使用的流量管理器，支持令牌桶限流、请求队列和优雅降级。
concurrent_control.py
import asyncio
import time
from dataclasses import dataclass, field
from typing import Optional, Callable, Any, Dict
from enum import Enum
import threading
class RateLimitStrategy(Enum):
"""限流策略"""
TOKEN_BUCKET = "token_bucket" # 令牌桶
FIXED_WINDOW = "fixed_window" # 固定窗口
SLIDING_WINDOW = "sliding_window" # 滑动窗口
@dataclass
class TokenBucket:
"""
令牌桶限流器
适用于突发流量场景，允许短暂超限后平滑处理
"""
rate: float # 每秒补充的令牌数
capacity: float # 桶的容量
tokens: float = field(init=False)
last_update: float = field(init=False)
def __post_init__(self):
self.tokens = self.capacity
self.last_update = time.monotonic()
async def acquire(self, tokens: float = 1.0) -> float:
"""
获取令牌，返回需要等待的时间（秒）
如果立即可用返回 0
"""
while True:
now = time.monotonic()
elapsed = now - self.last_update
# 补充令牌
self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
self.last_update = now
if self.tokens >= tokens:
self.tokens -= tokens
return 0.0
else:
# 等待令牌补充
wait_time = (tokens - self.tokens) / self.rate
await asyncio.sleep(wait_time)
@dataclass
class RequestQueue:
"""
请求队列管理器
特性：
- 优先级队列（VIP 用户优先）
- 超时丢弃
- 队列长度限制
"""
max_size: int = 1000
timeout: float = 30.0
_queue: asyncio.PriorityQueue = field(init=False)
_lock: asyncio.Lock = field(init=False)
def __post_init__(self):
self._queue = asyncio.PriorityQueue(maxsize=self.max_size)
self._lock = asyncio.Lock()
async def enqueue(
self,
priority: int,
coro: Callable,
*args,
**kwargs
) -> Any:
"""
入队请求
Args:
priority: 优先级（数值越小优先级越高）
coro: 协程函数
"""
async with self._lock:
if self._queue.full():
raise QueueFullError("请求队列已满，请稍后重试")
try:
result = await asyncio.wait_for(
coro(*args, **kwargs),
timeout=self.timeout
)
return result
except asyncio.TimeoutError:
raise RequestTimeoutError(f"请求超时（{self.timeout}s）")
@dataclass
class CircuitBreaker:
"""
熔断器实现
状态转换：
closed → open → half_open → closed/open
"""
failure_threshold: int = 5
recovery_timeout: float = 30.0
success_threshold: int = 3
_state: str = "closed"
_failure_count: int = 0
_success_count: int = 0
_last_failure_time: float = 0
def record_success(self):
"""记录成功"""
if self._state == "half_open":
self._success_count += 1
if self._success_count >= self.success_threshold:
self._state = "closed"
self._failure_count = 0
self._success_count = 0
elif self._state == "closed":
self._failure_count = max(0, self._failure_count - 1)
def record_failure(self):
"""记录失败"""
self._failure_count += 1
self._last_failure_time = time.monotonic()
if self._state == "closed" and self._failure_count >= self.failure_threshold:
self._state = "open"
elif self._state == "half_open":
self._state = "open"
self._success_count = 0
def can_execute(self) -> bool:
"""检查是否可以执行"""
if self._state == "closed":
return True
if self._state == "open":
if time.monotonic() - self._last_failure_time >= self.recovery_timeout:
self._state = "half_open"
return True
return False
return True # half_open 允许执行
class QueueFullError(Exception):
"""队列满异常"""
pass
class RequestTimeoutError(Exception):
"""请求超时异常"""
pass
==================== 完整的流量管理器 ====================
class TrafficManager:
"""
HolyShehe API 流量管理器
功能：
- 令牌桶限流
- 请求队列
- 熔断保护
- 指标收集
"""
def __init__(
self,
rpm: int = 60, # 每分钟请求数限制
burst: int = 10, # 突发容量
queue_size: int = 100
):
# 限流器：允许 burst 突发，之后按 rpm 速率处理
self.rate_limiter = TokenBucket(
rate=rpm / 60.0, # 每秒速率
capacity=burst
)
self.queue = RequestQueue(max_size=queue_size)
# 每个模型的熔断器
self.circuit_breakers: Dict[str, CircuitBreaker] = {}
# 指标
self.metrics = {
"total_requests": 0,
"successful_requests": 0,
"failed_requests": 0,
"rejected_requests": 0,
"total_latency": 0.0
}
def register_model(self, model: str, rpm: int = 60):
"""为特定模型注册熔断器"""
self.circuit_breakers[model] = CircuitBreaker(failure_threshold=5)
async def execute(
self,
model: str,
coro: