AI API를 실무에 도입할 때, "API가 잘 작동하고 있는가?"를 확인하는 것은 선택이 아닌 필수입니다. 저는 3년간 다양한 AI API를 운영하면서 요청 실패, 비용 초과, 응답 지연等问题으로 밤을 지새운 경험이 있습니다. 이번 가이드에서는 HolySheep AI를 기반으로 AI API의 가시성을 체계적으로 구축하는 방법을 초보자도 이해할 수 있도록 단계별로 설명드리겠습니다.
가시성이란 무엇인가?
간단히 말하면, AI API의 가시성(Observability)이란 시스템 내부에서 무슨 일이 일어나는지 밖에서 확인할 수 있는 능력을 의미합니다. 세 가지 핵심 요소로 구성됩니다:
- 메트릭(Metrics): 요청 수, 응답 시간, 에러율 등 숫자로 측정 가능한 데이터
- 로그(Logs): 각 요청의 상세 기록으로 문제 발생 시 원인 파악에 필수
- 트레이스(Traces): 요청의 흐름을 처음부터 끝까지 추적하는 기능
1단계: HolySheep AI 기본 설정
가시성 구축의 첫 번째 단계는 정상적인 API 연결을 확인하는 것입니다. HolySheep AI는 단일 API 키로 여러 모델을 지원하므로 중앙에서 모니터링하기에 최적화되어 있습니다.
1-1. API 키 발급 및 환경 설정
먼저 지금 가입하여 API 키를 발급받으세요. 가입 시 무료 크레딧이 제공되므로 즉시 테스트가 가능합니다.
# 환경 변수 설정 (.env 파일)
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export HOLYSHEEP_BASE_URL="https://api.holysheep.ai/v1"
Python 환경에서 사용
pip install openai requests python-dotenv1-2. 연결 확인 테스트
import openai
import os
from dotenv import load_dotenv
환경 변수 로드
load_dotenv()
HolySheep AI 클라이언트 설정
client = openai.OpenAI(
api_key=os.getenv("HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1"
)
연결 테스트 - 간단한 요청으로 응답 시간 측정
import time
start_time = time.time()
response = client.chat.completions.create(
model="gpt-4.1",
messages=[{"role": "user", "content": "안녕하세요"}],
max_tokens=50
)
end_time = time.time()
print(f"모델: gpt-4.1")
print(f"응답 시간: {(end_time - start_time) * 1000:.2f}ms")
print(f"토큰 사용량: {response.usage.total_tokens}")
print(f"예상 비용: ${response.usage.total_tokens / 1_000_000 * 8:.4f}")
print(f"응답 내용: {response.choices[0].message.content}")【스크린샷 힌트】위 코드를 실행하면 콘솔에 모델명, 응답 시간, 토큰 사용량, 예상 비용이 출력됩니다. 이 기본 테스트가 성공하면 API 연결은 정상입니다.
2단계: 메트릭 수집 시스템 구축
AI API의健康 상태를 파악하려면 핵심 메트릭을 자동으로 수집해야 합니다. HolySheep AI는 요청별로 토큰 사용량과 모델 정보를 제공하므로 이를 활용합니다.
2-1. 메트릭 수집 클래스 구현
import time
import json
from datetime import datetime
from collections import defaultdict
from typing import Dict, List, Optional
class AIMetricsCollector:
"""AI API 메트릭 수집기"""
def __init__(self):
self.metrics = {
"requests": 0,
"total_tokens": 0,
"total_cost": 0.0,
"response_times": [],
"errors": 0,
"model_usage": defaultdict(int)
}
# 모델별 가격표 (HolySheep AI 기준, $/1M 토큰)
self.model_prices = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
def record_request(self, model: str, usage, response_time_ms: float, error: Optional[str] = None):
"""API 요청 메트릭 기록"""
self.metrics["requests"] += 1
# 토큰 사용량 기록
prompt_tokens = usage.prompt_tokens if hasattr(usage, 'prompt_tokens') else 0
completion_tokens = usage.completion_tokens if hasattr(usage, 'completion_tokens') else 0
total_tokens = usage.total_tokens if hasattr(usage, 'total_tokens') else prompt_tokens + completion_tokens
self.metrics["total_tokens"] += total_tokens
self.metrics["model_usage"][model] += 1
# 비용 계산
price = self.model_prices.get(model, 8.0) # 기본값: $8/MTok
cost = (total_tokens / 1_000_000) * price
self.metrics["total_cost"] += cost
# 응답 시간 기록
self.metrics["response_times"].append(response_time_ms)
# 에러 기록
if error:
self.metrics["errors"] += 1
def get_summary(self) -> Dict:
"""메트릭 요약 반환"""
response_times = self.metrics["response_times"]
return {
"total_requests": self.metrics["requests"],
"total_tokens": self.metrics["total_tokens"],
"total_cost_usd": round(self.metrics["total_cost"], 4),
"error_rate": round(self.metrics["errors"] / max(self.metrics["requests"], 1) * 100, 2),
"avg_response_time_ms": round(sum(response_times) / max(len(response_times), 1), 2),
"p95_response_time_ms": round(sorted(response_times)[int(len(response_times) * 0.95)] if response_times else 0, 2),
"model_usage": dict(self.metrics["model_usage"])
}
메트릭 수집기 인스턴스 생성
metrics_collector = AIMetricsCollector()2-2. 모니터링 데코레이터 활용
import functools
import time
def monitor_api_call(collector, model_name: str):
"""API 호출을 모니터링하는 데코레이터"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
error_msg = None
usage = None
try:
result = func(*args, **kwargs)
# OpenAI SDK 응답 구조에서 usage 추출
if hasattr(result, 'usage'):
usage = result.usage
return result
except Exception as e:
error_msg = str(e)
raise
finally:
elapsed_ms = (time.time() - start) * 1000
if usage:
collector.record_request(model_name, usage, elapsed_ms, error_msg)
return wrapper
return decorator
실제 사용 예시
@monitor_api_call(metrics_collector, "gpt-4.1")
def call_gpt4(prompt: str):
return client.chat.completions.create(
model="gpt-4.1",
messages=[{"role": "user", "content": prompt}],
max_tokens=100
)
여러 모델 호출 테스트
test_prompts = ["인공지능이란?", "머신러닝 기초", "딥러닝이란?"]
for prompt in test_prompts:
call_gpt4(prompt)
결과 확인
print("=== HolySheep AI 모니터링 결과 ===")
summary = metrics_collector.get_summary()
for key, value in summary.items():
print(f"{key}: {value}")【스크린샷 힌트】위 코드를 실행하면 총 요청 수, 토큰 사용량, 비용, 에러율, 평균 응답 시간, P95 응답 시간, 모델별 사용량이 출력됩니다.
3단계: 비용 추적 및 알림 시스템
AI API 사용 시 가장 민감한 부분이 바로 비용입니다. HolySheep AI의 가격표를 기준으로 월별/일별 비용을 추적하고閾値 초과 시 알림을 받는 시스템을 구축합니다.
3-1. 비용 추적기 구현
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, List
class CostTracker:
"""AI API 비용 추적기 - SQLite 기반"""
def __init__(self, db_path: "ai_costs.db"):
self.conn = sqlite3.connect(db_path)
self.create_table()
def create_table(self):
"""비용 기록 테이블 생성"""
self.conn.execute("""
CREATE TABLE IF NOT EXISTS api_costs (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
model TEXT NOT NULL,
prompt_tokens INTEGER,
completion_tokens INTEGER,
total_tokens INTEGER,
cost_usd REAL,
response_time_ms REAL,
status TEXT
)
""")
self.conn.commit()
def log_request(self, model: str, usage, cost_usd: float, response_time_ms: float, status: str = "success"):
"""요청 비용 기록"""
self.conn.execute("""
INSERT INTO api_costs
(timestamp, model, prompt_tokens, completion_tokens, total_tokens, cost_usd, response_time_ms, status)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
datetime.now().isoformat(),
model,
usage.prompt_tokens if hasattr(usage, 'prompt_tokens') else 0,
usage.completion_tokens if hasattr(usage, 'completion_tokens') else 0,
usage.total_tokens if hasattr(usage, 'total_tokens') else 0,
cost_usd,
response_time_ms,
status
))
self.conn.commit()
def get_daily_cost(self, days: int = 7) -> List[Dict]:
"""최근 N일 일별 비용 조회"""
cursor = self.conn.execute("""
SELECT DATE(timestamp) as date,
SUM(cost_usd) as total_cost,
SUM(total_tokens) as total_tokens,
COUNT(*) as request_count
FROM api_costs
WHERE timestamp >= datetime('now', ?)
GROUP BY DATE(timestamp)
ORDER BY date DESC
""", (f'-{days} days',))
return [
{"date": row[0], "cost_usd": row[1], "total_tokens": row[2], "requests": row[3]}
for row in cursor.fetchall()
]
def get_cost_by_model(self, days: int = 30) -> Dict:
"""모델별 비용 분석"""
cursor = self.conn.execute("""
SELECT model,
SUM(cost_usd) as total_cost,
SUM(total_tokens) as total_tokens,
COUNT(*) as request_count,
AVG(response_time_ms) as avg_response_time
FROM api_costs
WHERE timestamp >= datetime('now', ?)
GROUP BY model
ORDER BY total_cost DESC
""", (f'-{days} days',))
return {row[0]: {
"cost_usd": round(row[1], 4),
"total_tokens": row[2],
"requests": row[3],
"avg_response_time_ms": round(row[4], 2)
} for row in cursor.fetchall()}
사용 예시
cost_tracker = CostTracker("ai_costs.db")
테스트 요청 및 비용 기록
test_models = [
("gpt-4.1", "gpt-4.1"),
("gemini-2.5-flash", "gemini-2.5-flash"),
("deepseek-v3.2", "deepseek-v3.2")
]
for model_id, model_name in test_models:
start = time.time()
try:
response = client.chat.completions.create(
model=model_id,
messages=[{"role": "user", "content": "AI API 모니터링 방법을 알려주세요"}],
max_tokens=100
)
elapsed = (time.time() - start) * 1000
cost_tracker.log_request(model_name, response.usage, 0.0008, elapsed)
print(f"{model_id}: 성공 - {elapsed:.2f}ms")
except Exception as e:
print(f"{model_id}: 실패 - {e}")
비용 보고서 출력
print("\n=== 모델별 비용 분석 ===")
model_costs = cost_tracker.get_cost_by_model()
for model, data in model_costs.items():
print(f"{model}: ${data['cost_usd']} ({data['requests']}회 요청)")3-2. 비용 알림 시스템
import os
import smtplib
from email.mime.text import MIMEText
from dataclasses import dataclass
@dataclass
class CostAlert:
"""비용 알림 설정"""
daily_limit: float = 10.0 # 일일 한도 ($)
weekly_limit: float = 50.0 # 주간 한도 ($)
per_request_limit: float = 1.0 # 요청당 한도 ($)
class CostAlertSystem:
"""비용 초과 알림 시스템"""
def __init__(self, alert_config: CostAlert):
self.config = alert_config
self.today_spending = 0.0
self.weekly_spending = 0.0
def check_and_alert(self, current_cost: float) -> List[str]:
"""비용 확인 및 알림 필요 시 알림 메시지 반환"""
alerts = []
self.today_spending += current_cost
if self.today_spending > self.config.daily_limit:
alerts.append(
f"⚠️ [경고] 일일 비용 한도 초과! "
f"현재: ${self.today_spending:.2f} / 한도: ${self.config.daily_limit}"
)
if current_cost > self.config.per_request_limit:
alerts.append(
f"⚠️ [경고] 요청당 비용 초과! "
f"현재 요청: ${current_cost:.2f} / 한도: ${self.config.per_request_limit}"
)
return alerts
def send_email_alert(self, message: str):
"""이메일 알림 전송 (구성 필요 시 활성화)"""
# 환경 변수에서 SMTP 설정 로드
smtp_server = os.getenv("SMTP_SERVER")
smtp_port = int(os.getenv("SMTP_PORT", "587"))
smtp_user = os.getenv("SMTP_USER")
smtp_password = os.getenv("SMTP_PASSWORD")
alert_recipient = os.getenv("ALERT_EMAIL")
if not all([smtp_server, smtp_user, smtp_password, alert_recipient]):
print("이메일 알림: SMTP 설정이 완료되지 않음 - 콘솔 출력만")
print(f" {message}")
return
msg = MIMEText(message)
msg['Subject'] = '[HolySheep AI] 비용 알림'
msg['From'] = smtp_user
msg['To'] = alert_recipient
try:
with smtplib.SMTP(smtp_server, smtp_port) as server:
server.starttls()
server.login(smtp_user, smtp_password)
server.send_message(msg)
print(f"이메일 알림 전송 완료: {alert_recipient}")
except Exception as e:
print(f"이메일 전송 실패: {e}")
사용 예시
alert_system = CostAlertSystem(
CostAlert(daily_limit=5.0, per_request_limit=0.50)
)
테스트 알림 발생
test_cost = 0.75
alerts = alert_system.check_and_alert(test_cost)
for alert in alerts:
alert_system.send_email_alert(alert)4단계: 응답 시간 모니터링 및 최적화
AI API의 응답 시간은 사용자 경험에 직접적인 영향을 미칩니다. HolySheep AI에서 여러 모델의 응답 시간을 비교하고 최적의 모델을 선택하는 방법을 살펴보겠습니다.
4-1. 모델별 응답 시간 벤치마크
import statistics
from typing import Tuple, List
class ModelBenchmark:
"""AI 모델 응답 시간 벤치마크"""
def __init__(self, client, metrics_collector):
self.client = client
self.collector = metrics_collector
self.test_prompt = "한국의 주요 관광지 3곳을 간단히 소개해주세요."
def benchmark_model(self, model: str, iterations: int = 5) -> Dict:
"""단일 모델 벤치마크 실행"""
response_times = []
successes = 0
errors = []
for i in range(iterations):
start = time.time()
try:
response = self.client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": self.test_prompt}],
max_tokens=150,
temperature=0.7
)
elapsed = (time.time() - start) * 1000
response_times.append(elapsed)
self.collector.record_request(model, response.usage, elapsed)
successes += 1
except Exception as e:
errors.append(str(e))
if response_times:
return {
"model": model,
"iterations": iterations,
"success_rate": successes / iterations * 100,
"avg_time_ms": round(statistics.mean(response_times), 2),
"min_time_ms": round(min(response_times), 2),
"max_time_ms": round(max(response_times), 2),
"std_dev_ms": round(statistics.stdev(response_times), 2) if len(response_times) > 1 else 0,
"errors": errors[:3] # 최대 3개 에러만 표시
}
return None
def run_full_benchmark(self) -> List[Dict]:
"""모든 모델 벤치마크 실행"""
models = [
"gpt-4.1",
"gemini-2.5-flash",
"deepseek-v3.2"
]
results = []
for model in models:
print(f"[ 벤치마크 중 ] {model}...")
result = self.benchmark_model(model, iterations=5)
if result:
results.append(result)
print(f" ✓ 평균 응답 시간: {result['avg_time_ms']}ms")
else:
print(f" ✗ 벤치마크 실패")
return sorted(results, key=lambda x: x['avg_time_ms'])
벤치마크 실행
benchmark = ModelBenchmark(client, metrics_collector)
print("=== HolySheep AI 모델 응답 시간 벤치마크 ===\n")
benchmark_results = benchmark.run_full_benchmark()
print("\n=== 결과 비교 ===")
print(f"{'모델':<20} {'평균(ms)':<12} {'최소(ms)':<12} {'최대(ms)':<12} {'성공률':<10}")
print("-" * 70)
for r in benchmark_results:
print(f"{r['model']:<20} {r['avg_time_ms']:<12} {r['min_time_ms']:<12} {r['max_time_ms']:<12} {r['success_rate']:.0f}%")4-2. 자동 모델 선택 시스템
from enum import Enum
from typing import Optional, Callable
class TaskPriority(Enum):
"""작업 우선순위"""
SPEED = "speed" # 속도 우선
QUALITY = "quality" # 품질 우선
COST = "cost" # 비용 우선
BALANCED = "balanced" # 균형
class ModelRouter:
"""작업 유형에 따른 자동 모델 선택"""
def __init__(self):
# HolySheep AI에서 사용 가능한 모델
self.model_configs = {
"gpt-4.1": {"speed": 3, "quality": 5, "cost": 2},
"claude-sonnet-4.5": {"speed": 3, "quality": 5, "cost": 2},
"gemini-2.5-flash": {"speed": 5, "quality": 3, "cost": 5},
"deepseek-v3.2": {"speed": 4, "quality": 3, "cost": 5}
}
def select_model(self, priority: TaskPriority, avg_response_times: dict = None) -> str:
"""우선순위에 맞는 최적 모델 선택"""
if priority == TaskPriority.SPEED:
# Gemini 2.5 Flash가 가장 빠름
return "gemini-2.5-flash"
elif priority == TaskPriority.QUALITY:
# GPT-4.1 또는 Claude Sonnet 4.5
if avg_response_times:
gpt_time = avg_response_times.get("gpt-4.1", 9999)
claude_time = avg_response_times.get("claude-sonnet-4.5", 9999)
return "gpt-4.1" if gpt_time <= claude_time else "claude-sonnet-4.5"
return "gpt-4.1"
elif priority == TaskPriority.COST:
# DeepSeek V3.2가 가장 저렴
return "deepseek-v3.2"
else: # BALANCED
# 응답 시간 기반 동적 선택
if avg_response_times:
times = {k: v for k, v in avg_response_times.items() if v < 2000}
if times:
return min(times, key=times.get)
return "gemini-2.5-flash"
def get_recommendation(self, task_type: str) -> dict:
"""작업 유형별 모델 추천"""
recommendations = {
"chatbot": {"model": "gemini-2.5-flash", "reason": "빠른 응답 시간"},
"code_generation": {"model": "gpt-4.1", "reason": "높은 코드 품질"},
"data_analysis": {"model": "gpt-4.1", "reason": "정확한 분석能力"},
"bulk_processing": {"model": "deepseek-v3.2", "reason": "최저 비용"},
"creative_writing": {"model": "claude-sonnet-4.5", "reason": "창의적 콘텐츠"}
}
return recommendations.get(task_type, {"model": "gemini-2.5-flash", "reason": "범용 최적"})
사용 예시
router = ModelRouter()
avg_times = {"gpt-4.1": 1500, "gemini-2.5-flash": 600, "deepseek-v3.2": 800}
print("=== 작업별 모델 추천 ===")
for task in ["chatbot", "code_generation", "bulk_processing"]:
rec = router.get_recommendation(task)
print(f"{task}: {rec['model']} ({rec['reason']})")
print("\n=== 우선순위별 모델 선택 ===")
for priority in TaskPriority:
model = router.select_model(priority, avg_times)
print(f"{priority.value}: {model}")5단계: 로깅 시스템 구축
문제가 발생했을 때 원인을 파악하려면 상세한 로그가 필수입니다. 요청과 응답의 모든 정보를 구조화하여 기록하는 시스템을 구축합니다.
5-1. 구조화 로깅 시스템
import json
import logging
from logging.handlers import RotatingFileHandler
from datetime import datetime
from pathlib import Path
class AILogger:
"""AI API 전용 로거"""
def __init__(self, log_dir: str = "logs", log_level: int = logging.INFO):
self.log_dir = Path(log_dir)
self.log_dir.mkdir(exist_ok=True)
# 로거 설정
self.logger = logging.getLogger("AI_API")
self.logger.setLevel(log_level)
self.logger.handlers = []
# 파일 핸들러 (일별 로테이션)
log_file = self.log_dir / f"ai_api_{datetime.now().strftime('%Y%m%d')}.log"
file_handler = RotatingFileHandler(
log_file, maxBytes=10_000_000, backupCount=30
)
file_handler.setLevel(log_level)
# 콘솔 핸들러
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.WARNING)
# 포맷터
formatter = logging.Formatter(
'%(asctime)s | %(levelname)-8s | %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)
self.logger.addHandler(file_handler)
self.logger.addHandler(console_handler)
def log_request(self, model: str, prompt: str, tokens: int, cost: float,
response_time_ms: float, response_length: int):
"""API 요청 로깅"""
log_data = {
"type": "request",
"model": model,
"prompt_length": len(prompt),
"tokens": tokens,
"cost_usd": round(cost, 6),
"response_time_ms": round(response_time_ms, 2),
"response_length": response_length,
"tokens_per_second": round(tokens / (response_time_ms / 1000), 2) if response_time_ms > 0 else 0
}
self.logger.info(json.dumps(log_data, ensure_ascii=False))
def log_error(self, model: str, error_type: str, error_message: str,
prompt: str = None):
"""에러 로깅"""
log_data = {
"type": "error",
"model": model,
"error_type": error_type,
"error_message": error_message[:500] if error_message else None,
"prompt_preview": prompt[:100] + "..." if prompt and len(prompt) > 100 else prompt
}
self.logger.error(json.dumps(log_data, ensure_ascii=False))
def log_performance_alert(self, model: str, response_time_ms: float, threshold_ms: float):
"""성능 경고 로깅"""
log_data = {
"type": "performance_alert",
"model": model,
"response_time_ms": response_time_ms,
"threshold_ms": threshold_ms,
"over_by_ms": round(response_time_ms - threshold_ms, 2)
}
self.logger.warning(json.dumps(log_data, ensure_ascii=False))
사용 예시
ai_logger = AILogger(log_level=logging.INFO)
정상 요청 로깅
ai_logger.log_request(
model="gpt-4.1",
prompt="한국의 역사와 문화에 대해 설명해주세요.",
tokens=2500,
cost=0.02,
response_time_ms=1250.5,
response_length=850
)
에러 로깅
ai_logger.log_error(
model="gpt-4.1",
error_type="RateLimitError",
error_message="요청 빈도가 제한을 초과했습니다. 60초 후에 다시 시도해주세요.",
prompt="긴급: 시스템 상태 확인 요청"
)
성능 경고 로깅
ai_logger.log_performance_alert(
model="gemini-2.5-flash",
response_time_ms=2500,
threshold_ms=2000
)
print("로그가 'logs/ai_api_YYYYMMDD.log' 파일에 기록되었습니다.")5-2. 로그 분석 및 보고서 생성
import re
from collections import Counter
class LogAnalyzer:
"""AI API 로그 분석기"""
def __init__(self, log_file: str):
self.log_file = log_file
def parse_logs(self) -> List[Dict]:
"""로그 파일 파싱"""
entries = []
try:
with open(self.log_file, 'r', encoding='utf-8') as f:
for line in f:
if ' | ' in line:
parts = line.strip().split(' | ', 2)
if len(parts) >= 3:
timestamp, level, message = parts
try:
data = json.loads(message)
data['timestamp'] = timestamp
data['level'] = level
entries.append(data)
except json.JSONDecodeError:
continue
except FileNotFoundError:
print(f"로그 파일을 찾을 수 없습니다: {self.log_file}")
return entries
def generate_report(self, entries: List[Dict]) -> Dict:
"""로그 분석 보고서 생성"""
if not entries:
return {"error": "분석할 로그가 없습니다"}
total_requests = sum(1 for e in entries if e.get('type') == 'request')
total_errors = sum(1 for e in entries if e.get('type') == 'error')
model_stats = Counter(e.get('model') for e in entries if e.get('model'))
response_times = [e['response_time_ms'] for e in entries
if e.get('type') == 'request' and 'response_time_ms' in e]
error_types = Counter(e.get('error_type') for e in entries
if e.get('type') == 'error')
return {
"summary": {
"total_logs": len(entries),
"total_requests": total_requests,
"total_errors": total_errors,
"error_rate": round(total_errors / max(total_requests, 1) * 100, 2)
},
"model_usage": dict(model_stats),
"performance": {
"avg_response_ms": round(sum(response_times) / max(len(response_times), 1), 2),
"max_response_ms": max(response_times) if response_times else 0,
"min_response_ms": min(response_times) if response_times else 0
},
"top_errors": dict(error_types.most_common(5))
}
def find_slow_requests(self, threshold_ms: float = 2000) -> List[Dict]:
"""느린 요청 찾기"""
entries = self.parse_logs()
return [
e for e in entries
if e.get('type') == 'request' and e.get('response_time_ms', 0) > threshold_ms
]
사용 예시 (실제 로그 파일이 있을 때)
print("=== 로그 분석 사용 방법 ===")
print("1. LogAnalyzer('logs/ai_api_20240115.log')로 초기화")
print("2. parse_logs()로 로그 파싱")
print("3. generate_report()로 분석 보고서 생성")
print("4. find_slow_requests(2000)로 느린 요청 조회")6단계: 종합 대시보드 구축
지금까지 구축한 모든 시스템을 하나로 통합하여 실시간 대시보드를 만들겠습니다. 터미널에서 실행 가능한 간단한 버전과 웹 대시보드용 프레임워크를 모두 제공합니다.
6-1. 실시간 모니터링 대시보드
import os
import sys
import time
from threading import Thread
from queue import Queue
class RealtimeDashboard:
"""실시간 AI API 모니터링 대시보드"""
def __init__(self, refresh_interval: int = 5):
self.refresh_interval = refresh_interval
self.metrics_queue = Queue()
self.running = False
# HolySheep AI 모델 가격
self.model_prices = {
"gpt-4.1": 8.0,
"claude-sonnet-4.5": 15.0,
"gemini-2.5-flash": 2.50,
"deepseek-v3.2": 0.42
}
def clear_screen(self):
"""화면 지우기 (크로스 플랫폼)"""
os.system('cls' if os.name == 'nt' else 'clear')
def format_meter(self, value: float, max_value: float, width: int = 20) -> str:
"""텍스트 기반 미터 생성"""
filled = int((value / max_value) * width)
empty = width - filled
bar = '█' * filled + '░' * empty
percentage = (value / max_value) * 100
return f"[{bar}] {percentage:.1f}%"
def format_bytes(self, bytes_value: float) -> str:
"""토큰数を読みやすい 단위로 변환"""
return f"{bytes_value:,.0f}"
def display_header(self):
"""대시보드 헤더 표시"""
print("=" * 70)
print(" HolySheep AI 실시간 모니터링 대시보드")
print("=" * 70)
print(f" 업데이트: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("-" * 70)
def display_metrics(self, metrics: Dict):
"""메트릭 정보 표시"""
print(f"\n📊 요청 통계")
print(f" ├─ 총 요청 수: {metrics.get('total_requests', 0):,}회")
print(f" ├─ 총 토큰 사용: {self.format_bytes(metrics.get('total_tokens', 0))}토큰")
print(f" └─ 총 비용: ${metrics.get('total_cost_usd', 0):.4f}")
print(f"\n⏱️ 응답 시간")
print(f" ├─ 평균: {metrics.get('avg_response_time_ms', 0):.2f}ms")
print(f" └─ P95: {metrics.get('p95_response_time_ms', 0):.2f}ms")
error_rate = metrics.get('error_rate', 0)
print(f"\n✅ 가동 상태")
if error_rate < 1:
status = "🟢 양호"
elif error_rate < 5:
status = "🟡 주의"
else:
status = "🔴 위험"
print(f" ├─ 상태: {status}")
print(f" └─ 에러율: {