어느 날 새벽 3시, 제 휴먼手机上突然响起刺耳的警报声。生产经营ダッシュ보드에서 확인한 결과,
ConnectionError: timeout exceeded 30s — 자사의 AI 챗봇 서비스가 완전히 응답하지 못하고 있었습니다. 수천 명의 사용자가 대기열에 갇힌 상태였고, 고객 지원팀에는 전화가，杀到炸裂되었고요. 이 사건으로 약 4시간간의 서비스 중단과 함께 예상치 못한 비용이 2배 이상 증가했습니다.
저는 이惨事の教训から、その後AI API 모니터링 시스템을 전면 개편했습니다. 이 글에서는 HolySheep AI를 포함한 AI API 게이트웨이에서 효과적인 SLI/SLO 기반 모니터링 및 경보 전략을 구현하는 방법을 실제 경험담과 함께分享합니다.
왜 AI API 모니터링이 중요한가?
일반적인 REST API와 달리, AI API는 다음과 같은 독특한 특성을 가집니다:
- 비용 변동성: 토큰 기반 과금으로 요청 규모에 따라 비용이 크게 변동
- 지연 시간 변동성: 모델 로딩, GPU 리소스 상태에 따라 응답 시간 편차가 큼
- Rate Limit 복잡성: 모델별, 계정별로 다른 제한 정책
- 다중 공급자 의존성: 단일 장애점(SPOF) 위험
HolySheep AI를 사용하면 단일 API 키로 여러 모델을 통합 관리할 수 있지만, 그럼에도 불구하고 각 모델의 성능 특성과 비용을 모니터링하는 것은 필수적입니다.
SLI/SLO 기본 개념과 AI API 적용
SLI (Service Level Indicator) 정의
SLI는 측정 가능한 서비스 수준의 지표입니다. AI API 환경에서는 다음과 같은 SLI를 설정해야 합니다:
// AI API 모니터링 SLI 정의 예시
const SLI_CONFIG = {
// 성공률 지표
success_rate: {
metric: "request_success_rate",
target: 99.5, // 99.5% 성공률 목표
window: "5m",
calculation: "(successful_requests / total_requests) * 100"
},
// 지연 시간 지표
latency: {
p50_target: 500, // 500ms 이내
p95_target: 2000, // 2초 이내
p99_target: 5000, // 5초 이내
window: "5m"
},
// 가용성 지표
availability: {
metric: "uptime_percentage",
target: 99.9, // 월간 99.9% 가용성
window: "1h"
},
// 오류율 지표
error_rate: {
metric: "error_request_rate",
critical_threshold: 5, // 5% 이상 시 심각
warning_threshold: 2, // 2% 이상 시 경고
window: "5m"
},
// Rate Limit 지표
rate_limit: {
utilization: "requests_used / rate_limit_max",
target: 80, // 80% 이상 시 경고
window: "1m"
},
// 토큰 사용량 지표
token_usage: {
metric: "tokens_per_dollar",
target: "optimized_cost_efficiency",
window: "1d"
}
};
SLO (Service Level Objective) 설정
SLO는 달성하려는 목표 수준입니다. HolySheep AI 게이트웨이에서 권장하는 SLO 설정 전략을 소개합니다:
# Python - HolySheep AI SLO 모니터링 구현
import httpx
import time
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import asyncio
@dataclass
class SLI_Metrics:
"""SLI 지표 데이터 클래스"""
timestamp: datetime
total_requests: int
successful_requests: int
failed_requests: int
timeout_requests: int
rate_limited_requests: int
avg_latency_ms: float
p95_latency_ms: float
p99_latency_ms: float
total_tokens: int
total_cost_cents: float
class HolySheepAIMonitor:
"""HolySheep AI API 모니터링 클래스"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
self.metrics_history: List[SLI_Metrics] = []
async def collect_metrics(self) -> SLI_Metrics:
"""실시간 메트릭 수집"""
async with httpx.AsyncClient(
base_url=self.base_url,
headers=self.headers,
timeout=30.0
) as client:
# 상태 확인 요청
start = time.time()
try:
response = await client.get("/models")
latency = (time.time() - start) * 1000
metrics = SLI_Metrics(
timestamp=datetime.now(),
total_requests=1,
successful_requests=1 if response.status_code == 200 else 0,
failed_requests=1 if response.status_code != 200 else 0,
timeout_requests=0,
rate_limited_requests=1 if response.status_code == 429 else 0,
avg_latency_ms=latency,
p95_latency_ms=latency,
p99_latency_ms=latency,
total_tokens=0,
total_cost_cents=0
)
self.metrics_history.append(metrics)
return metrics
except httpx.TimeoutException:
return SLI_Metrics(
timestamp=datetime.now(),
total_requests=1,
successful_requests=0,
failed_requests=1,
timeout_requests=1,
rate_limited_requests=0,
avg_latency_ms=30000,
p95_latency_ms=30000,
p99_latency_ms=30000,
total_tokens=0,
total_cost_cents=0
)
def calculate_sli(self, window_minutes: int = 5) -> Dict:
"""SLI 지표 계산"""
cutoff = datetime.now() - timedelta(minutes=window_minutes)
recent = [m for m in self.metrics_history if m.timestamp >= cutoff]
if not recent:
return {"error": "No recent data"}
total_requests = sum(m.total_requests for m in recent)
successful = sum(m.successful_requests for m in recent)
failed = sum(m.failed_requests for m in recent)
timeouts = sum(m.timeout_requests for m in recent)
rate_limited = sum(m.rate_limited_requests for m in recent)
return {
"success_rate": (successful / total_requests * 100) if total_requests > 0 else 0,
"error_rate": (failed / total_requests * 100) if total_requests > 0 else 0,
"timeout_rate": (timeouts / total_requests * 100) if total_requests > 0 else 0,
"rate_limit_rate": (rate_limited / total_requests * 100) if total_requests > 0 else 0,
"avg_latency_p99": max(m.p99_latency_ms for m in recent),
"total_cost_cents": sum(m.total_cost_cents for m in recent),
"window_minutes": window_minutes
}
def evaluate_slo(self, sli: Dict) -> Dict[str, str]:
"""SLO 충족 여부 평가"""
alerts = []
# 성공률 SLO (목표: 99.5%)
if sli.get("success_rate", 0) < 99.5:
alerts.append("CRITICAL: Success rate below 99.5%")
elif sli.get("success_rate", 0) < 99.9:
alerts.append("WARNING: Success rate below 99.9%")
# 지연 시간 SLO (목표: P99 < 5000ms)
if sli.get("avg_latency_p99", 0) > 5000:
alerts.append("CRITICAL: P99 latency exceeds 5000ms")
elif sli.get("avg_latency_p99", 0) > 2000:
alerts.append("WARNING: P99 latency exceeds 2000ms")
# 오류율 SLO (목표: < 0.5%)
if sli.get("error_rate", 0) > 5:
alerts.append("CRITICAL: Error rate exceeds 5%")
elif sli.get("error_rate", 0) > 2:
alerts.append("WARNING: Error rate exceeds 2%")
return {
"status": "CRITICAL" if any("CRITICAL" in a for a in alerts) else "WARNING" if alerts else "HEALTHY",
"alerts": alerts
}
사용 예시
async def main():
monitor = HolySheepAIMonitor("YOUR_HOLYSHEEP_API_KEY")
# 5분간 메트릭 수집 시뮬레이션
for _ in range(60):
await monitor.collect_metrics()
await asyncio.sleep(5)
# SLO 평가
sli = monitor.calculate_sli(window_minutes=5)
result = monitor.evaluate_slo(sli)
print(f"SLO Status: {result['status']}")
for alert in result['alerts']:
print(f" - {alert}")
return result
asyncio.run(main())
실시간 경보 시스템 구현
저는 실제로 사용하고 있는 경보 시스템 아키텍처를 공유합니다. HolySheep AI와 통합하여 PagerDuty, Slack, Discord 등으로 실시간 알림을 전송합니다:
# Python - 경보 시스템 전체 구현
import asyncio
import httpx
import smtplib
from email.mime.text import MIMEText
from typing import Callable, Optional, Dict, List
from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
import json
class AlertSeverity(Enum):
CRITICAL = "critical" # 즉시 대응 필요
WARNING = "warning" # 15분 내 대응 권장
INFO = "info" # 정보성 알림
@dataclass
class Alert:
"""경보 데이터 클래스"""
severity: AlertSeverity
title: str
message: str
metric_name: str
current_value: float
threshold: float
timestamp: datetime = field(default_factory=datetime.now)
metadata: Dict = field(default_factory=dict)
def to_dict(self) -> Dict:
return {
"severity": self.severity.value,
"title": self.title,
"message": self.message,
"metric_name": self.metric_name,
"current_value": self.current_value,
"threshold": self.threshold,
"timestamp": self.timestamp.isoformat(),
"metadata": self.metadata
}
class AlertChannel:
"""경보 채널 인터페이스"""
async def send(self, alert: Alert) -> bool:
raise NotImplementedError
class SlackAlertChannel(AlertChannel):
"""Slack 웹훅 경보 채널"""
def __init__(self, webhook_url: str, channel: str = "#ai-alerts"):
self.webhook_url = webhook_url
self.channel = channel
async def send(self, alert: Alert) -> bool:
"""Slack로 경보 전송"""
color_map = {
AlertSeverity.CRITICAL: "danger",
AlertSeverity.WARNING: "warning",
AlertSeverity.INFO: "good"
}
payload = {
"channel": self.channel,
"attachments": [{
"color": color_map[alert.severity],
"blocks": [
{
"type": "header",
"text": {
"type": "plain_text",
"text": f"🚨 {alert.title}"
}
},
{
"type": "section",
"fields": [
{"type": "mrkdwn", "text": f"*Severity:*\n{alert.severity.value.upper()}"},
{"type": "mrkdwn", "text": f"*Metric:*\n{alert.metric_name}"},
{"type": "mrkdwn", "text": f"*Current Value:*\n{alert.current_value:.2f}"},
{"type": "mrkdwn", "text": f"*Threshold:*\n{alert.threshold:.2f}"}
]
},
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"*Message:*\n{alert.message}"
}
},
{
"type": "context",
"elements": [
{
"type": "mrkdwn",
"text": f"⏰ {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S')} UTC"
}
]
}
]
}]
}
try:
async with httpx.AsyncClient() as client:
response = await client.post(
self.webhook_url,
json=payload,
timeout=10.0
)
return response.status_code == 200
except Exception as e:
print(f"Slack webhook error: {e}")
return False
class EmailAlertChannel(AlertChannel):
"""이메일 경보 채널"""
def __init__(self, smtp_host: str, smtp_port: int,
username: str, password: str,
from_addr: str, to_addrs: List[str]):
self.smtp_host = smtp_host
self.smtp_port = smtp_port
self.username = username
self.password = password
self.from_addr = from_addr
self.to_addrs = to_addrs
async def send(self, alert: Alert) -> bool:
"""이메일로 경보 전송"""
subject_map = {
AlertSeverity.CRITICAL: "[CRITICAL]",
AlertSeverity.WARNING: "[WARNING]",
AlertSeverity.INFO: "[INFO]"
}
subject = f"{subject_map[alert.severity]} AI API Alert: {alert.title}"
body = f"""
HolySheep AI API Monitoring Alert
=================================
Severity: {alert.severity.value.upper()}
Metric: {alert.metric_name}
Current Value: {alert.current_value:.2f}
Threshold: {alert.threshold:.2f}
Message:
{alert.message}
Timestamp: {alert.timestamp.strftime('%Y-%m-%d %H:%M:%S')} UTC
---
HolySheep AI Gateway Monitor
"""
msg = MIMEText(body)
msg['Subject'] = subject
msg['From'] = self.from_addr
msg['To'] = ', '.join(self.to_addrs)
try:
with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
server.starttls()
server.login(self.username, self.password)
server.send_message(msg)
return True
except Exception as e:
print(f"Email send error: {e}")
return False
class AlertManager:
"""경보 관리자 - 메인 모니터링 로직"""
# HolySheep AI 권장 임계값
THRESHOLDS = {
"success_rate": {
"critical": 95.0, # 95% 미만
"warning": 99.0, # 99% 미만
"unit": "%"
},
"latency_p99": {
"critical": 10000, # 10초 초과
"warning": 5000, # 5초 초과
"unit": "ms"
},
"error_rate": {
"critical": 5.0, # 5% 초과
"warning": 2.0, # 2% 초과
"unit": "%"
},
"rate_limit_utilization": {
"critical": 95.0, # 95% 이상
"warning": 80.0, # 80% 이상
"unit": "%"
},
"cost_per_hour": {
"critical": 10000, # $100/hour 초과
"warning": 5000, # $50/hour 초과
"unit": "cents"
}
}
def __init__(self):
self.channels: List[AlertChannel] = []
self.alert_history: List[Alert] = []
self.cooldown_seconds = 300 # 동일 경보 5분 간격 제한
def add_channel(self, channel: AlertChannel):
self.channels.append(channel)
async def check_and_alert(self, metrics: Dict) -> List[Alert]:
"""메트릭 확인 및 경보 발생"""
alerts = []
# 성공률 체크
success_rate = metrics.get("success_rate", 100)
threshold = self.THRESHOLDS["success_rate"]
if success_rate < threshold["critical"]:
alert = self._create_alert(
AlertSeverity.CRITICAL,
"Success Rate Critical",
f"Success rate dropped to {success_rate:.2f}%",
"success_rate",
success_rate,
threshold["critical"]
)
alerts.append(alert)
elif success_rate < threshold["warning"]:
alert = self._create_alert(
AlertSeverity.WARNING,
"Success Rate Warning",
f"Success rate dropped to {success_rate:.2f}%",
"success_rate",
success_rate,
threshold["warning"]
)
alerts.append(alert)
# P99 지연 시간 체크
latency_p99 = metrics.get("p99_latency_ms", 0)
threshold = self.THRESHOLDS["latency_p99"]
if latency_p99 > threshold["critical"]:
alert = self._create_alert(
AlertSeverity.CRITICAL,
"Latency Critical",
f"P99 latency is {latency_p99:.0f}ms",
"latency_p99",
latency_p99,
threshold["critical"]
)
alerts.append(alert)
elif latency_p99 > threshold["warning"]:
alert = self._create_alert(
AlertSeverity.WARNING,
"Latency Warning",
f"P99 latency is {latency_p99:.0f}ms",
"latency_p99",
latency_p99,
threshold["warning"]
)
alerts.append(alert)
# Rate Limit 체크
rate_limit_pct = metrics.get("rate_limit_utilization", 0)
threshold = self.THRESHOLDS["rate_limit_utilization"]
if rate_limit_pct > threshold["critical"]:
alert = self._create_alert(
AlertSeverity.CRITICAL,
"Rate Limit Critical",
f"Rate limit utilization at {rate_limit_pct:.1f}%",
"rate_limit_utilization",
rate_limit_pct,
threshold["critical"]
)
alerts.append(alert)
elif rate_limit_pct > threshold["warning"]:
alert = self._create_alert(
AlertSeverity.WARNING,
"Rate Limit Warning",
f"Rate limit utilization at {rate_limit_pct:.1f}%",
"rate_limit_utilization",
rate_limit_pct,
threshold["warning"]
)
alerts.append(alert)
# 비용 체크
cost_cents = metrics.get("cost_last_hour_cents", 0)
threshold = self.THRESHOLDS["cost_per_hour"]
if cost_cents > threshold["critical"]:
alert = self._create_alert(
AlertSeverity.CRITICAL,
"Cost Critical",
f"Hourly cost: ${cost_cents/100:.2f}",
"cost_per_hour",
cost_cents,
threshold["critical"]
)
alerts.append(alert)
elif cost_cents > threshold["warning"]:
alert = self._create_alert(
AlertSeverity.WARNING,
"Cost Warning",
f"Hourly cost: ${cost_cents/100:.2f}",
"cost_per_hour",
cost_cents,
threshold["warning"]
)
alerts.append(alert)
# 모든 채널로 경보 전송
for alert in alerts:
await self._dispatch_alert(alert)
return alerts
def _create_alert(self, severity: AlertSeverity, title: str,
message: str, metric_name: str,
current_value: float, threshold: float) -> Alert:
return Alert(
severity=severity,
title=title,
message=message,
metric_name=metric_name,
current_value=current_value,
threshold=threshold
)
async def _dispatch_alert(self, alert: Alert):
"""경보 전송 (冷却 시간 적용)"""
#冷却 확인
recent_same = [
a for a in self.alert_history[-10:]
if a.title == alert.title
and (datetime.now() - a.timestamp).total_seconds() < self.cooldown_seconds
]
if recent_same:
return #冷却 중
# 모든 채널로 전송
for channel in self.channels:
try:
await channel.send(alert)
except Exception as e:
print(f"Channel send error: {e}")
self.alert_history.append(alert)
사용 예시
async def main():
# 경보 관리자 초기화
manager = AlertManager()
# Slack 채널 추가
manager.add_channel(SlackAlertChannel(
webhook_url="https://hooks.slack.com/services/YOUR/WEBHOOK/URL",
channel="#ai-monitoring-alerts"
))
# 이메일 채널 추가 (선택적)
# manager.add_channel(EmailAlertChannel(...))
# 메트릭 수집 및 경보 확인
while True:
# 실제 환경에서는 HolySheep AI API에서 메트릭 수집
metrics = {
"success_rate": 98.5,
"p99_latency_ms": 3500,
"rate_limit_utilization": 75.0,
"cost_last_hour_cents": 4500
}
alerts = await manager.check_and_alert(metrics)
if alerts:
print(f"Generated {len(alerts)} alerts")
for alert in alerts:
print(f" [{alert.severity.value.upper()}] {alert.title}")
await asyncio.sleep(60) # 1분마다 체크
asyncio.run(main())
HolySheep AI 실전 모니터링 설정
저의 경험담을 바탕으로 HolySheep AI에서 실제 작동하는 완전한 모니터링 파이프라인을 구축했습니다. HolySheep AI의 다양한 모델(GPT-4.1, Claude Sonnet, Gemini 2.5 Flash, DeepSeek V3.2)을 통합 모니터링하는 설정입니다:
# Python - HolySheep AI 완전한 모니터링 시스템
import asyncio
import httpx
import json
from typing import Dict, List, Optional, Any
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from collections import defaultdict
import statistics
@dataclass
class ModelMetrics:
"""모델별 메트릭"""
model_name: str
request_count: int = 0
success_count: int = 0
error_count: int = 0
timeout_count: int = 0
rate_limit_count: int = 0
total_latency_ms: float = 0.0
latency_samples: List[float] = field(default_factory=list)
total_input_tokens: int = 0
total_output_tokens: int = 0
total_cost_cents: float = 0.0
first_request_time: Optional[datetime] = None
last_request_time: Optional[datetime] = None
class HolySheepMultiModelMonitor:
"""HolySheep AI 다중 모델 모니터링 시스템"""
# HolySheep AI 모델별 가격 (MTok당 센트 단위)
MODEL_PRICING = {
"gpt-4.1": {"input": 800, "output": 3200}, # $8/$32 per MTok
"gpt-4.1-mini": {"input": 150, "output": 600}, # $1.50/$6 per MTok
"claude-sonnet-4-20250514": {"input": 450, "output": 2250}, # $4.50/$22.50 per MTok
"claude-opus-4-20250514": {"input": 900, "output": 4500}, # $9/$45 per MTok
"gemini-2.5-flash": {"input": 250, "output": 1000}, # $2.50/$10 per MTok
"gemini-2.5-pro": {"input": 1500, "output": 6000}, # $15/$60 per MTok
"deepseek-v3.2": {"input": 42, "output": 168}, # $0.42/$1.68 per MTok
}
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.model_metrics: Dict[str, ModelMetrics] = defaultdict(
lambda: ModelMetrics(model_name="unknown")
)
self.request_log: List[Dict] = []
def _get_headers(self) -> Dict[str, str]:
return {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async def chat_completion(self, model: str, messages: List[Dict],
temperature: float = 0.7,
max_tokens: int = 1000) -> Dict[str, Any]:
"""HolySheep AI Chat Completion API 호출 및 메트릭 기록"""
start_time = datetime.now()
metrics = self.model_metrics[model]
metrics.first_request_time = metrics.first_request_time or start_time
request_data = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens
}
try:
async with httpx.AsyncClient(
base_url=self.base_url,
headers=self._get_headers(),
timeout=60.0 # 60초 타임아웃
) as client:
response = await client.post(
"/chat/completions",
json=request_data
)
end_time = datetime.now()
latency_ms = (end_time - start_time).total_seconds() * 1000
# 메트릭 업데이트
metrics.request_count += 1
metrics.last_request_time = end_time
metrics.latency_samples.append(latency_ms)
metrics.total_latency_ms += latency_ms
if response.status_code == 200:
data = response.json()
# 토큰 사용량 추출
usage = data.get("usage", {})
input_tokens = usage.get("prompt_tokens", 0)
output_tokens = usage.get("completion_tokens", 0)
metrics.total_input_tokens += input_tokens
metrics.total_output_tokens += output_tokens
# 비용 계산
pricing = self.MODEL_PRICING.get(model, {"input": 100, "output": 100})
input_cost = (input_tokens / 1_000_000) * pricing["input"]
output_cost = (output_tokens / 1_000_000) * pricing["output"]
metrics.total_cost_cents += (input_cost + output_cost) * 100
metrics.success_count += 1
self.request_log.append({
"timestamp": end_time.isoformat(),
"model": model,
"status": "success",
"latency_ms": latency_ms,
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost_cents": (input_cost + output_cost) * 100
})
return {"status": "success", "data": data, "latency_ms": latency_ms}
elif response.status_code == 429:
metrics.rate_limit_count += 1
metrics.error_count += 1
self.request_log.append({
"timestamp": end_time.isoformat(),
"model": model,
"status": "rate_limited",
"latency_ms": latency_ms
})
return {
"status": "rate_limited",
"error": f"HTTP {response.status_code}",
"latency_ms": latency_ms
}
else:
metrics.error_count += 1
self.request_log.append({
"timestamp": end_time.isoformat(),
"model": model,
"status": "error",
"latency_ms": latency_ms,
"error": f"HTTP {response.status_code}"
})
return {
"status": "error",
"error": f"HTTP {response.status_code}",
"latency_ms": latency_ms
}
except httpx.TimeoutException:
end_time = datetime.now()
latency_ms = (end_time - start_time).total_seconds() * 1000
metrics.timeout_count += 1
metrics.error_count += 1
metrics.last_request_time = end_time
self.request_log.append({
"timestamp": end_time.isoformat(),
"model": model,
"status": "timeout",
"latency_ms": latency_ms
})
return {
"status": "timeout",
"error": "Connection timeout after 60s",
"latency_ms": latency_ms
}
except httpx.ConnectError as e:
metrics.error_count += 1
self.request_log.append({
"timestamp": datetime.now().isoformat(),
"model": model,
"status": "connection_error",
"error": str(e)
})
return {
"status": "connection_error",
"error": f"ConnectionError: {str(e)}"
}
def get_model_summary(self, model: str) -> Dict[str, Any]:
"""모델별 요약 통계"""
metrics = self.model_metrics[model]
if metrics.request_count == 0:
return {"error": "No data for this model"}
success_rate = (metrics.success_count / metrics.request_count) * 100
error_rate = (metrics.error_count / metrics.request_count) * 100
timeout_rate = (metrics.timeout_count / metrics.request_count) * 100
rate_limit_rate = (metrics.rate_limit_count / metrics.request_count) * 100
avg_latency = metrics.total_latency_ms / metrics.request_count if metrics.request_count > 0 else 0
p50_latency = statistics.median(metrics.latency_samples) if metrics.latency_samples else 0
p95_latency = statistics.quantiles(metrics.latency_samples, n=20)[18] if len(metrics.latency_samples) > 20 else p50_latency
p99_latency = max(metrics.latency_samples) if metrics.latency_samples else 0
return {
"model": model,
"period": {
"start": metrics.first_request_time.isoformat() if metrics.first_request_time else None,
"end": metrics.last_request_time.isoformat() if metrics.last_request_time else None
},
"requests": {
"total": metrics.request_count,
"success": metrics.success_count,
"error": metrics.error_count,
"timeout": metrics.timeout_count,
"rate_limited": metrics.rate_limit_count
},
"rates": {
"success_rate": round(success_rate, 2),
"error_rate": round(error_rate, 2),
"timeout_rate": round(timeout_rate, 2),
"rate_limit_rate": round(rate_limit_rate, 2)
},
"latency_ms": {
"avg": round(avg_latency, 2),
"p50": round(p50_latency, 2),
"p95": round(p95_latency, 2),
"p99": round(p99_latency, 2)
},
"tokens": {
"input": metrics.total_input_tokens,
"output": metrics.total_output_tokens,
"total": metrics.total_input_tokens + metrics.total_output_tokens
},
"cost_cents": round(metrics.total_cost_cents, 2),
"cost_dollars": round(metrics.total_cost_cents / 100, 4)
}
def get_all_models_summary(self) -> List[Dict[str, Any]]:
"""모든 모델 요약"""
summaries = []
for model_name in self.model_metrics.keys():
summary = self.get_model_summary(model_name)
if "error" not in summary:
summaries.append(summary)
return summaries
def check_slo_compliance(self, target_success_rate: float = 99.5,
target_p99_latency: float = 5000) -> Dict[str, Any]:
"""SLO 준수 여부 확인"""
results = []
for model_name, metrics in self.model_metrics.items():
if metrics.request_count == 0:
continue
success_rate = (metrics.success_count / metrics.request_count) * 100
p99_latency = max(metrics.latency_samples) if metrics.latency_samples else 0
compliance = {
"model": model_name,
"success_rate_slo": {
"target": target_success_rate,
"actual": round(success_rate, 2),
"compliant": success_rate >= target_success_rate
},
"p99_latency_slo": {
"target": target_p99_latency,
"actual": round(p99_latency, 2),
"compliant": p99_latency <= target_p99_latency
}
}
results.append(compliance)
all_compliant = all(
r["success_rate_slo"]["compliant"] and r["p99_latency_slo"]["compliant"]
for r in results
)
return {
"overall_status": "COMPLIANT" if all_compliant else "NON_COMPLIANT",
"models": results
}
실전 사용 예시
async def main():
monitor = HolySheepMultiModelMonitor("YOUR_HOLYSHEEP_API_KEY")
# 다양한 모델로 테스트 요청
test_models = [
"deepseek-v3.2", # 저렴한 모델 먼저
"gemini-2.5-flash", # 빠른 모델
"gpt-4.1", # 고성능 모델
]
for model in test_models:
print(f"\nTesting {model}...")
# 10회 반복 테스트
for i in range(10):
result = await monitor.chat_completion(
model=model,
messages=[{"role": "user", "content": f"안녕하세요, 테스트 {i+1}입니다."}],
temperature=0.7,
max_tokens=100
)
if result["status"] == "success":
print(f" ✓ Request {i+1