Nếu bạn đang bắt đầu hành trình lập trình với AI API, chắc hẳn bạn đã từng gặp những lần gọi API thất bại vì mạng chậm, server quá tải, hoặc đơn giản là kết nối bị gián đoạn. Trong bài viết này, mình sẽ chia sẻ cách mình xây dựng hệ thống xử lý lỗi thông minh với retry hooks và circuit breaker — hai kỹ thuật cứu cánh giúp ứng dụng của bạn hoạt động ổn định ngay cả khi API gặp sự cố.
Tại sao cần Retry và Circuit Breaker?
Khi mới tìm hiểu về API, mình từng nghĩ đơn giản: gọi API, nhận kết quả, xong. Nhưng thực tế hoàn toàn khác. Mạng internet có thể chập chờn, server AI có lúc quá tải (đặc biệt vào giờ cao điểm), và nếu bạn cứ gọi đi gọi lại liên tục khi server đang down, bạn không những tốn tiền mà còn có thể bị chặn IP.
Retry (thử lại) giúp tự động gửi lại yêu cầu khi gặp lỗi tạm thời. Circuit Breaker (cầu dao) giúp "ngắt điện" khi API liên tục thất bại, tránh làm quá tải hệ thống. Hai kỹ thuật này kết hợp với nhau tạo thành một bộ đôi hoàn hảo cho bất kỳ ứng dụng nào sử dụng AI API.
Chuẩn bị môi trường
Trước tiên, bạn cần đăng ký tài khoản HolyShehe AI — nền tảng API AI với chi phí cực kỳ cạnh tranh, chỉ từ $0.42/MTok với DeepSeek V3.2, tiết kiệm đến 85% so với các nhà cung cấp phương Tây. Ngoài ra, HolySheep hỗ trợ thanh toán qua WeChat Pay và Alipay, độ trễ dưới 50ms, và tặng tín dụng miễn phí khi đăng ký. Đăng ký tại đây để nhận ngay $5 credit.
Cài đặt thư viện cần thiết
pip install requests tenacity httpx
tenacity: thư viện retry mạnh mẽ
httpx: HTTP client hỗ trợ async
requests: thư viện HTTP cơ bản
Retry Hooks — Tự động thử lại khi gặp lỗi
Mình sẽ hướng dẫn bạn cách cài đặt retry thông minh với thư viện tenacity. Điều đặc biệt là bạn có thể hook vào các giai đoạn khác nhau của quá trình retry: trước khi thử, sau mỗi lần thất bại, hoặc khi thành công.
import requests
from tenacity import (
retry, stop_after_attempt, wait_exponential,
retry_if_exception_type, before_sleep_log
)
import logging
import time
Cấu hình logging để theo dõi quá trình retry
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class HolySheepAIClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
def _make_request(self, endpoint: str, payload: dict) -> dict:
"""Gửi yêu cầu đến HolySheep AI API"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
response = requests.post(
f"{self.base_url}/{endpoint}",
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
return response.json()
@retry(
stop=stop_after_attempt(5), # Thử tối đa 5 lần
wait=wait_exponential(multiplier=1, min=2, max=30), # Đợi 2-30 giây, tăng theo cấp số nhân
retry=retry_if_exception_type((requests.exceptions.Timeout,
requests.exceptions.ConnectionError,
requests.exceptions.HTTPError)),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=True
)
def chat_completion(self, messages: list, model: str = "deepseek-chat") -> dict:
"""Gọi API chat completion với retry tự động"""
try:
payload = {
"model": model,
"messages": messages,
"temperature": 0.7,
"max_tokens": 1000
}
return self._make_request("chat/completions", payload)
except requests.exceptions.HTTPError as e:
# Không retry nếu là lỗi 4xx (lỗi client)
if 400 <= e.response.status_code < 500:
raise
# Retry nếu là lỗi 5xx (lỗi server)
raise
Cách sử dụng
client = HolySheepAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
messages = [{"role": "user", "content": "Xin chào, hãy giới thiệu về HolySheep AI"}]
try:
response = client.chat_completion(messages, model="deepseek-chat")
print(f"Kết quả: {response['choices'][0]['message']['content']}")
except Exception as e:
print(f"Lỗi sau khi retry 5 lần: {e}")Custom Retry Hook — Ghi log và theo dõi chi phí
Điều mình thích nhất ở tenacity là khả năng tùy chỉnh hooks. Bạn có thể gắn hành động vào bất kỳ giai đoạn nào: trước khi retry, sau khi thất bại, hoặc khi thành công. Mình thường dùng tính năng này để theo dõi chi phí API — mỗi lần gọi thành công, mình sẽ log số tokens đã sử dụng để tính toán chi phí thực tế.
from tenacity import (
retry, stop_after_attempt, wait_exponential,
retry_if_exception_type, after
)
from dataclasses import dataclass
from typing import Optional
import logging
logger = logging.getLogger(__name__)
Định nghĩa chi phí cho từng model (USD per 1M tokens)
MODEL_COSTS = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-chat": 0.42, # Siêu rẻ với HolySheep
}
@dataclass
class UsageTracker:
"""Theo dõi sử dụng và chi phí API"""
total_input_tokens: int = 0
total_output_tokens: int = 0
total_cost_usd: float = 0.0
retry_count: int = 0
last_error: Optional[str] = None
class SmartRetryClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.usage = UsageTracker()
def _calculate_cost(self, model: str, usage: dict) -> float:
"""Tính chi phí dựa trên số tokens"""
input_cost = (usage.get('prompt_tokens', 0) / 1_000_000) * MODEL_COSTS.get(model, 1)
output_cost = (usage.get('completion_tokens', 0) / 1_000_000) * MODEL_COSTS.get(model, 1)
return input_cost + output_cost
def _on_retry(self, attempt_state):
"""Hook được gọi mỗi khi retry xảy ra"""
self.usage.retry_count += 1
self.usage.last_error = str(attempt_state.outcome.exception())
logger.warning(
f"🔄 Retry lần {self.usage.retry_count}: "
f"Lỗi: {self.usage.last_error}"
)
def _on_success(self, attempt_state):
"""Hook được gọi khi request thành công"""
logger.info(f"✅ Request thành công sau {self.usage.retry_count} lần retry")
# Log chi phí
if hasattr(attempt_state.result, 'usage'):
usage = attempt_state.result.usage
cost = self._calculate_cost(attempt_state.kwargs.get('model', 'deepseek-chat'), usage)
self.usage.total_input_tokens += usage.get('prompt_tokens', 0)
self.usage.total_output_tokens += usage.get('completion_tokens', 0)
self.usage.total_cost_usd += cost
logger.info(
f"💰 Chi phí lần này: ${cost:.4f} | "
f"Tổng cộng: ${self.usage.total_cost_usd:.2f}"
)
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=2, max=30),
retry=retry_if_exception_type((requests.exceptions.Timeout,
requests.exceptions.ConnectionError)),
after=_on_retry,
after=_on_success
)
def call_api(self, model: str, messages: list) -> dict:
"""Gọi API với tracking chi phí"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": 0.7
}
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
response.raise_for_status()
return response.json()
Sử dụng với tracking chi phí
client = SmartRetryClient(api_key="YOUR_HOLYSHEEP_API_KEY")
DeepSeek V3.2 chỉ $0.42/MTok - tiết kiệm 85%+
response = client.call_api("deepseek-chat", [
{"role": "user", "content": "Giải thích retry mechanism"}
])
print(f"Tổng chi phí: ${client.usage.total_cost_usd:.4f}")
print(f"Tổng tokens: {client.usage.total_input_tokens + client.usage.total_output_tokens:,}")Circuit Breaker — Ngăn chặn thảm họa
Retry rất tốt, nhưng nếu API server bị down hoàn toàn trong 30 phút? Nếu bạn cứ retry liên tục, bạn sẽ tiêu tốn hàng triệu tokens và có thể bị banned. Đó là lý do cần Circuit Breaker — một "cầu dao" sẽ tự động ngắt khi API thất bại quá nhiều lần liên tiếp, cho server "nghỉ ngơi" và tránh tốn chi phí không cần thiết.
import time
from enum import Enum
from threading import Lock
from typing import Callable, Any
import requests
class CircuitState(Enum):
CLOSED = "closed" # Bình thường, cho phép request
OPEN = "open" # Đang bảo trì, chặn request
HALF_OPEN = "half_open" # Thử lại một lần
class CircuitBreaker:
"""
Circuit Breaker đơn giản với 3 trạng thái:
- CLOSED: Hoạt động bình thường
- OPEN: Quá nhiều lỗi, chặn request tạm thời
- HALF_OPEN: Thử cho 1 request để kiểm tra
"""
def __init__(
self,
failure_threshold: int = 5, # Mở cầu dao sau 5 lỗi liên tiếp
recovery_timeout: int = 60, # Chờ 60 giây trước khi thử lại
expected_exception: type = Exception
):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.expected_exception = expected_exception
self.failure_count = 0
self.last_failure_time = None
self.state = CircuitState.CLOSED
self._lock = Lock()
def call(self, func: Callable, *args, **kwargs) -> Any:
"""Thực thi hàm với circuit breaker protection"""
with self._lock:
if self.state == CircuitState.OPEN:
# Kiểm tra đã đủ thời gian chưa
if time.time() - self.last_failure_time >= self.recovery_timeout:
self.state = CircuitState.HALF_OPEN
print("🔔 Circuit: OPEN → HALF_OPEN (thử nghiệm)")
else:
raise Exception(
f"Circuit Breaker đang OPEN. "
f"Thử lại sau {self.recovery_timeout - (time.time() - self.last_failure_time):.0f}s"
)
try:
result = func(*args, **kwargs)
self._on_success()
return result
except self.expected_exception as e:
self._on_failure()
raise
def _on_success(self):
"""Xử lý khi request thành công"""
with self._lock:
if self.state == CircuitState.HALF_OPEN:
print("✅ Circuit: HALF_OPEN → CLOSED (khôi phục)")
self.failure_count = 0
self.state = CircuitState.CLOSED
def _on_failure(self):
"""Xử lý khi request thất bại"""
with self._lock:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN
print(f"⚠️ Circuit: CLOSED → OPEN (quá nhiều lỗi)")
class HolySheepCircuitBreakerClient:
"""Client kết hợp Retry + Circuit Breaker"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
# Khởi tạo circuit breaker với ngưỡng 3 lỗi, chờ 30 giây
self.circuit_breaker = CircuitBreaker(
failure_threshold=3,
recovery_timeout=30
)
def _do_request(self, messages: list, model: str) -> dict:
"""Thực hiện request thực tế"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": 0.7
}
# Thử 3 lần với exponential backoff
for attempt in range(3):
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
# Retry nếu server error (5xx)
if 500 <= response.status_code < 600:
wait_time = 2 ** attempt
print(f"🔄 Server error {response.status_code}, đợi {wait_time}s...")
time.sleep(wait_time)
continue
response.raise_for_status()
return response.json()
except (requests.exceptions.Timeout,
requests.exceptions.ConnectionError,
requests.exceptions.HTTPError) as e:
if attempt == 2:
raise
wait_time = 2 ** attempt
print(f"🔄 Lỗi kết nối, đợi {wait_time}s...")
time.sleep(wait_time)
raise Exception("Request thất bại sau 3 lần thử")
def chat(self, messages: list, model: str = "deepseek-chat") -> dict:
"""Gọi API an toàn với circuit breaker protection"""
try:
return self.circuit_breaker.call(self._do_request, messages, model)
except Exception as e:
print(f"❌ Yêu cầu bị chặn bởi Circuit Breaker: {e}")
raise
Demo sử dụng
client = HolySheepCircuitBreakerClient(api_key="YOUR_HOLYSHEEP_API_KEY")
Khi server ổn định
try:
response = client.chat([
{"role": "user", "content": "So sánh chi phí HolySheep vs OpenAI"}
])
print(f"Kết quả: {response['choices'][0]['message']['content']}")
except Exception as e:
print(f"Lỗi: {e}")Kết hợp Retry + Circuit Breaker hoàn chỉnh
Sau nhiều lần đau đớn vì API không ổn định, mình đã xây dựng một class kết hợp cả hai kỹ thuật. Mình cũng thêm vào tính năng fallback — nếu model chính thất bại, tự động chuyển sang model dự phòng rẻ hơn như DeepSeek V3.2 với chi phí chỉ $0.42/MTok.
import time
import logging
from threading import Lock
from typing import Optional, List, Callable
import requests
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class CircuitState:
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
class ResilientAIClient:
"""
Client AI API với:
- Retry thông minh (exponential backoff)
- Circuit Breaker (ngăn chặn cascading failures)
- Fallback model (tự động chuyển model khi lỗi)
- Rate limiting (giới hạn request/giây)
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
# Circuit Breaker settings
self.circuit_state = CircuitState.CLOSED
self.failure_count = 0
self.failure_threshold = 5
self.recovery_timeout = 60
self.last_failure_time = 0
self._lock = Lock()
# Models theo thứ tự ưu tiên (rẻ nhất đến đắt nhất)
self.models = ["deepseek-chat", "gemini-2.5-flash", "claude-sonnet-4.5"]
self.current_model_index = 0
@property
def current_model(self) -> str:
return self.models[self.current_model_index]
def _check_circuit(self):
"""Kiểm tra trạng thái circuit breaker"""
with self._lock:
if self.circuit_state == CircuitState.OPEN:
if time.time() - self.last_failure_time >= self.recovery_timeout:
self.circuit_state = CircuitState.HALF_OPEN
logger.info("Circuit: OPEN → HALF_OPEN")
else:
remaining = self.recovery_timeout - (time.time() - self.last_failure_time)
raise Exception(f"Circuit breaker OPEN. Thử lại sau {remaining:.0f}s")
def _record_success(self):
"""Ghi nhận thành công"""
with self._lock:
self.failure_count = 0
if self.circuit_state == CircuitState.HALF_OPEN:
logger.info("Circuit: HALF_OPEN → CLOSED")
self.circuit_state = CircuitState.CLOSED
self.current_model_index = 0 # Reset về model rẻ nhất
def _record_failure(self):
"""Ghi nhận thất bại"""
with self._lock:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.circuit_state = CircuitState.OPEN
logger.warning(f"Circuit: CLOSED → OPEN (sau {self.failure_count} lỗi)")
# Fallback sang model rẻ hơn
if self.current_model_index < len(self.models) - 1:
self.current_model_index += 1
logger.info(f"Fallback sang model: {self.current_model}")
def _execute_with_retry(self, payload: dict, max_retries: int = 3) -> dict:
"""Thực thi request với retry logic"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
last_error = None
for attempt in range(max_retries):
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
# Retry server errors
if 500 <= response.status_code < 600:
wait_time = 2 ** attempt
logger.warning(f"Server error {response.status_code}, retry sau {wait_time}s...")
time.sleep(wait_time)
continue
response.raise_for_status()
return response.json()
except (requests.exceptions.Timeout,
requests.exceptions.ConnectionError) as e:
last_error = e
wait_time = 2 ** attempt
logger.warning(f"Kết nối lỗi (lần {attempt + 1}), đợi {wait_time}s...")
time.sleep(wait_time)
raise last_error or Exception("Request thất bại")
def chat(self, messages: list, system_prompt: str = None) -> dict:
"""Gọi API với full resilience stack"""
self._check_circuit()
# Xây dựng payload với model hiện tại
if system_prompt:
full_messages = [{"role": "system", "content": system_prompt}] + messages
else:
full_messages = messages
payload = {
"model": self.current_model,
"messages": full_messages,
"temperature": 0.7,
"max_tokens": 2000
}
try:
result = self._execute_with_retry(payload)
self._record_success()
logger.info(
f"✅ Thành công với model {self.current_model} | "
f"Tokens: {result.get('usage', {}).get('total_tokens', 'N/A')}"
)
return result
except Exception as e:
self._record_failure()
logger.error(f"❌ Thất bại với model {self.current_model}: {e}")
# Thử model fallback nếu có
if self.current_model_index < len(self.models) - 1:
self.current_model_index += 1
logger.info(f"Thử model fallback: {self.current_model}")
return self.chat(messages, system_prompt)
raise Exception(f"Tất cả models đều thất bại: {e}")
============== DEMO ==============
if __name__ == "__main__":
client = ResilientAIClient(api_key="YOUR_HOLYSHEEP_API_KEY")
# Test với fallback tự động
try:
response = client.chat(
messages=[{"role": "user", "content": "Hello, AI API là gì?"}],
system_prompt="Bạn là trợ lý AI thân thiện, trả lời ngắn gọn."
)
content = response['choices'][0]['message']['content']
model = response.get('model', 'unknown')
tokens = response.get('usage', {}).get('total_tokens', 0)
print(f"\n📝 Model: {model}")
print(f"📊 Tokens: {tokens}")
print(f"💬 Response: {content}")
except Exception as e:
print(f"\n🚫 Lỗi nghiêm trọng: {e}")Best practices từ kinh nghiệm thực chiến
Qua 2 năm làm việc với AI API, đây là những bài học mà mình "đổ máu" mới có được:
- Luôn có fallback model: DeepSeek V3.2 chỉ $0.42/MTok — rẻ hơn 95% so với GPT-4. Mình luôn cài đặt thứ tự fallback: DeepSeek → Gemini Flash → Claude → GPT-4. Khi model đắt nhất fail, hệ thống tự động chuyển sang model rẻ hơn.
- Retry nhưng có giới hạn: Không retry vô hạn. Mình thường giới hạn 3-5 lần với exponential backoff (1s, 2s, 4s, 8s...). Nếu API down quá lâu, bạn sẽ chỉ tốn tiền oan.
- Đừng retry lỗi 4xx: Lỗi 400-499 là lỗi phía client (sai request, hết credit, invalid API key...). Retry không giải quyết được vấn đề, chỉ tốn thêm chi phí.
- Track chi phí theo thời gian thực: Với HolySheep, mình tiết kiệm được 85%+ chi phí so với OpenAI. Nhưng vẫn cần theo dõi sát sao. Mình log mỗi request thành công kèm số tokens và chi phí.
- Timeout hợp lý: 30 giây là con số mình hay dùng. API AI có thể mất 10-20 giây để xử lý request phức tạp, nhưng nếu lâu hơn thì có vấn đề.
Lỗi thường gặp và cách khắc phục
1. Lỗi "Connection timeout" liên tục
Nguyên nhân: Server API quá tải hoặc network không ổn định.
# Cách khắc phục: Tăng timeout và thêm retry với exponential backoff
import requests
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(5),
wait=wait_exponential(multiplier=1, min=4, max=60) # Đợi 4-60 giây
)
def call_api_with_longer_timeout():
response = requests.post(
"https://api.holysheep.ai/v1/chat/completions",
headers={"Authorization": f"Bearer {api_key}"},
json={"model": "deepseek-chat", "messages": messages},
timeout=60 # Tăng timeout lên 60 giây
)
return response.json()2. Lỗi "401 Unauthorized" hoặc "403 Forbidden"
Nguyên nhân: API key không đúng, hết credit, hoặc chưa kích hoạt API.
# Cách khắc phục: Kiểm tra API key và balance
import requests
def check_api_health(api_key: str):
"""Kiểm tra tài khoản trước khi gọi API"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Thử gọi endpoint kiểm tra credit
try:
response = requests.get(
"https://api.holysheep.ai/v1/usage",
headers=headers,
timeout=10
)
if response.status_code == 200:
data = response.json()
print(f"Tài khoản: {data.get('balance', 'N/A')} credit còn lại")
return True
elif response.status_code == 401:
print("❌ API key không hợp lệ")
return False
elif response.status_code == 403:
print("❌ Hết credit - vui lòng nạp thêm")
return False
except Exception as e:
print(f"Lỗi kiểm tra: {e}")
return False
Kiểm tra trước khi gọi chính
if check_api_health("YOUR_HOLYSHEEP_API_KEY"):
response = call_main_api()
else:
print("Không thể tiếp tục - vui lòng kiểm tra tài khoản")3. Lỗi "429 Too Many Requests" (Rate Limit)
Nguyên nhân: Gọi API quá nhanh, vượt quá giới hạn request/giây cho phép.
# Cách khắc phục: Thêm rate limiting và backoff thông minh
import time
import threading
from collections import deque
class RateLimiter:
"""Giới hạn số request trên giây"""
def __init__(self, max_requests: int = 10, time_window: int = 60):
self.max_requests = max_requests
self.time_window = time_window
self.requests = deque()
self._lock = threading.Lock()
def wait_if_needed(self):
"""Chờ nếu vượt rate limit"""
with self._lock:
now = time.time()
# Xóa request cũ khỏi window
while self.requests and self.requests[0] < now - self.time_window:
self.requests.popleft()
if len(self.requests) >= self.max_requests:
# Tính thời gian chờ
wait_time = self.requests[0] - (now - self.time_window)
print(f"⏳ Rate limit, chờ {wait_time:.1f}s...")
time.sleep(wait_time)
self.requests.append(now)
Sử dụng rate limiter
limiter = RateLimiter(max_requests=50, time_window=60) # 50 req/phút
def call_api_rate_limited(payload):
limiter.wait_if_needed()
response = requests.post(
"