오늘 아침 9시 23분, 당신의 AI 기반 고객 서비스 봇이 갑자기 아무 말도 하지 않았다. 모니터링 대시보드를 열어보니 원인은 명확했다. Claude API 응답 시간 8초 초과, Gemini API 503 Service Unavailable. 두 개의 주요 AI 모델이 동시에 장애를 겪고 있는 상황이다.
이 글은 HolySheep AI를 활용해 단일 API 키로 여러 AI 모델을 연동하고, 주요 모델이 실패할 때 자동으로 대체 모델로 전환하는Fallback 체계를 구축하는 방법을 단계별로 설명한다. 코드를 한 줄도 작성해보지 않은 초보자도 이 가이드를 따라하면 30분 만에 프로덕션 수준의 다중 모델 장애 대응 시스템을 완성할 수 있다.
왜 여러 AI 모델의 Fallback이 필요한가
2024년 11월, Anthropic의 Claude 서비스는 약 45분간 대규모 장애를 겪었다. 같은 달, Google Gemini도 예기치 않은 응답 지연으로 많은 개발자들이 영향을 받았다. 단일 AI 모델에 의존하는 시스템은 이러한 상황에서 완전히 무력화된다.
AI 서비스 장애는 크게 네 가지 유형으로 나눌 수 있다:
- 응답 지연: API 응답 시간이 평소 500ms에서 8초 이상으로 급증
- 서비스 불가: HTTP 503, 504 오류로 요청 자체를 받을 수 없는 상태
- 속도 제한: Rate Limit 초과로 인한 429 오류 빈발
- 품질 저하: 서비스는 정상이나 출력 품질이 눈에 띄게 떨어지는 경우
HolySheep AI는 이러한 문제들을 단일 API 엔드포인트 뒤에서 자동으로 해결한다. 하나의 API 키로 Claude, Gemini, GPT-4, DeepSeek 등 10개 이상의 모델에 접근하고, 장애 시 자동으로 Fallback을 수행한다.
HolySheep 다중 모델 Fallback 아키텍처 이해하기
HolySheep AI의 핵심 기능 중 하나는 Intelligent Routing이다. 이는 요청을 최적의 모델로 라우팅하고, 해당 모델이 실패할 경우 사전 정의된 순서대로 대체 모델로 자동 전환한다.
FallBack 체계를 만들기 전에 먼저 HolySheep의 단일 엔드포인트가 어떻게 작동하는지 이해해야 한다.
HolySheep 기본 구조
"""
HolySheep AI 기본 연동 구조
base_url: https://api.holysheep.ai/v1
"""
import openai
HolySheep API 키 설정
client = openai.OpenAI(
api_key="YOUR_HOLYSHEEP_API_KEY", # HolySheep에서 받은 API 키
base_url="https://api.holysheep.ai/v1" # HolySheep 전용 엔드포인트
)
기본 채팅 완료 요청
response = client.chat.completions.create(
model="gpt-4o", # 또는 claude-3-5-sonnet, gemini-2.0-flash 등
messages=[
{"role": "system", "content": "당신은 도움이 되는 도우미입니다."},
{"role": "user", "content": "안녕하세요, 현재 시간을 알려주세요."}
]
)
print(response.choices[0].message.content)
위 코드의 핵심은 base_url에 HolySheep 서버 주소(https://api.holysheep.ai/v1)만 입력하면, 내부적으로 최적의 모델로 라우팅된다는 점이다. 개발자는 단일 코드베이스로 여러 모델을 동일한 방식으로 호출할 수 있다.
실전 Fallback 시스템 구축: 5단계 가이드
1단계: 실패 감지 및 자동 전환 클래스 만들기
가장 기본적인 Fallback 시스템은 세 개의 모델을 우선순위 순서로 연결하는 것이다. 주 모델(Primary), 보조 모델(Secondary), 대안 모델(Tertiary) 세 단계를 구성한다.
"""
다중 모델 Fallback 시스템 v1.0
HolySheep AI를 활용한 자동 장애 대응
"""
import openai
import time
from typing import Optional, List
from dataclasses import dataclass
@dataclass
class ModelConfig:
"""모델 설정 데이터 클래스"""
name: str
priority: int # 1이 가장 높음
timeout: float # 초 단위 타임아웃
max_retries: int
class HolySheepMultiModelClient:
"""
HolySheep AI 다중 모델 Fallback 클라이언트
주 모델 실패 시 보조 → 대안 모델 순서로 자동 전환
"""
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
# Fallback 모델 목록 (우선순위 순서)
self.models = [
ModelConfig(name="claude-sonnet-4-20250514", priority=1, timeout=10, max_retries=2),
ModelConfig(name="gemini-2.0-flash", priority=2, timeout=8, max_retries=2),
ModelConfig(name="gpt-4o", priority=3, timeout=10, max_retries=2),
ModelConfig(name="deepseek-v3", priority=4, timeout=12, max_retries=1),
]
def chat_with_fallback(
self,
messages: List[dict],
temperature: float = 0.7,
max_tokens: int = 1000
) -> dict:
"""
Fallback이 적용된 채팅 완료 요청
Returns:
dict: {
'success': bool,
'response': str or None,
'model_used': str,
'error': str or None,
'fallback_count': int
}
"""
last_error = None
for i, model_config in enumerate(self.models):
try:
print(f"[INFO] {model_config.name} 시도 중 (우선순위 {model_config.priority})")
start_time = time.time()
response = self.client.chat.completions.create(
model=model_config.name,
messages=messages,
temperature=temperature,
max_tokens=max_tokens,
timeout=model_config.timeout
)
elapsed = time.time() - start_time
print(f"[성공] {model_config.name} 응답 시간: {elapsed:.2f}초")
return {
'success': True,
'response': response.choices[0].message.content,
'model_used': model_config.name,
'error': None,
'fallback_count': i,
'latency_ms': int(elapsed * 1000)
}
except openai.APITimeoutError:
last_error = f"{model_config.name} 타임아웃 ({model_config.timeout}초 초과)"
print(f"[경고] {last_error}")
continue
except openai.RateLimitError as e:
last_error = f"{model_config.name} 속도 제한 초과"
print(f"[경고] {last_error}")
time.sleep(2) # 2초 대기 후 재시도
continue
except openai.APIError as e:
last_error = f"{model_config.name} API 오류: {str(e)}"
print(f"[오류] {last_error}")
continue
# 모든 모델 실패
print(f"[실패] 모든 모델 Fallback 실패: {last_error}")
return {
'success': False,
'response': None,
'model_used': 'none',
'error': f"모든 모델 실패. 마지막 오류: {last_error}",
'fallback_count': len(self.models),
'latency_ms': 0
}
사용 예시
if __name__ == "__main__":
client = HolySheepMultiModelClient(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
messages = [
{"role": "system", "content": "당신은 친절한 고객 서비스 담당자입니다."},
{"role": "user", "content": "내 주문 상태를 확인해주세요. 주문번호: ORD-2024-12345"}
]
result = client.chat_with_fallback(messages)
if result['success']:
print(f"✅ 사용 모델: {result['model_used']}")
print(f"⏱️ 응답 시간: {result['latency_ms']}ms")
print(f"🔄 Fallback 횟수: {result['fallback_count']}")
print(f"📝 응답: {result['response']}")
else:
print(f"❌ 오류: {result['error']}")
위 코드는 HolySheep의 단일 엔드포인트를 활용하면서도 여러 모델에 대한 Fallback을 구현한다. 핵심 로직은 단순하다. 첫 번째 모델(Claude)로 요청을 시도하고, 타임아웃이나 API 오류가 발생하면 다음 모델(Gemini)로 자동 전환한다.
2단계: 응답 시간 기반 스마트 라우팅 추가하기
단순한 순차 Fallback보다 더 똑똑한 방법은 각 모델의 실제 응답 시간을 모니터링하고, 가장 빠른 모델을 우선적으로 사용하는 것이다. HolySheep는 내부적으로 이 기능을 지원하지만, 직접 구현하면 더 세밀한 제어가 가능하다.
"""
스마트 라우팅이 적용된 고급 Fallback 시스템
응답 시간 및 가용성을 기반으로 최적 모델 자동 선택
"""
import openai
import time
from collections import deque
from typing import Dict, List, Optional
import statistics
class SmartFallbackClient:
"""
HolySheep AI 스마트 라우팅 클라이언트
기능:
1. 모델별 응답 시간 추적 (Rolling Window)
2. 가용성 점수 기반 모델 선택
3. 연속 실패 시 모델 자동 비활성화
4. 자동 복구 감지
"""
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
# 모델별 메트릭 저장소
self.model_metrics: Dict[str, dict] = {
"claude-sonnet-4-20250514": {
"latencies": deque(maxlen=20), # 최근 20개 응답 시간
"failures": 0,
"total_requests": 0,
"is_healthy": True,
"disabled_until": 0
},
"gemini-2.0-flash": {
"latencies": deque(maxlen=20),
"failures": 0,
"total_requests": 0,
"is_healthy": True,
"disabled_until": 0
},
"gpt-4o": {
"latencies": deque(maxlen=20),
"failures": 0,
"total_requests": 0,
"is_healthy": True,
"disabled_until": 0
},
}
# 연속 실패 임계값
self.failure_threshold = 3
self.disable_duration = 60 # 60초간 비활성화
# Fallback 순서 (기본 우선순위)
self.fallback_order = [
"claude-sonnet-4-20250514",
"gemini-2.0-flash",
"gpt-4o"
]
def _get_model_score(self, model_name: str) -> float:
"""
모델 점수 계산 (높을수록 우선순위 높음)
점수 산정 요소:
- 평균 응답 시간 (빠를수록 높음)
- 가용성 (연속 실패 없을수록 높음)
- 요청 처리량
"""
if model_name not in self.model_metrics:
return 0.0
metrics = self.model_metrics[model_name]
current_time = time.time()
# 일시 비활성화된 모델은 점수 0
if metrics["disabled_until"] > current_time:
return 0.0
if not metrics["is_healthy"]:
return 0.0
# 응답 시간 점수 (평균 2초면 100점, 10초면 0점)
if metrics["latencies"]:
avg_latency = statistics.mean(metrics["latencies"])
latency_score = max(0, 100 - (avg_latency * 10))
else:
latency_score = 50 # 데이터 없으면 중간 점수
# 가용성 점수 (연속 실패 시 감점)
availability_score = max(0, 100 - (metrics["failures"] * 30))
# 최종 점수 = 응답 시간 60% + 가용성 40%
final_score = (latency_score * 0.6) + (availability_score * 0.4)
return final_score
def _select_best_model(self) -> str:
"""점수 기반 최적 모델 선택"""
model_scores = []
for model in self.fallback_order:
score = self._get_model_score(model)
model_scores.append((model, score))
# 점수 내림차순 정렬
model_scores.sort(key=lambda x: x[1], reverse=True)
return model_scores[0][0]
def _record_success(self, model_name: str, latency: float):
"""성공 응답 기록"""
if model_name in self.model_metrics:
metrics = self.model_metrics[model_name]
metrics["latencies"].append(latency)
metrics["total_requests"] += 1
metrics["failures"] = 0 # 성공 시 실패 카운터 리셋
metrics["is_healthy"] = True
def _record_failure(self, model_name: str):
"""실패 응답 기록"""
if model_name in self.model_metrics:
metrics = self.model_metrics[model_name]
metrics["failures"] += 1
metrics["total_requests"] += 1
# 연속 실패 임계값 초과 시 비활성화
if metrics["failures"] >= self.failure_threshold:
metrics["is_healthy"] = False
metrics["disabled_until"] = time.time() + self.disable_duration
print(f"[⚠️] {model_name} 일시 비활성화 ({self.disable_duration}초)")
def chat_with_smart_fallback(
self,
messages: List[dict],
user_id: str = "default"
) -> dict:
"""
스마트 라우팅이 적용된 채팅 완료
Returns:
{
'success': bool,
'response': str,
'model_used': str,
'latency_ms': int,
'models_tried': int
}
"""
# 1단계: 최적 모델 선택
primary_model = self._select_best_model()
print(f"[🧠] 선택된 최적 모델: {primary_model}")
# 2단계: 선택된 모델 우선 시도
tried_models = []
# 선택된 모델 우선, 그 다음 순서대로 Fallback
all_models = [primary_model] + [m for m in self.fallback_order if m != primary_model]
for model_name in all_models:
if model_name not in self.model_metrics:
continue
if not self.model_metrics[model_name]["is_healthy"]:
continue
tried_models.append(model_name)
try:
start_time = time.time()
response = self.client.chat.completions.create(
model=model_name,
messages=messages,
temperature=0.7,
max_tokens=1000,
timeout=15
)
latency = time.time() - start_time
self._record_success(model_name, latency)
return {
'success': True,
'response': response.choices[0].message.content,
'model_used': model_name,
'latency_ms': int(latency * 1000),
'models_tried': len(tried_models)
}
except Exception as e:
print(f"[❌] {model_name} 실패: {str(e)}")
self._record_failure(model_name)
continue
return {
'success': False,
'response': None,
'model_used': 'none',
'latency_ms': 0,
'models_tried': len(tried_models),
'error': '모든 모델 사용 불가'
}
def get_health_report(self) -> dict:
"""전체 모델 건강 상태 보고서"""
report = {}
for model_name, metrics in self.model_metrics.items():
latencies = list(metrics["latencies"])
report[model_name] = {
"평균 응답시간_ms": int(statistics.mean(latencies)) if latencies else None,
"최근 응답시간_ms": latencies[-1] * 1000 if latencies else None,
"총 요청수": metrics["total_requests"],
"연속 실패": metrics["failures"],
"상태": "정상" if metrics["is_healthy"] else "비활성화"
}
return report
사용 예시
if __name__ == "__main__":
client = SmartFallbackClient(api_key="YOUR_HOLYSHEEP_API_KEY")
messages = [
{"role": "user", "content": "서울의 오늘 날씨를 알려주세요."}
]
# 여러 번 요청하여 메트릭 수집
for i in range(5):
print(f"\n{'='*50}")
print(f"요청 #{i+1}")
result = client.chat_with_smart_fallback(messages)
if result['success']:
print(f"✅ 모델: {result['model_used']}")
print(f"⏱️ 지연: {result['latency_ms']}ms")
print(f"🔧 시도된 모델 수: {result['models_tried']}")
else:
print(f"❌ 실패: {result['error']}")
# 상태 보고서 출력
print(f"\n{'='*50}")
print("📊 모델 건강 상태 보고서")
print("="*50)
for model, status in client.get_health_report().items():
print(f"\n{model}:")
for key, value in status.items():
print(f" {key}: {value}")
이 코드에서 핵심은 _get_model_score() 함수다. 각 모델의 최근 응답 시간(최대 20개 샘플)을 추적하고, 평균 응답 시간이 빠를수록 높은 점수를 부여한다. 동시에 연속 실패 횟수가 많으면 점수를 낮추어 자동으로 해당 모델을 사용하지 않도록 한다.
3단계: 스트리밍 응답Fallback (실시간 챗봇용)
실시간 챗봇이나 AI 비서 서비스를 구축할 때는 스트리밍 응답이 필수적이다. 스트리밍 환경에서의 Fallback은 약간 더 복잡한 처리가 필요하다.
"""
스트리밍 응답 Fallback 시스템
실시간 챗봇 및 AI 비서 서비스용
"""
import openai
import time
import asyncio
from typing import AsyncGenerator, Optional
class StreamingFallbackClient:
"""
HolySheep AI 스트리밍 Fallback 클라이언트
스트리밍 환경에서는:
1. 첫 번째 토큰 수신까지의 시간을 모니터링
2. 지연이 임계값 초과 시 다음 모델로 전환
3. 전환 시 기존 스트림은 폐기하고 새로 시작
"""
def __init__(self, api_key: str):
self.client = openai.AsyncOpenAI(
api_key=api_key,
base_url="https://api.holysheep.ai/v1"
)
self.models = [
("claude-sonnet-4-20250514", 5), # 모델명, 첫 토큰 지연 임계값(초)
("gemini-2.0-flash", 4),
("gpt-4o", 6),
]
self.first_token_timeout = 10 # 첫 토큰 수신 최대 대기 시간
async def _stream_with_timeout(
self,
model: str,
messages: list
) -> AsyncGenerator[dict, None]:
"""
타임아웃이 있는 스트리밍 응답 생성기
첫 토큰이 임계값 내에 오지 않으면 예외 발생
"""
first_token_received = False
start_time = time.time()
try:
stream = await self.client.chat.completions.create(
model=model,
messages=messages,
stream=True,
temperature=0.7,
max_tokens=2000
)
async for chunk in stream:
# 첫 토큰 수신 시간 체크
if not first_token_received:
first_token_time = time.time() - start_time
first_token_received = True
if first_token_time > self.first_token_timeout:
# 첫 토큰 수신이 너무 늦었으면 스트림 중단
raise TimeoutError(
f"{model}: 첫 토큰 {first_token_time:.1f}초 소요 (임계값 초과)"
)
if chunk.choices and chunk.choices[0].delta.content:
yield {
"type": "content",
"content": chunk.choices[0].delta.content,
"model": model
}
# 스트림 완료
yield {
"type": "done",
"model": model,
"total_time": time.time() - start_time
}
except Exception as e:
yield {
"type": "error",
"model": model,
"error": str(e)
}
raise
async def chat_stream_with_fallback(
self,
messages: list
) -> AsyncGenerator[dict, None]:
"""
Fallback이 적용된 스트리밍 채팅
Yields:
dict: 토큰, 완료, 오류 이벤트
"""
for model_name, _ in self.models:
print(f"[INFO] {model_name} 시도 중...")
try:
async for event in self._stream_with_timeout(model_name, messages):
if event["type"] == "content":
# 정상 토큰 스트림
yield event
elif event["type"] == "done":
# 스트림 정상 완료
print(f"[SUCCESS] {model_name} 스트림 완료: {event['total_time']:.1f}초")
yield event
return
elif event["type"] == "error":
# 모델 오류, 다음 모델로 전환
print(f"[RETRY] {model_name} 실패: {event['error']}")
break # 다음 모델 시도
except (TimeoutError, Exception) as e:
print(f"[FALLBACK] {model_name} 타임아웃 또는 오류: {str(e)}")
continue # 다음 모델로 Fallback
# 모든 모델 실패
yield {
"type": "error",
"model": "none",
"error": "모든 모델 사용 불가"
}
asyncio 예시
async def main():
client = StreamingFallbackClient(api_key="YOUR_HOLYSHEEP_API_KEY")
messages = [
{"role": "user", "content": "인공지능의 미래에 대해 500자로 설명해주세요."}
]
print("=" * 50)
print("스트리밍 응답 시작")
print("=" * 50)
collected_response = ""
async for event in client.chat_stream_with_fallback(messages):
if event["type"] == "content":
print(event["content"], end="", flush=True)
collected_response += event["content"]
elif event["type"] == "done":
print(f"\n\n[완료] 사용 모델: {event['model']}")
print(f"[완료] 총 소요 시간: {event['total_time']:.1f}초")
elif event["type"] == "error":
print(f"\n[오류] {event['error']}")
if __name__ == "__main__":
asyncio.run(main())
4단계: 비즈니스 연속성 모니터링 대시보드 구축
실제 프로덕션 환경에서는 시스템의 상태를 실시간으로 모니터링해야 한다. Prometheus/Grafana 연동용 메트릭 수집 코드를 살펴보자.
"""
HolySheep AI 모니터링 및 알림 시스템
비즈니스 연속성을 위한 실시간 상태 추적
"""
import time
import threading
from dataclasses import dataclass, field
from typing import Dict, List, Callable, Optional
from collections import defaultdict
import statistics
@dataclass
class HealthMetrics:
"""모델별 건강 상태 메트릭"""
model_name: str
total_requests: int = 0
successful_requests: int = 0
failed_requests: int = 0
timeouts: int = 0
rate_limits: int = 0
latencies: List[float] = field(default_factory=list)
last_success_time: Optional[float] = None
last_failure_time: Optional[float] = None
consecutive_failures: int = 0
last_used: Optional[float] = None
class MonitoringDashboard:
"""
HolySheep AI 모니터링 대시보드
기능:
- 실시간 메트릭 수집 (SLA 추적)
- 자동 알림 (Webhook/Slack/Email)
- 장애 자동 보고서 생성
"""
def __init__(self):
self.metrics: Dict[str, HealthMetrics] = {}
self.alert_callbacks: List[Callable] = []
self.start_time = time.time()
# SLA 임계값 설정
self.sla_thresholds = {
"success_rate_min": 0.95, # 최소 95% 가용률
"latency_p95_max": 5000, # P95 지연시간 5초 이하
"timeout_rate_max": 0.05, # 타임아웃 5% 이하
}
def register_alert_callback(self, callback: Callable):
"""알림 콜백 등록 (webhook, slack 등)"""
self.alert_callbacks.append(callback)
def _send_alert(self, alert_type: str, message: str, severity: str = "warning"):
"""알림 발송"""
for callback in self.alert_callbacks:
try:
callback({
"type": alert_type,
"message": message,
"severity": severity,
"timestamp": time.time()
})
except Exception as e:
print(f"[알림 오류] {e}")
def record_request(
self,
model_name: str,
success: bool,
latency_ms: float,
error_type: Optional[str] = None
):
"""요청 결과 기록"""
if model_name not in self.metrics:
self.metrics[model_name] = HealthMetrics(model_name=model_name)
m = self.metrics[model_name]
m.total_requests += 1
m.last_used = time.time()
m.latencies.append(latency_ms)
# 최근 100개 latencies만 유지
if len(m.latencies) > 100:
m.latencies = m.latencies[-100:]
if success:
m.successful_requests += 1
m.consecutive_failures = 0
m.last_success_time = time.time()
else:
m.failed_requests += 1
m.consecutive_failures += 1
m.last_failure_time = time.time()
if error_type == "timeout":
m.timeouts += 1
elif error_type == "rate_limit":
m.rate_limits += 1
# 연속 실패 3회 이상 시 알림
if m.consecutive_failures >= 3:
self._send_alert(
"model_failure",
f"{model_name}: 연속 {m.consecutive_failures}회 실패",
severity="error"
)
def get_model_status(self, model_name: str) -> dict:
"""모델 상태 요약"""
if model_name not in self.metrics:
return {"status": "unknown", "message": "아직 데이터 없음"}
m = self.metrics[model_name]
if m.total_requests == 0:
return {"status": "no_data", "message": "요청 데이터 없음"}
success_rate = m.successful_requests / m.total_requests
avg_latency = statistics.mean(m.latencies) if m.latencies else 0
p95_latency = statistics.quantiles(m.latencies, n=20)[18] if len(m.latencies) >= 20 else avg_latency
timeout_rate = m.timeouts / m.total_requests
# 상태 판정
status = "healthy"
issues = []
if success_rate < self.sla_thresholds["success_rate_min"]:
status = "degraded"
issues.append(f"가용률 {success_rate:.1%} (임계값 {self.sla_thresholds['success_rate_min']:.1%})")
if p95_latency > self.sla_thresholds["latency_p95_max"]:
status = "degraded"
issues.append(f"P95 지연 {p95_latency:.0f}ms (임계값 {self.sla_thresholds['latency_p95_max']}ms)")
if timeout_rate > self.sla_thresholds["timeout_rate_max"]:
status = "degraded"
issues.append(f"타임아웃률 {timeout_rate:.1%} (임계값 {self.sla_thresholds['timeout_rate_max']:.1%})")
if m.consecutive_failures >= 3:
status = "unhealthy"
issues.append(f"연속 {m.consecutive_failures}회 실패")
return {
"status": status,
"success_rate": f"{success_rate:.1%}",
"avg_latency_ms": int(avg_latency),
"p95_latency_ms": int(p95_latency),
"total_requests": m.total_requests,
"issues": issues,
"last_success": m.last_success_time,
"last_failure": m.last_failure_time
}
def get_overall_health(self) -> dict:
"""전체 시스템 건강 상태"""
if not self.metrics:
return {"status": "unknown", "message": "모니터링 데이터 없음"}
statuses = [self.get_model_status(m)["status"] for m in self.metrics]
overall_status = "healthy"
if "unhealthy" in statuses:
overall_status = "unhealthy"
elif "degraded" in statuses:
overall_status = "degraded"
total_requests = sum(m.total_requests for m in self.metrics.values())
total_success = sum(m.successful_requests for m in self.metrics.values())
uptime = time.time() - self.start_time
return {
"status": overall_status,
"uptime_seconds": int(uptime),
"total_requests": total_requests,
"overall_success_rate": f"{total_success/total_requests:.1%}" if total_requests > 0 else "N/A",
"models": {name: self.get_model_status(name) for name in self.metrics.keys()}
}
def generate_report(self) -> str:
"""모니터링 보고서 생성 (MarkDown 형식)"""
health = self.get_overall_health()
report = f"""# HolySheep AI 모니터링 보고서
전체 상태: {health['status'].upper()}
**가동 시간**: {health['uptime_seconds']}초
**총 요청 수**: {health['total_requests']}
**전체 성공률**: {health['overall_success_rate']}
모델별 상태
| 모델 | 상태 | 성공률 | 평균 지연 | P95 지연 | 총 요청 |
|------|------|--------|----------|----------|---------|
"""
for model_name, status in health['models'].items():
status_icon = {"healthy": "✅", "degraded": "⚠️", "unhealthy": "❌", "unknown": "❓"}.get(status['status'], "❓")
report += f"| {model_name} | {status_icon} {status['status']} | {status['success_rate']} | {status['avg_latency_ms']}ms | {status['p95_latency_ms']}ms | {status['total_requests']} |\n"
return report
Slack 웹훅 예시
def slack_webhook_alert(alert_data: dict):
"""Slack으로 알림 발송"""
import json
import urllib.request
webhook_url = "YOUR_SLACK_WEBHOOK_URL" # Slack Webhook URL
color = {
"info": "#36a64f",
"warning": "#ff9800",
"error": "#f44336"
}.get(alert_data.get("severity", "info"), "#36a64f")
payload = {
"attachments": [{
"color": color,
"title": f"[{alert_data['severity'].upper()}] HolySheep AI Alert",
"text": alert_data['message'],
"footer": f"시간: {time.strftime('%Y-%m-%d %H:%M:%S')}"
}]
}
try:
data = json.dumps(payload).encode('utf-8')
req = urllib.request.Request(
webhook_url,
data=data,
headers={'Content-Type': 'application/json'}
)
urllib.request.urlopen(req, timeout=5)
print(f"[Slack] 알림 발송 완료: {alert_data['message']}")
except Exception as e:
print(f"[Slack] 알림 발송 실패: {e}")
사용 예시
if __name__ == "__main__":
dashboard = MonitoringDashboard()
dashboard.register_alert_callback(slack_webhook_alert)
# 시뮬레이션: 요청 기록
test_models = ["claude-sonnet-4-20250514", "gemini-2.0-flash", "gpt-4o"]
print("모니터링 시뮬레이션 시작...")
for i in range(20):
model = test_models[i % len(test_models)]
success = i % 10 != 0 # 10번째 요청은 실패
latency = 300 + (i % 5) * 100 # 300-700ms 지연
dashboard.record_request(
model_name=model,
success=success,
latency_ms=latency,
error_type="timeout" if i == 10 else None
)