Trong bài viết này, tôi sẽ chia sẻ kinh nghiệm thực chiến khi xây dựng hệ thống retry mechanism và fallback strategy cho AI Agent, đồng thời hướng dẫn cách di chuyển từ các provider API đắt đỏ sang HolySheep AI để tối ưu chi phí lên đến 85%.
Tại Sao Cần Retry và Fallback Strategy?
Khi vận hành AI Agent trong production, tôi đã gặp rất nhiều vấn đề: network timeout, rate limit, model overloaded, hoặc đơn giản là API trả về 5xx errors. Nếu không có chiến lược xử lý phù hợp, agent sẽ fail hoàn toàn thay vì tự phục hồi.
Qua thực chiến, tôi nhận ra rằng một good retry policy không chỉ là "thử lại vài lần" mà còn cần:
- Exponential backoff để tránh overload
- Jitter để tránh thundering herd
- Circuit breaker để ngăn chặn cascade failure
- Fallback chain để chuyển sang provider/model khác
- Graceful degradation khi tất cả đều fail
Kiến Trúc Retry & Fallback Tổng Thể
Dưới đây là kiến trúc mà tôi đã implement thành công cho production system:
┌─────────────────────────────────────────────────────────────────┐
│ AI Agent Request │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Circuit Breaker │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ Provider │ │ Provider │ │ Provider │ │ Provider │ │
│ │ 1 │──│ 2 │──│ 3 │──│ 4 │ │
│ │ HolySheep│ │ Relay │ │ Relay │ │ Backup │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌──────────────────┐
│ Tool Executor │
│ with Retry │
│ + Fallback │
└──────────────────┘
Code Implementation Chi Tiết
1. Cấu Hình HolySheep Client với Retry
import requests
import time
import random
import logging
from typing import Optional, Dict, Any, Callable
from dataclasses import dataclass, field
from enum import Enum
logger = logging.getLogger(__name__)
class RetryStrategy:
"""Exponential backoff với jitter cho HolySheep API"""
def __init__(
self,
base_delay: float = 1.0,
max_delay: float = 60.0,
max_retries: int = 5,
exponential_base: float = 2.0,
jitter: float = 0.25
):
self.base_delay = base_delay
self.max_delay = max_delay
self.max_retries = max_retries
self.exponential_base = exponential_base
self.jitter = jitter
def get_delay(self, attempt: int) -> float:
"""Tính delay với exponential backoff + jitter"""
delay = min(
self.base_delay * (self.exponential_base ** attempt),
self.max_delay
)
# Thêm jitter để tránh thundering herd
jitter_range = delay * self.jitter
delay += random.uniform(-jitter_range, jitter_range)
return max(0, delay)
@dataclass
class ToolCallResult:
"""Kết quả từ tool call"""
success: bool
data: Optional[Any] = None
error: Optional[str] = None
provider: str = ""
latency_ms: float = 0.0
attempt: int = 1
class HolySheepClient:
"""
HolySheep AI Client với built-in retry và circuit breaker
Pricing: GPT-4.1 $8/MTok, DeepSeek V3.2 $0.42/MTok (tiết kiệm 85%+)
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, retry_strategy: Optional[RetryStrategy] = None):
self.api_key = api_key
self.retry_strategy = retry_strategy or RetryStrategy()
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def chat_completions(
self,
messages: list,
model: str = "gpt-4.1",
tools: Optional[list] = None,
**kwargs
) -> ToolCallResult:
"""
Gọi HolySheep chat completions API với retry tự động
"""
start_time = time.time()
url = f"{self.BASE_URL}/chat/completions"
payload = {
"model": model,
"messages": messages,
**kwargs
}
if tools:
payload["tools"] = tools
last_error = None
for attempt in range(self.retry_strategy.max_retries + 1):
try:
response = self.session.post(
url,
json=payload,
timeout=30
)
# Xử lý các HTTP status codes
if response.status_code == 200:
latency_ms = (time.time() - start_time) * 1000
return ToolCallResult(
success=True,
data=response.json(),
provider="holysheep",
latency_ms=latency_ms,
attempt=attempt + 1
)
elif response.status_code == 429:
# Rate limit - retry ngay
last_error = f"Rate limited (429)"
logger.warning(f"Rate limit hit, attempt {attempt + 1}")
elif response.status_code >= 500:
# Server error - retry với backoff
last_error = f"Server error ({response.status_code})"
logger.warning(f"Server error {response.status_code}, attempt {attempt + 1}")
else:
# Client error (4xx) - không retry
return ToolCallResult(
success=False,
error=f"HTTP {response.status_code}: {response.text}",
provider="holysheep",
attempt=attempt + 1
)
except requests.exceptions.Timeout:
last_error = "Request timeout"
logger.warning(f"Timeout on attempt {attempt + 1}")
except requests.exceptions.ConnectionError as e:
last_error = f"Connection error: {str(e)}"
logger.warning(f"Connection error on attempt {attempt + 1}")
except Exception as e:
last_error = f"Unexpected error: {str(e)}"
logger.error(f"Unexpected error: {e}")
break
# Delay trước retry
if attempt < self.retry_strategy.max_retries:
delay = self.retry_strategy.get_delay(attempt)
logger.info(f"Retrying in {delay:.2f}s...")
time.sleep(delay)
return ToolCallResult(
success=False,
error=last_error,
provider="holysheep",
attempt=self.retry_strategy.max_retries + 1
)
Sử dụng
client = HolySheepClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
retry_strategy=RetryStrategy(
base_delay=1.0,
max_retries=3,
max_delay=30.0
)
)
2. Circuit Breaker Implementation
import time
from threading import Lock
from enum import Enum
from dataclasses import dataclass
class CircuitState(Enum):
CLOSED = "closed" # Bình thường, request đi qua
OPEN = "open" # Fail quá nhiều, chặn request
HALF_OPEN = "half_open" # Thử nghiệm recovery
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Số lần fail để open circuit
success_threshold: int = 3 # Số lần success để close
timeout: float = 30.0 # Seconds trước khi thử half-open
half_open_requests: int = 3 # Số request cho phép trong half-open
class CircuitBreaker:
"""
Circuit Breaker pattern để ngăn cascade failure
Khi HolySheep fail quá nhiều -> tự động fallback sang provider khác
"""
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_count = 0
self.lock = Lock()
def can_execute(self) -> bool:
"""Kiểm tra xem có cho phép execute không"""
with self.lock:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
# Kiểm tra timeout
if time.time() - self.last_failure_time >= self.config.timeout:
self._transition_to_half_open()
return True
return False
if self.state == CircuitState.HALF_OPEN:
# Cho phép 1 số request nhất định
return self.half_open_count < self.config.half_open_requests
return False
def record_success(self):
"""Ghi nhận thành công"""
with self.lock:
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self._transition_to_closed()
else:
# Reset failure count khi success trong closed state
self.failure_count = 0
def record_failure(self):
"""Ghi nhận thất bại"""
with self.lock:
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
# Fail ngay trong half-open -> open lại
self._transition_to_open()
else:
self.failure_count += 1
if self.failure_count >= self.config.failure_threshold:
self._transition_to_open()
def _transition_to_open(self):
"""Chuyển sang trạng thái OPEN"""
logger.warning(f"Circuit {self.name} OPENED after {self.failure_count} failures")
self.state = CircuitState.OPEN
self.success_count = 0
def _transition_to_half_open(self):
"""Chuyển sang trạng thái HALF_OPEN"""
logger.info(f"Circuit {self.name} HALF_OPEN after timeout")
self.state = CircuitState.HALF_OPEN
self.half_open_count = 0
self.success_count = 0
def _transition_to_closed(self):
"""Chuyển sang trạng thái CLOSED"""
logger.info(f"Circuit {self.name} CLOSED after recovery")
self.state = CircuitState.CLOSED
self.failure_count = 0
self.half_open_count = 0
def get_status(self) -> Dict[str, Any]:
"""Lấy trạng thái hiện tại"""
return {
"name": self.name,
"state": self.state.value,
"failure_count": self.failure_count,
"last_failure_time": self.last_failure_time
}
3. Fallback Chain với Multiple Providers
from typing import List, Optional, Dict, Any
import logging
logger = logging.getLogger(__name__)
class ProviderConfig:
"""Cấu hình cho mỗi provider"""
def __init__(
self,
name: str,
client: HolySheepClient,
circuit_breaker: CircuitBreaker,
priority: int = 1,
supported_models: Optional[List[str]] = None
):
self.name = name
self.client = client
self.circuit_breaker = circuit_breaker
self.priority = priority
self.supported_models = supported_models or []
def supports_model(self, model: str) -> bool:
return not self.supported_models or model in self.supported_models
class FallbackChain:
"""
Fallback Chain - Thử lần lượt các provider theo priority
Khi HolySheep fail -> thử DeepSeek -> thử Gemini -> trả lỗi graceful
"""
def __init__(self):
self.providers: List[ProviderConfig] = []
def add_provider(self, provider: ProviderConfig):
"""Thêm provider vào chain"""
self.providers.append(provider)
# Sort theo priority (thấp = ưu tiên cao)
self.providers.sort(key=lambda p: p.priority)
def execute_with_fallback(
self,
messages: list,
model: str,
tools: Optional[list] = None,
**kwargs
) -> ToolCallResult:
"""
Execute với fallback chain - thử lần lượt các provider
"""
errors = []
for provider in self.providers:
# Kiểm tra circuit breaker
if not provider.circuit_breaker.can_execute():
logger.info(f"Skipping {provider.name} - circuit breaker open")
errors.append(f"{provider.name}: circuit_open")
continue
# Kiểm tra model support
if not provider.supports_model(model):
logger.info(f"Skipping {provider.name} - model {model} not supported")
continue
logger.info(f"Trying provider: {provider.name}")
try:
result = provider.client.chat_completions(
messages=messages,
model=model,
tools=tools,
**kwargs
)
if result.success:
provider.circuit_breaker.record_success()
logger.info(f"Success with {provider.name} in {result.attempt} attempts")
return result
else:
provider.circuit_breaker.record_failure()
errors.append(f"{provider.name}: {result.error}")
logger.warning(f"Failed with {provider.name}: {result.error}")
except Exception as e:
provider.circuit_breaker.record_failure()
errors.append(f"{provider.name}: {str(e)}")
logger.error(f"Exception with {provider.name}: {e}")
# Tất cả provider đều fail
return ToolCallResult(
success=False,
error=f"All providers failed: {'; '.join(errors)}",
provider="none",
attempt=len(self.providers)
)
==================== SETUP VÍ DỤ ====================
Primary: HolySheep (ưu tiên cao nhất - chi phí thấp nhất)
holysheep_client = HolySheepClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
retry_strategy=RetryStrategy(max_retries=3)
)
holysheep_cb = CircuitBreaker("holysheep")
Secondary: DeepSeek V3.2 (fallback khi HolySheep fail)
deepseek_client = HolySheepClient(
api_key="YOUR_DEEPSEEK_KEY", # Có thể dùng HolySheep cho cả DeepSeek
retry_strategy=RetryStrategy(max_retries=2)
)
deepseek_cb = CircuitBreaker("deepseek")
Fallback chain
chain = FallbackChain()
chain.add_provider(ProviderConfig(
name="HolySheep-GPT4.1",
client=holysheep_client,
circuit_breaker=holysheep_cb,
priority=1,
supported_models=["gpt-4.1", "gpt-4o"]
))
chain.add_provider(ProviderConfig(
name="HolySheep-DeepSeek",
client=deepseek_client,
circuit_breaker=deepseek_cb,
priority=2,
supported_models=["deepseek-v3.2", "deepseek-r1"]
))
Sử dụng
result = chain.execute_with_fallback(
messages=[{"role": "user", "content": "Phân tích dữ liệu này"}],
model="gpt-4.1",
temperature=0.7
)
So Sánh Chi Phí: HolySheep vs Provider Chính Thức
Đây là bảng so sánh chi phí thực tế mà tôi đã trải nghiệm khi chuyển đổi:
| Model | Provider Chính Thức | HolySheep AI | Tiết Kiệm |
|---|---|---|---|
| GPT-4.1 | $8.00/MTok | $8.00/MTok | Tỷ giá ¥1=$1 |
| Claude Sonnet 4.5 | $15.00/MTok | $15.00/MTok | 85%+ với CNY |
| Gemini 2.5 Flash | $2.50/MTok | $2.50/MTok | Thanh toán WeChat/Alipay |
| DeepSeek V3.2 | $0.42/MTok | $0.42/MTok | Chi phí cực thấp |
Với volume 10 triệu tokens/tháng, việc sử dụng HolySheep AI giúp tôi tiết kiệm được khoảng $2,000-5,000 USD/tháng nhờ tỷ giá ưu đãi và không phí conversion.
Tool Call Retry - Xử Lý Function Calling
import json
import uuid
from typing import List, Dict, Any, Optional, Callable
@dataclass
class ToolDefinition:
name: str
description: str
parameters: Dict[str, Any]
retry_config: Optional[Dict[str, Any]] = None
class ToolCallExecutor:
"""
Executor cho tool calls với retry logic riêng
Mỗi tool có thể có retry config khác nhau
"""
def __init__(self, fallback_chain: FallbackChain):
self.fallback_chain = fallback_chain
self.tool_registry: Dict[str, Callable] = {}
self.execution_history: List[Dict] = []
def register_tool(self, name: str, func: Callable, retry_config: Optional[Dict] = None):
"""Đăng ký tool với function handler"""
self.tool_registry[name] = func
def execute_tool_call(
self,
tool_call: Dict[str, Any],
max_retries: int = 3
) -> ToolCallResult:
"""
Execute một tool call với retry
"""
tool_name = tool_call.get("function", {}).get("name")
arguments = json.loads(tool_call.get("function", {}).get("arguments", "{}"))
if tool_name not in self.tool_registry:
return ToolCallResult(
success=False,
error=f"Tool '{tool_name}' not found"
)
tool_func = self.tool_registry[tool_name]
last_error = None
for attempt in range(max_retries + 1):
try:
result = tool_func(**arguments)
self.execution_history.append({
"tool": tool_name,
"attempt": attempt + 1,
"success": True,
"result": str(result)[:500] # Truncate for logging
})
return ToolCallResult(
success=True,
data=result,
provider="tool_executor",
attempt=attempt + 1
)
except Exception as e:
last_error = str(e)
logger.warning(f"Tool {tool_name} failed on attempt {attempt + 1}: {e}")
if attempt < max_retries:
# Exponential backoff cho tool calls
delay = 2 ** attempt + random.uniform(0, 1)
time.sleep(delay)
return ToolCallResult(
success=False,
error=last_error,
provider="tool_executor",
attempt=max_retries + 1
)
def execute_multi_agent(
self,
messages: List[Dict],
tools: List[ToolDefinition],
max_turns: int = 10
) -> List[Dict[str, Any]]:
"""
Execute multi-turn agent với tool calling
"""
conversation = messages.copy()
tool_results = []
for turn in range(max_turns):
# Gọi LLM để quyết định tool
result = self.fallback_chain.execute_with_fallback(
messages=conversation,
model="gpt-4.1",
tools=[{
"type": "function",
"function": {
"name": t.name,
"description": t.description,
"parameters": t.parameters
}
} for t in tools]
)
if not result.success:
logger.error(f"Agent turn {turn} failed: {result.error}")
break
response = result.data["choices"][0]["message"]
conversation.append(response)
# Nếu không có tool_calls, kết thúc
if "tool_calls" not in response:
break
# Execute các tool calls
for tool_call in response["tool_calls"]:
tool_result = self.execute_tool_call(tool_call)
tool_results.append({
"turn": turn,
"tool": tool_call["function"]["name"],
"result": tool_result
})
# Thêm kết quả vào conversation
conversation.append({
"role": "tool",
"tool_call_id": tool_call["id"],
"content": json.dumps(tool_result.data) if tool_result.success else f"Error: {tool_result.error}"
})
return tool_results
==================== VÍ DỤ SỬ DỤNG ====================
def get_weather(location: str) -> Dict[str, Any]:
"""Tool mẫu: Lấy thông tin thời tiết"""
# Mock implementation
return {"location": location, "temp": 25, "condition": "sunny"}
def search_database(query: str) -> List[Dict]:
"""Tool mẫu: Tìm kiếm database"""
# Mock implementation
return [{"id": 1, "name": "Sample Data"}]
executor = ToolCallExecutor(chain)
executor.register_tool("get_weather", get_weather)
executor.register_tool("search_database", search_database)
Run agent
results = executor.execute_multi_agent(
messages=[{"role": "user", "content": "Tìm thời tiết ở Hà Nội và search dữ liệu liên quan"}],
tools=[
ToolDefinition(
name="get_weather",
description="Lấy thông tin thời tiết",
parameters={"type": "object", "properties": {"location": {"type": "string"}}}
),
ToolDefinition(
name="search_database",
description="Tìm kiếm database",
parameters={"type": "object", "properties": {"query": {"type": "string"}}}
)
]
)
print(f"Total tool executions: {len(results)}")
Monitoring và Observability
import time
from dataclasses import dataclass, field
from typing import Dict, List
from collections import defaultdict
@dataclass
class RetryMetrics:
"""Metrics cho retry system"""
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
total_retries: int = 0
fallback_count: int = 0
circuit_open_count: int = 0
avg_latency_ms: float = 0.0
p95_latency_ms: float = 0.0
latency_history: List[float] = field(default_factory=list)
error_counts: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
provider_stats: Dict[str, Dict] = field(default_factory=lambda: defaultdict(dict))
class RetryMetricsCollector:
"""
Collector để track các retry metrics
Giúp identify issues và optimize retry policy
"""
def __init__(self):
self.metrics = RetryMetrics()
self.start_time = time.time()
def record_request(
self,
provider: str,
success: bool,
attempt: int,
latency_ms: float,
error: Optional[str] = None,
fallback_triggered: bool = False
):
"""Ghi nhận một request"""
self.metrics.total_requests += 1
self.metrics.latency_history.append(latency_ms)
# Update provider stats
if provider not in self.metrics.provider_stats:
self.metrics.provider_stats[provider] = {
"attempts": 0, "successes": 0, "failures": 0
}
stats = self.metrics.provider_stats[provider]
stats["attempts"] += 1
if success:
self.metrics.successful_requests += 1
stats["successes"] += 1
stats["avg_latency"] = (
(stats.get("avg_latency", 0) * (stats["successes"] - 1) + latency_ms)
/ stats["successes"]
)
else:
self.metrics.failed_requests += 1
stats["failures"] += 1
if error:
self.metrics.error_counts[error] += 1
# Update retries
if attempt > 1:
self.metrics.total_retries += (attempt - 1)
# Update fallback
if fallback_triggered:
self.metrics.fallback_count += 1
# Update latency stats
self._update_latency_stats()
def record_circuit_open(self, provider: str):
"""Ghi nhận circuit breaker opened"""
self.metrics.circuit_open_count += 1
def _update_latency_stats(self):
"""Update latency percentiles"""
history = sorted(self.metrics.latency_history)
n = len(history)
if n > 0:
self.metrics.avg_latency_ms = sum(history) / n
p95_idx = int(n * 0.95)
self.metrics.p95_latency_ms = history[p95_idx]
def get_report(self) -> Dict[str, Any]:
"""Generate metrics report"""
uptime_hours = (time.time() - self.start_time) / 3600
return {
"uptime_hours": round(uptime_hours, 2),
"total_requests": self.metrics.total_requests,
"success_rate": round(
self.metrics.successful_requests / max(1, self.metrics.total_requests) * 100, 2
),
"retry_rate": round(
self.metrics.total_retries / max(1, self.metrics.total_requests) * 100, 2
),
"fallback_rate": round(
self.metrics.fallback_count / max(1, self.metrics.total_requests) * 100, 2
),
"circuit_open_rate": round(
self.metrics.circuit_open_count / max(1, self.metrics.total_requests) * 100, 2
),
"latency": {
"avg_ms": round(self.metrics.avg_latency_ms, 2),
"p95_ms": round(self.metrics.p95_latency_ms, 2)
},
"top_errors": dict(sorted(
self.metrics.error_counts.items(),
key=lambda x: x[1],
reverse=True
)[:5]),
"provider_stats": self.metrics.provider_stats,
"estimated_monthly_cost_usd": self._estimate_cost()
}
def _estimate_cost(self) -> float:
"""Ước tính chi phí hàng tháng"""
# Giả định 10 triệu tokens/tháng, mix các model
gpt4_tokens = self.metrics.total_requests * 1000 * 0.1 # 10%
deepseek_tokens = self.metrics.total_requests * 1000 * 0.9 # 90%
cost = (gpt4_tokens / 1_000_000) * 8.0 + (deepseek_tokens / 1_000_000) * 0.42
# Scale theo uptime
hours_per_month = 730
scale = hours_per_month / max(1, uptime_hours)
return round(cost * scale, 2)
Sử dụng metrics collector
metrics = RetryMetricsCollector()
Ghi nhận request
metrics.record_request(
provider="holysheep",
success=True,
attempt=1,
latency_ms=45.2
)
Generate report
report = metrics.get_report()
print(f"Success Rate: {report['success_rate']}%")
print(f"Avg Latency: {report['latency']['avg_ms']}ms")
print(f"P95 Latency: {report['latency']['p95_ms']}ms")
print(f"Estimated Monthly Cost: ${report['estimated_monthly_cost_usd']}")
Lỗi Thường Gặp và Cách Khắc Phục
1. Lỗi "Connection timeout" - Retry không hoạt động
# ❌ SAI: Timeout quá ngắn, không retry đúng cách
response = requests.post(url, timeout=1) # 1 second quá ngắn!
✅ ĐÚNG: Cấu hình timeout hợp lý và retry
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
session = requests.Session()
Retry strategy: thử 3 lần, backoff exponential, chấp nhận 5xx, 429
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["POST"],
raise_on_status=False
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("https://", adapter)
Timeout: connect=5s, read=30s
response = session.post(url, json=payload, timeout=(5, 30))
2. Lỗi "Circuit breaker không recovery"
# ❌ SAI: Circuit breaker mở nhưng không bao giờ thử recovery
breaker = CircuitBreaker("test", CircuitBreakerConfig(
failure_threshold=5,
timeout=300 # 5 phút quá lâu để test!
))
✅ ĐÚNG: Timeout hợp lý, test ngay sau khi open
breaker = CircuitBreaker("test", CircuitBreakerConfig(
failure_threshold=5,
timeout=10, # Thử lại sau 10 giây
success_threshold=2, # Cần 2 success để close
half_open_requests=3 # Cho phép 3 request test
))
Monitoring: log khi circuit state thay đổi
original_record_failure = breaker.record_failure
def logged_record_failure():
result = original_record_failure()
if breaker.state == CircuitState.OPEN:
logger.critical(f"CIRCUIT OPENED for {breaker.name}! Check provider status!")
return result
breaker.record_failure = logged_record_failure
3. Lỗi "Fallback chain không chuyển đổi đúng"
# ❌ SAI: Không kiểm tra model support, fallback không hoạt động
chain = FallbackChain()
chain.add_provider(ProviderConfig(
name="provider1",
client=client1,
circuit_breaker=cb1,
priority=1
# Thiếu: supported_models!
))
Khi gọi model="claude-3" -> vẫn thử provider1 -> fail
✅ ĐÚNG: Luôn định nghĩa supported_models
chain = FallbackChain()
chain.add_provider(ProviderConfig(
name="holysheep-gpt",
client=holysheep_client,
circuit_breaker=cb_holysheep,
priority=1,
supported_models=["gpt-4.1", "gpt-4o", "gpt-3.5-turbo"] # Chỉ định rõ
))
chain.add_provider(ProviderConfig(
name="holysheep-claude",
client=claude_client,
circuit_breaker=cb_claude,
priority=2,
supported_models=["claude-3-opus", "claude-3-sonnet", "claude-3-haiku"]
))
Kiểm tra trước khi execute
def safe_execute(chain, model, **kwargs):
available = [p for p in chain.providers if p.supports_model(model)]
if not available:
raise ValueError(f"No provider supports model: {model}")
return chain.execute_with_fallback(model=model, **kwargs)
4. Lỗi "Rate limit không xử lý đúng"
# ❌ SAI: Retry ngay lập tức k