9장: 모니터링 하네스 — 프로덕션 관측과 피드백 루프

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from collections import defaultdict
 
 
@dataclass
class TokenUsage:
    input_tokens: int
    output_tokens: int
    model: str
    timestamp: datetime
    request_id: str
    user_id: str | None = None
    feature: str | None = None  # 어떤 기능에서 사용했는지
 
 
# 모델별 토큰 단가 (USD per 1M tokens)
PRICING = {
    "claude-opus-4-20250514": {"input": 15.0, "output": 75.0},
    "claude-sonnet-4-20250514": {"input": 3.0, "output": 15.0},
    "claude-haiku-3-20250414": {"input": 0.25, "output": 1.25},
    "gpt-4o": {"input": 2.5, "output": 10.0},
    "gpt-4o-mini": {"input": 0.15, "output": 0.6},
}
 
 
class CostTracker:
    """토큰 사용량 및 비용 추적"""
 
    def __init__(self):
        self.usages: list[TokenUsage] = []
        self.budget_alerts: list[dict] = []
 
    def record(self, usage: TokenUsage):
        self.usages.append(usage)
        self._check_budget(usage)
 
    def calculate_cost(self, usage: TokenUsage) -> float:
        """단일 요청의 비용 계산"""
        pricing = PRICING.get(usage.model, {"input": 0, "output": 0})
        input_cost = usage.input_tokens * pricing["input"] / 1_000_000
        output_cost = usage.output_tokens * pricing["output"] / 1_000_000
        return input_cost + output_cost
 
    def daily_report(self, date: datetime | None = None) -> dict:
        """일일 비용 리포트"""
        target_date = (date or datetime.now()).date()
        daily = [
            u for u in self.usages
            if u.timestamp.date() == target_date
        ]
 
        by_model = defaultdict(lambda: {"tokens": 0, "cost": 0.0, "requests": 0})
        by_feature = defaultdict(lambda: {"tokens": 0, "cost": 0.0})
 
        for usage in daily:
            cost = self.calculate_cost(usage)
            total_tokens = usage.input_tokens + usage.output_tokens
 
            by_model[usage.model]["tokens"] += total_tokens
            by_model[usage.model]["cost"] += cost
            by_model[usage.model]["requests"] += 1
 
            if usage.feature:
                by_feature[usage.feature]["tokens"] += total_tokens
                by_feature[usage.feature]["cost"] += cost
 
        return {
            "date": str(target_date),
            "total_requests": len(daily),
            "total_cost": sum(self.calculate_cost(u) for u in daily),
            "by_model": dict(by_model),
            "by_feature": dict(by_feature),
        }
 
    def _check_budget(self, usage: TokenUsage):
        """예산 초과 확인"""
        today_cost = sum(
            self.calculate_cost(u) for u in self.usages
            if u.timestamp.date() == datetime.now().date()
        )
        # 일일 예산 50달러 기준
        daily_budget = 50.0
        if today_cost > daily_budget * 0.8:
            self.budget_alerts.append({
                "type": "budget_warning",
                "message": f"일일 예산의 80% 도달: ${today_cost:.2f}/${daily_budget}",
                "timestamp": datetime.now(),
            })

import time
from contextlib import contextmanager
from dataclasses import dataclass, field
 
 
@dataclass
class LatencyBreakdown:
    """지연시간 구간별 분해"""
    total_ms: float = 0
    preprocessing_ms: float = 0
    model_inference_ms: float = 0
    postprocessing_ms: float = 0
    tool_execution_ms: float = 0
    guardrail_ms: float = 0
 
 
class LatencyTracer:
    """구간별 지연시간 추적"""
 
    def __init__(self):
        self.breakdown = LatencyBreakdown()
        self._spans: dict[str, float] = {}
 
    @contextmanager
    def span(self, name: str):
        """구간 측정"""
        start = time.monotonic()
        try:
            yield
        finally:
            elapsed = (time.monotonic() - start) * 1000
            setattr(self.breakdown, f"{name}_ms", elapsed)
 
    def report(self) -> dict:
        self.breakdown.total_ms = (
            self.breakdown.preprocessing_ms
            + self.breakdown.model_inference_ms
            + self.breakdown.postprocessing_ms
            + self.breakdown.tool_execution_ms
            + self.breakdown.guardrail_ms
        )
        return {
            "total_ms": self.breakdown.total_ms,
            "breakdown": {
                "preprocessing": self.breakdown.preprocessing_ms,
                "model_inference": self.breakdown.model_inference_ms,
                "postprocessing": self.breakdown.postprocessing_ms,
                "tool_execution": self.breakdown.tool_execution_ms,
                "guardrail": self.breakdown.guardrail_ms,
            },
            "bottleneck": max(
                [
                    ("preprocessing", self.breakdown.preprocessing_ms),
                    ("model_inference", self.breakdown.model_inference_ms),
                    ("postprocessing", self.breakdown.postprocessing_ms),
                    ("tool_execution", self.breakdown.tool_execution_ms),
                    ("guardrail", self.breakdown.guardrail_ms),
                ],
                key=lambda x: x[1],
            )[0],
        }
 
 
# 사용 예시
async def process_request(request: str) -> str:
    tracer = LatencyTracer()
 
    with tracer.span("preprocessing"):
        processed = preprocess(request)
 
    with tracer.span("guardrail"):
        validated = check_guardrails(processed)
 
    with tracer.span("model_inference"):
        response = await call_model(validated)
 
    with tracer.span("postprocessing"):
        result = postprocess(response)
 
    latency_report = tracer.report()
    await send_metrics(latency_report)
 
    return result

import bisect
 
 
class PercentileMonitor:
    """백분위수 기반 지연시간 모니터링"""
 
    def __init__(self, window_size: int = 1000):
        self.values: list[float] = []
        self.window_size = window_size
 
    def record(self, value: float):
        bisect.insort(self.values, value)
        if len(self.values) > self.window_size:
            self.values.pop(0)
 
    def percentile(self, p: float) -> float:
        if not self.values:
            return 0
        idx = int(len(self.values) * p / 100)
        return self.values[min(idx, len(self.values) - 1)]
 
    def summary(self) -> dict:
        return {
            "p50": self.percentile(50),
            "p95": self.percentile(95),
            "p99": self.percentile(99),
            "min": self.values[0] if self.values else 0,
            "max": self.values[-1] if self.values else 0,
            "count": len(self.values),
        }

from collections import Counter
import math
 
 
class DriftDetector:
    """입출력 드리프트 감지"""
 
    def __init__(self, baseline_window: int = 1000):
        self.baseline_window = baseline_window
        self.baseline_distributions: dict[str, Counter] = {}
        self.current_distributions: dict[str, Counter] = {}
 
    def set_baseline(self, metric_name: str, values: list[str]):
        """기준 분포 설정"""
        self.baseline_distributions[metric_name] = Counter(values)
 
    def update_current(self, metric_name: str, value: str):
        """현재 분포 업데이트"""
        if metric_name not in self.current_distributions:
            self.current_distributions[metric_name] = Counter()
        self.current_distributions[metric_name][value] += 1
 
    def detect_drift(
        self, metric_name: str, threshold: float = 0.1
    ) -> dict:
        """KL 다이버전스 기반 드리프트 감지"""
        baseline = self.baseline_distributions.get(metric_name)
        current = self.current_distributions.get(metric_name)
 
        if not baseline or not current:
            return {"drift_detected": False, "reason": "데이터 부족"}
 
        kl_div = self._kl_divergence(baseline, current)
 
        return {
            "drift_detected": kl_div > threshold,
            "kl_divergence": kl_div,
            "threshold": threshold,
            "baseline_size": sum(baseline.values()),
            "current_size": sum(current.values()),
        }
 
    @staticmethod
    def _kl_divergence(p: Counter, q: Counter) -> float:
        """KL 다이버전스 계산"""
        all_keys = set(p.keys()) | set(q.keys())
        p_total = sum(p.values())
        q_total = sum(q.values())
 
        kl = 0.0
        for key in all_keys:
            p_prob = (p.get(key, 0) + 1) / (p_total + len(all_keys))
            q_prob = (q.get(key, 0) + 1) / (q_total + len(all_keys))
            if p_prob > 0:
                kl += p_prob * math.log(p_prob / q_prob)
 
        return kl
 
 
# 사용 예시
detector = DriftDetector()
 
# 기준 분포 설정 (지난 주의 질문 카테고리)
detector.set_baseline("question_category", [
    "product", "product", "billing", "support",
    "product", "billing", "product", "support",
    # ... 1000개의 기준 데이터
])
 
# 실시간 업데이트
detector.update_current("question_category", "product")
detector.update_current("question_category", "complaint")  # 새로운 카테고리
 
# 드리프트 확인
drift_result = detector.detect_drift("question_category")
if drift_result["drift_detected"]:
    alert("입력 드리프트 감지: 질문 패턴이 변경되었습니다")

from dataclasses import dataclass
 
 
@dataclass
class QualityMetrics:
    """품질 메트릭"""
    relevance_score: float  # 질문과 답변의 관련성
    completeness_score: float  # 답변의 완전성
    safety_score: float  # 안전성 점수
    user_rating: float | None  # 사용자 평점 (선택적)
 
 
class QualityMonitor:
    """실시간 품질 모니터링"""
 
    def __init__(self, min_quality_threshold: float = 0.7):
        self.threshold = min_quality_threshold
        self.scores: list[QualityMetrics] = []
        self.alert_count = 0
 
    async def evaluate(
        self,
        question: str,
        answer: str,
        evaluator_fn,
    ) -> QualityMetrics:
        """응답 품질 평가"""
        metrics = await evaluator_fn(question, answer)
        self.scores.append(metrics)
 
        # 품질 저하 감지
        avg_score = (
            metrics.relevance_score
            + metrics.completeness_score
            + metrics.safety_score
        ) / 3
 
        if avg_score < self.threshold:
            self.alert_count += 1
            await self._handle_quality_alert(
                question, answer, metrics
            )
 
        return metrics
 
    async def _handle_quality_alert(
        self,
        question: str,
        answer: str,
        metrics: QualityMetrics,
    ):
        """품질 저하 시 대응"""
        # 연속 알림이 임계값을 초과하면 에스컬레이션
        if self.alert_count >= 5:
            await escalate_to_team(
                "연속 품질 저하 감지",
                f"최근 5건의 응답이 품질 기준 미달. "
                f"마지막 점수: {metrics}",
            )

from enum import Enum
 
 
class AlertSeverity(Enum):
    CRITICAL = "critical"  # 즉시 대응 필요
    WARNING = "warning"    # 주의 관찰 필요
    INFO = "info"          # 참고 사항
 
 
class AlertChannel(Enum):
    PAGER = "pager"        # PagerDuty 등
    SLACK = "slack"
    EMAIL = "email"
    DASHBOARD = "dashboard"
 
 
# 알림 라우팅 규칙
ALERT_ROUTING = {
    AlertSeverity.CRITICAL: [
        AlertChannel.PAGER,
        AlertChannel.SLACK,
    ],
    AlertSeverity.WARNING: [
        AlertChannel.SLACK,
    ],
    AlertSeverity.INFO: [
        AlertChannel.DASHBOARD,
    ],
}
 
# AI 시스템 특화 알림 규칙
AI_ALERT_RULES = [
    {
        "name": "에러율 급증",
        "condition": "error_rate > 5%",
        "severity": AlertSeverity.CRITICAL,
        "cooldown_minutes": 15,
    },
    {
        "name": "비용 예산 초과",
        "condition": "daily_cost > budget * 0.9",
        "severity": AlertSeverity.WARNING,
        "cooldown_minutes": 60,
    },
    {
        "name": "지연시간 P99 증가",
        "condition": "p99_latency > 10000ms",
        "severity": AlertSeverity.WARNING,
        "cooldown_minutes": 30,
    },
    {
        "name": "품질 점수 하락",
        "condition": "avg_quality < 0.7 for 1h",
        "severity": AlertSeverity.CRITICAL,
        "cooldown_minutes": 30,
    },
    {
        "name": "드리프트 감지",
        "condition": "kl_divergence > 0.15",
        "severity": AlertSeverity.INFO,
        "cooldown_minutes": 360,
    },
]

from dataclasses import dataclass
from datetime import datetime
from enum import Enum
 
 
class FeedbackType(Enum):
    THUMBS_UP = "thumbs_up"
    THUMBS_DOWN = "thumbs_down"
    REPORT = "report"
    CORRECTION = "correction"
 
 
@dataclass
class UserFeedback:
    request_id: str
    feedback_type: FeedbackType
    comment: str | None = None
    correction: str | None = None
    timestamp: datetime = field(default_factory=datetime.now)
 
 
class FeedbackCollector:
    """사용자 피드백 수집 및 분석"""
 
    def __init__(self):
        self.feedbacks: list[UserFeedback] = []
 
    def record(self, feedback: UserFeedback):
        self.feedbacks.append(feedback)
 
    def satisfaction_rate(self, days: int = 7) -> float:
        """최근 N일간 만족도"""
        cutoff = datetime.now() - timedelta(days=days)
        recent = [
            f for f in self.feedbacks
            if f.timestamp > cutoff
        ]
        if not recent:
            return 0.0
 
        positive = sum(
            1 for f in recent
            if f.feedback_type == FeedbackType.THUMBS_UP
        )
        return positive / len(recent)
 
    def common_complaints(self, top_n: int = 10) -> list[dict]:
        """빈번한 불만 패턴 분석"""
        negative = [
            f for f in self.feedbacks
            if f.feedback_type in (
                FeedbackType.THUMBS_DOWN,
                FeedbackType.REPORT,
            )
            and f.comment
        ]
        # 실제로는 텍스트 클러스터링으로 패턴 추출
        return [
            {"comment": f.comment, "request_id": f.request_id}
            for f in negative[:top_n]
        ]