Trong bài viết này, tôi sẽ chia sẻ chi tiết cách triển khai hệ thống kiểm soát truy cập API cho các mô hình ngôn ngữ lớn (LLM) thông qua kinh nghiệm thực chiến khi di chuyển hạ tầng từ nhà cung cấp cũ sang HolySheep AI. Đây là playbook mà đội ngũ kỹ thuật của tôi đã sử dụng thành công, giúp tiết kiệm hơn 85% chi phí và cải thiện độ trễ từ 200ms xuống dưới 50ms.
Bối Cảnh Và Lý Do Chuyển Đổi
Trước khi đi vào chi tiết kỹ thuật, tôi muốn chia sẻ vì sao đội ngũ của tôi quyết định chuyển đổi. Sau 6 tháng vận hành hệ thống AI với chi phí API hàng tháng lên tới $3,500, chúng tôi nhận ra rằng:
- Chi phí tăng 40% mỗi quý do các nhà cung cấp lớn điều chỉnh giá
- Độ trễ trung bình 180-250ms không đáp ứng được yêu cầu real-time
- Thiếu cơ chế kiểm soát chi phí chi tiết theo từng team/dự án
- Không hỗ trợ thanh toán nội địa (WeChat/Alipay)
- Rate limiting không linh hoạt cho các use case khác nhau
Sau khi benchmark nhiều nhà cung cấp, HolySheep AI nổi bật với tỷ giá ¥1=$1, thời gian phản hồi dưới 50ms và hệ thống quản lý API key mạnh mẽ. Cụ thể, giá năm 2026 rất cạnh tranh: GPT-4.1 chỉ $8/MTok, Claude Sonnet 4.5 $15/MTok, Gemini 2.5 Flash $2.50/MTok, và DeepSeek V3.2 chỉ $0.42/MTok.
Kiến Trúc Hệ Thống Kiểm Soát Truy Cập
1. Tổng Quan Kiến Trúc
Hệ thống kiểm soát truy cập API LLM cần đảm bảo 4 yếu tố cốt lõi: Xác thực (Authentication), Ủy quyền (Authorization), Giới hạn tốc độ (Rate Limiting), và Theo dõi chi phí (Cost Tracking). Dưới đây là kiến trúc mà đội ngũ tôi đã triển khai thành công với HolySheep.
2. Triển Khhai Với HolySheep API
Điểm quan trọng nhất khi triển khai là cấu hình đúng base_url. HolySheep sử dụng endpoint https://api.holysheep.ai/v1 thay vì các endpoint cũ của nhà cung cấp khác. Dưới đây là code Python hoàn chỉnh cho module kiểm soát truy cập:
import os
import time
import hashlib
import hmac
import json
from datetime import datetime, timedelta
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from enum import Enum
import httpx
from collections import defaultdict
============== CẤU HÌNH HOLYSHEEP ==============
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.getenv("YOUR_HOLYSHEEP_API_KEY", "")
============== CÁC MODEL ĐƯỢC HỖ TRỢ ==============
class ModelType(Enum):
GPT_4_1 = "gpt-4.1"
CLAUDE_SONNET_4_5 = "claude-sonnet-4.5"
GEMINI_FLASH_2_5 = "gemini-2.5-flash"
DEEPSEEK_V3_2 = "deepseek-v3.2"
============== BẢNG GIÁ HOLYSHEEP 2026 (USD/MTok) ==============
MODEL_PRICING = {
ModelType.GPT_4_1.value: 8.0,
ModelType.CLAUDE_SONNET_4_5.value: 15.0,
ModelType.GEMINI_FLASH_2_5.value: 2.50,
ModelType.DEEPSEEK_V3_2.value: 0.42,
}
@dataclass
class RateLimitConfig:
"""Cấu hình giới hạn tốc độ cho mỗi API key"""
requests_per_minute: int = 60
requests_per_hour: int = 1000
tokens_per_day: int = 1_000_000
max_cost_per_month: float = 500.0
@dataclass
class UsageRecord:
"""Bản ghi sử dụng chi tiết"""
timestamp: datetime
model: str
input_tokens: int
output_tokens: int
cost: float
request_id: str
user_id: Optional[str] = None
project_id: Optional[str] = None
class APIKeyManager:
"""Quản lý API keys với kiểm soát truy cập chi tiết"""
def __init__(self, base_url: str = HOLYSHEEP_BASE_URL, api_key: str = HOLYSHEEP_API_KEY):
self.base_url = base_url
self.api_key = api_key
self.usage_db: Dict[str, List[UsageRecord]] = defaultdict(list)
self.rate_limits: Dict[str, RateLimitConfig] = {}
self.client = httpx.Client(
timeout=30.0,
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
def register_api_key(
self,
key_id: str,
rate_limit: RateLimitConfig
) -> bool:
"""Đăng ký API key mới với cấu hình rate limit"""
self.rate_limits[key_id] = rate_limit
return True
def check_rate_limit(self, key_id: str) -> tuple[bool, str]:
"""Kiểm tra giới hạn tốc độ - trả về (được_phép, thông_báo)"""
if key_id not in self.rate_limits:
return False, "API key chưa được đăng ký"
config = self.rate_limits[key_id]
now = datetime.now()
records = self.usage_db[key_id]
# Kiểm tra requests per minute
recent_minute = [
r for r in records
if now - r.timestamp < timedelta(minutes=1)
]
if len(recent_minute) >= config.requests_per_minute:
return False, f"Rate limit: Tối đa {config.requests_per_minute} req/phút"
# Kiểm tra requests per hour
recent_hour = [
r for r in records
if now - r.timestamp < timedelta(hours=1)
]
if len(recent_hour) >= config.requests_per_hour:
return False, f"Rate limit: Tối đa {config.requests_per_hour} req/giờ"
# Kiểm tra tokens per day
today_tokens = sum(
r.input_tokens + r.output_tokens
for r in records
if r.timestamp.date() == now.date()
)
if today_tokens >= config.tokens_per_day:
return False, f"Đã vượt quota: {config.tokens_per_day} tokens/ngày"
return True, "OK"
def calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
"""Tính chi phí dựa trên model và số tokens"""
price_per_mtok = MODEL_PRICING.get(model, 0)
total_tokens_mtok = (input_tokens + output_tokens) / 1_000_000
return round(total_tokens_mtok * price_per_mtok, 6)
def record_usage(
self,
key_id: str,
model: str,
input_tokens: int,
output_tokens: int,
request_id: str,
user_id: Optional[str] = None,
project_id: Optional[str] = None
) -> UsageRecord:
"""Ghi nhận việc sử dụng API"""
cost = self.calculate_cost(model, input_tokens, output_tokens)
record = UsageRecord(
timestamp=datetime.now(),
model=model,
input_tokens=input_tokens,
output_tokens=output_tokens,
cost=cost,
request_id=request_id,
user_id=user_id,
project_id=project_id
)
self.usage_db[key_id].append(record)
return record
def get_usage_summary(self, key_id: str, days: int = 30) -> Dict[str, Any]:
"""Lấy tổng hợp sử dụng trong N ngày"""
cutoff = datetime.now() - timedelta(days=days)
records = [r for r in self.usage_db[key_id] if r.timestamp >= cutoff]
total_cost = sum(r.cost for r in records)
total_input = sum(r.input_tokens for r in records)
total_output = sum(r.output_tokens for r in records)
request_count = len(records)
return {
"period_days": days,
"total_requests": request_count,
"total_input_tokens": total_input,
"total_output_tokens": total_output,
"total_cost_usd": round(total_cost, 2),
"avg_cost_per_request": round(total_cost / request_count, 4) if request_count > 0 else 0,
"model_breakdown": self._get_model_breakdown(records)
}
def _get_model_breakdown(self, records: List[UsageRecord]) -> Dict[str, Dict]:
breakdown = defaultdict(lambda: {"count": 0, "tokens": 0, "cost": 0.0})
for r in records:
breakdown[r.model]["count"] += 1
breakdown[r.model]["tokens"] += r.input_tokens + r.output_tokens
breakdown[r.model]["cost"] += r.cost
return dict(breakdown)
============== DEMO SỬ DỤNG ==============
if __name__ == "__main__":
manager = APIKeyManager()
# Đăng ký 3 API keys với cấu hình khác nhau
manager.register_api_key("key_prod", RateLimitConfig(
requests_per_minute=120,
requests_per_hour=5000,
tokens_per_day=5_000_000,
max_cost_per_month=2000.0
))
manager.register_api_key("key_dev", RateLimitConfig(
requests_per_minute=30,
requests_per_hour=500,
tokens_per_day=500_000,
max_cost_per_month=100.0
))
manager.register_api_key("key_test", RateLimitConfig(
requests_per_minute=10,
requests_per_hour=100,
tokens_per_day=50_000,
max_cost_per_month=20.0
))
# Mô phỏng một số request
for i in range(3):
allowed, msg = manager.check_rate_limit("key_prod")
if allowed:
record = manager.record_usage(
key_id="key_prod",
model="deepseek-v3.2", # Model rẻ nhất: $0.42/MTok
input_tokens=500,
output_tokens=200,
request_id=f"req_{i}"
)
print(f"✓ Request {i}: Chi phí ${record.cost:.4f}")
# Xem tổng hợp sau 30 ngày
summary = manager.get_usage_summary("key_prod", days=30)
print(f"\n📊 Tổng hợp 30 ngày:")
print(f" Tổng chi phí: ${summary['total_cost_usd']}")
print(f" Tổng requests: {summary['total_requests']}")Triển Khhai Middleware Kiểm Soát Truy Cập
Để tích hợp vào hệ thống production thực tế, tôi khuyến nghị sử dụng middleware. Dưới đây là implementation hoàn chỉnh với FastAPI và async support, tích hợp sẵn với HolySheep API:
import asyncio
from fastapi import FastAPI, HTTPException, Header, Request, Depends
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import Optional, List, Dict
import logging
from datetime import datetime, timedelta
import redis.asyncio as redis
============== CẤU HÌNH ==============
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
REDIS_URL = "redis://localhost:6379"
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
============== MODELS ==============
class ChatMessage(BaseModel):
role: str
content: str
class ChatCompletionRequest(BaseModel):
model: str = "deepseek-v3.2"
messages: List[ChatMessage]
temperature: Optional[float] = 0.7
max_tokens: Optional[int] = 2048
stream: Optional[bool] = False
class UsageInfo(BaseModel):
input_tokens: int
output_tokens: int
total_tokens: int
cost_usd: float
class APIKeyConfig(BaseModel):
"""Cấu hình chi tiết cho mỗi API key"""
key_id: str
name: str
active: bool = True
allowed_models: List[str] = []
rate_limit_rpm: int = 60
rate_limit_rpd: int = 50000
monthly_budget: float = 100.0
current_month_cost: float = 0.0
department: Optional[str] = None
project: Optional[str] = None
class AccessControlMiddleware:
"""Middleware kiểm soát truy cập toàn diện"""
def __init__(self):
self.redis_client: Optional[redis.Redis] = None
self.key_configs: Dict[str, APIKeyConfig] = {}
self.model_prices = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42,
}
async def initialize(self):
"""Khởi tạo kết nối Redis"""
try:
self.redis_client = await redis.from_url(
REDIS_URL,
encoding="utf-8",
decode_responses=True
)
logger.info("✓ Kết nối Redis thành công")
except Exception as e:
logger.warning(f"Không kết nối được Redis: {e}")
self.redis_client = None
async def validate_api_key(self, api_key: str) -> Optional[APIKeyConfig]:
"""Xác thực và lấy cấu hình API key"""
if api_key in self.key_configs:
return self.key_configs[api_key]
return None
async def check_rate_limit(
self,
key_id: str,
rpm_limit: int,
rpd_limit: int
) -> tuple[bool, Dict]:
"""Kiểm tra rate limit với sliding window"""
now = datetime.now()
if self.redis_client:
# Sliding window rate limiting với Redis
minute_key = f"ratelimit:{key_id}:minute:{now.strftime('%Y%m%d%H%M')}"
day_key = f"ratelimit:{key_id}:day:{now.strftime('%Y%m%d')}"
minute_count = await self.redis_client.incr(minute_key)
day_count = await self.redis_client.incr(day_key)
# Set expiry cho keys
await self.redis_client.expire(minute_key, 60)
await self.redis_client.expire(day_key, 86400)
if minute_count > rpm_limit:
ttl = await self.redis_client.ttl(minute_key)
return False, {
"error": "Rate limit exceeded",
"limit": rpm_limit,
"reset_in_seconds": ttl,
"type": "requests_per_minute"
}
if day_count > rpd_limit:
ttl = await self.redis_client.ttl(day_key)
return False, {
"error": "Daily limit exceeded",
"limit": rpd_limit,
"reset_in_seconds": ttl,
"type": "requests_per_day"
}
return True, {
"remaining_rpm": rpm_limit - minute_count,
"remaining_rpd": rpd_limit - day_count
}
else:
# Fallback: In-memory rate limiting (không khuyến khích cho production)
return True, {"note": "In-memory fallback mode"}
async def check_budget(
self,
key_config: APIKeyConfig,
estimated_cost: float
) -> bool:
"""Kiểm tra ngân sách còn lại"""
projected_cost = key_config.current_month_cost + estimated_cost
return projected_cost <= key_config.monthly_budget
async def record_usage(
self,
key_id: str,
model: str,
input_tokens: int,
output_tokens: int,
cost: float,
request_id: str
):
"""Ghi nhận usage vào Redis"""
if self.redis_client:
usage_key = f"usage:{key_id}:{datetime.now().strftime('%Y%m')}"
usage_data = {
"model": model,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost": cost,
"request_id": request_id,
"timestamp": datetime.now().isoformat()
}
await self.redis_client.lpush(usage_key, str(usage_data))
await self.redis_client.expire(usage_key, 2592000) # 30 ngày
async def get_usage_stats(self, key_id: str, days: int = 30) -> Dict:
"""Lấy thống kê sử dụng chi tiết"""
if not self.redis_client:
return {"error": "Redis not connected"}
total_cost = 0.0
total_input = 0
total_output = 0
request_count = 0
for day_offset in range(days):
date = (datetime.now() - timedelta(days=day_offset)).strftime('%Y%m')
usage_key = f"usage:{key_id}:{date}"
# Lấy tất cả usage records
records = await self.redis_client.lrange(usage_key, 0, -1)
for record in records:
try:
import json
data = json.loads(record)
total_cost += data.get("cost", 0)
total_input += data.get("input_tokens", 0)
total_output += data.get("output_tokens", 0)
request_count += 1
except:
continue
return {
"period_days": days,
"total_requests": request_count,
"total_input_tokens": total_input,
"total_output_tokens": total_output,
"total_cost_usd": round(total_cost, 4),
"avg_cost_per_request": round(total_cost / request_count, 6) if request_count > 0 else 0
}
class LLMProxy:
"""Proxy server để truy cập HolySheep với kiểm soát"""
def __init__(self, middleware: AccessControlMiddleware):
self.middleware = middleware
self.base_url = HOLYSHEEP_BASE_URL
self.client = httpx.AsyncClient(timeout=60.0)
async def chat_completion(
self,
api_key: str,
request: ChatCompletionRequest,
request_id: str
) -> Dict:
"""Xử lý request chat completion với đầy đủ kiểm soát"""
# Bước 1: Xác thực API key
key_config = await self.middleware.validate_api_key(api_key)
if not key_config or not key_config.active:
raise HTTPException(
status_code=401,
detail="API key không hợp lệ hoặc đã bị vô hiệu hóa"
)
# Bước 2: Kiểm tra model được phép
if key_config.allowed_models and request.model not in key_config.allowed_models:
raise HTTPException(
status_code=403,
detail=f"Model '{request.model}' không được phép. Danh sách: {key_config.allowed_models}"
)
# Bước 3: Kiểm tra rate limit
allowed, rate_info = await self.middleware.check_rate_limit(
key_config.key_id,
key_config.rate_limit_rpm,
key_config.rate_limit_rpd
)
if not allowed:
raise HTTPException(
status_code=429,
detail=rate_info
)
# Bước 4: Ước tính chi phí (ước tính sơ bộ)
estimated_tokens = sum(len(m.content.split()) * 1.3 for m in request.messages)
estimated_output = request.max_tokens or 1000
estimated_cost = self._estimate_cost(
request.model,
int(estimated_tokens),
estimated_output
)
# Bước 5: Kiểm tra ngân sách
budget_ok = await self.middleware.check_budget(key_config, estimated_cost)
if not budget_ok:
raise HTTPException(
status_code=402,
detail=f"Đã vượt ngân sách tháng. Giới hạn: ${key_config.monthly_budget}, "
f"Chi phí hiện tại: ${key_config.current_month_cost}"
)
# Bước 6: Gọi HolySheep API
start_time = time.time()
try:
response = await self._call_holysheep(request)
latency_ms = (time.time() - start_time) * 1000
# Bước 7: Ghi nhận usage thực tế
input_tokens = response.get("usage", {}).get("prompt_tokens", 0)
output_tokens = response.get("usage", {}).get("completion_tokens", 0)
actual_cost = self._calculate_cost(request.model, input_tokens, output_tokens)
await self.middleware.record_usage(
key_config.key_id,
request.model,
input_tokens,
output_tokens,
actual_cost,
request_id
)
return {
"success": True,
"data": response,
"usage": {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost_usd": actual_cost,
"latency_ms": round(latency_ms, 2)
}
}
except Exception as e:
logger.error(f"Lỗi khi gọi HolySheep: {e}")
raise HTTPException(status_code=500, detail=str(e))
async def _call_holysheep(self, request: ChatCompletionRequest) -> Dict:
"""Gọi trực tiếp HolySheep API"""
url = f"{self.base_url}/chat/completions"
payload = {
"model": request.model,
"messages": [m.model_dump() for m in request.messages],
"temperature": request.temperature,
"max_tokens": request.max_tokens,
"stream": request.stream
}
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
response = await self.client.post(url, json=payload, headers=headers)
response.raise_for_status()
return response.json()
def _estimate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
price = self.middleware.model_prices.get(model, 0)
total_tokens_mtok = (input_tokens + output_tokens) / 1_000_000
return total_tokens_mtok * price
def _calculate_cost(self, model: str, input_tokens: int, output_tokens: int) -> float:
return self._estimate_cost(model, input_tokens, output_tokens)
============== FASTAPI APPLICATION ==============
app = FastAPI(title="LLM API Gateway - HolySheep Edition")
middleware_instance = AccessControlMiddleware()
proxy = LLMProxy(middleware_instance)
@app.on_event("startup")
async def startup():
await middleware_instance.initialize()
@app.post("/v1/chat/completions")
async def chat_completions(
request: ChatCompletionRequest,
authorization: str = Header(...),
x_request_id: str = Header(default_factory=lambda: f"req_{int(time.time())}")
):
"""Endpoint chính cho chat completions với kiểm soát truy cập"""
api_key = authorization.replace("Bearer ", "")
return await proxy.chat_completion(api_key, request, x_request_id)
@app.get("/admin/usage/{key_id}")
async def get_usage(
key_id: str,
days: int = 30,
admin_key: str = Header(...)
):
"""Endpoint quản trị để xem usage statistics"""
if admin_key != os.getenv("ADMIN_SECRET"):
raise HTTPException(status_code=403, detail="Unauthorized")
return await middleware_instance.get_usage_stats(key_id, days)
============== DEMO ==============
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)Lỗi Thường Gặp Và Cách Khắc Phục
1. Lỗi Xác Thực API Key
Mô tả lỗi: Khi triển khai hệ thống kiểm soát truy cập, lỗi phổ biến nhất là nhận được HTTP 401 Unauthorized ngay cả khi API key có vẻ đúng.
Nguyên nhân: API key không được truyền đúng format trong header Authorization. HolySheep yêu cầu format: Authorization: Bearer YOUR_HOLYSHEEP_API_KEY
# ❌ CÁCH SAI - Gây lỗi 401
headers = {
"Authorization": api_key # Thiếu "Bearer " prefix
}
✓ CÁCH ĐÚNG
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}"
}
Hoặc sử dụng class helper
class HolySheepClient:
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
def _get_headers(self) -> Dict[str, str]:
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async def chat_completion(self, messages: List[Dict], model: str = "deepseek-v3.2"):
"""Gọi API với header đúng format"""
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.BASE_URL}/chat/completions",
json={
"model": model,
"messages": messages
},
headers=self._get_headers() # ✓ Header đúng format
)
if response.status_code == 401:
raise ValueError(
"Xác thực thất bại. Kiểm tra:\n"
"1. API key có đúng format không (bắt đầu bằng 'sk-...')\n"
"2. API key đã được kích hoạt chưa\n"
"3. Truy cập https://www.holysheep.ai/register để lấy key mới"
)
response.raise_for_status()
return response.json()
Sử dụng:
client = HolySheepClient("YOUR_HOLYSHEEP_API_KEY")
result = await client.chat_completion([
{"role": "user", "content": "Xin chào"}
])2. Lỗi Rate Limit Không Hoạt Động
Mô tả lỗi: Mặc dù đã cấu hình rate limit, hệ thống vẫn cho phép vượt quá giới hạn hoặc block sai request hợp lệ.
Nguyên nhân: Race condition khi nhiều request đến cùng lúc, hoặc logic kiểm tra không đồng bộ với việc ghi nhận usage.
import asyncio
from datetime import datetime
❌ CÁCH SAI - Có race condition
class BrokenRateLimiter:
def __init__(self):
self.counters = {}
async def check_and_increment(self, key: str, limit: int) -> bool:
current = self.counters.get(key, 0)
# ⚠️ Race condition: Giữa 2 request, giá trị không cập nhật kịp
if current >= limit:
return False
self.counters[key] = current + 1
return True
✓ CÁCH ĐÚNG - Sử dụng atomic operations với Redis
class ProductionRateLimiter:
def __init__(self, redis_url: str):
self.redis = redis.from_url(redis_url)
async def check_rate_limit(
self,
key: str,
limit: int,
window_seconds: int
) -> tuple[bool, Dict]:
"""
Sliding Window Rate Limiting với Redis Lua script
Đảm bảo atomic operation - không có race condition
"""
lua_script = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])
local window_start = now - window
-- Xóa các entries cũ
redis.call('ZREMRANGEBYSCORE', key, 0, window_start)
-- Đếm số request trong window
local count = redis.call('ZCARD', key)
if count >= limit then
-- Lấy thời gian request cũ nhất
local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
local reset_at = oldest[2] + window if oldest else now + window
return {0, count, reset_at}
end
-- Thêm request mới
redis.call('ZADD', key, now, now .. ':' .. math.random())
redis.call('EXPIRE', key, window)
return {1, count + 1, now + window}
"""
now = int(datetime.now().timestamp())
result = await self.redis.eval(
lua_script,
1, # number of keys
f"ratelimit:{key}", # key
str(limit),
str(window_seconds),
str(now)
)
allowed = bool(result[0])
remaining = limit - int(result[1])
retry_after = int(result[2]) - now if not allowed else 0
return allowed, {
"limit": limit,
"remaining": max(0, remaining),
"reset_in_seconds": retry_after
}
Sử dụng:
limiter = ProductionRateLimiter("redis://localhost:6379")
async def make_request_with_rate_limit():
allowed, info = await limiter.check_rate_limit(
key="user_123",