Mọi API AI đều có giới hạn tốc độ (rate limits). Gặp phải chúng trong quá trình phát triển thì thật phiền phức. Nhưng nếu gặp trong môi trường production, người dùng của bạn sẽ thấy lỗi, các luồng dữ liệu (stream) bị ngắt quãng và lỗi timeout có vẻ ngẫu nhiên cho đến khi bạn kiểm tra quy luật của chúng.
Sai lầm chính là coi giới hạn tốc độ chỉ là một vấn đề duy nhất. Thực tế, nó thường là bốn vấn đề khác nhau ẩn sau cùng một mã lỗi 429:
- requests per minute (số yêu cầu mỗi phút)
- tokens per minute (số token mỗi phút)
- concurrent in-flight requests (số yêu cầu đồng thời đang xử lý)
- account-level hoặc project-level quota exhaustion (hết hạn mức ở cấp độ tài khoản hoặc dự án)
Nếu bạn chỉ xây dựng giải pháp cho một trong số đó, những cái còn lại vẫn sẽ gây rắc rối cho bạn.
Nếu bạn vẫn đang ở giai đoạn chuyển đổi nhà cung cấp, hãy đọc hướng dẫn chuyển đổi trước. Nếu bạn đang đánh giá liệu một gateway có giúp ích cho việc dự phòng (fallback) và giảm bớt gánh nặng vận hành hay không, bài viết so sánh OpenRouter là tài liệu tham khảo tốt nhất.
Ý nghĩa thực sự của các giới hạn tốc độ
Giới hạn yêu cầu (Request limits)
Đây là lỗi hiển nhiên nhất. Bạn đã gửi quá nhiều yêu cầu trong một khoảng thời gian ngắn.
Giới hạn token (Token limits)
Đây là giới hạn mà các đội ngũ thường đánh giá thấp. Một prompt dài duy nhất có thể tiêu tốn ngân sách bằng nhiều yêu cầu nhỏ cộng lại. Nếu bạn đột ngột thêm một system prompt nặng 20 KB, số lượng yêu cầu có vẻ vẫn ổn định trong khi ngân sách token đã cạn kiệt.
Giới hạn đồng thời (Concurrency limits)
Một số nhà cung cấp và gateway hoàn toàn hài lòng với mức trung bình mỗi phút của bạn cho đến khi bạn mở năm mươi stream cùng một lúc. Gói dịch vụ thì ổn, nhưng hình thái bùng phát (burst shape) thì không.
Hết hạn mức hoặc số dư (Quota or balance exhaustion)
Vấn đề này thường xuất hiện dưới dạng triệu chứng "giới hạn tốc độ" trên bảng điều khiển vì kết quả vận hành là như nhau: các cuộc gọi API không còn thành công. Nhưng cách khắc phục thì khác. Cơ chế thử lại (backoff) sẽ vô dụng nếu vấn đề là số dư bằng không.
Cách các nhà cung cấp thường áp dụng giới hạn
Các con số cụ thể thay đổi theo thời gian, đó là lý do tại sao việc đưa một bảng giá công khai vào tài liệu ứng dụng của bạn sẽ nhanh chóng bị lỗi thời. Quy luật ổn định là như sau:
- Các nhà cung cấp kiểu OpenAI thường hiển thị các header về yêu cầu và token, đồng thời họ điều chỉnh mức trần của bạn dựa trên lịch sử tài khoản hoặc cấp độ sử dụng.
- Các nhà cung cấp kiểu Anthropic thường áp dụng cả lưu lượng (throughput) ở mức phút và các giới hạn dự án rộng hơn, đặc biệt là đối với các mô hình cao cấp.
- Các nhà cung cấp kiểu Google thường phân biệt hành vi giữa gói miễn phí và gói trả phí, và có thể thay đổi giới hạn rõ rệt theo từng dòng mô hình.
- Các bên tổng hợp (Aggregators) thêm một lớp giới hạn nữa bên trên các ràng buộc của nhà cung cấp gốc, nhưng đổi lại họ có thể điều hướng sang các kênh khác khi một nhà cung cấp gốc tạm thời bị quá tải.
Hãy coi các giới hạn của nhà cung cấp là cấu hình động, không phải là hằng số.
Đọc các Header giới hạn tốc độ
Tất cả các nhà cung cấp lớn đều trả về thông tin giới hạn tốc độ trong các response header:
x-ratelimit-limit-requests: 500
x-ratelimit-remaining-requests: 499
x-ratelimit-reset-requests: 60s
x-ratelimit-limit-tokens: 200000
x-ratelimit-remaining-tokens: 199500
Hãy sử dụng các header này một cách chủ động. Đừng đợi đến khi gặp lỗi 429 mới bắt đầu giảm tốc độ.
Thói quen vận hành mà bạn nên có rất đơn giản:
- Ghi nhật ký (log) các header khi thành công, không chỉ khi thất bại.
- Cảnh báo khi dung lượng còn lại xuống dưới một ngưỡng nhất định.
- Điều tiết lưu lượng (shape traffic) trước khi yêu cầu tiếp theo vượt quá giới hạn.
Nếu bạn chỉ xem xét các header sau khi thất bại, bạn đã chậm chân rồi.
Xây dựng logic thử lại (Retry Logic)
Cách làm sai
# Đừng làm thế này
import time
def call_api(messages):
while True:
try:
return client.chat.completions.create(
model="gpt-4.1",
messages=messages
)
except Exception:
time.sleep(1) # Độ trễ cố định, không có backoff, bắt mọi loại lỗi
Vấn đề: không có exponential backoff (thử lại lũy thừa), bắt cả những lỗi không thể thử lại, không có giới hạn số lần thử lại tối đa, không có jitter (độ nhiễu ngẫu nhiên).
Cách làm đúng
import time
import random
from openai import RateLimitError, APIError, APIConnectionError
def call_with_retry(messages, model="gpt-4.1", max_retries=3):
"""Thử lại với exponential backoff và jitter."""
for attempt in range(max_retries + 1):
try:
return client.chat.completions.create(
model=model,
messages=messages
)
except RateLimitError as e:
if attempt == max_retries:
raise
# Sử dụng retry_after từ response nếu có
wait = getattr(e, 'retry_after', None)
if wait is None:
wait = (2 ** attempt) + random.uniform(0, 1)
print(f"Bị giới hạn tốc độ. Đang chờ {wait:.1f}s (lần thử {attempt + 1})")
time.sleep(wait)
except APIConnectionError:
if attempt == max_retries:
raise
wait = (2 ** attempt) + random.uniform(0, 1)
time.sleep(wait)
except APIError as e:
# Không thử lại các lỗi phía client (400, 401, 403)
if e.status_code and 400 <= e.status_code < 500:
raise
if attempt == max_retries:
raise
time.sleep((2 ** attempt) + random.uniform(0, 1))
Các nguyên tắc chính:
- Exponential backoff: 1s, 2s, 4s, 8s
- Jitter: thêm 0-1s ngẫu nhiên để ngăn chặn hiện tượng "bầy đàn" (thundering herd)
- Tôn trọng header
retry_afterkhi được cung cấp - Không thử lại các lỗi phía client (yêu cầu sai, lỗi xác thực)
- Thiết lập số lần thử lại tối đa
Hai quy tắc production bổ sung quan trọng:
- không bao giờ thử lại mãi mãi trên một streaming endpoint
- không bao giờ thử lại một yêu cầu đã gắn liền với các tác dụng phụ (side effects) mà người dùng có thể thấy, trừ khi thao tác đó có tính lũy đẳng (idempotent)
Chat completions thường an toàn để thử lại. Các tác dụng phụ do tool kích hoạt thường thì không.
Phiên bản Async
import asyncio
import random
from openai import AsyncOpenAI, RateLimitError
async_client = AsyncOpenAI(
api_key="sk-lemon-xxx",
base_url="https://api.lemondata.cc/v1"
)
async def call_with_retry_async(messages, model="gpt-4.1", max_retries=3):
for attempt in range(max_retries + 1):
try:
return await async_client.chat.completions.create(
model=model,
messages=messages
)
except RateLimitError:
if attempt == max_retries:
raise
wait = (2 ** attempt) + random.uniform(0, 1)
await asyncio.sleep(wait)
Điều tiết lưu lượng trước khi nó trở thành một cơn bão thử lại
Logic thử lại chỉ là một nửa giải pháp. Nếu nhà cung cấp của bạn đã bị quá tải, việc thử lại có thể biến một đợt bùng phát thành một sự cố tự gây ra.
Ba biện pháp kiểm soát sau đây sẽ tạo nên sự khác biệt:
1. Xếp hàng theo tenant hoặc người dùng
Nếu một khách hàng bắt đầu một công việc xử lý hàng loạt (batch job) khổng lồ, bạn không muốn mọi khách hàng khác bị ảnh hưởng lây.
2. Giới hạn các luồng (stream) đồng thời
Các streaming endpoint rất dễ bị đánh giá thấp vì mỗi yêu cầu "có vẻ" rẻ trong khi nó vẫn mở trong một thời gian dài.
3. Cắt tỉa prompt trước khi gửi đi
Giới hạn token thường là mức trần thực tế. Một prompt dài gấp đôi sẽ cắt giảm lưu lượng an toàn xuống còn khoảng một nửa.
Token Bucket phía Client
Đối với các ứng dụng có lưu lượng cao, hãy triển khai giới hạn tốc độ phía client để tránh chạm tới giới hạn của server:
import time
import asyncio
class TokenBucket:
def __init__(self, rate: float, capacity: int):
self.rate = rate # số token mỗi giây
self.capacity = capacity # kích thước bùng phát tối đa
self.tokens = capacity
self.last_refill = time.monotonic()
async def acquire(self, tokens: int = 1):
while True:
now = time.monotonic()
elapsed = now - self.last_refill
self.tokens = min(
self.capacity,
self.tokens + elapsed * self.rate
)
self.last_refill = now
if self.tokens >= tokens:
self.tokens -= tokens
return
# Chờ cho đến khi có đủ token
wait = (tokens - self.tokens) / self.rate
await asyncio.sleep(wait)
# 500 yêu cầu mỗi phút = ~8.3 mỗi giây
limiter = TokenBucket(rate=8.0, capacity=20)
async def rate_limited_call(messages, model="gpt-4.1"):
await limiter.acquire()
return await async_client.chat.completions.create(
model=model,
messages=messages
)
Token bucket rất tốt khi bạn biết mức trần của mình. Chúng thậm chí còn tốt hơn khi bạn điều chỉnh chúng từ dữ liệu header quan sát được thay vì một con số dự đoán cứng nhắc.
Dự phòng mô hình (Model Fallback) khi bị giới hạn tốc độ
Khi mô hình chính của bạn bị giới hạn tốc độ, hãy chuyển sang một phương án thay thế:
FALLBACK_CHAIN = [
"claude-sonnet-4-6",
"gpt-4.1",
"gpt-4.1-mini",
]
async def call_with_fallback(messages):
for model in FALLBACK_CHAIN:
try:
return await async_client.chat.completions.create(
model=model,
messages=messages
)
except RateLimitError:
continue
raise Exception("Tất cả các mô hình đều bị giới hạn tốc độ")
Đây là lúc các model gateway phát huy tác dụng, nhưng chỉ khi việc dự phòng được tính toán kỹ lưỡng. Đừng âm thầm nhảy từ một mô hình suy luận cao cấp sang một mô hình giá rẻ tí hon mà không nghĩ đến tác động đối với người dùng.
Một chuỗi dự phòng hợp lý là:
- cùng một nhà cung cấp, mô hình anh em nhỏ hơn
- dòng mô hình tương đương từ một nhà cung cấp khác
- chỉ sau đó mới đến một mô hình rẻ hơn hoặc có ngữ cảnh thấp hơn
Nếu bạn trộn lẫn "dự phòng để đảm bảo tính sẵn sàng" với "dự phòng để tiết kiệm chi phí" trong cùng một bước, việc gỡ lỗi sẽ trở nên rất rắc rối.
Theo dõi mức sử dụng giới hạn tốc độ
Theo dõi mức tiêu thụ giới hạn tốc độ của bạn để phát hiện các vấn đề trước khi chúng ảnh hưởng đến người dùng:
import logging
def log_rate_limits(response):
headers = response.headers
remaining = headers.get("x-ratelimit-remaining-requests")
limit = headers.get("x-ratelimit-limit-requests")
if remaining and int(remaining) < int(limit) * 0.1:
logging.warning(
f"Cảnh báo giới hạn tốc độ: còn lại {remaining}/{limit} yêu cầu"
)
Thiết lập cảnh báo khi dung lượng còn lại giảm xuống dưới 10%. Điều này cho bạn thời gian để triển khai điều tiết (throttling) trước khi người dùng thấy lỗi 429.
Bạn cũng nên ghi nhật ký:
- request ID
- model
- ước tính kích thước đầu vào
- thời lượng stream
- số lần thử lại
- kết quả cuối cùng (
success,rate_limited,network_error,quota_exhausted)
Nếu không có những trường này, các sự cố giới hạn tốc độ chỉ là những phỏng đoán.
Danh sách kiểm tra Production đơn giản
Trước khi gọi chatbot hoặc agent của bạn là "an toàn với giới hạn tốc độ", hãy xác minh năm mục sau:
- Có chính sách thử lại có giới hạn cho cả luồng sync và async.
- Bạn ghi nhật ký các header giới hạn tốc độ trên các phản hồi thành công.
- Có cơ chế điều tiết theo từng người dùng hoặc từng tenant trước khi gọi API gốc.
- Có ít nhất một mô hình dự phòng đã được xác thực.
- Frontend nhận được trạng thái lỗi rõ ràng thay vì một stream bị treo.
Nếu bạn đang xây dựng toàn bộ ứng dụng chứ không chỉ là các hàm thử lại cơ bản, hướng dẫn xây dựng chatbot với một key sẽ cho thấy các mảnh ghép này khớp với nhau như thế nào trong một dịch vụ FastAPI thực tế.
Tóm tắt
Giới hạn tốc độ không phải là một trường hợp hy hữu. Đó là điều kiện hoạt động bình thường của bất kỳ sản phẩm AI nào có lượng người dùng thực tế. Những đội ngũ xử lý tốt vấn đề này không phải vì họ có các giới hạn cao hơn một cách thần kỳ. Họ coi lưu lượng, thử lại và dự phòng là một phần của thiết kế ứng dụng ngay từ đầu.
Hãy tạo một API key tại LemonData, kiểm tra luồng thử lại của bạn trước khi có lưu lượng thực tế, và chuẩn bị cho lỗi 429 tiếp theo trước khi nó xảy ra.
| Chiến lược | Khi nào nên sử dụng |
|---|---|
| Exponential backoff | Luôn luôn (cơ bản) |
| Client-side rate limiter | Ứng dụng lưu lượng cao (>100 RPM) |
| Model fallback | Ứng dụng production với yêu cầu SLA |
| Giám sát chủ động | Bất kỳ triển khai production nào |
| Batch API | Các khối lượng công việc không yêu cầu thời gian thực |
Mục tiêu không phải là tránh hoàn toàn các giới hạn tốc độ. Mục tiêu là xử lý chúng một cách mượt mà để người dùng của bạn không bao giờ nhận ra.
Xây dựng các ứng dụng AI bền bỉ: lemondata.cc cung cấp khả năng điều hướng đa kênh, tự động xử lý các giới hạn tốc độ của nhà cung cấp gốc. Một API key, hơn 300 mô hình.