Đêm qua, hệ thống production của tôi bị sập hoàn toàn. Lỗi ConnectionError: timeout after 30s xuất hiện liên tục trên dashboard monitoring. Sau 3 tiếng debug căng thẳng, nguyên nhân được tìm ra: không có cơ chế rate limiting, một client bên thứ ba gửi 10,000 request/giây vào API — gấp 100 lần giới hạn thiết kế. Từ thất bại đó, tôi đã xây dựng một AI API Gateway hoàn chỉnh với FastAPI, và hôm nay sẽ chia sẻ toàn bộ source code cùng best practices.
Tại Sao Cần AI API Gateway?
Khi triển khai AI vào production, bạn sẽ gặp ngay các vấn đề: quản lý nhiều model (GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash, DeepSeek V3.2), bảo mật API key, kiểm soát chi phí, và giới hạn tốc độ. Một AI API Gateway đóng vai trò trung gian, xử lý authentication, routing thông minh, và bảo vệ upstream services khỏi overload.
Kiến Trúc Tổng Quan
┌─────────────────────────────────────────────────────────────────┐
│ Client Applications │
│ (Web App, Mobile, IoT Devices, Third-party) │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ FastAPI AI Gateway │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌─────────────────┐ │
│ │ Auth │ │ Router │ │ Rate │ │ Logging │ │
│ │ Layer │→→│ Engine │→→│ Limiter │→→│ & Metrics │ │
│ └──────────┘ └──────────┘ └──────────┘ └─────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
┌─────────────────────┼─────────────────────┐
▼ ▼ ▼
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ HolySheep AI │ │ OpenAI API │ │ Custom ML │
│ Gateway │ │ Fallback │ │ Models │
│ ¥1=$1 rate │ │ │ │ │
└──────────────┘ └──────────────┘ └──────────────┘
Cài Đặt Môi Trường Và Dependencies
# requirements.txt
fastapi==0.115.0
uvicorn[standard]==0.30.0
pydantic==2.8.2
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
httpx==0.27.0
redis[hiredis]==5.0.8
slowapi==0.1.9
python-multipart==0.0.9
tenacity==8.3.0
Tạo virtual environment và cài đặt
python -m venv venv
source venv/bin/activate # Linux/Mac
Hoặc: venv\Scripts\activate # Windows
pip install -r requirements.txtXây Dựng Cấu Hình Core
# config.py
import os
from pydantic_settings import BaseSettings
from typing import Dict
class Settings(BaseSettings):
# HolySheep AI Configuration
HOLYSHEEP_API_KEY: str = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL: str = "https://api.holysheep.ai/v1"
# Model Pricing (USD per 1M tokens - 2026)
MODEL_PRICING: Dict[str, Dict[str, float]] = {
"gpt-4.1": {"input": 8.0, "output": 8.0},
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42}, # Tiết kiệm 85%+
}
# Rate Limiting
RATE_LIMIT_PER_MINUTE: int = 60
RATE_LIMIT_PER_HOUR: int = 1000
BURST_LIMIT: int = 10
# Redis Configuration
REDIS_HOST: str = os.getenv("REDIS_HOST", "localhost")
REDIS_PORT: int = int(os.getenv("REDIS_PORT", "6379"))
REDIS_DB: int = int(os.getenv("REDIS_DB", "0"))
# JWT Configuration
SECRET_KEY: str = os.getenv("SECRET_KEY", "your-super-secret-key-change-in-production")
ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
class Config:
env_file = ".env"
settings = Settings()Module Xác Thực JWT Token
# auth/jwt_handler.py
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from .config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
security = HTTPBearer()
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password: str) -> str:
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
def decode_token(token: str) -> dict:
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM])
return payload
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token không hợp lệ hoặc đã hết hạn",
headers={"WWW-Authenticate": "Bearer"},
)
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security)
) -> dict:
token = credentials.credentials
payload = decode_token(token)
user_id = payload.get("sub")
if user_id is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token không chứa thông tin user"
)
return {"user_id": user_id, "tier": payload.get("tier", "free")}Database Models Cho Users Và API Keys
# models/database.py
from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime
import os
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./gateway.db")
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True, nullable=False)
hashed_password = Column(String, nullable=False)
tier = Column(String, default="free") # free, pro, enterprise
credits = Column(Float, default=10.0) # USD credits
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
class APIKey(Base):
__tablename__ = "api_keys"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, nullable=False)
key_hash = Column(String, unique=True, index=True, nullable=False)
key_prefix = Column(String, nullable=False) # First 8 chars for identification
name = Column(String, nullable=True)
requests_today = Column(Integer, default=0)
tokens_today = Column(Integer, default=0)
cost_today = Column(Float, default=0.0)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
Base.metadata.create_all(bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()Rate Limiter Với Redis Integration
# middleware/rate_limiter.py
import time
from typing import Optional, Tuple
import redis.asyncio as redis
from fastapi import Request, HTTPException, status
from slowapi import Limiter
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from .config import settings
class RedisRateLimiter:
def __init__(self):
self.redis_client: Optional[redis.Redis] = None
async def connect(self):
try:
self.redis_client = redis.Redis(
host=settings.REDIS_HOST,
port=settings.REDIS_PORT,
db=settings.REDIS_DB,
decode_responses=True
)
await self.redis_client.ping()
print("✅ Kết nối Redis thành công")
except Exception as e:
print(f"⚠️ Không thể kết nối Redis: {e}. Sử dụng in-memory fallback.")
self.redis_client = None
async def check_rate_limit(
self,
identifier: str,
limit: int,
window: int
) -> Tuple[bool, int, int]:
"""
Kiểm tra rate limit với sliding window algorithm
Returns: (is_allowed, remaining, reset_time)
"""
if self.redis_client:
return await self._redis_check(identifier, limit, window)
return self._memory_check(identifier, limit, window)
async def _redis_check(
self,
identifier: str,
limit: int,
window: int
) -> Tuple[bool, int, int]:
key = f"rate_limit:{identifier}"
current_time = int(time.time())
window_start = current_time - window
pipe = self.redis_client.pipeline()
pipe.zremrangebyscore(key, 0, window_start)
pipe.zadd(key, {str(current_time): current_time})
pipe.zcard(key)
pipe.expire(key, window)
results = await pipe.execute()
request_count = results[2]
remaining = max(0, limit - request_count)
reset_time = current_time + window
if request_count > limit:
return False, 0, reset_time
return True, remaining, reset_time
def _memory_check(
self,
identifier: str,
limit: int,
window: int
) -> Tuple[bool, int, int]:
# Fallback đơn giản khi không có Redis
if not hasattr(self._memory_store, identifier):
self._memory_store[identifier] = []
current_time = time.time()
requests = self._memory_store[identifier]
# Lọc requests trong window
self._memory_store[identifier] = [
t for t in requests if current_time - t < window
]
if len(self._memory_store[identifier]) >= limit:
return False, 0, int(current_time + window)
self._memory_store[identifier].append(current_time)
return True, limit - len(self._memory_store[identifier]), int(current_time + window)
async def record_usage(
self,
identifier: str,
tokens: int,
cost: float
):
"""Ghi nhận usage cho billing"""
if self.redis_client:
key = f"usage:{identifier}:{time.strftime('%Y%m%d')}"
pipe = self.redis_client.pipeline()
pipe.hincrby(key, "requests", 1)
pipe.hincrby(key, "tokens", tokens)
pipe.hincrbyfloat(key, "cost", cost)
pipe.expire(key, 86400 * 7) # Lưu 7 ngày
await pipe.execute()
_memory_store = {}
rate_limiter = RedisRateLimiter()Smart Router Engine
# routers/ai_router.py
import httpx
from typing import Optional, Dict, Any
from fastapi import APIRouter, Depends, HTTPException, status, Request
from pydantic import BaseModel, Field
from ..auth.jwt_handler import get_current_user
from ..middleware.rate_limiter import rate_limiter
from ..config import settings
router = APIRouter(prefix="/ai", tags=["AI Gateway"])
class ChatRequest(BaseModel):
model: str = Field(..., description="Model: gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2")
messages: list = Field(..., description="Chat messages")
temperature: float = Field(default=0.7, ge=0, le=2)
max_tokens: int = Field(default=1000, ge=1, le=32000)
stream: bool = Field(default=False)
class ChatResponse(BaseModel):
model: str
content: str
usage: Dict[str, int]
latency_ms: float
cost_usd: float
MODEL_MAPPING = {
"gpt-4.1": "gpt-4.1",
"claude-sonnet-4.5": "claude-sonnet-4.5",
"gemini-2.5-flash": "gemini-2.5-flash",
"deepseek-v3.2": "deepseek-v3.2",
}
async def forward_to_provider(
request_data: ChatRequest,
user_tier: str
) -> Dict[str, Any]:
"""Forward request đến HolySheep AI Gateway"""
start_time = time.time()
headers = {
"Authorization": f"Bearer {settings.HOLYSHEEP_API_KEY}",
"Content-Type": "application/json",
}
payload = {
"model": MODEL_MAPPING.get(request_data.model, request_data.model),
"messages": request_data.messages,
"temperature": request_data.temperature,
"max_tokens": request_data.max_tokens,
}
try:
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{settings.HOLYSHEEP_BASE_URL}/chat/completions",
json=payload,
headers=headers
)
if response.status_code == 429:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Rate limit exceeded. Vui lòng giảm tần suất request."
)
response.raise_for_status()
result = response.json()
# Tính toán chi phí
latency_ms = (time.time() - start_time) * 1000
input_tokens = result.get("usage", {}).get("prompt_tokens", 0)
output_tokens = result.get("usage", {}).get("completion_tokens", 0)
pricing = settings.MODEL_PRICING.get(request_data.model, {"input": 0, "output": 0})
cost_usd = (input_tokens / 1_000_000 * pricing["input"] +
output_tokens / 1_000_000 * pricing["output"])
return {
"model": request_data.model,
"content": result["choices"][0]["message"]["content"],
"usage": {
"prompt_tokens": input_tokens,
"completion_tokens": output_tokens,
"total_tokens": input_tokens + output_tokens
},
"latency_ms": round(latency_ms, 2),
"cost_usd": round(cost_usd, 6)
}
except httpx.HTTPStatusError as e:
raise HTTPException(
status_code=e.response.status_code,
detail=f"Lỗi từ upstream: {e.response.text}"
)
except httpx.RequestError as e:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=f"Không thể kết nối đến AI provider: {str(e)}"
)
@router.post("/chat", response_model=ChatResponse)
async def chat_completions(
request: ChatRequest,
current_user: dict = Depends(get_current_user),
request_state: Request = None
):
# Rate limit check
user_id = current_user["user_id"]
tier = current_user["tier"]
limits = {
"free": (20, 60), # 20 req/min
"pro": (100, 60), # 100 req/min
"enterprise": (1000, 60)
}
limit, window = limits.get(tier, limits["free"])
is_allowed, remaining, reset_time = await rate_limiter.check_rate_limit(
f"user:{user_id}", limit, window
)
if not is_allowed:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail=f"Đã vượt quá giới hạn {limit} requests/phút. Reset lúc {reset_time}",
headers={"X-RateLimit-Limit": str(limit), "X-RateLimit-Remaining": "0"}
)
result = await forward_to_provider(request, tier)
# Ghi nhận usage
total_tokens = result["usage"]["total_tokens"]
await rate_limiter.record_usage(f"user:{user_id}", total_tokens, result["cost_usd"])
return result
import timeMain Application Entry Point
# main.py
from fastapi import FastAPI, Request, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from contextlib import asynccontextmanager
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from models.database import engine, Base
from auth.jwt_handler import router as auth_router, create_access_token
from routers.ai_router import router as ai_router
from middleware.rate_limiter import rate_limiter
from config import settings
limiter = Limiter(key_func=get_remote_address)
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
print("🚀 Khởi động AI Gateway...")
await rate_limiter.connect()
# Tạo database tables
Base.metadata.create_all(bind=engine)
print("✅ AI Gateway đã sẵn sàng!")
print(f"📊 HolySheep Base URL: {settings.HOLYSHEEP_BASE_URL}")
yield
# Shutdown
print("👋 Tắt AI Gateway...")
app = FastAPI(
title="HolySheep AI Gateway",
description="API Gateway thông minh cho AI services - Routing, Auth & Rate Limiting",
version="1.0.0",
lifespan=lifespan
)
CORS Middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Thay đổi trong production
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
Rate Limiter
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
Include routers
app.include_router(auth_router, prefix="/auth", tags=["Authentication"])
app.include_router(ai_router, prefix="/api/v1", tags=["AI Services"])
@app.get("/health")
async def health_check():
return {
"status": "healthy",
"gateway": "HolySheep AI Gateway",
"version": "1.0.0",
"upstream": settings.HOLYSHEEP_BASE_URL
}
@app.get("/")
async def root():
return {
"message": "Chào mừng đến với HolySheep AI Gateway",
"docs": "/docs",
"models": list(settings.MODEL_PRICING.keys())
}
if __name__ == "__main__":
import uvicorn
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)Authentication Endpoints
# auth/routes.py
from fastapi import APIRouter, Depends, HTTPException, status, Depends
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from datetime import timedelta
from pydantic import BaseModel, EmailStr
from ..models.database import get_db, User
from ..auth.jwt_handler import (
create_access_token,
verify_password,
get_password_hash,
get_current_user
)
from ..config import settings
router = APIRouter()
class UserCreate(BaseModel):
email: EmailStr
password: str
full_name: str = None
class UserResponse(BaseModel):
id: int
email: str
tier: str
credits: float
is_active: bool
class Token(BaseModel):
access_token: str
token_type: str
expires_in: int
@router.post("/register", response_model=UserResponse)
async def register(user: UserCreate, db: Session = Depends(get_db)):
# Kiểm tra email đã tồn tại
existing_user = db.query(User).filter(User.email == user.email).first()
if existing_user:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Email đã được đăng ký"
)
# Tạo user mới với 10 USD credits miễn phí
hashed_password = get_password_hash(user.password)
new_user = User(
email=user.email,
hashed_password=hashed_password,
tier="free",
credits=10.0 # Tín dụng miễn phí khi đăng ký
)
db.add(new_user)
db.commit()
db.refresh(new_user)
return UserResponse(
id=new_user.id,
email=new_user.email,
tier=new_user.tier,
credits=new_user.credits,
is_active=new_user.is_active
)
@router.post("/login", response_model=Token)
async def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
user = db.query(User).filter(User.email == form_data.username).first()
if not user or not verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Email hoặc mật khẩu không đúng",
headers={"WWW-Authenticate": "Bearer"},
)
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Tài khoản đã bị vô hiệu hóa"
)
access_token = create_access_token(
data={"sub": str(user.id), "tier": user.tier},
expires_delta=timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
)
return Token(
access_token=access_token,
token_type="bearer",
expires_in=settings.ACCESS_TOKEN_EXPIRE_MINUTES * 60
)
@router.get("/me", response_model=UserResponse)
async def get_me(
current_user: dict = Depends(get_current_user),
db: Session = Depends(get_db)
):
user = db.query(User).filter(User.id == int(current_user["user_id"])).first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return UserResponse(
id=user.id,
email=user.email,
tier=user.tier,
credits=user.credits,
is_active=user.is_active
)Cách Sử Dụng API Gateway
# client_example.py
import httpx
import asyncio
BASE_URL = "https://your-gateway-domain.com" # Thay bằng domain của bạn
API_KEY = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9..." # Token từ /auth/login
async def chat_example():
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2", # Model tiết kiệm nhất - chỉ $0.42/MTok
"messages": [
{"role": "system", "content": "Bạn là trợ lý AI tiếng Việt."},
{"role": "user", "content": "Giải thích về FastAPI Rate Limiting"}
],
"temperature": 0.7,
"max_tokens": 500
}
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{BASE_URL}/api/v1/ai/chat",
json=payload,
headers=headers
)
if response.status_code == 200:
result = response.json()
print(f"✅ Model: {result['model']}")
print(f"⏱️ Latency: {result['latency_ms']}ms")
print(f"💰 Cost: ${result['cost_usd']}")
print(f"📝 Response:\n{result['content']}")
elif response.status_code == 429:
print("⛔ Rate limit exceeded! Thử lại sau.")
retry_after = response.headers.get("Retry-After", 60)
print(f" Retry sau {retry_after} giây")
else:
print(f"❌ Lỗi: {response.status_code} - {response.text}")
async def check_health():
async with httpx.AsyncClient() as client:
response = await client.get(f"{BASE_URL}/health")
print(response.json())
asyncio.run(check_health())
asyncio.run(chat_example())So Sánh Chi Phí Và Lợi Ích
| Model | Giá gốc (OpenAI) | HolySheep AI | Tiết kiệm |
|---|---|---|---|
| GPT-4.1 | $60/MTok | $8/MTok | 86% |
| Claude Sonnet 4.5 | $90/MTok | $15/MTok | 83% |
| Gemini 2.5 Flash | $7.50/MTok | $2.50/MTok | 67% |
| DeepSeek V3.2 | $2.80/MTok | $0.42/MTok | 85% |
Tỷ giá ¥1 = $1 cùng hỗ trợ WeChat/Alipay giúp thanh toán dễ dàng cho developers Trung Quốc. Độ trễ trung bình <50ms đảm bảo trải nghiệm mượt mà. Đăng ký tại đây để nhận 10 USD tín dụng miễn phí!
Triển Khai Production Với Docker
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
Cài đặt dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
Copy source code
COPY . .
Expose port
EXPOSE 8000
Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD python -c "import httpx; httpx.get('http://localhost:8000/health').raise_for_status()"
Run
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
docker-compose.yml
version: '3.8'
services:
gateway:
build: .
ports:
- "8000:8000"
environment:
- HOLYSHEEP_API_KEY=${HOLYSHEEP_API_KEY}
- REDIS_HOST=redis
- DATABASE_URL=postgresql://user:pass@postgres:5432/gateway
depends_on:
- redis
- postgres
restart: unless-stopped
redis:
image: redis:7-alpine
volumes:
- redis_data:/data
restart: unless-stopped
postgres:
image: postgres:15-alpine
environment:
- POSTGRES_DB=gateway
- POSTGRES_USER=user
- POSTGRES_PASSWORD=pass
volumes:
- postgres_data:/var/lib/postgresql/data
restart: unless-stopped
volumes:
redis_data:
postgres_data:Lỗi Thường Gặp Và Cách Khắc Phục
1. Lỗi 401 Unauthorized - Token Hết Hạn Hoặc Không Hợp Lệ
# ❌ Sai: Không kiểm tra token expiration
@app.get("/api/data")
async def get_data(token: str):
return {"data": decode_without_verification(token)}
✅ Đúng: Kiểm tra kỹ token và handle exceptions
from fastapi import HTTPException, status
async def verify_and_get_data(token: str):
try:
payload = jwt.decode(
token,
settings.SECRET_KEY,
algorithms=[settings.ALGORITHM]
)
user_id = payload.get("sub")
exp = payload.get("exp")
if not user_id:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token không chứa user ID"
)
if exp and datetime.utcnow().timestamp() > exp:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token đã hết hạn. Vui lòng đăng nhập lại."
)
return {"user_id": user_id, "payload": payload}
except JWTError as e:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=f"Xác thực thất bại: {str(e)}"
)2. Lỗi 429 Too Many Requests - Rate Limit Vượt Quá
# ❌ Sai: Không có rate limiting, dẫn đến overload như sự cố đầu bài
@app.post("/api/chat")
async def chat(request: ChatRequest):
result = await call_ai(request) # Không giới hạn!
return result
✅ Đúng: Implement rate limiting với retry logic
from tenacity import retry, stop_after_attempt, wait_exponential
@router.post("/chat", response_model=ChatResponse)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def chat_with_limit(
request: ChatRequest,
current_user: dict = Depends(get_current_user)
):
user_id = current_user["user_id"]
tier = current_user["tier"]
# Lấy limits theo tier
limits = {"free": 20, "pro": 100, "enterprise": 1000}
limit = limits.get(tier, 20)
# Check với Redis
is_allowed, remaining, reset_time = await rate_limiter.check_rate_limit(
f"user:{user_id}", limit, 60
)
if not is_allowed:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail={
"message": "Đã vượt quá giới hạn request",
"limit": limit,
"reset_at": reset_time,
"tier": tier,
"upgrade_url": "/billing/upgrade"
},
headers={
"X-RateLimit-Limit": str(limit),
"X-RateLimit-Remaining": "0",
"X-RateLimit-Reset": str(reset_time),
"Retry-After": str(reset_time - int(time.time()))
}
)
return await forward_to_provider(request, user_id)3. Lỗi Timeout - Upstream Service Không Phản Hồi
# ❌ Sai: Timeout quá ngắn hoặc không có retry
async def call_ai(request):
async with httpx.AsyncClient() as client:
response = await client.post(url, json=request, timeout=5.0)
return response.json()
✅ Đúng: Timeout hợp lý v