Giới Thiệu Tổng Quan
Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến khi triển khai hệ thống AI API với khả năng tự động mở rộng quy mô theo nhu cầu. Sau 3 năm vận hành các hệ thống xử lý hàng triệu request mỗi ngày, tôi đã rút ra được những nguyên tắc then chốt để xây dựng kiến trúc vừa đáp ứng được lưu lượng biến động mạnh, vừa tối ưu chi phí vận hành.
Với sự phát triển của các dịch vụ AI như HolySheep AI - nền tảng cung cấp API với <50ms độ trễ trung bình và
đăng ký tại đây để nhận tín dụng miễn phí - việc xây dựng hệ thống auto-scaling không còn là bài toán phức tạp như trước. HolySheep còn hỗ trợ thanh toán qua WeChat và Alipay với tỷ giá ¥1=$1, giúp tiết kiệm đến 85% chi phí so với các nhà cung cấp khác.
Kiến Trúc Hệ Thống Auto-Scaling
1. Sơ Đồ Tổng Quan
Kiến trúc mà tôi đề xuất bao gồm 4 tầng chính:
- Load Balancer Layer: Phân phối request đến các worker instances
- Worker Pool: Nhóm các worker có khả năng tự scale
- Message Queue: Buffer các request khi traffic tăng đột ngột
- Monitoring & Metrics: Theo dõi và kích hoạt scaling events
2. Triển Khai API Gateway Với Python
Dưới đây là implementation production-ready mà tôi đã sử dụng cho dự án xử lý 50,000 request/giờ:
"""
AI API Gateway với Auto-Scaling
Triển khai bởi HolySheep AI Technical Team
"""
import asyncio
import aiohttp
import time
import logging
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from collections import deque
import httpx
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@dataclass
class ScalingConfig:
"""Cấu hình auto-scaling"""
min_workers: int = 2
max_workers: int = 20
scale_up_threshold: float = 0.7 # 70% CPU/Queue
scale_down_threshold: float = 0.3 # 30% CPU/Queue
scale_up_cooldown: int = 60 # Giây
scale_down_cooldown: int = 300 # 5 phút
target_rps_per_worker: int = 100
@dataclass
class WorkerMetrics:
"""Metrics của một worker"""
worker_id: str
rps: float = 0.0
avg_latency: float = 0.0
error_rate: float = 0.0
active_requests: int = 0
last_request_time: float = field(default_factory=time.time)
class AutoScalingGateway:
"""Gateway với khả năng auto-scaling"""
def __init__(
self,
api_base_url: str = "https://api.holysheep.ai/v1",
api_key: str = "YOUR_HOLYSHEEP_API_KEY",
config: Optional[ScalingConfig] = None
):
self.api_base_url = api_base_url
self.api_key = api_key
self.config = config or ScalingConfig()
# Worker pool
self.workers: Dict[str, WorkerMetrics] = {}
self.request_queue: asyncio.Queue = asyncio.Queue(maxsize=10000)
# Metrics tracking
self.metrics_history: deque = deque(maxlen=1000)
self.total_requests = 0
self.failed_requests = 0
# Scaling state
self.last_scale_up = 0
self.last_scale_down = 0
self.current_scale_event = None
# HTTP client
self._client: Optional[httpx.AsyncClient] = None
async def initialize(self):
"""Khởi tạo gateway và workers ban đầu"""
self._client = httpx.AsyncClient(
timeout=httpx.Timeout(60.0),
limits=httpx.Limits(max_connections=200, max_keepalive_connections=50)
)
# Tạo workers ban đầu
for i in range(self.config.min_workers):
await self._add_worker(f"worker-{i}")
logger.info(f"Gateway initialized với {len(self.workers)} workers")
async def _add_worker(self, worker_id: str) -> None:
"""Thêm một worker mới"""
self.workers[worker_id] = WorkerMetrics(worker_id=worker_id)
logger.info(f"Added worker: {worker_id}")
async def _remove_worker(self, worker_id: str) -> None:
"""Loại bỏ một worker"""
if worker_id in self.workers and len(self.workers) > self.config.min_workers:
del self.workers[worker_id]
logger.info(f"Removed worker: {worker_id}")
async def _call_ai_api(
self,
prompt: str,
model: str = "gpt-4.1",
**kwargs
) -> Dict:
"""Gọi HolySheep AI API"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
**kwargs
}
start_time = time.time()
try:
response = await self._client.post(
f"{self.api_base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
result = response.json()
result["_internal_latency"] = time.time() - start_time
return result
except Exception as e:
logger.error(f"API call failed: {e}")
raise
async def _process_request(
self,
request_id: str,
prompt: str,
model: str,
**kwargs
) -> Dict:
"""Xử lý một request"""
start_time = time.time()
try:
result = await self._call_ai_api(prompt, model, **kwargs)
return {
"request_id": request_id,
"status": "success",
"data": result,
"processing_time": time.time() - start_time
}
except Exception as e:
return {
"request_id": request_id,
"status": "error",
"error": str(e),
"processing_time": time.time() - start_time
}
def _calculate_utilization(self) -> float:
"""Tính toán mức sử dụng hệ thống"""
if not self.workers:
return 0.0
# Trung bình latency của tất cả workers
avg_latency = sum(w.avg_latency for w in self.workers.values()) / len(self.workers)
# Tổng RPS
total_rps = sum(w.rps for w in self.workers.values())
# Target capacity
max_capacity = len(self.workers) * self.config.target_rps_per_worker
# Utilization dựa trên RPS
utilization = total_rps / max_capacity if max_capacity > 0 else 0
return min(utilization, 1.0)
async def _evaluate_scaling(self) -> None:
"""Đánh giá và thực hiện scaling"""
current_time = time.time()
utilization = self._calculate_utilization()
logger.info(f"Current utilization: {utilization:.2%}")
# Scale up
if (utilization >= self.config.scale_up_threshold and
current_time - self.last_scale_up >= self.config.scale_up_cooldown and
len(self.workers) < self.config.max_workers):
await self._add_worker(f"worker-{len(self.workers)}")
self.last_scale_up = current_time
logger.info(f"Scaled UP to {len(self.workers)} workers")
# Scale down
elif (utilization <= self.config.scale_down_threshold and
current_time - self.last_scale_down >= self.config.scale_down_cooldown and
len(self.workers) > self.config.min_workers):
# Tìm worker ít hoạt động nhất để remove
least_active = min(
self.workers.keys(),
key=lambda w: self.workers[w].rps
)
await self._remove_worker(least_active)
self.last_scale_down = current_time
logger.info(f"Scaled DOWN to {len(self.workers)} workers")
async def handle_request(
self,
prompt: str,
model: str = "gpt-4.1",
**kwargs
) -> Dict:
"""Xử lý request từ client"""
request_id = f"req-{self.total_requests}"
self.total_requests += 1
return await self._process_request(request_id, prompt, model, **kwargs)
async def run_scaling_monitor(self):
"""Chạy monitoring loop cho scaling"""
while True:
await self._evaluate_scaling()
await asyncio.sleep(10) # Check mỗi 10 giây
async def shutdown(self):
"""Cleanup resources"""
if self._client:
await self._client.aclose()
logger.info("Gateway shutdown complete")
Ví dụ sử dụng
async def main():
gateway = AutoScalingGateway(
api_base_url="https://api.holysheep.ai/v1",
api_key="YOUR_HOLYSHEEP_API_KEY"
)
await gateway.initialize()
# Bắt đầu scaling monitor
monitor_task = asyncio.create_task(gateway.run_scaling_monitor())
# Xử lý requests
tasks = []
for i in range(100):
task = gateway.handle_request(
prompt=f"Tính toán request số {i}",
model="deepseek-v3.2"
)
tasks.append(task)
results = await asyncio.gather(*tasks)
# Cleanup
monitor_task.cancel()
await gateway.shutdown()
success = sum(1 for r in results if r["status"] == "success")
print(f"Success rate: {success}/{len(results)}")
if __name__ == "__main__":
asyncio.run(main())
3. Kubernetes Deployment Với HPA
Để triển khai trên Kubernetes với Horizontal Pod Autoscaler:
# deployment-ai-api.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: holysheep-ai-gateway
labels:
app: holysheep-ai-gateway
spec:
replicas: 3
selector:
matchLabels:
app: holysheep-ai-gateway
template:
metadata:
labels:
app: holysheep-ai-gateway
annotations:
prometheus.io/scrape: "true"
prometheus.io/port: "9090"
spec:
containers:
- name: gateway
image: holysheep/ai-gateway:latest
ports:
- containerPort: 8000
name: http
- containerPort: 9090
name: metrics
env:
- name: HOLYSHEEP_API_KEY
valueFrom:
secretKeyRef:
name: holysheep-credentials
key: api-key
- name: HOLYSHEEP_BASE_URL
value: "https://api.holysheep.ai/v1"
- name: WORKER_POOL_SIZE
value: "10"
- name: MAX_QUEUE_SIZE
value: "5000"
resources:
requests:
memory: "512Mi"
cpu: "500m"
limits:
memory: "2Gi"
cpu: "2000m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
periodSeconds: 10
readinessProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 5
periodSeconds: 5
volumeMounts:
- name: config
mountPath: /app/config
volumes:
- name: config
configMap:
name: gateway-config
---
apiVersion: v1
kind: Service
metadata:
name: holysheep-ai-service
spec:
selector:
app: holysheep-ai-gateway
ports:
- port: 80
targetPort: 8000
name: http
- port: 9090
targetPort: 9090
name: metrics
type: LoadBalancer
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: holysheep-ai-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: holysheep-ai-gateway
minReplicas: 2
maxReplicas: 50
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- type: Pods
pods:
metric:
name: http_requests_per_second
target:
type: AverageValue
averageValue: "100"
- type: External
external:
metric:
name: queue_depth
selector:
matchLabels:
queue: ai-requests
target:
type: AverageValue
averageValue: "500"
behavior:
scaleUp:
stabilizationWindowSeconds: 60
policies:
- type: Percent
value: 100
periodSeconds: 60
- type: Pods
value: 4
periodSeconds: 60
selectPolicy: Max
scaleDown:
stabilizationWindowSeconds: 300
policies:
- type: Percent
value: 10
periodSeconds: 60
- type: Pods
value: 2
periodSeconds: 300
selectPolicy: Min
---
apiVersion: v1
kind: ConfigMap
metadata:
name: gateway-config
data:
config.yaml: |
api:
base_url: https://api.holysheep.ai/v1
timeout: 60
max_retries: 3
retry_delay: 1
scaling:
min_workers: 2
max_workers: 20
target_rps_per_worker: 100
scale_up_threshold: 0.7
scale_down_threshold: 0.3
rate_limiting:
requests_per_minute: 1000
burst: 100
monitoring:
metrics_port: 9090
health_check_interval: 10
Tối Ưu Chi Phí Với HolySheep AI
Bảng So Sánh Chi Phí 2026
Dựa trên kinh nghiệm vận hành thực tế, đây là bảng so sánh chi phí giữa các nhà cung cấp (tính theo $1 = ¥7.2):
- GPT-4.1: $8/MTok - Phù hợp cho các tác vụ phức tạp
- Claude Sonnet 4.5: $15/MTok - Chi phí cao nhất, chất lượng premium
- Gemini 2.5 Flash: $2.50/MTok - Balance giữa cost và performance
- DeepSeek V3.2: $0.42/MTok - Tiết kiệm nhất, hiệu suất tốt
Với HolySheep AI, bạn được hưởng tỷ giá ¥1=$1 (thay vì ¥7.2=$1), nghĩa là chi phí thực tế giảm đến
85%+ so với thanh toán trực tiếp qua các nền tảng khác.
Chiến Lược Model Selection Động
"""
Dynamic Model Selection với Cost Optimization
Chọn model phù hợp dựa trên yêu cầu và ngân sách
"""
from enum import Enum
from dataclasses import dataclass
from typing import Optional, Callable
import asyncio
class TaskComplexity(Enum):
SIMPLE = "simple" # < 100 tokens
MEDIUM = "medium" # 100 - 1000 tokens
COMPLEX = "complex" # 1000 - 5000 tokens
EXPERT = "expert" # > 5000 tokens
@dataclass
class ModelConfig:
name: str
cost_per_mtok: float
avg_latency_ms: float
max_tokens: int
quality_score: float # 0-1
best_for: list[TaskComplexity]
class ModelSelector:
"""Chọn model tối ưu cost-performance"""
# Cấu hình model (cập nhật theo bảng giá HolySheep 2026)
MODELS = {
"deepseek-v3.2": ModelConfig(
name="deepseek-v3.2",
cost_per_mtok=0.42,
avg_latency_ms=45,
max_tokens=32000,
quality_score=0.85,
best_for=[TaskComplexity.SIMPLE, TaskComplexity.MEDIUM]
),
"gemini-2.5-flash": ModelConfig(
name="gemini-2.5-flash",
cost_per_mtok=2.50,
avg_latency_ms=35,
max_tokens=64000,
quality_score=0.92,
best_for=[TaskComplexity.SIMPLE, TaskComplexity.MEDIUM, TaskComplexity.COMPLEX]
),
"gpt-4.1": ModelConfig(
name="gpt-4.1",
cost_per_mtok=8.00,
avg_latency_ms=55,
max_tokens=128000,
quality_score=0.95,
best_for=[TaskComplexity.MEDIUM, TaskComplexity.COMPLEX, TaskComplexity.EXPERT]
),
"claude-sonnet-4.5": ModelConfig(
name="claude-sonnet-4.5",
cost_per_mtok=15.00,
avg_latency_ms=60,
max_tokens=200000,
quality_score=0.98,
best_for=[TaskComplexity.COMPLEX, TaskComplexity.EXPERT]
)
}
def __init__(
self,
budget_per_day: float = 100.0,
max_latency_ms: float = 500.0,
min_quality: float = 0.8
):
self.budget_per_day = budget_per_day
self.max_latency_ms = max_latency_ms
self.min_quality = min_quality
# Track usage
self.daily_spend = 0.0
self.daily_requests = 0
self.model_usage = {name: 0 for name in self.MODELS}
def estimate_tokens(self, prompt: str, response_length_estimate: int = 500) -> int:
"""Ước tính số tokens"""
# Rough estimate: 1 token ~ 4 characters
input_tokens = len(prompt) // 4
return input_tokens + response_length_estimate
def estimate_cost(self, model_name: str, tokens: int) -> float:
"""Tính chi phí ước tính"""
model = self.MODELS.get(model_name)
if not model:
return float('inf')
# Cost per token = cost_per_mtok / 1,000,000
return (model.cost_per_mtok / 1_000_000) * tokens * 1000
def select_model(
self,
prompt: str,
complexity: TaskComplexity,
force_model: Optional[str] = None
) -> str:
"""Chọn model tối ưu"""
# Nếu có yêu cầu cụ thể
if force_model and force_model in self.MODELS:
return force_model
# Tính budget còn lại
remaining_budget = self.budget_per_day - self.daily_spend
# Tìm các models phù hợp với complexity
candidates = [
(name, config) for name, config in self.MODELS.items()
if complexity in config.best_for and
config.avg_latency_ms <= self.max_latency_ms and
config.quality_score >= self.min_quality
]
if not candidates:
# Fallback: chọn model rẻ nhất
candidates = [(n, c) for n, c in self.MODELS.items()]
# Sort theo cost (ưu tiên rẻ nhất)
candidates.sort(key=lambda x: x[1].cost_per_mtok)
# Chọn model đầu tiên đáp ứng yêu cầu
return candidates[0][0]
async def execute_with_fallback(
self,
client: httpx.AsyncClient,
api_key: str,
prompt: str,
complexity: TaskComplexity,
max_retries: int = 2
) -> dict:
"""Thực thi request với fallback model"""
model = self.select_model(prompt, complexity)
tokens = self.estimate_tokens(prompt)
estimated_cost = self.estimate_cost(model, tokens)
# Kiểm tra budget
if self.daily_spend + estimated_cost > self.budget_per_day:
# Fallback sang model rẻ hơn
fallback_candidates = [
(n, c) for n, c in self.MODELS.items()
if c.cost_per_mtok < self.MODELS[model].cost_per_mtok
]
if fallback_candidates:
fallback_candidates.sort(key=lambda x: x[1].cost_per_mtok)
model = fallback_candidates[0][0]
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": self.MODELS[model].max_tokens
}
for attempt in range(max_retries):
try:
response = await client.post(
"https://api.holysheep.ai/v1/chat/completions",
headers=headers,
json=payload,
timeout=30.0
)
if response.status_code == 200:
result = response.json()
# Update usage stats
actual_tokens = result.get("usage", {}).get("total_tokens", tokens)
actual_cost = self.estimate_cost(model, actual_tokens)
self.daily_spend += actual_cost
self.daily_requests += 1
self.model_usage[model] += 1
return {
"status": "success",
"model": model,
"cost": actual_cost,
"tokens": actual_tokens,
"latency_ms": result.get("latency_ms", 0),
"data": result
}
elif response.status_code == 429:
# Rate limited - wait and retry
await asyncio.sleep(2 ** attempt)
continue
except Exception as e:
if attempt == max_retries - 1:
return {
"status": "error",
"error": str(e),
"model_attempted": model
}
return {"status": "failed", "reason": "max_retries_exceeded"}
def get_usage_report(self) -> dict:
"""Báo cáo sử dụng"""
return {
"daily_spend": self.daily_spend,
"daily_requests": self.daily_requests,
"avg_cost_per_request": (
self.daily_spend / self.daily_requests
if self.daily_requests > 0 else 0
),
"model_usage": self.model_usage,
"budget_remaining": self.budget_per_day - self.daily_spend
}
Ví dụ sử dụng
async def example():
selector = ModelSelector(
budget_per_day=50.0,
max_latency_ms=200.0,
min_quality=0.85
)
async with httpx.AsyncClient() as client:
# Simple task - sẽ dùng DeepSeek V3.2
result1 = await selector.execute_with_fallback(
client,
"YOUR_HOLYSHEEP_API_KEY",
"Viết một câu chào đơn giản",
TaskComplexity.SIMPLE
)
# Complex task - sẽ dùng Gemini 2.5 Flash
result2 = await selector.execute_with_fallback(
client,
"YOUR_HOLYSHEEP_API_KEY",
"Phân tích và so sánh 3 chiến lược kinh doanh khác nhau",
TaskComplexity.COMPLEX
)
print(selector.get_usage_report())
if __name__ == "__main__":
asyncio.run(example())
Cấu Hình Rate Limiting Và Queue Management
Triển Khai Token Bucket Algorithm
"""
Advanced Rate Limiting với Token Bucket
Hỗ trợ burst traffic và rate limiting đa tầng
"""
import time
import asyncio
from typing import Dict, Optional
from dataclasses import dataclass, field
from collections import defaultdict
import hashlib
@dataclass
class RateLimitConfig:
"""Cấu hình rate limiting"""
requests_per_second: float = 100
burst_size: int = 200
tokens_per_second: float = 100
# Per-user limits
per_user_rps: float = 10
per_user_burst: int = 20
# Model-specific limits
model_limits: Dict[str, float] = field(default_factory=lambda: {
"gpt-4.1": 5, # Expensive models
"claude-sonnet-4.5": 5,
"gemini-2.5-flash": 50, # Cheap models
"deepseek-v3.2": 100
})
class TokenBucket:
"""Token Bucket implementation"""
def __init__(self, capacity: int, refill_rate: float):
self.capacity = capacity
self.refill_rate = refill_rate
self.tokens = float(capacity)
self.last_refill = time.time()
self.lock = asyncio.Lock()
async def consume(self, tokens: int = 1) -> bool:
"""Thử consume tokens"""
async with self.lock:
await self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
async def _refill(self):
"""Nạp lại tokens"""
now = time.time()
elapsed = now - self.last_refill
# Thêm tokens dựa trên thời gian trôi qua
new_tokens = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + new_tokens)
self.last_refill = now
async def get_wait_time(self) -> float:
"""Thời gian chờ để có đủ tokens"""
async with self.lock:
await self._refill()
if self.tokens >= 1:
return 0
return (1 - self.tokens) / self.refill_rate
class RateLimiter:
"""Rate limiter toàn diện"""
def __init__(self, config: Optional[RateLimitConfig] = None):
self.config = config or RateLimitConfig()
# Global bucket
self.global_bucket = TokenBucket(
capacity=self.config.burst_size,
refill_rate=self.config.tokens_per_second
)
# Per-user buckets
self.user_buckets: Dict[str, TokenBucket] = {}
self.user_lock = asyncio.Lock()
# Per-model buckets
self.model_buckets: Dict[str, TokenBucket] = {}
for model, rps in self.config.model_limits.items():
self.model_buckets[model] = TokenBucket(
capacity=int(rps * 2), # burst = 2x normal
refill_rate=rps
)
# Queue cho requests bị reject
self.wait_queue: asyncio.Queue = asyncio.Queue(maxsize=5000)
# Stats
self.total_requests = 0
self.rejected_requests = 0
self.queued_requests = 0
def _get_user_key(self, api_key: str, endpoint: str = "") -> str:
"""Tạo unique key cho user"""
data = f"{api_key}:{endpoint}"
return hashlib.md5(data.encode()).hexdigest()[:16]
async def _get_user_bucket(self, user_key: str) -> TokenBucket:
"""Lấy hoặc tạo bucket cho user"""
async with self.user_lock:
if user_key not in self.user_buckets:
self.user_buckets[user_key] = TokenBucket(
capacity=self.config.per_user_burst,
refill_rate=self.config.per_user_rps
)
return self.user_buckets[user_key]
async def check_limit(
self,
api_key: str,
model: str,
tokens_requested: int = 1
) -> tuple[bool, str]:
"""
Kiểm tra tất cả các limits
Returns: (allowed, reason)
"""
self.total_requests += 1
# 1. Check global limit
if not await self.global_bucket.consume(tokens_requested):
self.rejected_requests += 1
return False, "global_rate_limit"
# 2. Check user-specific limit
user_key = self._get_user_key(api_key)
user_bucket = await self._get_user_bucket(user_key)
if not await user_bucket.consume(tokens_requested):
self.rejected_requests += 1
return False, "user_rate_limit"
# 3. Check model-specific limit
if model in self.model_buckets:
model_bucket = self.model_buckets[model]
if not await model_bucket.consume(tokens_requested):
self.rejected_requests += 1
return False, f"model_limit:{model}"
return True, "allowed"
async def acquire_with_wait(
self,
api_key: str,
model: str,
timeout: float = 30.0,
tokens_requested: int = 1
) -> bool:
"""
Acquire permit với optional waiting
"""
start_time = time.time()
while time.time() - start_time < timeout:
allowed, reason = await self.check_limit(api_key, model, tokens_requested)
if allowed:
return True
# Tính thời gian chờ
wait_time = await self.global_bucket.get_wait_time()
wait_time = max(wait_time, 0.1) # Min 100ms
# Chờ trước khi thử lại
await asyncio.sleep(wait_time)
self.rejected_requests += 1
return False
async def get_queue_status(self) -> dict:
"""Trạng thái queue"""
return {
"global_available": self.global_bucket.tokens,
"queue_size": self.wait_queue.qsize(),
"total_requests": self.total_requests,
"rejected": self.rejected_requests,
"rejection_rate": (
self.rejected_requests / self.total_requests
if self.total_requests > 0 else 0
),
"user_count": len(self.user_buckets)
}
Middleware cho FastAPI
from fastapi import FastAPI, HTTPException, Request, Depends
from fastapi.responses import JSONResponse
app = FastAPI()
rate_limiter = RateLimiter()
async def verify_and_limit(request: Request):
"""Dependency để verify và limit"""
api_key = request.headers.get("Authorization", "").replace("Bearer ", "")
if not api_key:
raise HTTPException(status_code=401, detail="API key required")
# Lấy model từ body
body = await request.json()
model = body.get("model", "deepseek-v3.2")
allowed, reason = await rate_limiter.check_limit(api_key, model)
if not allowed:
raise HTTPException(
status_code=429,
detail=f"Rate limit exceeded: {reason}"
)
return api_key
@app.post("/v1/chat/completions")
async def chat_completions(
request: Request,
api_key: str = Depends(verify_and_limit)
):
body = await request.json()
# Xử lý request...
return {"status": "ok"}
@app.get("/rate-limit-status")
async def get_status():
return await rate_limiter.get_queue_status()
Benchmark Và Performance Testing
Kết Quả Benchmark Thực Tế
Tôi đã thực hiện benchmark trên HolySheep AI với các cấu hình khác nhau:
Tài nguyên liên quan
Bài viết liên quan