Nếu bạn đang vận hành hệ thống AI dựa trên Claude Function Calling, chắc chắn bạn đã từng đối mặt với những lỗi không mong muốn làm gián đoạn dịch vụ. Bài viết này sẽ chia sẻ kinh nghiệm thực chiến từ một dự án di chuyển thực tế, kèm theo code mẫu và chiến lược xử lý lỗi production-ready.
Nghiên Cứu Điển Hình: Startup AI Ở Hà Nội
Bối Cảnh Kinh Doanh
Một startup AI tại Hà Nội chuyên cung cấp chatbot hỗ trợ khách hàng cho các sàn thương mại điện tử đã sử dụng Claude Function Calling để xử lý đơn hàng tự động. Hệ thống ban đầu xử lý khoảng 50,000 yêu cầu mỗi ngày với độ trễ trung bình 420ms.
Điểm Đau Của Nhà Cung Cấp Cũ
- Chi phí quá cao: Hóa đơn hàng tháng lên đến $4,200 với mức giá $15/MTok cho Claude Sonnet 4.5
- Độ trễ không ổn định: Thời gian phản hồi dao động từ 300ms đến 800ms vào giờ cao điểm
- Quản lý API key phức tạp: Không có cơ chế xoay vòng key tự động
- Hỗ trợ kỹ thuật chậm: Thời gian phản hồi ticket trung bình 48 giờ
Lý Do Chọn HolySheep AI
Sau khi đánh giá nhiều nhà cung cấp, startup này quyết định đăng ký tại đây HolySheep AI vì:
- Tỷ giá chỉ ¥1 = $1 — tiết kiệm 85%+ so với các nhà cung cấp khác
- Hỗ trợ WeChat và Alipay thanh toán
- Độ trễ trung bình dưới 50ms
- Tín dụng miễn phí khi đăng ký để trải nghiệm
Các Bước Di Chuyển Cụ Thể
Bước 1: Thay Đổi Base URL
# Trước đây (sai)
BASE_URL = "https://api.anthropic.com/v1"
Sau khi chuyển sang HolySheep AI (đúng)
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"Bước 2: Triển Khai Canary Deploy
import random
from typing import Optional
class LoadBalancer:
def __init__(self, holysheep_key: str, anthropic_key: str, canary_ratio: float = 0.1):
self.holysheep_key = holysheep_key
self.anthropic_key = anthropic_key
self.canary_ratio = canary_ratio
def get_api_key(self) -> tuple[str, str]:
"""Trả về (base_url, api_key) dựa trên canary ratio"""
if random.random() < self.canary_ratio:
return "https://api.holysheep.ai/v1", self.holysheep_key
return "https://api.anthropic.com/v1", self.anthropic_key
Khởi tạo load balancer với 10% traffic sang HolySheep
lb = LoadBalancer(
holysheep_key="YOUR_HOLYSHEEP_API_KEY",
anthropic_key="sk-ant-...",
canary_ratio=0.1
)Bước 3: Xoay Vòng API Key Tự Động
import time
from threading import Lock
from dataclasses import dataclass
from typing import List
@dataclass
class APIKeyConfig:
keys: List[str]
current_index: int = 0
last_rotation: float = 0
rotation_interval: int = 3600 # Xoay mỗi giờ
class KeyRotator:
def __init__(self, keys: List[str]):
self.config = APIKeyConfig(keys=keys)
self.lock = Lock()
def get_next_key(self) -> str:
with self.lock:
current_time = time.time()
# Xoay key nếu đã đến interval
if current_time - self.config.last_rotation > self.config.rotation_interval:
self.config.current_index = (self.config.current_index + 1) % len(self.config.keys)
self.config.last_rotation = current_time
print(f"[KeyRotator] Đã xoay sang key index: {self.config.current_index}")
return self.config.keys[self.config.current_index]
Sử dụng với nhiều API key
rotator = KeyRotator([
"YOUR_HOLYSHEEP_API_KEY_1",
"YOUR_HOLYSHEEP_API_KEY_2",
"YOUR_HOLYSHEEP_API_KEY_3"
])Kết Quả 30 Ngày Sau Go-Live
| Chỉ Số | Trước Di Chuyển | Sau Di Chuyển | Cải Thiện |
|---|---|---|---|
| Độ trễ trung bình | 420ms | 180ms | -57% |
| Hóa đơn hàng tháng | $4,200 | $680 | -84% |
| Tỷ lệ lỗi | 2.3% | 0.4% | -83% |
| Uptime | 99.2% | 99.95% | +0.75% |
Kiến Trúc Xử Lý Lỗi Toàn Diện
1. Retry Logic Với Exponential Backoff
import asyncio
import aiohttp
from typing import Any, Optional
from dataclasses import dataclass
import time
@dataclass
class RetryConfig:
max_retries: int = 3
base_delay: float = 1.0
max_delay: float = 60.0
exponential_base: float = 2.0
retryable_status_codes: tuple = (429, 500, 502, 503, 504)
class ClaudeFunctionCaller:
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.retry_config = RetryConfig()
async def call_with_retry(
self,
session: aiohttp.ClientSession,
messages: list,
tools: list,
tool_choice: Optional[str] = None
) -> dict:
last_exception = None
for attempt in range(self.retry_config.max_retries):
try:
payload = {
"model": "claude-sonnet-4-20250514",
"messages": messages,
"tools": tools,
"max_tokens": 1024
}
if tool_choice:
payload["tool_choice"] = {"type": "function", "name": tool_choice}
headers = {
"x-api-key": self.api_key,
"anthropic-version": "2023-06-01",
"content-type": "application/json"
}
async with session.post(
f"{self.base_url}/messages",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 200:
return await response.json()
elif response.status == 429:
# Rate limit - cần retry
retry_after = response.headers.get("Retry-After", "1")
await asyncio.sleep(float(retry_after))
continue
elif response.status in self.retry_config.retryable_status_codes:
# Server error - retry với backoff
delay = min(
self.retry_config.base_delay * (self.retry_config.exponential_base ** attempt),
self.retry_config.max_delay
)
print(f"[Retry] Attempt {attempt + 1}: Status {response.status}, delay {delay}s")
await asyncio.sleep(delay)
continue
else:
# Lỗi không retry được
error_body = await response.text()
raise Exception(f"API Error {response.status}: {error_body}")
except aiohttp.ClientError as e:
last_exception = e
delay = min(
self.retry_config.base_delay * (self.retry_config.exponential_base ** attempt),
self.retry_config.max_delay
)
print(f"[Retry] Network error: {e}, retry in {delay}s")
await asyncio.sleep(delay)
raise Exception(f"Max retries exceeded. Last error: {last_exception}")
Sử dụng
async def main():
caller = ClaudeFunctionCaller(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
async with aiohttp.ClientSession() as session:
result = await caller.call_with_retry(
session=session,
messages=[{"role": "user", "content": "Tính tổng 123 + 456"}],
tools=[{
"name": "calculator",
"description": "Thực hiện phép tính",
"input_schema": {
"type": "object",
"properties": {
"expression": {"type": "string"}
}
}
}]
)
print(result)
asyncio.run(main())
2. Circuit Breaker Pattern
import time
from enum import Enum
from threading import Lock
from dataclasses import dataclass
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ỗi liên tiếp để mở circuit
success_threshold: int = 3 # Số thành công để đóng circuit (half_open)
timeout: float = 30.0 # Thời gian chờ trước khi thử lại (giây)
half_open_max_calls: int = 3 # Số call tối đa trong half_open
class CircuitBreaker:
def __init__(self, name: str, config: 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: float = 0
self.half_open_calls = 0
self.lock = Lock()
def can_execute(self) -> bool:
with self.lock:
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
elapsed = time.time() - self.last_failure_time
if elapsed >= self.config.timeout:
print(f"[CircuitBreaker] {self.name}: Chuyển sang HALF_OPEN sau {elapsed:.1f}s")
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
return True
return False
if self.state == CircuitState.HALF_OPEN:
if self.half_open_calls < self.config.half_open_max_calls:
self.half_open_calls += 1
return True
return False
return False
def record_success(self):
with self.lock:
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
print(f"[CircuitBreaker] {self.name}: Đóng circuit sau {self.success_count} thành công")
self._reset()
else:
self.failure_count = 0
def record_failure(self):
with self.lock:
self.last_failure_time = time.time()
self.failure_count += 1
if self.state == CircuitState.HALF_OPEN:
print(f"[CircuitBreaker] {self.name}: Thất bại trong HALF_OPEN, mở lại circuit")
self.state = CircuitState.OPEN
self.success_count = 0
elif self.failure_count >= self.config.failure_threshold:
print(f"[CircuitBreaker] {self.name}: Mở circuit sau {self.failure_count} lỗi")
self.state = CircuitState.OPEN
def _reset(self):
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.half_open_calls = 0
Sử dụng với Function Calling
class ResilientClaudeCaller:
def __init__(self, api_key: str):
self.api_key = api_key
self.circuit_breaker = CircuitBreaker(
name="claude-api",
config=CircuitBreakerConfig(
failure_threshold=5,
success_threshold=2,
timeout=30.0
)
)
async def call(self, session, payload):
if not self.circuit_breaker.can_execute():
raise Exception("Circuit breaker OPEN - API không khả dụng")
try:
# Gọi API...
result = await self._do_api_call(session, payload)
self.circuit_breaker.record_success()
return result
except Exception as e:
self.circuit_breaker.record_failure()
raise3. Function Calling Error Classification
from enum import Enum
from typing import Optional, Any
from dataclasses import dataclass
import json
class FunctionErrorType(Enum):
TOOL_NOT_FOUND = "tool_not_found"
INVALID_PARAMETERS = "invalid_parameters"
TOOL_EXECUTION_FAILED = "tool_execution_failed"
RATE_LIMIT = "rate_limit"
AUTHENTICATION_ERROR = "authentication_error"
TIMEOUT = "timeout"
UNKNOWN = "unknown"
@dataclass
class FunctionCallError:
error_type: FunctionErrorType
message: str
tool_name: Optional[str] = None
raw_error: Optional[Any] = None
retry_possible: bool = False
fallback_available: bool = False
def classify_function_error(error: Exception, tool_name: str = None) -> FunctionCallError:
error_message = str(error).lower()
# Tool not found
if "tool" in error_message and ("not found" in error_message or "does not exist" in error_message):
return FunctionCallError(
error_type=FunctionErrorType.TOOL_NOT_FOUND,
message=f"Tool '{tool_name}' không tồn tại",
tool_name=tool_name,
retry_possible=False,
fallback_available=True
)
# Invalid parameters
if "parameter" in error_message or "schema" in error_message or "validation" in error_message:
return FunctionCallError(
error_type=FunctionErrorType.INVALID_PARAMETERS,
message=f"Tham số không hợp lệ cho tool '{tool_name}'",
tool_name=tool_name,
retry_possible=False,
fallback_available=True
)
# Rate limit
if "429" in error_message or "rate limit" in error_message or "quota" in error_message:
return FunctionCallError(
error_type=FunctionErrorType.RATE_LIMIT,
message="Đã vượt quá giới hạn rate",
tool_name=tool_name,
retry_possible=True,
fallback_available=True
)
# Authentication
if "401" in error_message or "unauthorized" in error_message or "authentication" in error_message:
return FunctionCallError(
error_type=FunctionErrorType.AUTHENTICATION_ERROR,
message="Lỗi xác thực API key",
tool_name=tool_name,
retry_possible=False,
fallback_available=False
)
# Timeout
if "timeout" in error_message or "timed out" in error_message:
return FunctionCallError(
error_type=FunctionErrorType.TIMEOUT,
message="Yêu cầu bị timeout",
tool_name=tool_name,
retry_possible=True,
fallback_available=True
)
# Default unknown
return FunctionCallError(
error_type=FunctionErrorType.UNKNOWN,
message=f"Lỗi không xác định: {error}",
tool_name=tool_name,
retry_possible=True,
fallback_available=True
)
def handle_classified_error(error: FunctionCallError) -> dict:
"""Xử lý lỗi đã được phân loại và trả về response phù hợp"""
if error.error_type == FunctionErrorType.TOOL_NOT_FOUND:
return {
"status": "error",
"error_type": "tool_not_found",
"message": error.message,
"suggestion": "Kiểm tra lại tên tool trong function definitions"
}
if error.error_type == FunctionErrorType.INVALID_PARAMETERS:
return {
"status": "error",
"error_type": "invalid_parameters",
"message": error.message,
"suggestion": "Kiểm tra schema và định dạng tham số"
}
if error.error_type == FunctionErrorType.RATE_LIMIT:
return {
"status": "error",
"error_type": "rate_limit",
"message": error.message,
"retry_after": 60,
"suggestion": "Thử lại sau 60 giây hoặc nâng cấp gói dịch vụ"
}
if error.error_type == FunctionErrorType.AUTHENTICATION_ERROR:
return {
"status": "error",
"error_type": "authentication_error",
"message": error.message,
"action_required": "Kiểm tra và cập nhật API key"
}
return {
"status": "error",
"error_type": "unknown",
"message": error.message
}Lỗi Thường Gặp Và Cách Khắc Phục
1. Lỗi "tool_use_failed" - Tool Execution Timeout
Mô tả: Khi function execution vượt quá thời gian cho phép, Claude trả về lỗi tool_use_failed.
# Vấn đề: Tool execution quá chậm
{
"type": "error",
"error": {
"type": "tool_use_failed",
"message": "Tool execution exceeded timeout of 30 seconds"
}
}
Giải pháp: Sử dụng async execution với timeout rõ ràng
import asyncio
from concurrent.futures import ThreadPoolExecutor
async def execute_tool_with_timeout(tool_func, params, timeout_seconds=25):
"""Execute tool với timeout an toàn"""
try:
loop = asyncio.get_event_loop()
# Chạy sync function trong thread pool
with ThreadPoolExecutor() as pool:
result = await asyncio.wait_for(
loop.run_in_executor(pool, lambda: tool_func(**params)),
timeout=timeout_seconds
)
return {"status": "success", "result": result}
except asyncio.TimeoutError:
return {
"status": "error",
"error_type": "timeout",
"message": f"Tool execution exceeded {timeout_seconds}s"
}
except Exception as e:
return {
"status": "error",
"error_type": "execution_failed",
"message": str(e)
}
Sử dụng với Claude response
async def process_claude_response(response, tool_registry):
if response.stop_reason == "tool_use":
for content in response.content:
if content.type == "tool_use":
tool_name = content.name
tool_input = content.input
if tool_name in tool_registry:
result = await execute_tool_with_timeout(
tool_func=tool_registry[tool_name],
params=tool_input,
timeout_seconds=25
)
# Tiếp tục với result...
else:
print(f"Tool '{tool_name}' not found in registry")2. Lỗi "invalid_request_error" - Sai Tool Schema
Mô tả: Schema định nghĩa trong tool không khớp với tham số Claude gửi.
# Vấn đề: Schema không đúng định dạng
TOOLS_WRONG = [
{
"name": "get_weather",
"description": "Lấy thông tin thời tiết",
"input_schema": {
"properties": {
"location": "string" # Thiếu "type"
}
}
}
]
Giải pháp: Schema chuẩn OpenAI format
TOOLS_CORRECT = [
{
"name": "get_weather",
"description": "Lấy thông tin thời tiết theo thành phố",
"input_schema": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "Tên thành phố cần tra cứu thời tiết"
},
"units": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"default": "celsius"
}
},
"required": ["location"]
}
}
]
Hàm validate schema trước khi gửi request
def validate_tool_schema(tools: list) -> tuple[bool, list]:
"""Validate tất cả tool schemas"""
errors = []
for tool in tools:
name = tool.get("name", "unknown")
schema = tool.get("input_schema", {})
# Kiểm tra type của schema
if schema.get("type") != "object":
errors.append(f"Tool '{name}': input_schema phải có type='object'")
# Kiểm tra properties có định dạng đúng
properties = schema.get("properties", {})
for prop_name, prop_def in properties.items():
if not isinstance(prop_def, dict):
errors.append(f"Tool '{name}': property '{prop_name}' phải là object")
elif "type" not in prop_def:
errors.append(f"Tool '{name}': property '{prop_name}' thiếu 'type'")
# Kiểm tra required array
if "required" in schema and not isinstance(schema["required"], list):
errors.append(f"Tool '{name}': required phải là array")
return len(errors) == 0, errors
Validate trước khi sử dụng
is_valid, validation_errors = validate_tool_schema(TOOLS_CORRECT)
if not is_valid:
for error in validation_errors:
print(f"Validation Error: {error}")
else:
print("Tất cả tools đã hợp lệ")3. Lỗi Authentication - Invalid hoặc Expired API Key
Mô tả: API key không hợp lệ hoặc đã hết hạn, thường gặp khi sử dụng key test trong production.
# Vấn đề: Không handle authentication error riêng
async def call_claude_basic(messages, tools):
headers = {
"x-api-key": os.getenv("CLAUDE_API_KEY"),
"anthropic-version": "2023-06-01"
}
async with aiohttp.ClientSession() as session:
response = await session.post(
"https://api.holysheep.ai/v1/messages",
json={"model": "claude-sonnet-4-20250514", "messages": messages, "tools": tools},
headers=headers
)
# Không check response status!
return await response.json()
Giải pháp: Handle authentication với retry logic riêng
from aiohttp import ClientResponse
async def call_claude_robust(
messages: list,
tools: list,
api_key: str,
max_retries: int = 3
) -> dict:
"""Gọi Claude API với handle auth error toàn diện"""
headers = {
"x-api-key": api_key,
"anthropic-version": "2023-06-01",
"content-type": "application/json"
}
payload = {
"model": "claude-sonnet-4-20250514",
"messages": messages,
"tools": tools,
"max_tokens": 1024
}
async with aiohttp.ClientSession() as session:
for attempt in range(max_retries):
try:
async with session.post(
"https://api.holysheep.ai/v1/messages",
json=payload,
headers=headers,
timeout=aiohttp.ClientTimeout(total=30)
) as response:
if response.status == 200:
return await response.json()
# Authentication error (401, 403)
if response.status in (401, 403):
error_body = await response.text()
print(f"[Auth Error] Mã {response.status}: {error_body}")
# Log và alert ngay lập tức
await log_auth_error(
status=response.status,
error=error_body,
attempt=attempt
)
# Không retry authentication errors
if attempt == 0:
raise AuthenticationError(
"API key không hợp lệ. Vui lòng kiểm tra API key tại "
"https://www.holysheep.ai/register"
)
break
# Rate limit (429)
if response.status == 429:
retry_after = response.headers.get("Retry-After", "60")
wait_time = int(retry_after)
print(f"[Rate Limit] Chờ {wait_time}s trước khi retry...")
await asyncio.sleep(wait_time)
continue
# Server errors (5xx)
if 500 <= response.status < 600:
delay = 2 ** attempt
print(f"[Server Error {response.status}] Retry sau {delay}s...")
await asyncio.sleep(delay)
continue
# Client errors (4xx khác)
error_body = await response.text()
raise APIError(f"Mã lỗi {response.status}: {error_body}")
except aiohttp.ClientConnectorError as e:
print(f"[Connection Error] {e}")
await asyncio.sleep(2 ** attempt)
except asyncio.TimeoutError:
print(f"[Timeout] Attempt {attempt + 1} timeout")
if attempt == max_retries - 1:
raise TimeoutError("Claude API không phản hồi sau nhiều lần thử")
raise MaximumRetriesExceeded(f"Đã thử {max_retries} lần không thành công")
class AuthenticationError(Exception):
"""Exception riêng cho authentication errors"""
pass
class APIError(Exception):
"""Exception chung cho API errors"""
pass
class MaximumRetriesExceeded(Exception):
"""Exception khi vượt quá số lần retry"""
passBest Practices Tổng Hợp
1. Monitoring Và Alerting
from dataclasses import dataclass
from typing import Dict, List
import time
from collections import defaultdict
@dataclass
class FunctionCallMetrics:
total_calls: int = 0
successful_calls: int = 0
failed_calls: int = 0
total_latency_ms: float = 0
error_counts: Dict[str, int] = None
def __post_init__(self):
self.error_counts = defaultdict(int)
def record_success(self, latency_ms: float):
self.total_calls += 1
self.successful_calls += 1
self.total_latency_ms += latency_ms
def record_failure(self, error_type: str, latency_ms: float = 0):
self.total_calls += 1
self.failed_calls += 1
self.error_counts[error_type] += 1
if latency_ms:
self.total_latency_ms += latency_ms
@property
def success_rate(self) -> float:
if self.total_calls == 0:
return 0
return self.successful_calls / self.total_calls * 100
@property
def avg_latency_ms(self) -> float:
if self.total_calls == 0:
return 0
return self.total_latency_ms / self.total_calls
def get_report(self) -> dict:
return {
"total_calls": self.total_calls,
"success_rate": f"{self.success_rate:.2f}%",
"avg_latency_ms": f"{self.avg_latency_ms:.2f}ms",
"error_breakdown": dict(self.error_counts),
"timestamp": time.time()
}
class MetricsCollector:
def __init__(self):
self.metrics: Dict[str, FunctionCallMetrics] = {}
self.lock = time = __import__('time')
def get_metrics(self, tool_name: str) -> FunctionCallMetrics:
if tool_name not in self.metrics:
self.metrics[tool_name] = FunctionCallMetrics()
return self.metrics[tool_name]
def record(self, tool_name: str, success: bool, latency_ms: float, error_type: str = None):
metrics = self.get_metrics(tool_name)
if success:
metrics.record_success(latency_ms)
else:
metrics.record_failure(error_type or "unknown", latency_ms)
def get_all_reports(self) -> Dict[str, dict]:
return {
tool_name: metrics.get_report()
for tool_name, metrics in self.metrics.items()
}
def check_alerts(self) -> List[str]:
"""Kiểm tra và trả về các alert cần thiết"""
alerts = []
for tool_name, metrics in self.metrics.items():
# Alert nếu success rate < 95%
if metrics.success_rate < 95:
alerts.append(
f"[ALERT] {tool_name}: Success rate {metrics.success_rate:.1f}% "
f"(threshold: 95%)"
)
# Alert nếu latency > 500ms
if metrics.avg_latency_ms > 500:
alerts.append(
f"[ALERT] {tool_name}: Avg latency {metrics.avg_latency_ms:.1f}ms "
f"(threshold: 500ms)"
)
return alerts
Sử dụng trong ứng dụng
metrics_collector = MetricsCollector()
async def monitored_function_call(tool_name: str, func, *args, **kwargs):
"""Wrapper để monitor function calls"""
start_time = time.time()
try:
result = await func(*args, **kwargs)
latency_ms = (time.time() - start_time) * 1000
metrics_collector.record(tool_name, success=True, latency_ms=latency_ms)
return result
except Exception as e:
latency_ms = (time.time() - start_time) * 1000
error_type = type(e).__name__
metrics_collector.record(tool_name, success=False, latency_ms=latency_ms, error_type=error_type)
raise2. Fallback Strategy
from typing import Callable, Any, Optional
from dataclasses import dataclass
import asyncio
@dataclass
class FallbackConfig:
enable_fallback: bool = True
fallback_models: list = None
max_fallback_attempts: int = 2
class ClaudeWithFallback:
def __init__(self, api_key: str, config: FallbackConfig = None):
self.api_key = api_key
self.config = config or FallbackConfig()
self.config.fallback_models = self.config.fallback_models or [
("claude-sonnet-4-20250514", "https://api.holysheep.ai/v1"),
("claude-haiku-4-20250514", "https://api.holysheep.ai/v1"),
]
async def call_with_fallback(
self,
messages: list,
tools: list,
primary_model: str = "claude-sonnet-4-20250514"
) -> dict:
"""Gọi API với fallback chain"""
# Thử primary model trước
models_to_try = [(primary_model, "https://api.holysheep.ai/v1")]
models_to_try.extend(self.config.fallback_models[:self.config.max_fallback_attempts])
last_error = None
for model, base_url in models_to_try:
try:
print(f"[Fallback] Thử model: {model}")
result = await self._call_api(
base_url=base_url,
model=model,
messages=messages,
tools=tools
)
print(f"[Fallback] Th