Trong một đêm cao điểm production, hệ thống chatbot của tôi đột nhiên chết hoàn toàn. ConnectionError: timeout exceeded 30s — hàng nghìn người dùng không thể truy cập. Đó là khoảnh khắc tôi nhận ra rằng kiến trúc single-point-of-failure là con dao hai lưỡi. Bài viết này là toàn bộ bài học xương máu của tôi khi xây dựng hệ thống Claude API high-availability với HolySheep AI.
Tại Sao Cần Kiến Trúc High-Availability?
Với HolySheep AI, chi phí chỉ $15/MTok cho Claude Sonnet 4.5 (so với $80+ tại các provider khác), nhưng downtime vẫn có thể gây thiệt hại nghiêm trọng. Một kiến trúc HA đúng cách đảm bảo:
- Uptime 99.9%+ — thời gian chết dưới 8.7 giờ/năm
- Auto-failover — chuyển đổi endpoint trong 500ms
- Load balancing — phân phối request đều khắp các instance
- Rate limit handling — không miss request quan trọng
Kiến Trúc Tổng Quan
+------------------+ +-------------------+ +------------------+
| Load Balancer |----->| API Gateway |----->| HolySheep API |
| (nginx/haproxy)| | (Health Check) | | api.holysheep.ai|
+------------------+ +-------------------+ +------------------+
| |
v v
+------------------+ +-------------------+
| Circuit Breaker | | Response Cache |
| (Resilience4j) | | (Redis/TTL) |
+------------------+ +-------------------+
|
v
+------------------+ +-------------------+
| Fallback Models |----->| Queue System |
| (GPT-4.1 $8) | | (Redis Queue) |
+------------------+ +-------------------+1. Cấu Hình Connection Pool Và Retry Logic
Đây là lớp nền tảng quan trọng nhất. Tôi đã mất 3 ngày debug một lỗi 401 Unauthorized chỉ vì không config đúng timeout.
# config/holy_sheep_client.py
import httpx
from typing import Optional, Dict, Any
from tenacity import retry, stop_after_attempt, wait_exponential
import asyncio
class HolySheepClaudeClient:
"""Client cấp doanh nghiệp với high-availability"""
def __init__(
self,
api_key: str = "YOUR_HOLYSHEEP_API_KEY",
base_url: str = "https://api.holysheep.ai/v1",
max_connections: int = 100,
max_keepalive_connections: int = 20,
timeout: float = 60.0,
max_retries: int = 3
):
self.api_key = api_key
self.base_url = base_url
# Connection pool config — điều này QUAN TRỌNG
limits = httpx.Limits(
max_connections=max_connections,
max_keepalive_connections=max_keepalive_connections,
keepalive_expiry=30.0
)
# Timeout strategy phân tầng
self.timeout = httpx.Timeout(
timeout,
connect=10.0, # Connect timeout
read=45.0, # Read timeout
write=10.0, # Write timeout
pool=5.0 # Pool acquisition timeout
)
self.client = httpx.AsyncClient(
limits=limits,
timeout=self.timeout,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"X-Request-ID": "", # Tracing ID
}
)
# Circuit breaker state
self._failure_count = 0
self._circuit_open = False
self._last_failure_time = None
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
async def chat_completions(
self,
messages: list,
model: str = "claude-sonnet-4-20250514",
temperature: float = 0.7,
max_tokens: int = 4096,
**kwargs
) -> Dict[str, Any]:
"""
Gọi Claude API với retry logic tự động.
"""
if self._circuit_open:
raise CircuitBreakerOpenError(
"Circuit breaker is OPEN. Fallback activated."
)
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
**kwargs
}
try:
response = await self.client.post(
f"{self.base_url}/chat/completions",
json=payload
)
# Xử lý HTTP status code
if response.status_code == 200:
self._on_success()
return response.json()
elif response.status_code == 401:
# KHÔNG retry 401 — API key lỗi
raise AuthenticationError(
f"Invalid API key: {response.text}"
)
elif response.status_code == 429:
# Rate limit — retry với exponential backoff
retry_after = int(
response.headers.get("Retry-After", 5)
)
await asyncio.sleep(retry_after)
raise RateLimitError("Rate limit exceeded")
else:
raise APIError(
f"HTTP {response.status_code}: {response.text}"
)
except httpx.TimeoutException as e:
self._on_failure()
raise TimeoutError(f"Request timeout: {e}")
except httpx.ConnectError as e:
self._on_failure()
raise ConnectionError(f"Connection failed: {e}")
def _on_success(self):
"""Reset circuit breaker khi thành công"""
self._failure_count = 0
self._circuit_open = False
def _on_failure(self):
"""Tăng failure count, open circuit nếu vượt threshold"""
self._failure_count += 1
self._last_failure_time = asyncio.get_event_loop().time()
# Open circuit sau 5 failures liên tiếp
if self._failure_count >= 5:
self._circuit_open = True
print(f"[WARNING] Circuit breaker OPENED at {self._failure_count} failures")
class CircuitBreakerOpenError(Exception):
pass
class RateLimitError(Exception):
pass
class AuthenticationError(Exception):
pass
class APIError(Exception):
pass2. Multi-Endpoint Load Balancer Với Health Check
Tôi triển khai một load balancer thông minh kiểm tra health endpoint mỗi 10 giây và tự động loại bỏ endpoint không khả dụng.
# services/load_balancer.py
import asyncio
import time
from dataclasses import dataclass, field
from typing import List, Optional
import httpx
from collections import deque
@dataclass
class Endpoint:
url: str
name: str
healthy: bool = True
latency_ms: float = 0.0
failure_streak: int = 0
last_check: float = field(default_factory=time.time)
response_times: deque = field(default_factory=lambda: deque(maxlen=10))
@property
def avg_latency(self) -> float:
if not self.response_times:
return 0.0
return sum(self.response_times) / len(self.response_times)
class HolySheepLoadBalancer:
"""
Load balancer với weighted round-robin và health check.
Ưu tiên endpoint có latency thấp nhất.
"""
def __init__(
self,
endpoints: List[str],
health_check_interval: int = 10,
failure_threshold: int = 3
):
self.endpoints = [
Endpoint(url=url, name=f"endpoint-{i}")
for i, url in enumerate(endpoints)
]
self.health_check_interval = health_check_interval
self.failure_threshold = failure_threshold
self._health_check_task = None
self._lock = asyncio.Lock()
# Primary endpoint index
self._current_idx = 0
async def start(self):
"""Khởi động background health check"""
self._health_check_task = asyncio.create_task(
self._health_check_loop()
)
print("[LOAD_BALANCER] Started with health check")
async def stop(self):
"""Dừng health check"""
if self._health_check_task:
self._health_check_task.cancel()
async def get_endpoint(self) -> Endpoint:
"""
Chọn endpoint tốt nhất dựa trên:
1. Health status
2. Average latency
3. Failure streak
"""
async with self._lock:
# Lọc endpoint healthy
healthy = [ep for ep in self.endpoints if ep.healthy]
if not healthy:
# Fallback: chọn endpoint có ít failure streak nhất
fallback = min(
self.endpoints,
key=lambda x: x.failure_streak
)
print(f"[WARNING] No healthy endpoints. Using fallback: {fallback.name}")
return fallback
# Weighted selection: ưu tiên latency thấp
# Endpoint có latency < 100ms được ưu tiên 3x
weights = []
for ep in healthy:
if ep.avg_latency < 100:
weights.append(3)
elif ep.avg_latency < 300:
weights.append(2)
else:
weights.append(1)
total_weight = sum(weights)
import random
rand_val = random.uniform(0, total_weight)
cumulative = 0
for i, ep in enumerate(healthy):
cumulative += weights[i]
if rand_val <= cumulative:
return ep
return healthy[0]
async def _health_check_loop(self):
"""Background health check task"""
while True:
try:
await asyncio.sleep(self.health_check_interval)
await self._check_all_endpoints()
except asyncio.CancelledError:
break
except Exception as e:
print(f"[HEALTH_CHECK] Error: {e}")
async def _check_all_endpoints(self):
"""Kiểm tra health của tất cả endpoints"""
tasks = [self._check_endpoint(ep) for ep in self.endpoints]
await asyncio.gather(*tasks, return_exceptions=True)
async def _check_endpoint(self, endpoint: Endpoint):
"""Health check đơn lẻ"""
health_url = f"{endpoint.url}/health"
try:
start = time.perf_counter()
async with httpx.AsyncClient() as client:
response = await client.get(
health_url,
timeout=5.0
)
latency = (time.perf_counter() - start) * 1000
endpoint.latency_ms = latency
endpoint.response_times.append(latency)
if response.status_code == 200:
endpoint.healthy = True
endpoint.failure_streak = 0
print(f"[HEALTH] {endpoint.name}: OK ({latency:.1f}ms)")
else:
await self._mark_unhealthy(endpoint)
except Exception as e:
await self._mark_unhealthy(endpoint)
async def _mark_unhealthy(self, endpoint: Endpoint):
"""Đánh dấu endpoint không khả dụng"""
endpoint.failure_streak += 1
endpoint.healthy = endpoint.failure_streak < self.failure_threshold
if not endpoint.healthy:
print(f"[HEALTH] {endpoint.name}: FAILED ({endpoint.failure_streak} streaks)")
Usage
async def main():
lb = HolySheepLoadBalancer(
endpoints=[
"https://api.holysheep.ai/v1",
"https://backup.holysheep.ai/v1", # Backup region
],
health_check_interval=10,
failure_threshold=3
)
await lb.start()
# Test lấy endpoint
endpoint = await lb.get_endpoint()
print(f"Selected endpoint: {endpoint.name} ({endpoint.url})")
await lb.stop()
if __name__ == "__main__":
asyncio.run(main())3. Circuit Breaker Pattern Với Fallback
Khi HolySheep API gặp sự cố, hệ thống phải tự động chuyển sang model fallback mà không ảnh hưởng trải nghiệm người dùng. Tôi sử dụng GPT-4.1 ($8/MTok) như fallback chính.
# services/circuit_breaker.py
import asyncio
import time
from enum import Enum
from typing import Callable, Any, Optional
from dataclasses import dataclass
class CircuitState(Enum):
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject requests
HALF_OPEN = "half_open" # Testing recovery
@dataclass
class CircuitBreakerConfig:
failure_threshold: int = 5 # Open after N failures
recovery_timeout: int = 60 # Try recovery after N seconds
half_open_max_calls: int = 3 # Max test calls in half-open
class ClaudeCircuitBreaker:
"""
Circuit breaker cho Claude API với multi-model fallback.
Fallback chain:
Claude Sonnet 4.5 ($15/MTok)
→ GPT-4.1 ($8/MTok)
→ DeepSeek V3.2 ($0.42/MTok)
"""
def __init__(self, config: Optional[CircuitBreakerConfig] = None):
self.config = config or CircuitBreakerConfig()
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time = None
self.half_open_calls = 0
# Model configs với pricing
self.models = [
{
"name": "claude-sonnet-4-20250514",
"provider": "holy_sheep",
"price_per_mtok": 15.0,
"quality": "high"
},
{
"name": "gpt-4.1",
"provider": "holy_sheep",
"price_per_mtok": 8.0,
"quality": "high"
},
{
"name": "deepseek-v3.2",
"provider": "holy_sheep",
"price_per_mtok": 0.42,
"quality": "medium"
}
]
def _should_allow_request(self) -> bool:
"""Kiểm tra xem có nên cho request đi qua không"""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
# Kiểm tra đã đến lúc thử recovery chưa
if time.time() - self.last_failure_time >= self.config.recovery_timeout:
self.state = CircuitState.HALF_OPEN
self.half_open_calls = 0
print("[CIRCUIT] OPEN → HALF_OPEN (recovery test)")
return True
return False
if self.state == CircuitState.HALF_OPEN:
return self.half_open_calls < self.config.half_open_max_calls
return False
async def call(
self,
client: Any,
messages: list,
current_model_idx: int = 0
) -> Any:
"""
Gọi API với circuit breaker và fallback tự động.
"""
if not self._should_allow_request():
# Chuyển sang model tiếp theo
return await self._fallback_to_next_model(
client, messages, current_model_idx
)
model = self.models[current_model_idx]
try:
if self.state == CircuitState.HALF_OPEN:
self.half_open_calls += 1
response = await client.chat_completions(
messages=messages,
model=model["name"]
)
self._on_success()
return response
except Exception as e:
print(f"[CIRCUIT] Model {model['name']} failed: {e}")
self._on_failure()
# Thử model tiếp theo
return await self._fallback_to_next_model(
client, messages, current_model_idx
)
async def _fallback_to_next_model(
self,
client: Any,
messages: list,
current_idx: int
) -> Any:
"""Fallback sang model rẻ hơn"""
next_idx = current_idx + 1
if next_idx >= len(self.models):
# Đã thử hết models, raise error
raise AllModelsFailedError(
"All Claude models failed. Last resort: queue for retry."
)
next_model = self.models[next_idx]
print(f"[FALLBACK] Switching to {next_model['name']} "
f"(${next_model['price_per_mtok']}/MTok)")
return await self.call(client, messages, next_idx)
def _on_success(self):
"""Xử lý khi call thành công"""
self.failure_count = 0
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.half_open_max_calls:
self.state = CircuitState.CLOSED
self.success_count = 0
print("[CIRCUIT] HALF_OPEN → CLOSED (recovery successful)")
else:
self.state = CircuitState.CLOSED
def _on_failure(self):
"""Xử lý khi call thất bại"""
self.failure_count += 1
self.success_count = 0
self.last_failure_time = time.time()
if self.state == CircuitState.HALF_OPEN:
# Thất bại trong recovery test → open lại
self.state = CircuitState.OPEN
print("[CIRCUIT] HALF_OPEN → OPEN (recovery failed)")
elif self.failure_count >= self.config.failure_threshold:
self.state = CircuitState.OPEN
print(f"[CIRCUIT] CLOSED → OPEN (after {self.failure_count} failures)")
def get_status(self) -> dict:
"""Lấy trạng thái circuit breaker"""
return {
"state": self.state.value,
"failure_count": self.failure_count,
"last_failure": self.last_failure_time,
"current_primary_model": self.models[0]["name"]
}
class AllModelsFailedError(Exception):
pass
Usage trong main application
async def process_with_circuit_breaker():
from config.holy_sheep_client import HolySheepClaudeClient
client = HolySheepClaudeClient(
api_key="YOUR_HOLYSHEEP_API_KEY",
base_url="https://api.holysheep.ai/v1"
)
circuit_breaker = ClaudeCircuitBreaker(
CircuitBreakerConfig(
failure_threshold=5,
recovery_timeout=60,
half_open_max_calls=3
)
)
messages = [
{"role": "user", "content": "Giải thích high-availability architecture"}
]
try:
response = await circuit_breaker.call(client, messages)
print(f"Response: {response}")
print(f"Circuit status: {circuit_breaker.get_status()}")
except AllModelsFailedError as e:
# Queue request để retry sau
print(f"ALL MODELS DOWN: {e}")4. Caching Layer Với Redis
Với những request có nội dung tương tự, cache có thể tiết kiệm đến 80% chi phí. Tôi implement semantic cache với Redis.
# services/semantic_cache.py
import hashlib
import json
import redis.asyncio as redis
from typing import Optional, Any
import json_repair
class SemanticCache:
"""
Semantic caching cho Claude API responses.
Cache key được tạo từ content hash + model + params.
"""
def __init__(
self,
redis_url: str = "redis://localhost:6379/0",
default_ttl: int = 3600, # 1 hour
similarity_threshold: float = 0.95
):
self.redis = redis.from_url(redis_url)
self.default_ttl = default_ttl
self.similarity_threshold = similarity_threshold
def _generate_cache_key(
self,
messages: list,
model: str,
temperature: float,
**kwargs
) -> str:
"""Tạo cache key từ request params"""
# Compact messages to content only
content_parts = []
for msg in messages:
if isinstance(msg, dict):
role = msg.get("role", "")
content = msg.get("content", "")
content_parts.append(f"{role}:{content}")
else:
content_parts.append(str(msg))
content_hash = hashlib.sha256(
"|".join(content_parts).encode()
).hexdigest()[:16]
params_hash = hashlib.md5(
json.dumps({"model": model, "temp": temperature}, sort_keys=True).encode()
).hexdigest()
return f"claude_cache:{content_hash}:{params_hash}"
async def get(
self,
messages: list,
model: str,
temperature: float = 0.7,
**kwargs
) -> Optional[dict]:
"""Lấy cached response nếu có"""
cache_key = self._generate_cache_key(
messages, model, temperature, **kwargs
)
cached = await self.redis.get(cache_key)
if cached:
data = json.loads(cached)
print(f"[CACHE] HIT for {cache_key[:30]}...")
return data
print(f"[CACHE] MISS for {cache_key[:30]}...")
return None
async def set(
self,
messages: list,
model: str,
response: dict,
temperature: float = 0.7,
ttl: Optional[int] = None,
**kwargs
):
"""Lưu response vào cache"""
cache_key = self._generate_cache_key(
messages, model, temperature, **kwargs
)
# Metadata để track usage
cache_data = {
"response": response,
"cached_at": self.redis.time()[0],
"model": model,
"tokens_used": response.get("usage", {}).get("total_tokens", 0)
}
await self.redis.setex(
cache_key,
ttl or self.default_ttl,
json.dumps(cache_data)
)
print(f"[CACHE] Stored response for {cache_key[:30]}... "
f"(TTL: {ttl or self.default_ttl}s)")
async def invalidate_pattern(self, pattern: str):
"""Xóa cache theo pattern"""
keys = []
async for key in self.redis.scan_iter(match=pattern):
keys.append(key)
if keys:
await self.redis.delete(*keys)
print(f"[CACHE] Invalidated {len(keys)} keys matching {pattern}")
async def get_stats(self) -> dict:
"""Lấy cache statistics"""
info = await self.redis.info("stats")
keys_count = await self.redis.dbsize()
# Estimate savings
total_savings = 0
async for key in self.redis.scan_iter(match="claude_cache:*"):
cached = await self.redis.get(key)
if cached:
data = json.loads(cached)
# Claude Sonnet: $15/MTok
# Giả định average 500 tokens per request
total_savings += (500 / 1_000_000) * 15
return {
"total_keys": keys_count,
"cache_hits": info.get("keyspace_hits", 0),
"cache_misses": info.get("keyspace_misses", 0),
"estimated_savings_usd": round(total_savings, 2)
}
Integration với main client
async def cached_claude_call(
client: Any,
cache: SemanticCache,
messages: list,
model: str = "claude-sonnet-4-20250514",
use_cache: bool = True,
**kwargs
):
"""
Claude call với automatic caching.
"""
if use_cache:
cached_response = await cache.get(messages, model, **kwargs)
if cached_response:
return {
**cached_response["response"],
"cached": True,
"cache_hit": True
}
# Gọi API
response = await client.chat_completions(
messages=messages,
model=model,
**kwargs
)
# Cache response
if use_cache:
await cache.set(messages, model, response, **kwargs)
return {**response, "cached": False}
Demo
async def demo():
cache = SemanticCache(
redis_url="redis://localhost:6379/0",
default_ttl=1800
)
# Test cache hit/miss
messages = [{"role": "user", "content": "What is AI?"}]
result1 = await cached_claude_call(
client=None, # Skip actual API call
cache=cache,
messages=messages,
model="claude-sonnet-4-20250514"
)
stats = await cache.get_stats()
print(f"Cache stats: {stats}")
if __name__ == "__main__":
import asyncio
asyncio.run(demo())5. Monitoring Và Alerting
Đây là phần quan trọng nhất mà nhiều người bỏ qua. Monitoring không chỉ là watchdog mà còn là early warning system.
# services/monitoring.py
import time
import asyncio
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from collections import deque
import statistics
@dataclass
class MetricsSnapshot:
timestamp: float
endpoint: str
latency_ms: float
status_code: int
tokens_used: int = 0
cached: bool = False
error: Optional[str] = None
class ClaudeAPIMonitor:
"""
Real-time monitoring cho Claude API calls.
Track latency, error rates, token usage và cost.
"""
def __init__(
self,
window_size: int = 1000,
alert_thresholds: Optional[Dict] = None
):
self.metrics: deque = deque(maxlen=window_size)
self.alert_thresholds = alert_thresholds or {
"error_rate_pct": 5.0, # Alert if >5% errors
"p99_latency_ms": 2000, # Alert if p99 > 2s
"rate_limit_pct": 10.0, # Alert if >10% rate limited
}
self._alert_history: deque = deque(maxlen=100)
self._lock = asyncio.Lock()
async def record(self, snapshot: MetricsSnapshot):
"""Ghi nhận một API call"""
async with self._lock:
self.metrics.append(snapshot)
# Check alerts
await self._check_alerts(snapshot)
async def _check_alerts(self, snapshot: MetricsSnapshot):
"""Kiểm tra và trigger alerts"""
stats = await self.get_stats()
alerts_triggered = []
# Error rate alert
if stats["error_rate_pct"] > self.alert_thresholds["error_rate_pct"]:
alerts_triggered.append({
"type": "high_error_rate",
"message": f"Error rate {stats['error_rate_pct']:.1f}% "
f"exceeds threshold {self.alert_thresholds['error_rate_pct']}%",
"severity": "critical"
})
# Latency alert
if stats["p99_latency_ms"] > self.alert_thresholds["p99_latency_ms"]:
alerts_triggered.append({
"type": "high_latency",
"message": f"P99 latency {stats['p99_latency_ms']:.0f}ms "
f"exceeds threshold {self.alert_thresholds['p99_latency_ms']}ms",
"severity": "warning"
})
# Rate limit alert
if stats["rate_limit_pct"] > self.alert_thresholds["rate_limit_pct"]:
alerts_triggered.append({
"type": "rate_limiting",
"message": f"Rate limit % {stats['rate_limit_pct']:.1f}% "
f"indicates need for quota increase",
"severity": "warning"
})
for alert in alerts_triggered:
alert["timestamp"] = time.time()
self._alert_history.append(alert)
await self._send_alert(alert)
async def _send_alert(self, alert: dict):
"""Gửi alert (Slack, PagerDuty, etc.)"""
# In production, integrate with Slack/PagerDuty
emoji = "🔴" if alert["severity"] == "critical" else "🟡"
print(f"{emoji} ALERT [{alert['type']}]: {alert['message']}")
async def get_stats(self) -> Dict:
"""Lấy statistics hiện tại"""
if not self.metrics:
return self._empty_stats()
latencies = [m.latency_ms for m in self.metrics]
errors = [m for m in self.metrics if m.error or m.status_code >= 400]
rate_limited = [m for m in self.metrics if m.status_code == 429]
cached = [m for m in self.metrics if m.cached]
total_tokens = sum(m.tokens_used for m in self.metrics)
# Cost calculation (Claude Sonnet 4.5: $15/MTok)
input_cost = (total_tokens * 0.3 / 1_000_000) * 15
output_cost = (total_tokens * 0.7 / 1_000_000) * 15
total_cost = input_cost + output_cost
return {
"total_requests": len(self.metrics),
"error_count": len(errors),
"error_rate_pct": len(errors) / len(self.metrics) * 100,
"rate_limit_count": len(rate_limited),
"rate_limit_pct": len(rate_limited) / len(self.metrics) * 100,
"cache_hit_rate": len(cached) / len(self.metrics) * 100 if self.metrics else 0,
"avg_latency_ms": statistics.mean(latencies),
"p50_latency_ms": statistics.median(latencies),
"p95_latency_ms": statistics.quantiles(latencies, n=20)[18] if len(latencies) >= 20 else statistics.mean(latencies),
"p99_latency_ms": statistics.quantiles(latencies, n=100)[98] if len(latencies) >= 100 else statistics.mean(latencies),
"total_tokens": total_tokens,
"estimated_cost_usd": round(total_cost, 4),
"requests_per_minute": len([m for m in self.metrics if time.time() - m.timestamp < 60])
}
def _empty_stats(self) -> Dict:
return {
"total_requests": 0,
"error_rate_pct": 0,
"avg_latency_ms": 0,
"p99_latency_ms": 0,
"total_tokens": 0,
"estimated_cost_usd": 0
}
async def get_endpoint_health(self) -> Dict[str, Dict]:
"""Health status của từng endpoint"""
endpoint_stats: Dict[str, Dict] = {}
for m in self.metrics:
if m.endpoint not in endpoint_stats:
endpoint_stats[m.endpoint] = {
"requests": 0,
"errors": 0,
"latencies": []
}
ep = endpoint_stats[m.endpoint]
ep["requests"] += 1
ep["latencies"].append(m.latency_ms)
if m.error:
ep["errors"] += 1
# Calculate health score
for ep, stats in endpoint_stats.items():
error_rate = stats["errors"] / stats["requests"] * 100
avg_latency = statistics.mean(stats["latencies"])
# Health score: 100 - error_rate*5 - latency_penalty
latency_penalty = min(30, avg_latency / 100)
stats["health_score"] = max(0, 100 - error_rate * 5 - latency_penalty)
stats["error_rate"] = error_rate
stats["avg_latency"] = avg_latency
return endpoint_stats
Singleton monitor
monitor = ClaudeAPIMonitor()
Usage
async def demo_monitoring():
# Simulate some requests
for i in range(100):
snapshot = MetricsSnapshot(
timestamp=time.time(),
endpoint="api.holysheep.ai",
latency_ms=100 + (i % 50), # 100-150ms typical
status_code=200,
tokens_used=500,
cached=(i % 5 == 0) # 20% cache hit
)
await monitor.record(snapshot)
stats = await monitor.get_stats()
print("\n📊 Claude API Monitoring Stats:")
print(f" Total Requests: {stats['total_requests']}")
print(f" Error Rate: {stats['error_rate_pct']:.2f}%")
print(f" Avg Latency: {stats['avg_latency_ms']:.1f}ms")
print(f" P99 Latency: {stats['p99_latency_ms']:.1f}ms")
print(f" Cache Hit Rate: {stats['cache_hit_rate']:.1f}%")
print(f" Estimated Cost: ${stats['estimated_cost_usd']:.4f}")
health = await monitor.get_endpoint_health()
for ep, h in health.items():
print(f"\n Endpoint: {