Chào các bạn, mình là Minh — Senior Backend Engineer với 6 năm kinh nghiệm xây dựng hệ thống xử lý AI tự động. Tuần vừa rồi, đội ngũ mình vừa hoàn thành một cuộc di chuyển lớn: chuyển toàn bộ API từ nhà cung cấp cũ sang HolySheep AI với mục tiêu tiết kiệm chi phí và tăng độ ổn định hệ thống. Trong bài viết này, mình sẽ chia sẻ chi tiết cách implement Circuit Breaker và Bulkhead Pattern — hai pattern quan trọng nhất giúp hệ thống AI API của chúng mình đạt uptime 99.7%.
Bối Cảnh: Vì Sao Chúng Tôi Cần Tối Ưu?
Trước khi đi vào chi tiết kỹ thuật, mình muốn kể cho các bạn nghe về bài toán thực tế mà đội ngũ đã gặp phải. Hồi tháng 8/2025, hệ thống chatbot AI của công ty mình phục vụ khoảng 50,000 requests mỗi ngày. Chúng tôi đang dùng một nhà cung cấp API với chi phí khá cao: GPT-4o Mini chạy $0.15/1K tokens, và mỗi tháng chúng tôi phải chi khoảng $3,200 chỉ riêng tiền API.
Vấn đề không chỉ nằm ở chi phí. Hệ thống cũ thường xuyên gặp tình trạng:
- Response time không ổn định, có lúc lên tới 8-12 giây
- Timeout không kiểm soát được, ảnh hưởng trực tiếp tới trải nghiệm người dùng
- Khi một service bị lỗi, toàn bộ hệ thống bị cascade failure
- Không có cơ chế fallback tự động
Sau khi research và benchmark nhiều giải pháp, chúng tôi quyết định chuyển sang HolySheep AI với các lý do chính:
- Chi phí thấp hơn 85%: Tỷ giá ¥1=$1 có nghĩa là các model giá rẻ hơn rất nhiều so với giá quốc tế. DeepSeek V3.2 chỉ $0.42/MTok so với $2-3 của các nhà cung cấp khác.
- Độ trễ thấp: Dưới 50ms latency, đảm bảo trải nghiệm mượt mà cho người dùng.
- Hỗ trợ thanh toán địa phương: WeChat Pay, Alipay — thuận tiện cho các doanh nghiệp Trung Quốc.
- Tín dụng miễn phí khi đăng ký: Có thể test trước khi commit.
Kiến Trúc Hệ Thống Với Circuit Breaker Pattern
Circuit Breaker Pattern Là Gì?
Circuit Breaker Pattern hoạt động giống như автоматический выключатель (automatic circuit breaker) trong điện lưới. Khi một service gọi API liên tục thất bại, circuit breaker sẽ "nhảy" để ngăn các request tiếp tục đổ vào service đó, thay vào đó trả về response fallback ngay lập tức.
Ba trạng thái của Circuit Breaker:
- CLOSED: Hoạt động bình thường, tất cả request được đi qua.
- OPEN: Circuit "nhảy", request được chặn, trả về fallback ngay.
- HALF-OPEN: Cho phép một số request thử lại để kiểm tra service đã hồi phục chưa.
Implement Circuit Breaker Với HolySheep AI
Dưới đây là implementation hoàn chỉnh Circuit Breaker Pattern bằng Python sử dụng tenacity library và kết nối với HolySheep AI API:
import httpx
import asyncio
from tenacity import (
retry,
stop_after_attempt,
wait_exponential,
retry_if_exception_type
)
from enum import Enum
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional, Dict, Any
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
Cấu hình HolySheep AI
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
class CircuitState(Enum):
CLOSED = "closed"
OPEN = "open"
HALF_OPEN = "half_open"
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Số lần thất bại để open circuit
success_threshold: int = 2 # Số lần thành công để close circuit
timeout_duration: int = 30 # Thời gian (giây) trước khi thử half-open
half_open_max_calls: int = 3 # Số call được phép trong half-open
@dataclass
class CircuitBreakerMetrics:
failures: int = 0
successes: int = 0
last_failure_time: Optional[datetime] = None
state: CircuitState = CircuitState.CLOSED
half_open_calls: int = 0
class CircuitBreaker:
def __init__(self, name: str, config: CircuitBreakerConfig = None):
self.name = name
self.config = config or CircuitBreakerConfig()
self.metrics = CircuitBreakerMetrics()
self._lock = asyncio.Lock()
async def call(self, func, *args, **kwargs):
"""Execute function với circuit breaker protection"""
async with self._lock:
# Kiểm tra timeout để chuyển sang half-open
if self.metrics.state == CircuitState.OPEN:
if self._should_attempt_reset():
logger.info(f"Circuit {self.name}: OPEN -> HALF_OPEN")
self.metrics.state = CircuitState.HALF_OPEN
self.metrics.half_open_calls = 0
# Nếu đang OPEN và chưa timeout, reject ngay
if self.metrics.state == CircuitState.OPEN:
raise CircuitBreakerOpenError(f"Circuit {self.name} is OPEN")
# Nếu đang HALF_OPEN và đã đạt limit, reject
if self.metrics.state == CircuitState.HALF_OPEN:
if self.metrics.half_open_calls >= self.config.half_open_max_calls:
raise CircuitBreakerOpenError(
f"Circuit {self.name} half-open limit reached"
)
self.metrics.half_open_calls += 1
# Thực thi request
try:
result = await func(*args, **kwargs)
await self._on_success()
return result
except Exception as e:
await self._on_failure()
raise
async def _on_success(self):
async with self._lock:
self.metrics.failures = 0
self.metrics.successes += 1
if self.metrics.state == CircuitState.HALF_OPEN:
if self.metrics.successes >= self.config.success_threshold:
logger.info(f"Circuit {self.name}: HALF_OPEN -> CLOSED")
self.metrics.state = CircuitState.CLOSED
self.metrics.successes = 0
elif self.metrics.state == CircuitState.CLOSED:
# Reset counter khi có success liên tiếp
pass
async def _on_failure(self):
async with self._lock:
self.metrics.failures += 1
self.metrics.last_failure_time = datetime.now()
self.metrics.successes = 0
if self.metrics.state == CircuitState.HALF_OPEN:
logger.warning(f"Circuit {self.name}: HALF_OPEN -> OPEN (failed)")
self.metrics.state = CircuitState.OPEN
elif (self.metrics.failures >= self.config.failure_threshold and
self.metrics.state == CircuitState.CLOSED):
logger.warning(f"Circuit {self.name}: CLOSED -> OPEN")
self.metrics.state = CircuitState.OPEN
def _should_attempt_reset(self) -> bool:
if self.metrics.last_failure_time is None:
return True
elapsed = datetime.now() - self.metrics.last_failure_time
return elapsed.total_seconds() >= self.config.timeout_duration
class CircuitBreakerOpenError(Exception):
pass
Client cho HolySheep AI
class HolySheepAIClient:
def __init__(self, api_key: str):
self.api_key = api_key
self.circuit_breaker = CircuitBreaker(
name="holysheep_main",
config=CircuitBreakerConfig(
failure_threshold=5,
success_threshold=2,
timeout_duration=30
)
)
self.client = httpx.AsyncClient(
base_url=HOLYSHEEP_BASE_URL,
timeout=30.0
)
async def chat_completion(
self,
model: str,
messages: list,
temperature: float = 0.7,
fallback_response: str = "Xin lỗi, hệ thống đang bận. Vui lòng thử lại sau."
) -> Dict[str, Any]:
"""Gọi API với circuit breaker và fallback"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": temperature
}
async def _make_request():
response = await self.client.post(
"/chat/completions",
json=payload,
headers=headers
)
response.raise_for_status()
return response.json()
try:
result = await self.circuit_breaker.call(_make_request)
logger.info(f"Success: {result.get('model', 'unknown')}")
return result
except CircuitBreakerOpenError:
logger.warning("Circuit breaker open - returning fallback")
return {
"fallback": True,
"content": fallback_response,
"error": "Service temporarily unavailable"
}
except httpx.HTTPStatusError as e:
logger.error(f"HTTP Error: {e.response.status_code}")
raise
except Exception as e:
logger.error(f"Unexpected error: {str(e)}")
raise
Sử dụng
async def main():
client = HolySheepAIClient(HOLYSHEEP_API_KEY)
messages = [
{"role": "system", "content": "Bạn là trợ lý AI thân thiện."},
{"role": "user", "content": "Cho tôi biết về xu hướng AI năm 2026"}
]
try:
# Sử dụng DeepSeek V3.2 - giá chỉ $0.42/MTok
result = await client.chat_completion(
model="deepseek-v3.2",
messages=messages
)
print(result)
except Exception as e:
print(f"Lỗi: {e}")
if __name__ == "__main__":
asyncio.run(main())Bulkhead Pattern: Cô Lập Resource Để Tránh Cascade Failure
Tại Sao Cần Bulkhead Pattern?
Bulkhead Pattern lấy cảm hứng từ vách ngăn trong thân tàu. Khi một ngăn bị thủng nước sẽ không tràn sang các ngăn khác. Trong hệ thống AI API, điều này có nghĩa là: nếu một loại request ngốn quá nhiều resource, nó không được phép ảnh hưởng đến các request khác.
Ví dụ thực tế tại công ty mình:
- Priority Queue: Request từ khách hàng VIP được ưu tiên, có dedicated connection pool riêng.
- Batch Processing: Các job xử lý hàng loạt chạy trong pool riêng, không ảnh hưởng realtime chat.
- Health Check: Monitoring requests không bị ảnh hưởng bởi traffic spike.
Implement Bulkhead Pattern Với Connection Pooling
Đây là implementation Bulkhead Pattern với separate connection pools cho từng loại request:
import asyncio
import httpx
from dataclasses import dataclass
from typing import Dict, Optional
from contextlib import asynccontextmanager
import time
@dataclass
class BulkheadConfig:
max_connections: int # Tổng số connection tối đa
max_keepalive_connections: int # Số connection keep-alive
keepalive_expiry: int # Thời gian sống của connection (giây)
request_timeout: float # Timeout cho mỗi request
class BulkheadPool:
"""Một bulkhead - pool riêng biệt cho một loại request"""
def __init__(self, name: str, config: BulkheadConfig):
self.name = name
self.config = config
self._client: Optional[httpx.AsyncClient] = None
self._semaphore: asyncio.Semaphore = None
self._metrics = {
"active_requests": 0,
"total_requests": 0,
"rejected_requests": 0,
"total_latency": 0.0
}
self._lock = asyncio.Lock()
async def initialize(self):
"""Khởi tạo connection pool"""
self._semaphore = asyncio.Semaphore(self.config.max_connections)
self._client = httpx.AsyncClient(
timeout=httpx.Timeout(self.config.request_timeout),
limits=httpx.Limits(
max_connections=self.config.max_connections,
max_keepalive_connections=self.config.max_keepalive_connections,
keepalive_expiry=self.config.keepalive_expiry
)
)
print(f"Bulkhead '{self.name}' initialized: "
f"{self.config.max_connections} connections")
async def close(self):
"""Đóng connection pool"""
if self._client:
await self._client.aclose()
@asynccontextmanager
async def acquire(self):
"""Acquire một connection từ pool với semaphore"""
async with self._lock:
self._metrics["active_requests"] += 1
self._metrics["total_requests"] += 1
start_time = time.time()
try:
# Thử acquire với timeout
async with asyncio.timeout(5.0): # 5s timeout chờ
await self._semaphore.acquire()
try:
yield self._client
finally:
self._semaphore.release()
except asyncio.TimeoutError:
async with self._lock:
self._metrics["rejected_requests"] += 1
raise BulkheadCapacityError(
f"Bulkhead '{self.name}' at capacity - request rejected"
)
finally:
async with self._lock:
self._metrics["active_requests"] -= 1
elapsed = time.time() - start_time
self._metrics["total_latency"] += elapsed
def get_metrics(self) -> Dict:
return {
"bulkhead": self.name,
**self._metrics,
"avg_latency": (
self._metrics["total_latency"] / self._metrics["total_requests"]
if self._metrics["total_requests"] > 0 else 0
)
}
class BulkheadCapacityError(Exception):
pass
class HolySheepBulkheadManager:
"""Manager quản lý nhiều bulkhead pools cho HolySheep AI"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self._pools: Dict[str, BulkheadPool] = {}
self._initialize_pools()
def _initialize_pools(self):
"""Khởi tạo các bulkhead pools riêng biệt"""
# Pool cho request VIP - ưu tiên cao nhất
self._pools["vip"] = BulkheadPool(
name="vip_requests",
config=BulkheadConfig(
max_connections=50,
max_keepalive_connections=20,
keepalive_expiry=300,
request_timeout=60.0
)
)
# Pool cho request thường - standard tier
self._pools["standard"] = BulkheadPool(
name="standard_requests",
config=BulkheadConfig(
max_connections=100,
max_keepalive_connections=40,
keepalive_expiry=120,
request_timeout=30.0
)
)
# Pool cho batch processing - chạy nền
self._pools["batch"] = BulkheadPool(
name="batch_processing",
config=BulkheadConfig(
max_connections=30,
max_keepalive_connections=10,
keepalive_expiry=600,
request_timeout=120.0
)
)
async def initialize(self):
"""Initialize tất cả pools"""
await asyncio.gather(*[pool.initialize() for pool in self._pools.values()])
print("All bulkhead pools initialized")
async def close(self):
"""Đóng tất cả pools"""
await asyncio.gather(*[pool.close() for pool in self._pools.values()])
async def chat_completion(
self,
model: str,
messages: list,
priority: str = "standard"
) -> dict:
"""
Gửi chat completion request với bulkhead isolation
Args:
model: Model AI (deepseek-v3.2, gpt-4.1, claude-sonnet-4.5, etc.)
messages: Danh sách messages
priority: 'vip', 'standard', hoặc 'batch'
"""
if priority not in self._pools:
priority = "standard"
pool = self._pools[priority]
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": 0.7
}
async with pool.acquire() as client:
response = await client.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers
)
response.raise_for_status()
return response.json()
def get_all_metrics(self) -> Dict:
"""Lấy metrics từ tất cả pools"""
return {
name: pool.get_metrics()
for name, pool in self._pools.items()
}
Demo sử dụng
async def demo_bulkhead():
"""Demo sử dụng bulkhead pattern"""
client = HolySheepBulkheadManager("YOUR_HOLYSHEEP_API_KEY")
await client.initialize()
try:
# VIP request - priority cao
vip_messages = [
{"role": "user", "content": "Phân tích chiến lược kinh doanh Q4 2026"}
]
result = await client.chat_completion(
model="deepseek-v3.2", # $0.42/MTok - tiết kiệm 85%
messages=vip_messages,
priority="vip"
)
print(f"VIP Result: {result}")
# Batch processing - priority thấp
batch_messages = [
{"role": "user", "content": "Tạo 10 captions cho bài đăng mạng xã hội"}
]
result = await client.chat_completion(
model="gemini-2.5-flash", # $2.50/MTok - nhanh và rẻ
messages=batch_messages,
priority="batch"
)
print(f"Batch Result: {result}")
# Print metrics
print("\n=== Bulkhead Metrics ===")
for name, metrics in client.get_all_metrics().items():
print(f"{name}: {metrics}")
finally:
await client.close()
if __name__ == "__main__":
asyncio.run(demo_bulkhead())Kết Hợp Circuit Breaker + Bulkhead + Rate Limiting
Đây là phần mình muốn chia sẻ kinh nghiệm quý báu: không nên chỉ dùng một pattern đơn lẻ, mà kết hợp cả ba để tạo thành một "defense in depth" hoàn chỉnh. Dưới đây là production-ready implementation:
import asyncio
import httpx
import time
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Callable
from enum import Enum
from collections import defaultdict
import hashlib
Cấu hình HolySheep AI
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
@dataclass
class RateLimitConfig:
requests_per_minute: int = 60
requests_per_hour: int = 1000
tokens_per_minute: int = 100000
class TokenBucket:
"""Token Bucket algorithm cho rate limiting"""
def __init__(self, capacity: int, refill_rate: float):
self.capacity = capacity
self.tokens = capacity
self.refill_rate = refill_rate # tokens/second
self.last_refill = time.time()
self._lock = asyncio.Lock()
async def acquire(self, tokens: int = 1) -> bool:
"""Try to acquire tokens, return True if successful"""
async with self._lock:
self._refill()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
def _refill(self):
"""Refill tokens based on elapsed time"""
now = time.time()
elapsed = now - self.last_refill
refill_amount = elapsed * self.refill_rate
self.tokens = min(self.capacity, self.tokens + refill_amount)
self.last_refill = now
@dataclass
class RequestMetrics:
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
rejected_requests: int = 0
total_latency: float = 0.0
total_cost: float = 0.0
by_model: Dict[str, int] = field(default_factory=lambda: defaultdict(int))
class HolySheepResilienceFramework:
"""
Production-grade resilience framework kết hợp:
- Circuit Breaker Pattern
- Bulkhead Pattern
- Rate Limiting
- Fallback Strategy
- Retry Logic
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
# Circuit breakers cho từng model
self.circuit_breakers: Dict[str, 'SimpleCircuitBreaker'] = {}
self._init_circuit_breakers()
# Bulkhead pools
self.bulkhead_pools: Dict[str, asyncio.Semaphore] = {}
self._init_bulkhead_pools()
# Rate limiters
self.rate_limiters: Dict[str, TokenBucket] = {}
self._init_rate_limiters()
# Metrics
self.metrics = RequestMetrics()
# HTTP client
self.client = httpx.AsyncClient(
timeout=httpx.Timeout(60.0),
limits=httpx.Limits(max_connections=200, max_keepalive_connections=50)
)
# Pricing (2026/MTok)
self.model_pricing = {
"deepseek-v3.2": 0.42, # $0.42/MTok
"gpt-4.1": 8.0, # $8/MTok
"claude-sonnet-4.5": 15.0, # $15/MTok
"gemini-2.5-flash": 2.50, # $2.50/MTok
}
# Fallback chain
self.fallback_chain = [
"deepseek-v3.2", # Ưu tiên - rẻ nhất
"gemini-2.5-flash", # Fallback 1
"gpt-4.1", # Fallback 2
]
def _init_circuit_breakers(self):
"""Khởi tạo circuit breaker cho mỗi model"""
for model in self.model_pricing.keys():
self.circuit_breakers[model] = SimpleCircuitBreaker(
name=model,
failure_threshold=3,
timeout=30
)
def _init_bulkhead_pools(self):
"""Khởi tạo bulkhead pools"""
self.bulkhead_pools = {
"vip": asyncio.Semaphore(50),
"standard": asyncio.Semaphore(100),
"batch": asyncio.Semaphore(30),
}
def _init_rate_limiters(self):
"""Khởi tạo rate limiters"""
self.rate_limiters = {
"global": TokenBucket(capacity=60, refill_rate=1.0), # 60 req/min
"deepseek": TokenBucket(capacity=120, refill_rate=2.0), # 120 req/min
"gpt": TokenBucket(capacity=30, refill_rate=0.5),
"claude": TokenBucket(capacity=20, refill_rate=0.33),
}
async def close(self):
await self.client.aclose()
async def call_with_resilience(
self,
messages: List[Dict],
model: Optional[str] = None,
priority: str = "standard",
use_fallback: bool = True,
max_retries: int = 2
) -> Dict:
"""
Gọi API với đầy đủ resilience patterns
Args:
messages: Danh sách chat messages
model: Model cụ thể, hoặc None để dùng fallback chain
priority: 'vip', 'standard', hoặc 'batch'
use_fallback: Có dùng fallback chain không
max_retries: Số lần retry tối đa
"""
self.metrics.total_requests += 1
# Chọn model và fallback chain
if model:
models_to_try = [model]
else:
models_to_try = self.fallback_chain if use_fallback else [self.fallback_chain[0]]
# Acquire bulkhead
bulkhead = self.bulkhead_pools.get(priority, self.bulkhead_pools["standard"])
async with bulkhead:
# Thử từng model trong chain
for attempt_model in models_to_try:
try:
result = await self._attempt_request(
attempt_model,
messages,
max_retries
)
# Update metrics
self.metrics.successful_requests += 1
self.metrics.by_model[attempt_model] += 1
# Reset circuit breaker của model thành công
if attempt_model in self.circuit_breakers:
self.circuit_breakers[attempt_model].record_success()
return result
except CircuitOpenError:
continue # Thử model tiếp theo
except RateLimitError:
await asyncio.sleep(1) # Chờ rồi thử model tiếp theo
continue
except Exception as e:
# Record failure cho circuit breaker
if attempt_model in self.circuit_breakers:
self.circuit_breakers[attempt_model].record_failure()
continue
# Tất cả đều thất bại
self.metrics.rejected_requests += 1
return {
"error": "All models unavailable",
"fallback_content": "Xin lỗi, hệ thống đang quá tải. "
"Vui lòng thử lại sau 5 phút."
}
async def _attempt_request(
self,
model: str,
messages: List[Dict],
max_retries: int
) -> Dict:
"""Thực hiện một request với retry logic"""
# Kiểm tra circuit breaker
cb = self.circuit_breakers.get(model)
if cb and not cb.can_execute():
raise CircuitOpenError(f"Circuit open for {model}")
# Kiểm tra rate limit
rate_key = model.split("-")[0] if "-" in model else "global"
limiter = self.rate_limiters.get(rate_key, self.rate_limiters["global"])
if not await limiter.acquire(1):
raise RateLimitError(f"Rate limit exceeded for {rate_key}")
# Retry loop
last_error = None
for attempt in range(max_retries + 1):
start_time = time.time()
try:
result = await self._make_request(model, messages)
# Tính cost
latency = time.time() - start_time
prompt_tokens = result.get("usage", {}).get("prompt_tokens", 0)
completion_tokens = result.get("usage", {}).get("completion_tokens", 0)
total_tokens = prompt_tokens + completion_tokens
cost = (total_tokens / 1_000_000) * self.model_pricing.get(model, 1.0)
# Update metrics
self.metrics.total_latency += latency
self.metrics.total_cost += cost
return result
except httpx.TimeoutException:
last_error = "Timeout"
await asyncio.sleep(0.5 * (attempt + 1)) # Exponential backoff
except httpx.HTTPStatusError as e:
if e.response.status_code == 429:
last_error = "Rate limited"
await asyncio.sleep(2 ** attempt) # Longer backoff
elif e.response.status_code >= 500:
last_error = f"Server error: {e.response.status_code}"
else:
raise
except Exception as e:
last_error = str(e)
raise
raise Exception(f"All retries failed: {last_error}")
async def _make_request(self, model: str, messages: List[Dict]) -> Dict:
"""Thực hiện HTTP request đến HolySheep AI"""
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
"temperature": 0.7
}
response = await self.client.post(
f"{self.base_url}/chat/completions",
json=payload,
headers=headers
)
response.raise_for_status()
return response.json()
def get_cost_report(self) -> Dict:
"""Generate báo cáo chi phí"""
return {
"total_requests": self.metrics.total_requests,
"successful_requests": self.metrics.successful_requests,
"failed_requests": self.metrics.failed_requests,
"rejected_requests": self.metrics.rejected_requests,
"success_rate": (
self.metrics.successful_requests / self.metrics.total_requests * 100
if self.metrics.total_requests > 0 else 0
),
"total_cost_usd": self.metrics.total_cost,
"avg_latency_ms": (
self.metrics.total_latency / self.metrics.total_requests * 1000
if self.metrics.total_requests > 0 else 0
),
"by_model": dict(self.metrics.by_model),
"estimated_monthly_cost": self.metrics.total_cost * 30 # Ước tính
}
Helper classes
class SimpleCircuitBreaker: