Là một senior backend engineer với 7 năm kinh nghiệm triển khai hệ thống AI tại các doanh nghiệp tài chính và fintech, tôi đã trải qua vô số lần "production down" vì environment confusion giữa development, staging và production. Bài viết này là playbook thực chiến mà tôi đã đúc kết sau hàng trăm lần deployment, giúp bạn xây dựng hệ thống AI API isolated hoàn chỉnh với chi phí tối ưu.
Vì Sao Cần AI API Environment Isolation?
Khi đội ngũ của tôi mở rộng từ 3 lên 30 developers trong dự án AI chatbot cho ngân hàng số, chúng tôi đối mặt với bài toán kinh điển: làm sao để test API mới mà không đốt ngân sách production? Cách tiếp cận cũ — dùng cùng một API key cho tất cả môi trường — dẫn đến hóa đơn $12,000/tháng thay vì $2,000. Kể từ đó, tôi luôn áp dụng strict environment isolation.
Bài Toán Thực Tế Mà HolySheep AI Giải Quyết
- Cost Leakage: Development environment vô tình gọi production API với giá cao
- Data Security: Dữ liệu test có thể lộn sang production dataset
- Rate Limit Collision: Staging và production tranh nhau quota
- Rollback Complexity: Không thể nhanh chóng revert về baseline
HolySheep AI cung cấp hệ thống API keys riêng biệt cho từng môi trường, tích hợp thanh toán qua WeChat/Alipay, và đặc biệt là độ trễ trung bình dưới 50ms — lý tưởng cho kiến trúc multi-environment.
Kiến Trúc AI API Environment Isolation
1. Phân Chia Môi Trường 3 Tầng
┌─────────────────────────────────────────────────────────┐
│ ARCHITECTURE LAYERS │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Development │ │ Staging │ │ Production │ │
│ │ (Local) │ │ (Pre-prod) │ │ (Live) │ │
│ └──────┬──────┘ └──────┬──────┘ └──────┬──────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ DEV_KEY_* │ │ STG_KEY_* │ │ PRD_KEY_* │ │
│ │ (Limited) │ │ (Standard) │ │ (Unlimited) │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │ │ │ │
│ └──────────────────┼──────────────────┘ │
│ ▼ │
│ ┌─────────────────────────┐ │
│ │ HolySheep API Proxy │ │
│ │ https://api.holysheep │ │
│ │ .ai/v1 │ │
│ └─────────────────────────┘ │
│ │ │
│ ┌──────────────────┼──────────────────┐ │
│ ▼ ▼ ▼ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ GPT-4.1 │ │ Claude 4.5 │ │ DeepSeek V3 │ │
│ │ $8/MTok │ │ $15/MTok │ │ $0.42/MTok │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘2. Cấu Hình Config cho Từng Môi Trường
# =============================================
config/environments.py
Author: HolySheep AI Integration Guide
=============================================
from enum import Enum
from pydantic_settings import BaseSettings
from functools import lru_cache
class Environment(str, Enum):
DEVELOPMENT = "development"
STAGING = "staging"
PRODUCTION = "production"
class HolySheepConfig(BaseSettings):
"""Cấu hình HolySheep API — KHÔNG sử dụng OpenAI/ Anthropic endpoints"""
# Base URL bắt buộc cho HolySheep
base_url: str = "https://api.holysheep.ai/v1"
# API Keys riêng cho từng môi trường
api_key_dev: str = "YOUR_DEV_HOLYSHEEP_KEY"
api_key_stg: str = "YOUR_STAGING_HOLYSHEEP_KEY"
api_key_prd: str = "YOUR_PRODUCTION_HOLYSHEEP_KEY"
# Timeout và retry configuration
timeout: int = 30
max_retries: int = 3
retry_delay: float = 1.0
# Rate limiting per environment
rate_limit_rpm: dict = {
Environment.DEVELOPMENT: 60,
Environment.STAGING: 300,
Environment.PRODUCTION: 1000
}
# Budget constraints
monthly_budget_usd: dict = {
Environment.DEVELOPMENT: 50,
Environment.STAGING: 500,
Environment.PRODUCTION: 10000
}
class Config:
env_file = ".env"
env_prefix = "HOLYSHEEP_"
@lru_cache()
def get_holy_sheep_config(env: Environment) -> HolySheepConfig:
"""Factory pattern để lấy config theo environment"""
config = HolySheepConfig()
# Inject correct API key
config.api_key = {
Environment.DEVELOPMENT: config.api_key_dev,
Environment.STAGING: config.api_key_stg,
Environment.PRODUCTION: config.api_key_prd
}[env]
# Environment-specific base URLs (nếu cần)
config.rate_limit = config.rate_limit_rpm[env]
config.budget = config.monthly_budget_usd[env]
return configTriển Khai HolySheep SDK Wrapper
Đây là phần quan trọng nhất — tôi sẽ chia sẻ SDK wrapper mà đội ngũ đã optimize qua 2 năm sử dụng thực tế. Wrapper này bao gồm automatic failover, cost tracking, và environment-aware routing.
# =============================================
services/ai_client.py
HolySheep AI SDK Wrapper với Environment Isolation
=============================================
import os
import time
import logging
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import defaultdict
import hashlib
Third-party imports
import httpx
from openai import OpenAI, AsyncOpenAI
from tenacity import retry, stop_after_attempt, wait_exponential
from config.environments import Environment, get_holy_sheep_config
logger = logging.getLogger(__name__)
@dataclass
class UsageMetrics:
"""Theo dõi chi phí và usage theo từng môi trường"""
total_tokens: int = 0
prompt_tokens: int = 0
completion_tokens: int = 0
total_cost_usd: float = 0.0
request_count: int = 0
error_count: int = 0
avg_latency_ms: float = 0.0
last_updated: datetime = field(default_factory=datetime.now)
class HolySheepAIClient:
"""
Production-ready AI client với HolySheep integration.
Tính năng:
- Environment isolation
- Automatic cost tracking
- Rate limiting
- Circuit breaker pattern
- Cost budget enforcement
"""
# Pricing from HolySheep (2026/MTok)
PRICING = {
"gpt-4.1": {"prompt": 8.0, "completion": 8.0}, # $8/MTok
"claude-sonnet-4.5": {"prompt": 15.0, "completion": 15.0}, # $15/MTok
"gemini-2.5-flash": {"prompt": 2.5, "completion": 2.5}, # $2.50/MTok
"deepseek-v3.2": {"prompt": 0.42, "completion": 0.42}, # $0.42/MTok
}
def __init__(self, environment: Environment = Environment.DEVELOPMENT):
self.env = environment
self.config = get_holy_sheep_config(environment)
self.metrics = UsageMetrics()
# Initialize HolySheep client — base_url từ config
self.client = OpenAI(
api_key=self.config.api_key,
base_url=self.config.base_url, # https://api.holysheep.ai/v1
timeout=self.config.timeout,
max_retries=0 # We handle retries manually
)
self.async_client = AsyncOpenAI(
api_key=self.config.api_key,
base_url=self.config.base_url,
timeout=self.config.timeout
)
# Circuit breaker state
self._failure_count = 0
self._circuit_open = False
self._circuit_open_time = None
self.CIRCUIT_BREAKER_THRESHOLD = 5
self.CIRCUIT_BREAKER_TIMEOUT = 60
logger.info(
f"HolySheep client initialized for {environment.value} | "
f"Rate limit: {self.config.rate_limit} RPM | "
f"Budget: ${self.config.budget}/month"
)
def _check_circuit_breaker(self):
"""Kiểm tra circuit breaker trước mỗi request"""
if self._circuit_open:
if time.time() - self._circuit_open_time > self.CIRCUIT_BREAKER_TIMEOUT:
self._circuit_open = False
self._failure_count = 0
logger.info("Circuit breaker reset - resuming requests")
else:
raise RuntimeError("Circuit breaker is OPEN - too many failures")
def _record_failure(self):
"""Ghi nhận failure cho circuit breaker"""
self._failure_count += 1
self.metrics.error_count += 1
if self._failure_count >= self.CIRCUIT_BREAKER_THRESHOLD:
self._circuit_open = True
self._circuit_open_time = time.time()
logger.error(f"Circuit breaker OPENED after {self._failure_count} failures")
def _record_success(self):
"""Ghi nhận success"""
self._failure_count = max(0, self._failure_count - 1)
def _calculate_cost(self, model: str, usage: Dict[str, int]) -> float:
"""Tính chi phí dựa trên pricing HolySheep"""
pricing = self.PRICING.get(model, {"prompt": 8.0, "completion": 8.0})
prompt_cost = (usage.get("prompt_tokens", 0) / 1_000_000) * pricing["prompt"]
completion_cost = (usage.get("completion_tokens", 0) / 1_000_000) * pricing["completion"]
return prompt_cost + completion_cost
def _enforce_budget(self, estimated_cost: float):
"""Kiểm tra và enforce budget constraints"""
if self.metrics.total_cost_usd + estimated_cost > self.config.budget:
raise PermissionError(
f"Budget exceeded for {self.env.value}: "
f"${self.metrics.total_cost_usd:.2f}/${self.config.budget:.2f}"
)
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10))
def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
temperature: float = 0.7,
max_tokens: Optional[int] = None,
**kwargs
) -> Dict[str, Any]:
"""
Gọi chat completion qua HolySheep API.
Args:
model: Model name (gpt-4.1, claude-sonnet-4.5, deepseek-v3.2, etc.)
messages: List of message objects
temperature: Sampling temperature
max_tokens: Maximum tokens in response
Returns:
OpenAI-style response dictionary
"""
self._check_circuit_breaker()
start_time = time.time()
try:
# Budget check
estimated_cost = self._calculate_cost(model, {
"prompt_tokens": sum(len(m["content"].split()) * 1.3 for m in messages),
"completion_tokens": max_tokens or 1000
})
self._enforce_budget(estimated_cost)
# Make request to HolySheep
response = self.client.chat.completions.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
**kwargs
)
# Calculate actual cost
actual_cost = self._calculate_cost(model, response.usage.model_dump())
# Update metrics
self.metrics.total_tokens += response.usage.total_tokens
self.metrics.prompt_tokens += response.usage.prompt_tokens
self.metrics.completion_tokens += response.usage.completion_tokens
self.metrics.total_cost_usd += actual_cost
self.metrics.request_count += 1
self.metrics.avg_latency_ms = (
(self.metrics.avg_latency_ms * (self.metrics.request_count - 1) +
(time.time() - start_time) * 1000) / self.metrics.request_count
)
self._record_success()
logger.debug(
f"[{self.env.value}] {model} | "
f"Tokens: {response.usage.total_tokens} | "
f"Cost: ${actual_cost:.4f} | "
f"Latency: {(time.time() - start_time)*1000:.0f}ms"
)
return response.model_dump()
except Exception as e:
self._record_failure()
logger.error(f"[{self.env.value}] Request failed: {str(e)}")
raise
def get_usage_report(self) -> Dict[str, Any]:
"""Generate usage report cho environment hiện tại"""
return {
"environment": self.env.value,
"total_requests": self.metrics.request_count,
"total_tokens": self.metrics.total_tokens,
"prompt_tokens": self.metrics.prompt_tokens,
"completion_tokens": self.metrics.completion_tokens,
"total_cost_usd": round(self.metrics.total_cost_usd, 4),
"budget_remaining_usd": round(
self.config.budget - self.metrics.total_cost_usd, 4
),
"avg_latency_ms": round(self.metrics.avg_latency_ms, 2),
"error_rate": round(
self.metrics.error_count / max(1, self.metrics.request_count) * 100, 2
)
}
=============================================
Sử dụng trong ứng dụng
=============================================
Factory function
def create_ai_client(env: Optional[str] = None) -> HolySheepAIClient:
"""Tạo AI client dựa trên environment variable hoặc parameter"""
env = env or os.getenv("APP_ENV", "development")
return HolySheepAIClient(Environment(env))
Singleton cho dependency injection
_ai_client: Optional[HolySheepAIClient] = None
def get_ai_client() -> HolySheepAIClient:
global _ai_client
if _ai_client is None:
_ai_client = create_ai_client()
return _ai_clientChiến Lược Migration Từ API Cũ Sang HolySheep
Bước 1: Assessment và Inventory
# =============================================
scripts/migration/inventory_api_usage.py
Scan toàn bộ codebase để inventory API usage
=============================================
import ast
import os
import re
from pathlib import Path
from typing import Dict, List, Set
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class APIEndpoint:
file_path: str
line_number: int
endpoint_type: str # openai, anthropic, custom
method: str # chat.completion, embedding, etc.
model: str
is_critical: bool
class APIUsageInventory:
"""Inventory tất cả API usage trong codebase"""
# Patterns để detect API calls
PATTERNS = {
"openai": [
r'openai\.(?:api_key|base_url)',
r'client\.chat\.completions',
r'OpenAI\(',
r'azure_openai',
],
"anthropic": [
r'anthropic\.(?:api_key|base_url)',
r'client\.messages',
r'Anthropic\(',
],
"holy_sheep": [
r'api\.holysheep\.ai',
r'HolySheep',
r'HOLYSHEEP',
]
}
def __init__(self, project_root: str):
self.project_root = Path(project_root)
self.endpoints: List[APIEndpoint] = []
self.stats: Dict[str, int] = defaultdict(int)
def scan_file(self, file_path: Path) -> List[APIEndpoint]:
"""Scan một file Python để tìm API usage"""
endpoints = []
try:
content = file_path.read_text(encoding='utf-8')
lines = content.split('\n')
for i, line in enumerate(lines, 1):
# Check for old API patterns
if re.search(r'api\.openai\.com', line, re.IGNORECASE):
endpoints.append(APIEndpoint(
file_path=str(file_path),
line_number=i,
endpoint_type="openai",
method=self._extract_method(line),
model=self._extract_model(line),
is_critical=self._is_critical_path(line)
))
elif re.search(r'api\.anthropic\.com', line, re.IGNORECASE):
endpoints.append(APIEndpoint(
file_path=str(file_path),
line_number=i,
endpoint_type="anthropic",
method=self._extract_method(line),
model=self._extract_model(line),
is_critical=self._is_critical_path(line)
))
except Exception as e:
print(f"Error scanning {file_path}: {e}")
return endpoints
def _extract_method(self, line: str) -> str:
"""Extract method name từ line"""
methods = ["chat.completions.create", "embeddings.create",
"images.generate", "client.messages.create"]
for m in methods:
if m in line:
return m
return "unknown"
def _extract_model(self, line: str) -> str:
"""Extract model name từ line"""
models = ["gpt-4", "gpt-3.5", "claude-3", "claude-2"]
for m in models:
if m in line.lower():
return m
return "unknown"
def _is_critical_path(self, line: str) -> bool:
"""Check nếu đây là critical path (production, payment, etc.)"""
critical_indicators = ["production", "payment", "order", "transaction", "critical"]
return any(ind in line.lower() for ind in critical_indicators)
def run_scan(self) -> Dict:
"""Run full scan của project"""
print(f"Scanning {self.project_root}...")
for py_file in self.project_root.rglob("*.py"):
# Skip venv, node_modules, etc.
if any(skip in str(py_file) for skip in ['venv', 'node_modules', '.git', '__pycache__']):
continue
endpoints = self.scan_file(py_file)
self.endpoints.extend(endpoints)
# Generate statistics
self.stats["total_files"] = len(set(e.file_path for e in self.endpoints))
self.stats["total_calls"] = len(self.endpoints)
self.stats["openai_calls"] = len([e for e in self.endpoints if e.endpoint_type == "openai"])
self.stats["anthropic_calls"] = len([e for e in self.endpoints if e.endpoint_type == "anthropic"])
self.stats["critical_paths"] = len([e for e in self.endpoints if e.is_critical])
return {
"statistics": dict(self.stats),
"endpoints": [e.__dict__ for e in self.endpoints]
}
Sử dụng
if __name__ == "__main__":
inventory = APIUsageInventory("./your_project")
results = inventory.run_scan()
print("\n" + "="*60)
print("API USAGE INVENTORY REPORT")
print("="*60)
print(f"Total files with API calls: {results['statistics']['total_files']}")
print(f"Total API calls found: {results['statistics']['total_calls']}")
print(f"OpenAI calls to migrate: {results['statistics']['openai_calls']}")
print(f"Anthropic calls to migrate: {results['statistics']['anthropic_calls']}")
print(f"Critical paths requiring testing: {results['statistics']['critical_paths']}")
# Export to JSON for further processing
import json
with open("api_inventory_report.json", "w") as f:
json.dump(results, f, indent=2)Bước 2: Migration Script Tự Động
# =============================================
scripts/migration/migrate_to_holysheep.py
Automated migration script với rollback capability
=============================================
import os
import re
import shutil
import subprocess
from pathlib import Path
from datetime import datetime
from typing import Dict, List, Optional
import json
class HolySheepMigration:
"""
Migration tool để chuyển từ OpenAI/Anthropic sang HolySheep.
Hỗ trợ dry-run mode và automatic rollback.
"""
# Migration rules
REPLACEMENTS = {
# OpenAI -> HolySheep
r'api\.openai\.com': 'api.holysheep.ai',
r'https?://api\.openai\.com/v1': 'https://api.holysheep.ai/v1',
r'"openai"': '"holysheep"',
r"'openai'": "'holysheep'",
# Anthropic -> HolySheep
r'api\.anthropic\.com': 'api.holysheep.ai',
r'https?://api\.anthropic\.com/v1': 'https://api.holysheep.ai/v1',
# Model name mapping
r'gpt-4': 'gpt-4.1',
r'gpt-3\.5-turbo': 'gpt-3.5-turbo',
r'claude-3-opus': 'claude-sonnet-4.5',
r'claude-3-sonnet': 'claude-sonnet-4.5',
r'claude-3-haiku': 'claude-sonnet-4.5',
}
def __init__(self, project_root: str, dry_run: bool = True):
self.project_root = Path(project_root)
self.dry_run = dry_run
self.backup_dir = self.project_root / f".migration_backup_{datetime.now():%Y%m%d_%H%M%S}"
self.changes_made: List[Dict] = []
def _create_backup(self):
"""Tạo backup trước khi migrate"""
if self.dry_run:
print("[DRY RUN] Skipping backup creation")
return
print(f"Creating backup at {self.backup_dir}")
shutil.copytree(
self.project_root,
self.backup_dir,
ignore=shutil.ignore_patterns(
'__pycache__', '*.pyc', '.git', 'node_modules',
'venv', '.venv', '*.log'
),
dirs_exist_ok=True
)
print("Backup created successfully")
def _apply_replacements(self, content: str, file_path: Path) -> tuple[str, List[Dict]]:
"""Apply tất cả replacements cho một file"""
new_content = content
changes = []
for pattern, replacement in self.REPLACEMENTS.items():
matches = list(re.finditer(pattern, new_content))
if matches:
new_content = re.sub(pattern, replacement, new_content)
changes.append({
"file": str(file_path),
"pattern": pattern,
"replacement": replacement,
"count": len(matches)
})
return new_content, changes
def migrate_file(self, file_path: Path) -> Optional[Dict]:
"""Migrate một file"""
try:
content = file_path.read_text(encoding='utf-8')
new_content, changes = self._apply_replacements(content, file_path)
if not changes:
return None
if not self.dry_run:
file_path.write_text(new_content, encoding='utf-8')
return {
"file": str(file_path),
"changes": changes,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
return {
"file": str(file_path),
"error": str(e)
}
def run(self) -> Dict:
"""Run full migration"""
print("="*60)
print("HOLYSHEEP MIGRATION TOOL")
print("="*60)
print(f"Mode: {'DRY RUN' if self.dry_run else 'LIVE MIGRATION'}")
print(f"Project: {self.project_root}")
print("="*60)
self._create_backup()
# Find all Python files
py_files = list(self.project_root.rglob("*.py"))
py_files = [f for f in py_files if '__pycache__' not in str(f)]
print(f"\nFound {len(py_files)} Python files to scan")
# Scan and migrate
results = {
"dry_run": self.dry_run,
"timestamp": datetime.now().isoformat(),
"files_scanned": 0,
"files_modified": 0,
"total_changes": 0,
"details": []
}
for py_file in py_files:
results["files_scanned"] += 1
migration_result = self.migrate_file(py_file)
if migration_result:
results["files_modified"] += 1
results["total_changes"] += sum(
c["count"] for c in migration_result["changes"]
)
results["details"].append(migration_result)
if self.dry_run:
print(f" [WOULD MODIFY] {py_file}")
for change in migration_result["changes"]:
print(f" {change['pattern']} -> {change['replacement']} ({change['count']} occurrences)")
else:
print(f" [MODIFIED] {py_file}")
# Summary
print("\n" + "="*60)
print("MIGRATION SUMMARY")
print("="*60)
print(f"Files scanned: {results['files_scanned']}")
print(f"Files modified: {results['files_modified']}")
print(f"Total replacements: {results['total_changes']}")
if self.dry_run:
print("\nTo run actual migration, set dry_run=False")
print("Backup available at:", self.backup_dir)
# Save report
report_path = self.project_root / f"migration_report_{datetime.now():%Y%m%d_%H%M%S}.json"
with open(report_path, "w") as f:
json.dump(results, f, indent=2)
print(f"\nReport saved to: {report_path}")
return results
def rollback(self):
"""Rollback migration từ backup"""
if not self.backup_dir.exists():
print("No backup found to rollback")
return
print(f"Rolling back from {self.backup_dir}")
# Restore files
for item in self.backup_dir.rglob("*"):
if item.is_file():
relative = item.relative_to(self.backup_dir)
target = self.project_root / relative
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(item, target)
print("Rollback completed successfully")
Sử dụng
if __name__ == "__main__":
import sys
project_path = sys.argv[1] if len(sys.argv) > 1 else "."
dry_mode = "--dry-run" in sys.argv
migration = HolySheepMigration(project_path, dry_run=dry_mode)
results = migration.run()
# Rollback nếu cần
if "--rollback" in sys.argv:
migration.rollback()Rollback Plan và Disaster Recovery
Một trong những bài học đắt giá nhất của tôi là: luôn có rollback plan trước khi deploy. Dưới đây là checklist và implementation đã được test qua nhiều lần incident thực tế.
# =============================================
deployment/rollback_manager.py
Rollback manager với automatic failover
=============================================
import os
import time
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, Optional, Callable
from dataclasses import dataclass, asdict
from enum import Enum
from pathlib import Path
import httpx
from openai import OpenAI
from config.environments import Environment
logger = logging.getLogger(__name__)
class RollbackTrigger(str, Enum):
HIGH_ERROR_RATE = "high_error_rate"
HIGH_LATENCY = "high_latency"
BUDGET_EXCEEDED = "budget_exceeded"
MANUAL = "manual"
HEALTH_CHECK_FAILED = "health_check_failed"
@dataclass
class HealthCheckResult:
status: str # healthy, degraded, unhealthy
latency_ms: float
error_rate: float
timestamp: datetime
details: Dict
class HolySheepRollbackManager:
"""
Rollback manager với automatic health checking và failover.
Hỗ trợ chuyển đổi nhanh giữa HolySheep và fallback providers.
"""
def __init__(
self,
primary_base_url: str = "https://api.holysheep.ai/v1",
fallback_base_url: Optional[str] = None,
health_check_interval: int = 60,
error_threshold: float = 0.05,
latency_threshold_ms: float = 5000
):
self.primary_base_url = primary_base_url
self.fallback_base_url = fallback_base_url
self.health_check_interval = health_check_interval
self.error_threshold = error_threshold
self.latency_threshold_ms = latency_threshold_ms
self.is_primary_active = True
self.health_history: list[HealthCheckResult] = []
self.rollback_history: list[Dict] = []
# Metrics
self.total_requests = 0
self.failed_requests = 0
self.rollbacks_triggered = 0
async def health_check(self, base_url: str) -> HealthCheckResult:
"""Kiểm tra health của một endpoint"""
start_time = time.time()
errors = 0
try:
# Test với simple completion
client = OpenAI(api_key="test", base_url=base_url, timeout=10)
async with httpx.AsyncClient(timeout=10) as http_client:
response = await http_client.post(
f"{base_url}/chat/completions",
json={
"model": "gpt-3.5-turbo",
"messages": [{"role": "user", "content": "ping"}],
"max_tokens": 1
},
headers={"Authorization": f"Bearer test"}
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code in [200, 401, 403]: # Auth errors are OK for health check
status = "healthy"
else:
status = "degraded"
errors = 1
except Exception as e:
latency_ms = (time.time() - start_time) * 1000
status = "unhealthy"
errors = 1
logger.error(f"Health check failed: {e}")
result = HealthCheckResult(
status=status,
latency_ms=latency_ms,
error_rate=1.0 if errors > 0 else 0.0,
timestamp=datetime.now(),
details={"base_url": base_url, "errors": errors}
)
self.health_history.append(result)
# Keep only last 100 checks
self.health_history = self.health_history[-100:]
return result
def should_rollback(self) -> tuple[bool, Optional[RollbackTrigger]]:
"""Kiểm tra xem có nên rollback không"""
if not self.health_history:
return False, None
recent_checks = [
h for h in self.health_history
if h.timestamp > datetime.now() - timedelta(minutes=5)
]
if not recent_checks:
return False, None
# Check error rate
avg_error_rate = sum(h.error_rate for h in recent_checks) / len(recent_checks)
if avg_error_rate > self.error_threshold:
return True, RollbackTrigger.HIGH_ERROR_RATE
# Check latency
avg_latency = sum(h.latency_ms for h in recent_checks) / len(recent_checks)
if avg_latency > self.latency_threshold_ms:
return True, RollbackTrigger.HIGH_LATENCY
# Check consecutive failures
recent_statuses = [h.status for h in recent_checks[-3:]]
if all(s == "unhealthy" for s in recent_statuses):
return True, RollbackTrigger.HEALTH_CHECK_FAILED