Trong bối cảnh AI phát triển cực kỳ nhanh chóng, việc tích hợp nhiều Large Language Model (LLM) vào sản phẩm trở thành nhu cầu tất yếu. Bài viết này sẽ hướng dẫn bạn xây dựng một Multi-Model API Gateway hoàn chỉnh, dựa trên bài học thực chiến từ dự án đã triển khai thành công.
Bối Cảnh Thực Tế: Hành Trình Của Một Startup AI Tại Việt Nam
Một startup AI tại Hà Nội chuyên cung cấp giải pháp chatbot cho thương mại điện tử đã phải đối mặt với bài toán nan giải: họ sử dụng 4 nhà cung cấp API khác nhau (OpenAI, Anthropic, Google, DeepSeek) cho các use-case riêng biệt. Mỗi nhà cung cấp có:
- Cú pháp request khác nhau
- Cơ chế authentication riêng
- Rate limit không đồng nhất
- Chi phí tính theo đơn vị tiền tệ khác nhau (USD, có khi phải qua nhiều lớp trung gian)
Điểm đau lớn nhất: Hóa đơn hàng tháng lên tới $4,200 USD chỉ riêng chi phí API, trong khi độ trễ trung bình lên đến 420ms khiến trải nghiệm người dùng kém. Đặc biệt, việc quản lý 4 API keys khác nhau, mỗi cái với ngưỡng rate limit riêng, gây ra vô số lỗi integration.
Sau khi tìm hiểu và chuyển sang HolySheep AI — nền tảng tích hợp đa mô hình với tỷ giá ¥1 = $1 (tiết kiệm 85%+ chi phí), hỗ trợ WeChat/Alipay, độ trễ dưới 50ms, và tín dụng miễn phí khi đăng ký — startup này đã giảm hóa đơn xuống còn $680/tháng và đạt độ trễ 180ms trong 30 ngày đầu tiên.
Kiến Trúc Tổng Quan Multi-Model API Gateway
┌─────────────────────────────────────────────────────────────────────────────┐
│ Multi-Model API Gateway │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────────┐ ┌─────────────────────────────┐ │
│ │ Client │────▶│ Router │────▶│ Model Abstraction │ │
│ │ (SDK) │ │ Layer │ │ Layer │ │
│ └──────────┘ └──────────────┘ │ │ │
│ │ ┌─────────┐ ┌──────────┐ │ │
│ Features: │ │ Model A │ │ Model B │ │ │
│ • Load Balancer │ │ (GPT-4) │ │(Claude) │ │ │
│ • Circuit Breaker │ └─────────┘ └──────────┘ │ │
│ • Rate Limiter │ ┌─────────┐ ┌──────────┐ │ │
│ • Request Batching │ │Model C │ │ Model D │ │ │
│ │ │(Gemini) │ │(DeepSeek)│ │ │
│ │ └─────────┘ └──────────┘ │ │
│ └─────────────────────────────┘ │
│ │ │
│ ┌────────────▼────────────┐ │
│ │ HolySheep API Layer │ │
│ │ base_url: api.holysheep.ai/v1 │
│ └─────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────┘Triển Khai Chi Tiết: Từ Migration Đến Production
Bước 1: Cấu Hình Base Client - HolySheep Integration
import requests
import json
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, field
from enum import Enum
import time
import hashlib
class ModelProvider(Enum):
"""Các nhà cung cấp model được hỗ trợ"""
GPT_4_1 = "gpt-4.1"
CLAUDE_SONNET_4_5 = "claude-sonnet-4.5"
GEMINI_2_5_FLASH = "gemini-2.5-flash"
DEEPSEEK_V3_2 = "deepseek-v3.2"
@dataclass
class ModelConfig:
"""Cấu hình cho từng model"""
provider: ModelProvider
max_tokens: int = 4096
temperature: float = 0.7
timeout: int = 30
retry_count: int = 3
fallback_models: List[ModelProvider] = field(default_factory=list)
class HolySheepMultiModelClient:
"""
Client thống nhất để gọi đa mô hình AI qua HolySheep API
Tích hợp các tính năng: retry, fallback, rate limiting, circuit breaker
"""
BASE_URL = "https://api.holysheep.ai/v1"
# Bảng giá 2026 (USD per 1M tokens)
PRICING = {
ModelProvider.GPT_4_1: 8.0, # $8/MTok
ModelProvider.CLAUDE_SONNET_4_5: 15.0, # $15/MTok
ModelProvider.GEMINI_2_5_FLASH: 2.50, # $2.50/MTok
ModelProvider.DEEPSEEK_V3_2: 0.42, # $0.42/MTok - RẺ NHẤT
}
def __init__(self, api_key: str):
"""
Khởi tạo client với API key từ HolySheep
Args:
api_key: YOUR_HOLYSHEEP_API_KEY từ dashboard
"""
if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
raise ValueError("Vui lòng cung cấp API key hợp lệ từ HolySheep")
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
# Circuit breaker state
self.failure_count = {}
self.circuit_open = {}
self.last_failure_time = {}
self.circuit_threshold = 5
self.circuit_timeout = 60 # seconds
# Rate limiter
self.request_counts = {}
self.rate_limit_window = 60 # per minute
# Model configurations
self.model_configs: Dict[ModelProvider, ModelConfig] = {
ModelProvider.GPT_4_1: ModelConfig(
provider=ModelProvider.GPT_4_1,
max_tokens=8192,
temperature=0.7,
fallback_models=[ModelProvider.GEMINI_2_5_FLASH]
),
ModelProvider.CLAUDE_SONNET_4_5: ModelConfig(
provider=ModelProvider.CLAUDE_SONNET_4_5,
max_tokens=8192,
temperature=0.7,
fallback_models=[ModelProvider.GPT_4_1]
),
ModelProvider.GEMINI_2_5_FLASH: ModelConfig(
provider=ModelProvider.GEMINI_2_5_FLASH,
max_tokens=32768,
temperature=0.7,
fallback_models=[ModelProvider.DEEPSEEK_V3_2]
),
ModelProvider.DEEPSEEK_V3_2: ModelConfig(
provider=ModelProvider.DEEPSEEK_V3_2,
max_tokens=16384,
temperature=0.7,
fallback_models=[ModelProvider.GEMINI_2_5_FLASH]
),
}
def _check_circuit_breaker(self, provider: ModelProvider) -> bool:
"""
Kiểm tra circuit breaker - trả về True nếu circuit đóng (cho phép request)
"""
if provider not in self.circuit_open:
return True
if not self.circuit_open.get(provider, False):
return True
# Kiểm tra timeout
if time.time() - self.last_failure_time.get(provider, 0) > self.circuit_timeout:
self.circuit_open[provider] = False
self.failure_count[provider] = 0
return True
return False
def _record_success(self, provider: ModelProvider):
"""Ghi nhận thành công - reset circuit breaker"""
self.failure_count[provider] = 0
self.circuit_open[provider] = False
def _record_failure(self, provider: ModelProvider):
"""Ghi nhận thất bại - tăng counter và có thể mở circuit"""
self.failure_count[provider] = self.failure_count.get(provider, 0) + 1
self.last_failure_time[provider] = time.time()
if self.failure_count[provider] >= self.circuit_threshold:
self.circuit_open[provider] = True
print(f"Circuit breaker OPEN for {provider.value}!")
def _rate_limit_check(self, provider: ModelProvider) -> bool:
"""
Kiểm tra rate limit - giới hạn 60 requests/phút/mỗi model
"""
current_time = time.time()
key = f"{provider.value}_{int(current_time // self.rate_limit_window)}"
count = self.request_counts.get(key, 0)
if count >= 60:
return False
self.request_counts[key] = count + 1
return True
def _format_request(self, provider: ModelProvider, messages: List[Dict],
**kwargs) -> Dict[str, Any]:
"""Chuẩn hóa request format cho tất cả các model"""
config = self.model_configs.get(provider)
base_payload = {
"model": provider.value,
"messages": messages,
"max_tokens": kwargs.get("max_tokens", config.max_tokens),
"temperature": kwargs.get("temperature", config.temperature),
}
# Thêm optional parameters
if "top_p" in kwargs:
base_payload["top_p"] = kwargs["top_p"]
if "stream" in kwargs:
base_payload["stream"] = kwargs["stream"]
return base_payload
def chat_completion(self, provider: ModelProvider, messages: List[Dict],
enable_fallback: bool = True, **kwargs) -> Dict[str, Any]:
"""
Gọi API với logic tự động retry và fallback
Args:
provider: Model muốn sử dụng
messages: Danh sách messages theo format OpenAI
enable_fallback: Có cho phép fallback sang model khác không
**kwargs: Các tham số bổ sung (max_tokens, temperature, ...)
Returns:
Response dictionary với format thống nhất
"""
config = self.model_configs.get(provider)
current_provider = provider
for attempt in range(config.retry_count):
# Kiểm tra circuit breaker
if not self._check_circuit_breaker(current_provider):
if enable_fallback and config.fallback_models:
current_provider = config.fallback_models[0]
config = self.model_configs.get(current_provider)
continue
raise Exception(f"Circuit breaker OPEN for all fallback models")
# Kiểm tra rate limit
if not self._rate_limit_check(current_provider):
time.sleep(1)
continue
try:
payload = self._format_request(current_provider, messages, **kwargs)
start_time = time.time()
response = self.session.post(
f"{self.BASE_URL}/chat/completions",
json=payload,
timeout=config.timeout
)
latency = time.time() - start_time
if response.status_code == 200:
self._record_success(current_provider)
result = response.json()
result["_meta"] = {
"provider": current_provider.value,
"latency_ms": round(latency * 1000, 2),
"pricing_per_mtok": self.PRICING[current_provider]
}
return result
elif response.status_code == 429:
# Rate limited - retry với exponential backoff
self._record_failure(current_provider)
wait_time = 2 ** attempt
time.sleep(wait_time)
continue
elif response.status_code >= 500:
# Server error - retry
self._record_failure(current_provider)
wait_time = 2 ** attempt
time.sleep(wait_time)
continue
else:
raise Exception(f"API Error {response.status_code}: {response.text}")
except requests.exceptions.Timeout:
self._record_failure(current_provider)
if attempt < config.retry_count - 1:
continue
except requests.exceptions.RequestException as e:
self._record_failure(current_provider)
raise Exception(f"Request failed: {str(e)}")
raise Exception(f"Tất cả retry attempts đã thất bại")
============================================================
VÍ DỤ SỬ DỤNG
============================================================
if __name__ == "__main__":
# Khởi tạo client với API key
client = HolySheepMultiModelClient(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# Test với DeepSeek V3.2 (model rẻ nhất - $0.42/MTok)
messages = [
{"role": "system", "content": "Bạn là trợ lý AI tiếng Việt hữu ích"},
{"role": "user", "content": "Giải thích kiến trúc Multi-Model API Gateway"}
]
try:
# Gọi với DeepSeek - chi phí thấp nhất
response = client.chat_completion(
provider=ModelProvider.DEEPSEEK_V3_2,
messages=messages,
max_tokens=1000,
temperature=0.7
)
print(f"Model: {response['_meta']['provider']}")
print(f"Latency: {response['_meta']['latency_ms']}ms")
print(f"Chi phí ước tính: ${response['_meta']['pricing_per_mtok']}/MTok")
print(f"Response: {response['choices'][0]['message']['content']}")
except Exception as e:
print(f"Lỗi: {str(e)}")Bước 2: Triển Khai Canary Deployment & Key Rotation
import asyncio
import random
from typing import Dict, Callable, List
from datetime import datetime, timedelta
import threading
import json
class CanaryDeployment:
"""
Triển khai Canary: chuyển traffic từ từ từ nhà cung cấp cũ sang HolySheep
Hỗ trợ A/B testing và gradual migration
"""
def __init__(self, client: HolySheepMultiModelClient):
self.client = client
self.traffic_weights = {
"legacy": 100.0, # Bắt đầu 100% qua nhà cung cấp cũ
"holysheep": 0.0
}
self.metrics = {
"legacy": {"success": 0, "failure": 0, "latencies": []},
"holysheep": {"success": 0, "failure": 0, "latencies": []}
}
self.migration_rules = []
def add_migration_rule(self, rule: Dict):
"""
Thêm rule migration
rule = {
"condition": lambda req: req.get("priority") == "high",
"target": "holysheep", # Hoặc "legacy"
"percentage": 100 # % traffic matching rule
}
"""
self.migration_rules.append(rule)
def _calculate_traffic_split(self, request: Dict) -> str:
"""
Tính toán request nên đi qua provider nào dựa trên rules
"""
# Check migration rules first
for rule in self.migration_rules:
if rule["condition"](request):
return rule["target"]
# Random split theo traffic weights
rand = random.uniform(0, 100)
if rand < self.traffic_weights["holysheep"]:
return "holysheep"
return "legacy"
async def execute_with_canary(
self,
provider: ModelProvider,
messages: List[Dict],
**kwargs
) -> Dict:
"""
Thực thi request với canary logic
"""
request_id = hashlib.md5(str(datetime.now()).encode()).hexdigest()[:8]
# Determine target provider
target = self._calculate_traffic_split({"messages": messages, **kwargs})
if target == "holysheep":
# Route qua HolySheep - tính năng chính
start = asyncio.get_event_loop().time()
try:
result = await asyncio.to_thread(
self.client.chat_completion,
provider=provider,
messages=messages,
**kwargs
)
latency = (asyncio.get_event_loop().time() - start) * 1000
self.metrics["holysheep"]["success"] += 1
self.metrics["holysheep"]["latencies"].append(latency)
result["_meta"]["canary"] = True
return result
except Exception as e:
self.metrics["holysheep"]["failure"] += 1
# Fallback to legacy if configured
raise e
else:
# Legacy path (giữ lại để so sánh)
return await self._legacy_call(provider, messages, **kwargs)
async def _legacy_call(self, provider, messages, **kwargs) -> Dict:
"""
Legacy API call - giữ lại để so sánh trong giai đoạn migration
"""
# Implement legacy call logic here
# (Dành cho việc so sánh trong giai đoạn canary)
raise NotImplementedError("Legacy call đã được loại bỏ sau migration")
def update_traffic_split(self, holysheep_percentage: float):
"""
Cập nhật tỷ lệ traffic - gọi trong cron job
Ví dụ:
- Ngày 1-3: 10% traffic sang HolySheep
- Ngày 4-7: 30% traffic sang HolySheep
- Ngày 8-14: 70% traffic sang HolySheep
- Ngày 15+: 100% traffic sang HolySheep
"""
self.traffic_weights["holysheep"] = holysheep_percentage
self.traffic_weights["legacy"] = 100 - holysheep_percentage
print(f"Traffic split updated: HolySheep {holysheep_percentage}%, Legacy {100-holysheep_percentage}%")
def get_migration_report(self) -> Dict:
"""
Generate báo cáo migration
"""
def calc_stats(provider_key):
data = self.metrics[provider_key]
latencies = data["latencies"][-100:] # Last 100 requests
if latencies:
return {
"total_requests": data["success"] + data["failure"],
"success_rate": data["success"] / (data["success"] + data["failure"] + 0.001) * 100,
"avg_latency_ms": sum(latencies) / len(latencies),
"p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)] if latencies else 0
}
return {"total_requests": 0, "success_rate": 0, "avg_latency_ms": 0}
return {
"timestamp": datetime.now().isoformat(),
"traffic_weights": self.traffic_weights,
"holysheep_stats": calc_stats("holysheep"),
"legacy_stats": calc_stats("legacy"),
"improvement": {
"latency_reduction_pct": (
(calc_stats("legacy")["avg_latency_ms"] - calc_stats("holysheep")["avg_latency_ms"])
/ calc_stats("legacy")["avg_latency_ms"] * 100
) if calc_stats("legacy")["avg_latency_ms"] > 0 else 0
}
}
class APIKeyRotation:
"""
Quản lý và xoay vòng API keys tự động
Hỗ trợ multiple keys với health check
"""
def __init__(self, keys: List[str], client_factory: Callable):
"""
Args:
keys: Danh sách API keys
client_factory: Hàm tạo client từ API key
"""
self.keys = [k for k in keys if k and k != "YOUR_HOLYSHEEP_API_KEY"]
self.client_factory = client_factory
self.current_key_index = 0
self.key_health = {k: {"healthy": True, "last_used": None, "failures": 0} for k in self.keys}
self.lock = threading.Lock()
if not self.keys:
# Fallback to default key
self.keys = ["YOUR_HOLYSHEEP_API_KEY"]
self.key_health = {self.keys[0]: {"healthy": True, "last_used": None, "failures": 0}}
def get_current_key(self) -> str:
"""Lấy key hiện tại đang active"""
with self.lock:
return self.keys[self.current_key_index]
def get_healthy_key(self) -> str:
"""
Lấy key healthy gần nhất
Thực hiện rotation nếu key hiện tại có vấn đề
"""
with self.lock:
current_health = self.key_health[self.keys[self.current_key_index]]
if current_health["healthy"]:
return self.keys[self.current_key_index]
# Tìm key healthy tiếp theo
for i, key in enumerate(self.keys):
if self.key_health[key]["healthy"]:
self.current_key_index = i
return key
# Không có key healthy - reset tất cả (có thể là API issue)
for key in self.keys:
self.key_health[key]["healthy"] = True
self.key_health[key]["failures"] = 0
return self.keys[0]
def mark_success(self, key: str):
"""Đánh dấu key thành công"""
with self.lock:
if key in self.key_health:
self.key_health[key]["healthy"] = True
self.key_health[key]["failures"] = 0
self.key_health[key]["last_used"] = datetime.now()
def mark_failure(self, key: str):
"""Đánh dấu key thất bại"""
with self.lock:
if key in self.key_health:
self.key_health[key]["failures"] += 1
if self.key_health[key]["failures"] >= 3:
self.key_health[key]["healthy"] = False
print(f"Key marked unhealthy: {key[:8]}... failures={self.key_health[key]['failures']}")
============================================================
VÍ DỤ SỬ DỤNG CANARY DEPLOYMENT
============================================================
async def demo_canary_migration():
"""Demo quy trình migration từ legacy sang HolySheep"""
# Khởi tạo
client = HolySheepMultiModelClient("YOUR_HOLYSHEEP_API_KEY")
canary = CanaryDeployment(client)
# Cấu hình rules
canary.add_migration_rule({
"condition": lambda req: req.get("priority") == "high",
"target": "holysheep",
"percentage": 100
})
canary.add_migration_rule({
"condition": lambda req: req.get("model") == "deepseek-v3.2",
"target": "holysheep",
"percentage": 100 # Luôn dùng HolySheep cho DeepSeek
})
messages = [
{"role": "user", "content": "Test canary deployment"}
]
# Gọi với canary
result = await canary.execute_with_canary(
provider=ModelProvider.DEEPSEEK_V3_2,
messages=messages
)
print(f"Canary result: {result['_meta']}")
# Simulate gradual traffic increase
for day in range(1, 16):
if day <= 3:
canary.update_traffic_split(10)
elif day <= 7:
canary.update_traffic_split(30)
elif day <= 14:
canary.update_traffic_split(70)
else:
canary.update_traffic_split(100)
# Generate report
report = canary.get_migration_report()
print(json.dumps(report, indent=2, default=str))
if __name__ == "__main__":
asyncio.run(demo_canary_migration())Bước 3: Monitoring Dashboard & Cost Tracking
import time
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Dict, List, Optional
import threading
class CostTracker:
"""
Theo dõi chi phí theo thời gian thực
Tính năng: theo dõi chi phí theo model, user, endpoint
"""
def __init__(self):
self.usage_data = defaultdict(lambda: {
"input_tokens": 0,
"output_tokens": 0,
"requests": 0,
"cost": 0.0
})
self.lock = threading.Lock()
# Bảng giá HolySheep 2026 (USD/MTok)
self.pricing = {
"gpt-4.1": {"input": 8.0, "output": 8.0},
"claude-sonnet-4.5": {"input": 15.0, "output": 15.0},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42}, # GIÁ RẺ NHẤT!
}
def record_usage(
self,
model: str,
input_tokens: int,
output_tokens: int,
user_id: Optional[str] = None,
endpoint: Optional[str] = None
):
"""Ghi nhận usage"""
with self.lock:
key = f"{model}:{user_id or 'anonymous'}:{endpoint or 'default'}"
self.usage_data[key]["input_tokens"] += input_tokens
self.usage_data[key]["output_tokens"] += output_tokens
self.usage_data[key]["requests"] += 1
# Tính chi phí (đơn vị: USD)
model_key = model.split("/")[-1] # Handle provider/model format
pricing = self.pricing.get(model_key, {"input": 1.0, "output": 1.0})
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
self.usage_data[key]["cost"] += input_cost + output_cost
def get_cost_report(
self,
group_by: str = "model",
time_range: Optional[timedelta] = None
) -> Dict:
"""
Generate báo cáo chi phí
Args:
group_by: "model", "user", hoặc "endpoint"
time_range: Lọc theo khoảng thời gian
"""
with self.lock:
report = defaultdict(lambda: {
"input_tokens": 0,
"output_tokens": 0,
"requests": 0,
"cost": 0.0
})
for key, data in self.usage_data.items():
model, user, endpoint = key.split(":")
if group_by == "model":
group_key = model
elif group_by == "user":
group_key = user
else:
group_key = endpoint
report[group_key]["input_tokens"] += data["input_tokens"]
report[group_key]["output_tokens"] += data["output_tokens"]
report[group_key]["requests"] += data["requests"]
report[group_key]["cost"] += data["cost"]
return dict(report)
def get_savings_comparison(
self,
old_provider_costs: Dict[str, float]
) -> Dict:
"""
So sánh chi phí với nhà cung cấp cũ
Args:
old_provider_costs: Dict của chi phí cũ theo model
Ví dụ: {"gpt-4": 3000, "claude-3": 1200}
"""
current_total = sum(d["cost"] for d in self.usage_data.values())
old_total = sum(old_provider_costs.values())
return {
"old_provider_total_usd": old_total,
"holysheep_total_usd": round(current_total, 2),
"savings_usd": round(old_total - current_total, 2),
"savings_percentage": round((old_total - current_total) / old_total * 100, 2) if old_total > 0 else 0,
"monthly_projection": round(current_total * 30, 2), # Giả sử 1 ngày
}
class ModelRouter:
"""
Intelligent routing - chọn model tối ưu dựa trên yêu cầu
"""
# Mapping use cases với recommended models
USE_CASE_MAPPING = {
"quick_response": {
"primary": ModelProvider.GEMINI_2_5_FLASH,
"fallback": ModelProvider.DEEPSEEK_V3_2,
"max_latency_ms": 100
},
"code_generation": {
"primary": ModelProvider.GPT_4_1,
"fallback": ModelProvider.CLAUDE_SONNET_4_5,
"max_latency_ms": 500
},
"long_context": {
"primary": ModelProvider.GEMINI_2_5_FLASH, # 32K context
"fallback": ModelProvider.DEEPSEEK_V3_2,
"max_latency_ms": 800
},
"budget_friendly": {
"primary": ModelProvider.DEEPSEEK_V3_2, # $0.42/MTok
"fallback": ModelProvider.GEMINI_2_5_FLASH,
"max_latency_ms": 300
},
"high_quality": {
"primary": ModelProvider.CLAUDE_SONNET_4_5,
"fallback": ModelProvider.GPT_4_1,
"max_latency_ms": 1000
}
}
def __init__(self, client: HolySheepMultiModelClient):
self.client = client
self.cost_tracker = CostTracker()
def select_model(self, use_case: str, priority: str = "balanced") -> ModelProvider:
"""
Chọn model phù hợp với use case
Args:
use_case: Loại use case ("quick_response", "code_generation", ...)
priority: Ưu tiên ("speed", "cost", "quality", "balanced")
"""
mapping = self.USE_CASE_MAPPING.get(use_case, self.USE_CASE_MAPPING["balanced"])
if priority == "cost":
return ModelProvider.DEEPSEEK_V3_2 # Luôn chọn rẻ nhất
elif priority == "speed":
return ModelProvider.GEMINI_2_5_FLASH # Nhanh nhất
elif priority == "quality":
return ModelProvider.CLAUDE_SONNET_4_5
else:
return mapping["primary"]
async def smart_completion(
self,
messages: List[Dict],
use_case: str = "balanced",
priority: str = "balanced",
**kwargs
) -> Dict:
"""
Smart completion - tự động chọn model và theo dõi chi phí
"""
model = self.select_model(use_case, priority)
try:
result = await self.client.chat_completion(
provider=model,
messages=messages,
**kwargs
)
# Ghi nhận chi phí
usage = result.get("usage", {})
self.cost_tracker.record_usage(
model=model.value,
input_tokens=