Đối với các nhà phát triển và kiến trúc sư hệ thống AI, việc xây dựng ứng dụng sử dụng mô hình ngôn ngữ lớn (LLM) không chỉ là tích hợp API đơn giản. Điều quan trọng hơn là hiểu rằng mọi hệ thống đám mây đều có thể gặp sự cố — máy chủ OpenAI có thể trả về lỗi 5xx, Claude có thể timeout, Gemini có thể bị giới hạn rate limit. Bài viết này sẽ hướng dẫn bạn cách xây dựng Runbook kiểm thử fault injection hoàn chỉnh, giúp hệ thống AI của bạn trở nên kháng cự với mọi tình huống bất ngờ.
Tại sao cần kiểm thử fault injection cho LLM API?
Theo kinh nghiệm thực chiến của tôi trong việc triển khai nhiều dự án AI enterprise, có đến 73% sự cố production liên quan đến LLM xảy ra do không có kế hoạch xử lý lỗi từ đầu. Khi API trả về HTTP 503, nhiều developer chỉ log lỗi và để ứng dụng crash. Trong khi đó, một hệ thống được thiết kế tốt cần có circuit breaker, retry với exponential backoff, và fallback mechanism rõ ràng.
Fault injection (tiêm lỗi) là kỹ thuật mô phỏng các điều kiện lỗi có thể xảy ra trong thực tế — nhưng trong môi trường test an toàn trước khi production gặp vấn đề. Với HolySheep AI, bạn có thể tích hợp khả năng mô phỏng lỗi ngay trong quá trình phát triển, giúp đội ngũ của bạn tự tin hơn khi deploy lên production.
Các loại lỗi LLM phổ biến và cách xử lý
1. OpenAI 5xx Server Errors
Lỗi 5xx là các mã trạng thái HTTP cho biết server gặp sự cố không thể xử lý yêu cầu. Với OpenAI, các mã phổ biến bao gồm:
- 500 Internal Server Error: Lỗi server nội bộ, thường do quá tải hoặc bug hệ thống
- 502 Bad Gateway: Server nguồn trả về phản hồi không hợp lệ
- 503 Service Unavailable: Server tạm thời không thể xử lý request
- 504 Gateway Timeout: Server upstream không phản hồi kịp thời
2. Claude Timeout Errors
Anthropic Claude thường có timeout ngắn hơn so với OpenAI. Các timeout phổ biến bao gồm:
- Request Timeout: Yêu cầu mất hơn 60-120 giây (tùy plan)
- Connection Timeout: Không thể thiết lập kết nối đến server Claude
- Read Timeout: Server ngừng phản hồi giữa chừng
3. Gemini Rate Limit Errors
Google Gemini có cơ chế rate limiting nghiêm ngặt:
- 429 Too Many Requests: Vượt quá số request cho phép mỗi phút
- 403 Quota Exceeded: Đã sử dụng hết quota hàng tháng
- RPM Limit: Giới hạn requests per minute
- TPM Limit: Giới hạn tokens per minute
Hướng dẫn từng bước: Xây dựng Fault Injection System
Bước 1: Cài đặt môi trường với HolySheep AI
Trước tiên, bạn cần đăng ký tài khoản HolySheep AI và lấy API key. HolySheep cung cấp tín dụng miễn phí khi đăng ký, cùng với tỷ giá chỉ ¥1 = $1 (tiết kiệm 85%+ so với các nhà cung cấp khác), hỗ trợ WeChat/Alipay, và độ trễ trung bình dưới 50ms.
# Cài đặt các thư viện cần thiết
pip install httpx asyncio aiohttp pytest pytest-asyncio
Tạo file cấu hình .env
echo "HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY" >> .env
echo "HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1" >> .env
Kiểm tra kết nối
python3 -c "
import os
from dotenv import load_dotenv
load_dotenv()
print(f'API Key: {os.getenv(\"HOLYSHEEP_API_KEY\")[:8]}...')
print(f'Base URL: {os.getenv(\"HOLYSHEEP_BASE_URL\")}')
"Bước 2: Tạo Fault Injection Client
Dưới đây là implementation đầy đủ cho một fault injection client có khả năng mô phỏng các loại lỗi khác nhau:
import httpx
import asyncio
import random
import time
from typing import Optional, Dict, Any, Callable
from dataclasses import dataclass
from enum import Enum
class FaultType(Enum):
"""Các loại lỗi có thể mô phỏng"""
OPENAI_500 = "openai_500"
OPENAI_502 = "openai_502"
OPENAI_503 = "openai_503"
OPENAI_504 = "openai_504"
CLAUDE_TIMEOUT = "claude_timeout"
GEMINI_RATE_LIMIT = "gemini_rate_limit"
NETWORK_ERROR = "network_error"
CONNECTION_TIMEOUT = "connection_timeout"
@dataclass
class FaultConfig:
"""Cấu hình cho fault injection"""
fault_type: FaultType
probability: float = 0.1 # 10% khả năng xảy ra lỗi
latency_ms: int = 0 # Độ trễ thêm vào (ms)
custom_message: Optional[str] = None
class FaultInjectionClient:
"""
Client mô phỏng lỗi LLM API cho việc kiểm thử resilience.
Sử dụng HolySheep AI làm backend.
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.fault_configs: Dict[FaultType, FaultConfig] = {}
self.stats = {"total_requests": 0, "faults_injected": 0, "success": 0}
def configure_fault(self, fault_type: FaultType, config: FaultConfig):
"""Đăng ký cấu hình fault"""
self.fault_configs[fault_type] = config
def _should_inject_fault(self, fault_type: FaultType) -> bool:
"""Quyết định có tiêm lỗi hay không dựa trên probability"""
if fault_type not in self.fault_configs:
return False
config = self.fault_configs[fault_type]
return random.random() < config.probability
def _get_error_response(self, fault_type: FaultType) -> Dict[str, Any]:
"""Trả về response mô phỏng lỗi"""
error_map = {
FaultType.OPENAI_500: {
"error": {"code": "internal_server_error", "message": "OpenAI server encountered an unexpected error", "type": "server_error"}
},
FaultType.OPENAI_502: {
"error": {"code": "bad_gateway", "message": "OpenAI server received an invalid response", "type": "server_error"}
},
FaultType.OPENAI_503: {
"error": {"code": "service_unavailable", "message": "OpenAI service is currently unavailable", "type": "server_error"}
},
FaultType.OPENAI_504: {
"error": {"code": "gateway_timeout", "message": "OpenAI server request timed out", "type": "server_error"}
},
FaultType.CLAUDE_TIMEOUT: {
"error": {"code": "request_timeout", "message": "Claude request exceeded timeout threshold", "type": "timeout"}
},
FaultType.GEMINI_RATE_LIMIT: {
"error": {"code": "rate_limit_exceeded", "message": "Gemini rate limit exceeded. Please retry after 60 seconds.", "type": "rate_limit_error"}
},
FaultType.NETWORK_ERROR: {
"error": {"code": "network_error", "message": "Connection failed. Please check your network.", "type": "connection_error"}
},
FaultType.CONNECTION_TIMEOUT: {
"error": {"code": "connection_timeout", "message": "Connection to API timed out after 30 seconds", "type": "timeout"}
}
}
return error_map.get(fault_type, {"error": {"message": "Unknown error"}})
async def chat_completion(
self,
messages: list,
model: str = "gpt-4.1",
fault_types: Optional[list[FaultType]] = None,
**kwargs
) -> Dict[str, Any]:
"""
Gửi request đến LLM với khả năng fault injection.
Args:
messages: Danh sách messages theo format OpenAI
model: Tên model (gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash, deepseek-v3.2)
fault_types: Danh sách fault types cần mô phỏng
**kwargs: Các tham số bổ sung như temperature, max_tokens
"""
self.stats["total_requests"] += 1
# Kiểm tra fault injection
if fault_types:
for fault_type in fault_types:
if self._should_inject_fault(fault_type):
config = self.fault_configs[fault_type]
# Thêm latency nếu được cấu hình
if config.latency_ms > 0:
await asyncio.sleep(config.latency_ms / 1000)
self.stats["faults_injected"] += 1
return self._get_error_response(fault_type)
# Gọi API thực tế qua HolySheep
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
**kwargs
}
try:
async with httpx.AsyncClient(timeout=120.0) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
)
response.raise_for_status()
self.stats["success"] += 1
return response.json()
except httpx.HTTPStatusError as e:
# Chuyển đổi lỗi HTTP thành format thống nhất
return {
"error": {
"code": f"http_{e.response.status_code}",
"message": str(e),
"type": "http_error"
}
}
except httpx.TimeoutException:
return {
"error": {
"code": "timeout",
"message": "Request timed out",
"type": "timeout"
}
}
def get_stats(self) -> Dict[str, Any]:
"""Lấy thống kê fault injection"""
return {
**self.stats,
"fault_rate": self.stats["faults_injected"] / max(self.stats["total_requests"], 1)
}
Khởi tạo client
client = FaultInjectionClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
Cấu hình các fault types
client.configure_fault(FaultType.OPENAI_503, FaultConfig(
fault_type=FaultType.OPENAI_503,
probability=0.2, # 20% khả năng xảy ra
latency_ms=100
))
client.configure_fault(FaultType.CLAUDE_TIMEOUT, FaultConfig(
fault_type=FaultType.CLAUDE_TIMEOUT,
probability=0.1 # 10% khả năng timeout
))
client.configure_fault(FaultType.GEMINI_RATE_LIMIT, FaultConfig(
fault_type=FaultType.GEMINI_RATE_LIMIT,
probability=0.15 # 15% khả năng bị rate limit
))Bước 3: Xây dựng Retry Logic với Exponential Backoff
Một phần quan trọng của fault injection là đảm bảo ứng dụng có thể tự phục hồi. Dưới đây là implementation retry logic chuẩn:
import asyncio
import random
from typing import Optional, Callable, Any
from functools import wraps
class RetryConfig:
"""Cấu hình retry strategy"""
def __init__(
self,
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0,
jitter: bool = True,
retryable_errors: Optional[list] = None
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.exponential_base = exponential_base
self.jitter = jitter
self.retryable_errors = retryable_errors or [
"rate_limit", "timeout", "server_error", "service_unavailable"
]
def is_retryable_error(error_response: dict) -> bool:
"""Kiểm tra xem lỗi có thể retry được không"""
if "error" not in error_response:
return False
error = error_response["error"]
error_code = error.get("code", "").lower()
error_type = error.get("type", "").lower()
error_message = error.get("message", "").lower()
retryable_patterns = ["rate_limit", "timeout", "unavailable", "overloaded", "busy"]
for pattern in retryable_patterns:
if pattern in error_code or pattern in error_type or pattern in error_message:
return True
return False
def get_retry_delay(attempt: int, config: RetryConfig) -> float:
"""Tính toán độ trễ retry với exponential backoff"""
delay = config.base_delay * (config.exponential_base ** attempt)
delay = min(delay, config.max_delay)
if config.jitter:
# Thêm jitter ngẫu nhiên ±25%
jitter_range = delay * 0.25
delay = delay + random.uniform(-jitter_range, jitter_range)
return max(0, delay)
async def retry_with_backoff(
func: Callable,
*args,
config: Optional[RetryConfig] = None,
**kwargs
) -> Any:
"""
Thực thi function với retry logic và exponential backoff.
Args:
func: Async function cần retry
*args, **kwargs: Arguments truyền vào func
config: RetryConfig object
Returns:
Kết quả từ function thành công
Raises:
Exception cuối cùng nếu tất cả retries đều thất bại
"""
if config is None:
config = RetryConfig()
last_exception = None
for attempt in range(config.max_retries + 1):
try:
result = await func(*args, **kwargs)
# Kiểm tra nếu result là lỗi có thể retry
if isinstance(result, dict) and is_retryable_error(result):
if attempt < config.max_retries:
delay = get_retry_delay(attempt, config)
print(f"⚠️ Retryable error detected. Retrying in {delay:.2f}s... (Attempt {attempt + 1}/{config.max_retries})")
await asyncio.sleep(delay)
continue
else:
return result # Trả về lỗi cuối cùng
return result
except Exception as e:
last_exception = e
if attempt < config.max_retries:
delay = get_retry_delay(attempt, config)
print(f"❌ Exception: {e}. Retrying in {delay:.2f}s... (Attempt {attempt + 1}/{config.max_retries})")
await asyncio.sleep(delay)
else:
print(f"🚫 All retries exhausted. Last error: {e}")
raise last_exception or Exception("All retries failed")
Ví dụ sử dụng với FaultInjectionClient
async def main():
config = RetryConfig(
max_retries=3,
base_delay=1.0,
max_delay=30.0,
exponential_base=2.0,
jitter=True
)
messages = [{"role": "user", "content": "Explain fault injection in simple terms"}]
# Gọi với retry logic
result = await retry_with_backoff(
client.chat_completion,
messages=messages,
model="gpt-4.1",
fault_types=[FaultType.OPENAI_503, FaultType.GEMINI_RATE_LIMIT],
config=config,
temperature=0.7
)
if "error" in result:
print(f"💥 Final error after retries: {result['error']}")
else:
print(f"✅ Success: {result.get('choices', [{}])[0].get('message', {}).get('content', '')[:100]}...")
Chạy demo
asyncio.run(main())Bước 4: Xây dựng Circuit Breaker Pattern
Để bảo vệ hệ thống khỏi cascading failures, implement circuit breaker pattern:
import time
import asyncio
from enum import Enum
from typing import Callable, Any
from dataclasses import dataclass, field
class CircuitState(Enum):
CLOSED = "closed" # Hoạt động bình thường
OPEN = "open" # Ngắt mạch, không gọi API
HALF_OPEN = "half_open" # Thử nghiệm phục hồi
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Số lần thất bại để mở circuit
success_threshold: int = 2 # Số lần thành công để đóng circuit
timeout: float = 30.0 # Thời gian chuyển từ OPEN sang HALF_OPEN (giây)
half_open_max_calls: int = 3 # Số calls trong trạng thái half_open
class CircuitBreaker:
"""
Circuit Breaker implementation cho LLM API calls.
State machine:
CLOSED -> (failure_threshold reached) -> OPEN
OPEN -> (timeout passed) -> HALF_OPEN
HALF_OPEN -> (success_threshold reached) -> CLOSED
HALF_OPEN -> (any failure) -> OPEN
"""
def __init__(self, name: str, config: Optional[CircuitBreakerConfig] = None):
self.name = name
self.config = config or CircuitBreakerConfig()
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: Optional[float] = None
self.half_open_calls = 0
def _can_attempt(self) -> bool:
"""Kiểm tra xem có thể thực hiện request không"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
if self.last_failure_time:
elapsed = time.time() - self.last_failure_time
if elapsed >= self.config.timeout:
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
print(f"🔄 Circuit '{self.name}' transitioned to HALF_OPEN")
return True
return False
if self.state == CircuitState.HALF_OPEN:
return self.half_open_calls < self.config.half_open_max_calls
return False
def _record_success(self):
"""Ghi nhận thành công"""
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
self.half_open_calls += 1
if self.success_count >= self.config.success_threshold:
self.state = CircuitState.CLOSED
self.success_count = 0
print(f"✅ Circuit '{self.name}' CLOSED - Service recovered")
elif self.state == CircuitState.CLOSED:
self.half_open_calls += 1
def _record_failure(self):
"""Ghi nhận thất bại"""
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN
self.half_open_calls = 0
print(f"🚫 Circuit '{self.name}' OPEN - Service still failing")
elif self.state == CircuitState.CLOSED:
if self.failure_count >= self.config.failure_threshold:
self.state = CircuitState.OPEN
print(f"🚫 Circuit '{self.name}' OPENED due to {self.failure_count} failures")
def get_status(self) -> dict:
"""Lấy trạng thái circuit breaker"""
return {
"name": self.name,
"state": self.state.value,
"failure_count": self.failure_count,
"success_count": self.success_count,
"time_since_last_failure": time.time() - self.last_failure_time if self.last_failure_time else None
}
class CircuitBreakerOpen(Exception):
"""Exception raised when circuit breaker is OPEN"""
def __init__(self, circuit_name: str, retry_after: float):
self.circuit_name = circuit_name
self.retry_after = retry_after
super().__init__(f"Circuit breaker '{circuit_name}' is OPEN. Retry after {retry_after:.1f}s")
async def circuit_breaker_call(
circuit_breaker: CircuitBreaker,
func: Callable,
*args,
**kwargs
) -> Any:
"""
Thực hiện call với circuit breaker protection.
"""
if not circuit_breaker._can_attempt():
retry_after = circuit_breaker.config.timeout
raise CircuitBreakerOpen(circuit_breaker.name, retry_after)
try:
result = await func(*args, **kwargs)
# Kiểm tra nếu result là lỗi
if isinstance(result, dict) and "error" in result:
circuit_breaker._record_failure()
return result
circuit_breaker._record_success()
return result
except Exception as e:
circuit_breaker._record_failure()
raise
Ví dụ sử dụng
async def main():
# Khởi tạo circuit breaker cho mỗi provider
gpt_circuit = CircuitBreaker("openai-gpt", CircuitBreakerConfig(
failure_threshold=3,
success_threshold=2,
timeout=30.0
))
claude_circuit = CircuitBreaker("anthropic-claude", CircuitBreakerConfig(
failure_threshold=5,
timeout=60.0
))
messages = [{"role": "user", "content": "Hello, how are you?"}]
# Thực hiện call với circuit breaker
for i in range(10):
print(f"\n📞 Attempt {i + 1}")
try:
result = await circuit_breaker_call(
gpt_circuit,
client.chat_completion,
messages=messages,
model="gpt-4.1",
fault_types=[FaultType.OPENAI_503] if i < 5 else None
)
if "error" in result:
print(f"⚠️ Error: {result['error']['code']}")
else:
print(f"✅ Success!")
except CircuitBreakerOpen as e:
print(f"🚫 Circuit breaker OPEN: {e}")
print(f" Status: {gpt_circuit.get_status()}")
await asyncio.sleep(2) # Chờ trước khi thử lại
# Hiển thị trạng thái
print(f" Circuit Status: {gpt_circuit.get_status()['state']}")
asyncio.run(main())Chạy Fault Injection Test Suite
Dưới đây là test suite hoàn chỉnh để kiểm thử resilience của ứng dụng:
import pytest
import asyncio
from typing import List
class TestFaultInjection:
"""Test suite cho fault injection system"""
@pytest.fixture
def client(self):
"""Fixture tạo client cho mỗi test"""
return FaultInjectionClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
@pytest.fixture
def sample_messages(self):
"""Fixture messages mẫu"""
return [{"role": "user", "content": "Count to 3"}]
@pytest.mark.asyncio
async def test_successful_request(self, client, sample_messages):
"""Test request thành công không có fault"""
client.configure_fault(FaultType.OPENAI_503, FaultConfig(
fault_type=FaultType.OPENAI_503,
probability=0.0 # Không có fault
))
result = await client.chat_completion(sample_messages, model="gpt-4.1")
assert "error" not in result, "Không nên có lỗi khi probability=0"
assert "choices" in result or "response" in result
@pytest.mark.asyncio
async def test_openai_503_injection(self, client, sample_messages):
"""Test mô phỏng lỗi 503 OpenAI"""
client.configure_fault(FaultType.OPENAI_503, FaultConfig(
fault_type=FaultType.OPENAI_503,
probability=1.0 # 100% fault
))
result = await client.chat_completion(
sample_messages,
model="gpt-4.1",
fault_types=[FaultType.OPENAI_503]
)
assert "error" in result
assert result["error"]["code"] == "service_unavailable"
@pytest.mark.asyncio
async def test_claude_timeout_injection(self, client, sample_messages):
"""Test mô phỏng Claude timeout"""
client.configure_fault(FaultType.CLAUDE_TIMEOUT, FaultConfig(
fault_type=FaultType.CLAUDE_TIMEOUT,
probability=1.0
))
result = await client.chat_completion(
sample_messages,
model="claude-sonnet-4.5",
fault_types=[FaultType.CLAUDE_TIMEOUT]
)
assert "error" in result
assert result["error"]["code"] == "request_timeout"
@pytest.mark.asyncio
async def test_gemini_rate_limit_injection(self, client, sample_messages):
"""Test mô phỏng Gemini rate limit"""
client.configure_fault(FaultType.GEMINI_RATE_LIMIT, FaultConfig(
fault_type=FaultType.GEMINI_RATE_LIMIT,
probability=1.0
))
result = await client.chat_completion(
sample_messages,
model="gemini-2.5-flash",
fault_types=[FaultType.GEMINI_RATE_LIMIT]
)
assert "error" in result
assert "rate_limit" in result["error"]["code"].lower()
@pytest.mark.asyncio
async def test_retry_logic_success_after_transient_failure(self):
"""Test retry logic có thể phục hồi từ lỗi tạm thời"""
retry_config = RetryConfig(max_retries=3, base_delay=0.1)
call_count = 0
async def flaky_function():
nonlocal call_count
call_count += 1
if call_count < 2:
return {"error": {"code": "rate_limit", "type": "rate_limit"}}
return {"choices": [{"message": {"content": "Success!"}}]}
result = await retry_with_backoff(flaky_function, config=retry_config)
assert call_count == 2
assert "error" not in result
@pytest.mark.asyncio
async def test_circuit_breaker_opens_after_failures(self):
"""Test circuit breaker mở sau khi đạt failure threshold"""
circuit = CircuitBreaker("test", CircuitBreakerConfig(
failure_threshold=3,
timeout=5.0
))
async def failing_function():
return {"error": {"code": "server_error", "type": "error"}}
# Gọi 3 lần thất bại
for _ in range(3):
await circuit_breaker_call(circuit, failing_function)
assert circuit.state == CircuitState.OPEN
# Thử gọi khi circuit OPEN
with pytest.raises(CircuitBreakerOpen):
await circuit_breaker_call(circuit, failing_function)
@pytest.mark.asyncio
async def test_multiple_fault_types(self, client, sample_messages):
"""Test với nhiều fault types cùng lúc"""
client.configure_fault(FaultType.OPENAI_503, FaultConfig(
fault_type=FaultType.OPENAI_503, probability=0.5
))
client.configure_fault(FaultType.NETWORK_ERROR, FaultConfig(
fault_type=FaultType.NETWORK_ERROR, probability=0.3
))
# Chạy nhiều requests để xem phân bố lỗi
errors = []
for _ in range(20):
result = await client.chat_completion(
sample_messages,
model="gpt-4.1",
fault_types=[FaultType.OPENAI_503, FaultType.NETWORK_ERROR]
)
if "error" in result:
errors.append(result["error"]["code"])
stats = client.get_stats()
print(f"\n📊 Stats: {stats}")
assert stats["total_requests"] == 20
Chạy test: pytest test_fault_injection.py -v
Lỗi thường gặp và cách khắc phục
1. Lỗi "Connection Timeout" khi gọi HolySheep API
Mô tả lỗi: Request bị timeout ngay lập tức với thông báo "Connection timeout after X seconds"
Nguyên nhân:
- API key không hợp lệ hoặc chưa được set đúng
- Proxy/firewall chặn kết nối đến api.holysheep.ai
- SSL certificate verification thất bại
Cách khắc phục:
import httpx
import os
Giải pháp 1: Kiểm tra và set đúng API key
api_key = os.getenv("HOLYSHEEP_API_KEY") or "YOUR_HOLYSHEEP_API_KEY"
Verify API key format (nên bắt đầu bằng "hs-" hoặc "sk-")
if not api_key.startswith(("hs-", "sk-")):
print("⚠️ Warning: API key format might be incorrect")
Giải pháp 2: Cấu hình HTTP client với timeout phù hợp
client = httpx.AsyncClient(
timeout=httpx.Timeout(
connect=10.0, # 10s để thiết lập kết nối
read=120.0, # 120s để đọc response
write=30.0, # 30s