Đêm qua, đội ngũ kỹ thuật của một công ty fintech lớn đã trải qua 3 tiếng đồng hồ debugging một lỗi kinh điển: ConnectionError: timeout after 30s xảy ra liên tục khi production AI agent gọi API. Sau khi kiểm tra, họ phát hiện nguyên nhân là OpenAI rate limit — ứng dụng chỉ có duy nhất một API key và không có fallback mechanism. 5.000 khách hàng bị gián đoạn dịch vụ, doanh thu thiệt hại ước tính 12.000 USD trong đêm.
Bài viết này là hướng dẫn toàn diện giúp bạn xây dựng Enterprise Agent Platform với các tính năng: unified API key management, intelligent fallback, invoice tracking và model-level audit logging. Tất cả được triển khai thực tế với HolySheep AI — nền tảng tích hợp 20+ mô hình AI với chi phí tiết kiệm đến 85%.
Tại Sao Enterprise Cần Unified Agent Platform?
Khi doanh nghiệp mở rộng AI implementation, họ thường gặp các vấn đề phổ biến:
- API Key Hell: Mỗi nhà cung cấp (OpenAI, Anthropic, Google) có API key riêng, mỗi key có rate limit và pricing khác nhau
- Single Point of Failure: Chỉ dùng một provider duy nhất → khi provider downtime, toàn bộ hệ thống ngừng hoạt động
- Cost Blindness: Không ai biết team nào dùng model gì, tiêu tốn bao nhiêu tiền
- Audit Trail Missing: Không thể truy vết request nào thuộc về user/department nào
- Compliance Risk: Không có log đầy đủ cho audit quarterly hay regulatory compliance
Kiến Trúc Tổng Quan Enterprise Agent Platform
┌─────────────────────────────────────────────────────────────────┐
│ ENTERPRISE AGENT PLATFORM │
├─────────────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
│ │ API Gateway │──│ Load Balancer│──│ Intelligent Router │ │
│ │ (Unified) │ │ │ │ - Fallback Strategy │ │
│ └─────────────┘ └─────────────┘ │ - Cost Optimizer │ │
│ │ - Latency Aware │ │
│ └─────────────────────────┘ │
│ ┌─────────────────────────────────────────────────────────────┐│
│ │ UNIFIED API KEY MANAGER ││
│ │ • Single credential for all providers ││
│ │ • Automatic key rotation ││
│ │ • Rate limit management ││
│ └─────────────────────────────────────────────────────────────┘│
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ HolySheep │ │ OpenAI │ │ Anthropic │ │
│ │ API (20+) │ │ Compatible │ │ Compatible │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
│ ┌─────────────────────────────────────────────────────────────┐│
│ │ AUDIT & INVOICE LAYER ││
│ │ • Real-time cost tracking per model/user/department ││
│ │ • Invoice generation ││
│ │ • Compliance-ready logs ││
│ └─────────────────────────────────────────────────────────────┘│
└─────────────────────────────────────────────────────────────────┘
Triển Khai Chi Tiết: Unified API Key Management
HolySheep cung cấp unified API key duy nhất truy cập 20+ models từ các nhà cung cấp khác nhau. Bạn không cần quản lý nhiều key, không cần theo dõi rate limit riêng cho từng provider.
# ========================================
UNIFIED API KEY CLIENT - HolySheep AI
Base URL: https://api.holysheep.ai/v1
========================================
import requests
import json
from typing import Optional, Dict, Any, List
from datetime import datetime
import hashlib
class UnifiedAgentClient:
"""
Enterprise-grade unified client cho HolySheep AI Platform.
- Single API key truy cập 20+ models
- Automatic fallback khi provider fail
- Real-time cost tracking
- Audit logging tự động
"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json',
'X-Client-Version': 'enterprise-v2.0',
'X-Platform': 'HolySheep-Unified'
})
# Audit log storage
self.audit_logs: List[Dict] = []
# Available models với priority và cost
self.models = {
'gpt-4.1': {'provider': 'openai', 'cost_per_1k': 8.00, 'priority': 1, 'latency_ms': 800},
'claude-sonnet-4.5': {'provider': 'anthropic', 'cost_per_1k': 15.00, 'priority': 2, 'latency_ms': 1200},
'gemini-2.5-flash': {'provider': 'google', 'cost_per_1k': 2.50, 'priority': 3, 'latency_ms': 400},
'deepseek-v3.2': {'provider': 'deepseek', 'cost_per_1k': 0.42, 'priority': 4, 'latency_ms': 600}
}
def _log_audit(self, model: str, request_data: Dict, response_data: Dict,
latency_ms: float, cost_usd: float, status: str):
"""Tự động log mọi request cho audit trail"""
audit_entry = {
'timestamp': datetime.utcnow().isoformat(),
'model': model,
'request_tokens': request_data.get('tokens', 0),
'response_tokens': response_data.get('tokens', 0),
'latency_ms': round(latency_ms, 2),
'cost_usd': round(cost_usd, 4),
'status': status,
'request_id': hashlib.md5(f"{datetime.now()}{model}".encode()).hexdigest()[:12]
}
self.audit_logs.append(audit_entry)
print(f"[AUDIT] {audit_entry['timestamp']} | Model: {model} | "
f"Latency: {latency_ms}ms | Cost: ${cost_usd:.4f} | Status: {status}")
return audit_entry['request_id']
def chat_completion(self, messages: List[Dict], model: str = 'deepseek-v3.2',
fallback_enabled: bool = True, user_id: Optional[str] = None,
department: Optional[str] = None) -> Dict[str, Any]:
"""
Unified chat completion với intelligent fallback.
Args:
messages: List of message dicts [{role, content}]
model: Primary model (default: deepseek-v3.2 - cheapest)
fallback_enabled: Enable automatic fallback khi primary fail
user_id: User identifier for audit
department: Department for cost allocation
"""
import time
start_time = time.time()
attempt_order = [model] + [m for m in self.models.keys() if m != model]
last_error = None
for attempt_model in attempt_order:
if not fallback_enabled and attempt_model != model:
break
try:
print(f"[INFO] Requesting model: {attempt_model} "
f"(Fallback: {attempt_model != model})")
# Build request
payload = {
'model': attempt_model,
'messages': messages,
'temperature': 0.7,
'max_tokens': 2000
}
# Add metadata for tracking
if user_id or department:
payload['metadata'] = {
'user_id': user_id,
'department': department
}
# Make request
response = self.session.post(
f'{self.base_url}/chat/completions',
json=payload,
timeout=30
)
response.raise_for_status()
result = response.json()
# Calculate cost
tokens_used = result.get('usage', {}).get('total_tokens', 1000)
cost = (tokens_used / 1000) * self.models[attempt_model]['cost_per_1k']
# Log successful request
latency_ms = (time.time() - start_time) * 1000
request_id = self._log_audit(
model=attempt_model,
request_data={'tokens': tokens_used},
response_data=result,
latency_ms=latency_ms,
cost_usd=cost,
status='SUCCESS'
)
result['request_id'] = request_id
result['actual_model'] = attempt_model
result['fallback_used'] = attempt_model != model
result['cost_usd'] = cost
return result
except requests.exceptions.Timeout:
last_error = f"Timeout calling {attempt_model}"
print(f"[WARNING] {last_error}")
continue
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
last_error = f"Rate limit for {attempt_model}"
print(f"[WARNING] {last_error}")
continue
elif e.response.status_code == 401:
raise Exception(f"Invalid API key. Check your HolySheep key.")
else:
last_error = f"HTTP {e.response.status_code}: {str(e)}"
print(f"[ERROR] {last_error}")
if not fallback_enabled:
raise
continue
except Exception as e:
last_error = str(e)
print(f"[ERROR] Unexpected error: {last_error}")
if not fallback_enabled:
raise
continue
# All models failed
raise Exception(f"All models failed. Last error: {last_error}")
def get_cost_report(self, start_date: Optional[str] = None,
end_date: Optional[str] = None) -> Dict[str, Any]:
"""Generate cost report từ audit logs"""
filtered_logs = self.audit_logs
if start_date:
filtered_logs = [l for l in filtered_logs if l['timestamp'] >= start_date]
if end_date:
filtered_logs = [l for l in filtered_logs if l['timestamp'] <= end_date]
report = {
'total_requests': len(filtered_logs),
'total_cost_usd': sum(l['cost_usd'] for l in filtered_logs),
'total_latency_ms': sum(l['latency_ms'] for l in filtered_logs),
'by_model': {},
'avg_cost_per_request': 0,
'avg_latency_ms': 0
}
for log in filtered_logs:
model = log['model']
if model not in report['by_model']:
report['by_model'][model] = {
'requests': 0,
'total_cost': 0,
'total_latency': 0
}
report['by_model'][model]['requests'] += 1
report['by_model'][model]['total_cost'] += log['cost_usd']
report['by_model'][model]['total_latency'] += log['latency_ms']
if filtered_logs:
report['avg_cost_per_request'] = report['total_cost_usd'] / len(filtered_logs)
report['avg_latency_ms'] = report['total_latency_ms'] / len(filtered_logs)
return report
========================================
SỬ DỤNG THỰC TẾ
========================================
Khởi tạo client với unified API key
client = UnifiedAgentClient(
api_key="YOUR_HOLYSHEEP_API_KEY" # Key duy nhất cho 20+ models
)
Test unified completion
messages = [
{"role": "system", "content": "Bạn là trợ lý AI chuyên nghiệp cho doanh nghiệp."},
{"role": "user", "content": "Giải thích điều gì xảy ra khi OpenAI rate limit và cách xử lý?"}
]
try:
response = client.chat_completion(
messages=messages,
model='deepseek-v3.2', # Primary: rẻ nhất
fallback_enabled=True, # Tự động fallback khi fail
user_id='user_12345',
department='engineering'
)
print(f"\n✓ Response from: {response['actual_model']}")
print(f"✓ Fallback used: {response['fallback_used']}")
print(f"✓ Request ID: {response['request_id']}")
print(f"✓ Cost: ${response['cost_usd']:.4f}")
print(f"\nResponse:\n{response['choices'][0]['message']['content']}")
except Exception as e:
print(f"✗ Error: {e}")
Generate cost report
report = client.get_cost_report()
print(f"\n=== COST REPORT ===")
print(f"Total Requests: {report['total_requests']}")
print(f"Total Cost: ${report['total_cost_usd']:.2f}")
print(f"Avg Latency: {report['avg_latency_ms']:.0f}ms")
Intelligent Fallback Strategy: Không Bao Giờ Down
Một hệ thống enterprise thực sự cần fallback thông minh, không chỉ đơn giản là "thử cái khác". Dưới đây là triển khai production-ready với các chiến lược fallback đa tầng:
# ========================================
INTELLIGENT FALLBACK SYSTEM
Production-ready với multi-layer fallback
========================================
import time
import asyncio
from typing import List, Dict, Callable, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
from collections import defaultdict
import statistics
class FallbackStrategy(Enum):
COST_OPTIMIZED = "cost_optimized" # Ưu tiên model rẻ nhất
LATENCY_OPTIMIZED = "latency" # Ưu tiên model nhanh nhất
RELIABILITY = "reliability" # Ưu tiên model đáng tin cậy nhất
BALANCED = "balanced" # Cân bằng tất cả
@dataclass
class ModelMetrics:
"""Theo dõi performance của từng model"""
name: str
provider: str
success_count: int = 0
failure_count: int = 0
timeout_count: int = 0
total_latency_ms: List[float] = field(default_factory=list)
last_success: Optional[float] = None
last_failure: Optional[float] = None
@property
def success_rate(self) -> float:
total = self.success_count + self.failure_count
return self.success_count / total if total > 0 else 0.0
@property
def avg_latency(self) -> float:
return statistics.mean(self.total_latency_ms) if self.total_latency_ms else 999999
@dataclass
class FallbackConfig:
"""Configuration cho fallback system"""
max_retries: int = 3
timeout_ms: int = 30000
circuit_breaker_threshold: int = 5 # Failures before circuit opens
circuit_breaker_timeout: int = 60 # Seconds before retry
enable_adaptive_routing: bool = True
strategy: FallbackStrategy = FallbackStrategy.BALANCED
class IntelligentFallbackRouter:
"""
Production-grade fallback router với:
- Circuit breaker pattern
- Adaptive routing dựa trên real-time metrics
- Cost-aware fallback
- Latency-aware routing
"""
def __init__(self, api_key: str, config: FallbackConfig = None):
self.api_key = api_key
self.config = config or FallbackConfig()
self.base_url = "https://api.holysheep.ai/v1"
# Model registry với pricing và base characteristics
self.model_registry = {
'deepseek-v3.2': {
'provider': 'deepseek',
'cost_per_1k': 0.42,
'base_latency_ms': 600,
'reliability_score': 0.95,
'context_window': 64000
},
'gemini-2.5-flash': {
'provider': 'google',
'cost_per_1k': 2.50,
'base_latency_ms': 400,
'reliability_score': 0.98,
'context_window': 1000000
},
'gpt-4.1': {
'provider': 'openai',
'cost_per_1k': 8.00,
'base_latency_ms': 800,
'reliability_score': 0.92,
'context_window': 128000
},
'claude-sonnet-4.5': {
'provider': 'anthropic',
'cost_per_1k': 15.00,
'base_latency_ms': 1200,
'reliability_score': 0.97,
'context_window': 200000
}
}
# Real-time metrics tracking
self.model_metrics: Dict[str, ModelMetrics] = {
name: ModelMetrics(name=name, provider=data['provider'])
for name, data in self.model_registry.items()
}
# Circuit breaker state
self.circuit_state: Dict[str, str] = {name: 'closed' for name in self.model_registry}
self.circuit_opened_at: Dict[str, float] = {}
# Fallback chain definitions
self.fallback_chains = {
'cheap': ['deepseek-v3.2', 'gemini-2.5-flash', 'gpt-4.1'],
'fast': ['gemini-2.5-flash', 'deepseek-v3.2', 'gpt-4.1'],
'reliable': ['gemini-2.5-flash', 'claude-sonnet-4.5', 'gpt-4.1'],
'balanced': ['deepseek-v3.2', 'gemini-2.5-flash', 'claude-sonnet-4.5', 'gpt-4.1']
}
def _calculate_model_score(self, model_name: str) -> float:
"""
Tính điểm cho model dựa trên:
- Success rate (40%)
- Latency (30%)
- Cost efficiency (30%)
"""
metrics = self.model_metrics[model_name]
data = self.model_registry[model_name]
# Success rate score (higher is better)
success_score = metrics.success_rate * 100
# Latency score (lower is better, normalized)
latency_score = max(0, 100 - (metrics.avg_latency / 10))
# Cost score (lower is better, normalized)
cost_score = max(0, 100 - (data['cost_per_1k'] * 5))
# Weighted score
weighted_score = (success_score * 0.4) + (latency_score * 0.3) + (cost_score * 0.3)
return weighted_score
def _get_ranked_models(self, chain_type: str = 'balanced') -> List[str]:
"""Lấy danh sách models đã rank theo strategy"""
if self.config.enable_adaptive_routing:
# Adaptive: sắp xếp theo real-time score
models_with_scores = [
(name, self._calculate_model_score(name))
for name in self.model_registry.keys()
if self.circuit_state.get(name, 'closed') == 'closed'
]
models_with_scores.sort(key=lambda x: x[1], reverse=True)
return [m[0] for m in models_with_scores]
else:
# Static chain
return self.fallback_chains.get(chain_type, self.fallback_chains['balanced'])
def _update_metrics(self, model_name: str, success: bool, latency_ms: float):
"""Cập nhật metrics sau mỗi request"""
metrics = self.model_metrics[model_name]
if success:
metrics.success_count += 1
metrics.last_success = time.time()
else:
metrics.failure_count += 1
metrics.last_failure = time.time()
metrics.total_latency_ms.append(latency_ms)
# Keep only last 100 measurements
if len(metrics.total_latency_ms) > 100:
metrics.total_latency_ms = metrics.total_latency_ms[-100:]
# Update circuit breaker
if metrics.failure_count >= self.config.circuit_breaker_threshold:
self.circuit_state[model_name] = 'open'
self.circuit_opened_at[model_name] = time.time()
print(f"[CIRCUIT BREAKER] Opened for {model_name}")
def _check_circuit_breaker(self, model_name: str) -> bool:
"""Kiểm tra và quản lý circuit breaker"""
if self.circuit_state.get(model_name) == 'open':
opened_at = self.circuit_opened_at.get(model_name, 0)
if time.time() - opened_at > self.config.circuit_breaker_timeout:
self.circuit_state[model_name] = 'half-open'
print(f"[CIRCUIT BREAKER] Half-open for {model_name}")
return True
return False
return True
async def chat_completion_async(
self,
messages: List[Dict],
primary_model: str = 'deepseek-v3.2',
chain_type: str = 'balanced',
user_metadata: Optional[Dict] = None
) -> Dict[str, Any]:
"""
Async completion với intelligent fallback
"""
ranked_models = self._get_ranked_models(chain_type)
# Ensure primary model is first
if primary_model in ranked_models:
ranked_models.remove(primary_model)
ranked_models.insert(0, primary_model)
last_error = None
results = []
for model_name in ranked_models:
if not self._check_circuit_breaker(model_name):
print(f"[SKIP] Circuit breaker open for {model_name}")
continue
start_time = time.time()
try:
# Simulate API call (thay bằng actual request)
response = await self._make_request(
model=model_name,
messages=messages,
timeout=self.config.timeout_ms / 1000,
metadata=user_metadata
)
latency_ms = (time.time() - start_time) * 1000
self._update_metrics(model_name, success=True, latency_ms=latency_ms)
results.append({
'model': model_name,
'response': response,
'latency_ms': latency_ms,
'success': True,
'fallback_count': len(results)
})
return results[-1]
except asyncio.TimeoutError:
latency_ms = (time.time() - start_time) * 1000
self._update_metrics(model_name, success=False, latency_ms=latency_ms)
last_error = f"Timeout for {model_name}"
print(f"[WARNING] {last_error}")
continue
except Exception as e:
latency_ms = (time.time() - start_time) * 1000
self._update_metrics(model_name, success=False, latency_ms=latency_ms)
last_error = str(e)
print(f"[ERROR] {model_name}: {last_error}")
continue
# All models failed
raise Exception(f"All fallback models exhausted. Last error: {last_error}")
async def _make_request(
self,
model: str,
messages: List[Dict],
timeout: float,
metadata: Optional[Dict]
) -> Dict[str, Any]:
"""Simulate API request - thay bằng actual implementation"""
# Simulate network call
await asyncio.sleep(0.1) # Simulated latency
# Simulate occasional failures for testing
import random
if random.random() < 0.1: # 10% failure rate simulation
raise Exception(f"Simulated failure for {model}")
return {
'id': f'chatcmpl-{model}-{int(time.time())}',
'model': model,
'choices': [{
'index': 0,
'message': {
'role': 'assistant',
'content': f'Simulated response from {model}'
},
'finish_reason': 'stop'
}],
'usage': {
'prompt_tokens': 100,
'completion_tokens': 50,
'total_tokens': 150
}
}
def get_health_report(self) -> Dict[str, Any]:
"""Generate real-time health report cho tất cả models"""
return {
'models': {
name: {
'status': self.circuit_state[name],
'success_rate': f"{metrics.success_rate * 100:.1f}%",
'avg_latency_ms': f"{metrics.avg_latency:.0f}",
'score': f"{self._calculate_model_score(name):.1f}",
'total_requests': metrics.success_count + metrics.failure_count
}
for name, metrics in self.model_metrics.items()
},
'recommendations': self._get_recommendations()
}
def _get_recommendations(self) -> List[str]:
"""Đưa ra recommendations dựa trên metrics"""
recommendations = []
for name, metrics in self.model_metrics.items():
if metrics.failure_count > 10 and metrics.success_rate < 0.8:
recommendations.append(
f"{name} có success rate thấp ({metrics.success_rate * 100:.1f}%). "
f"Cân nhắc giảm priority hoặc kiểm tra provider status."
)
if metrics.avg_latency > 3000:
recommendations.append(
f"{name} có latency cao ({metrics.avg_latency:.0f}ms). "
f"Xem xét chỉ dùng làm fallback."
)
return recommendations if recommendations else ["Tất cả models hoạt động bình thường."]
========================================
DEMO: SỬ DỤNG INTELLIGENT FALLBACK
========================================
async def demo_intelligent_fallback():
"""Demo đầy đủ intelligent fallback system"""
router = IntelligentFallbackRouter(
api_key="YOUR_HOLYSHEEP_API_KEY",
config=FallbackConfig(
max_retries=3,
circuit_breaker_threshold=3,
enable_adaptive_routing=True,
strategy=FallbackStrategy.BALANCED
)
)
messages = [
{"role": "user", "content": "Phân tích các yếu tố ảnh hưởng đến giá USD/VND trong tháng 5/2026"}
]
# Test với fallback
print("=" * 60)
print("INTELLIGENT FALLBACK DEMO")
print("=" * 60)
try:
result = await router.chat_completion_async(
messages=messages,
primary_model='deepseek-v3.2', # Model rẻ nhất
chain_type='balanced',
user_metadata={
'user_id': 'enterprise_user_001',
'department': 'finance',
'request_type': 'analysis'
}
)
print(f"\n✓ SUCCESS")
print(f" Model: {result['model']}")
print(f" Latency: {result['latency_ms']:.0f}ms")
print(f" Fallback attempts: {result['fallback_count']}")
print(f" Response: {result['response']['choices'][0]['message']['content']}")
except Exception as e:
print(f"\n✗ FAILED: {e}")
# Show health report
print("\n" + "=" * 60)
print("MODEL HEALTH REPORT")
print("=" * 60)
health = router.get_health_report()
for model, status in health['models'].items():
print(f"\n{model}:")
print(f" Status: {status['status']}")
print(f" Success Rate: {status['success_rate']}")
print(f" Avg Latency: {status['avg_latency_ms']}ms")
print(f" Score: {status['score']}")
print("\nRecommendations:")
for rec in health['recommendations']:
print(f" • {rec}")
Run demo
if __name__ == "__main__":
asyncio.run(demo_intelligent_fallback())
Invoice Management Và Cost Allocation
Với HolySheep, bạn nhận được hóa đơn rõ ràng theo từng model, department, và user. Tất cả transactions được track chi tiết đến cent.
# ========================================
INVOICE MANAGEMENT & COST ALLOCATION
HolySheep Enterprise Billing System
========================================
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timedelta
from decimal import Decimal
import json
@dataclass
class Transaction:
"""Một transaction riêng lẻ"""
timestamp: datetime
model: str
tokens_used: int
cost_usd: Decimal
user_id: Optional[str]
department: Optional[str]
project: Optional[str]
request_id: str
latency_ms: float
status: str
@dataclass
class InvoiceLine:
"""Một dòng trên hóa đơn"""
description: str
quantity: int
unit_price: Decimal
total_usd: Decimal
line_type: str # 'usage', 'subscription', 'credit'
class InvoiceManager:
"""
Quản lý invoice và cost allocation cho enterprise
- Theo dõi chi tiêu theo department, user, project
- Tạo invoice reports
- Áp dụng discounts và credits
"""
# HolySheep Pricing (Updated 2026-05-19)
PRICING = {
'deepseek-v3.2': Decimal('0.42'), # $0.42/1M tokens
'gemini-2.5-flash': Decimal('2.50'), # $2.50/1M tokens
'gpt-4.1': Decimal('8.00'), # $8.00/1M tokens
'claude-sonnet-4.5': Decimal('15.00'), # $15.00/1M tokens
# ... thêm 20+ models khác
}
def __init__(self, billing_email: str, company_name: str):
self.billing_email = billing_email
self.company_name = company_name
self.transactions: List[Transaction] = []
self.department_budgets: Dict[str, Decimal] = {}
self.user_credits: Dict[str, Decimal] = {}
def add_transaction(self, transaction: Transaction):
"""Thêm transaction mới"""
self.transactions.append(transaction)
def calculate_cost(self, model: str, tokens: int) -> Decimal:
"""Tính chi phí cho một request"""
price_per_million = self.PRICING.get(model, Decimal('1.00'))
return (Decimal(tokens) / Decimal('1000000')) * price_per_million
def get_department_spending(self, department: str,
start_date: datetime = None,
end_date: datetime = None) -> Dict:
"""Lấy chi tiêu theo department"""
filtered = self.transactions
if start_date:
filtered = [t for t in filtered if t.timestamp >= start_date]
if end_date:
filtered = [t for t in filtered if t.timestamp <= end_date]
dept_transactions = [t for t in filtered if t.department == department]
total_cost = sum(t.cost_usd for t in dept_transactions)
total_tokens = sum(t.tokens_used for t in dept_transactions)
total_requests = len(dept_transactions)
# Breakdown by model
by_model = {}
for t in dept_transactions:
if t.model not in by_model:
by_model[t.model] = {'cost': Decimal('0'), 'tokens': 0, 'requests': 0}
by_model[t.model]['cost'] += t.cost_usd
by_model[t.model]['tokens'] += t.tokens_used
by_model[t.model]['requests'] += 1
return {
'department': department,
'total_cost_usd': total_cost,
'total_tokens': total_tokens,
'total_requests': total_requests,
'budget': self.department_budgets.get(department, Decimal('0')),
'budget_used_percent': (total_cost / self.department_budgets[department] *