Khi làm việc với Claude API trong môi trường production, lỗi 429 Too Many Requests là "người bạn không mong muốn" xuất hiện vào đúng lúc quan trọng nhất. Tôi đã từng mất 3 tiếng để debug một pipeline xử lý 10,000 tài liệu vì không hiểu rõ cơ chế rate limit của Anthropic. Bài viết này sẽ giúp bạn không phải đi lại con đường đó.
429 Rate Limit Là Gì Và Tại Sao Nó Xảy Ra?
Mã lỗi 429 означает Anthropic đang nhận quá nhiều request từ API key của bạn trong một khoảng thời gian ngắn. Cơ chế này bảo vệ hệ thống khỏi bị quá tải và đảm bảo công bằng cho tất cả người dùng.
Các loại Rate Limit của Claude API
- Requests per minute (RPM) — Giới hạn số lượng request mỗi phút
- Tokens per minute (TPM) — Giới hạn tổng tokens xử lý mỗi phút
- Concurrent requests — Số request đồng thời tối đa
- Daily usage limit — Giới hạn sử dụng hàng ngày theo gói subscription
Khi bạn nhận được response 429, response body sẽ chứa header Retry-After cho biết số giây cần chờ trước khi thử lại.
{
"error": {
"type": "rate_limit_error",
"message": "Too many requests. Please wait before retrying.",
"retry_after": 30
}
}Chiến Lược Xử Lý 429 Toàn Diện
1. Exponential Backoff — Chiến Lược Cơ Bản
Đây là phương pháp tăng dần thời gian chờ giữa các lần retry, giúp giảm áp lực lên API và tăng khả năng thành công.
import requests
import time
import random
def claude_request_with_backoff(messages, model="claude-sonnet-4-20250514", max_retries=5):
"""
Gửi request đến Claude API với exponential backoff
base_url sử dụng HolySheep AI thay vì Anthropic trực tiếp
"""
base_url = "https://api.holysheep.ai/v1"
api_key = "YOUR_HOLYSHEEP_API_KEY" # Thay bằng API key của bạn
url = f"{base_url}/chat/completions"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"max_tokens": 4096
}
for attempt in range(max_retries):
try:
response = requests.post(url, json=payload, headers=headers, timeout=60)
if response.status_code == 200:
return response.json()
elif response.status_code == 429:
# Lấy retry_after từ response hoặc tính toán
retry_after = response.headers.get('Retry-After', 2 ** attempt)
wait_time = int(retry_after) + random.uniform(0, 1)
print(f"Rate limited. Chờ {wait_time:.2f} giây (attempt {attempt + 1}/{max_retries})")
time.sleep(wait_time)
else:
raise Exception(f"API Error: {response.status_code} - {response.text}")
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
wait_time = 2 ** attempt + random.uniform(0, 1)
print(f"Request failed: {e}. Chờ {wait_time:.2f} giây...")
time.sleep(wait_time)
raise Exception("Max retries exceeded")
Ví dụ sử dụng
messages = [{"role": "user", "content": "Giải thích cơ chế rate limiting"}]
result = claude_request_with_backoff(messages)
print(result['choices'][0]['message']['content'])2. Token Bucket Algorithm — Xử Lý Chuyên Nghiệp
Đối với ứng dụng cần xử lý số lượng lớn request, Token Bucket là lựa chọn tối ưu. Thuật toán này kiểm soát tốc độ request một cách chính xác và hiệu quả.
import time
import threading
from collections import deque
from typing import Optional
class TokenBucketRateLimiter:
"""
Token Bucket Rate Limiter cho Claude API
Kiểm soát RPM và TPM một cách chính xác
"""
def __init__(self, rpm: int = 50, tpm: int = 100000):
self.rpm = rpm
self.tpm = tpm
self.refill_rate_rpm = rpm / 60 # Tokens mới mỗi giây
self.refill_rate_tpm = tpm / 60
self.rpm_buckets = {} # Mỗi model một bucket
self.lock = threading.Lock()
def acquire(self, model: str, tokens: int, timeout: float = 60) -> bool:
"""
Chờ cho đến khi có đủ quota để gửi request
Returns: True nếu acquired thành công, False nếu timeout
"""
start_time = time.time()
while True:
if time.time() - start_time > timeout:
return False
with self.lock:
if model not in self.rpm_buckets: