챗봇 성능 모니터링과 대화 품질 분석: 메트릭 설계부터 A/B 테스트 자동화까지

import time
import hashlib
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime
from collections import defaultdict
import json

@dataclass
class ConversationMetrics:
    """단일 대화 턴의 메트릭을 수집하는 데이터 클래스"""
    conversation_id: str
    turn_id: int
    timestamp: datetime = field(default_factory=datetime.utcnow)

    # 지연 시간 메트릭
    ttft_ms: float = 0.0            # Time to First Token
    e2e_latency_ms: float = 0.0     # End-to-End 지연 시간
    retrieval_latency_ms: float = 0.0  # RAG 검색 시간
    tool_latency_ms: float = 0.0    # 도구 호출 시간

    # 토큰 및 비용 메트릭
    input_tokens: int = 0
    output_tokens: int = 0
    total_cost_usd: float = 0.0
    model_name: str = ""
    cache_hit: bool = False

    # 품질 메트릭 (후처리로 채워짐)
    relevance_score: Optional[float] = None
    faithfulness_score: Optional[float] = None
    is_hallucination: Optional[bool] = None
    safety_passed: Optional[bool] = None

    # 사용자 피드백
    user_feedback: Optional[str] = None  # "thumbs_up", "thumbs_down", None
    regenerated: bool = False


class ChatbotMetricsCollector:
    """챗봇 메트릭 수집 및 집계 엔진"""

    # 모델별 토큰 단가 (USD per 1K tokens)
    PRICING = {
        "gpt-4o": {"input": 0.0025, "output": 0.01},
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
        "claude-sonnet-4-20250514": {"input": 0.003, "output": 0.015},
        "claude-haiku-4-20250414": {"input": 0.0008, "output": 0.004},
    }

    def __init__(self, storage_backend="postgres"):
        self.storage_backend = storage_backend
        self.metrics_buffer: list[ConversationMetrics] = []
        self.aggregated = defaultdict(list)

    def start_turn(self, conversation_id: str, turn_id: int, model: str) -> dict:
        """대화 턴 시작 시 타이머를 초기화한다"""
        return {
            "conversation_id": conversation_id,
            "turn_id": turn_id,
            "model": model,
            "start_time": time.monotonic(),
            "ttft_recorded": False,
            "retrieval_start": None,
        }

    def record_ttft(self, context: dict) -> float:
        """첫 번째 토큰 수신 시점을 기록한다"""
        ttft = (time.monotonic() - context["start_time"]) * 1000
        context["ttft_recorded"] = True
        return ttft

    def finalize_turn(
        self,
        context: dict,
        input_tokens: int,
        output_tokens: int,
        retrieval_ms: float = 0.0,
        tool_ms: float = 0.0,
    ) -> ConversationMetrics:
        """대화 턴 종료 시 전체 메트릭을 확정한다"""
        e2e = (time.monotonic() - context["start_time"]) * 1000
        model = context["model"]
        pricing = self.PRICING.get(model, {"input": 0.0, "output": 0.0})
        cost = (
            input_tokens / 1000 * pricing["input"]
            + output_tokens / 1000 * pricing["output"]
        )

        metrics = ConversationMetrics(
            conversation_id=context["conversation_id"],
            turn_id=context["turn_id"],
            e2e_latency_ms=e2e,
            retrieval_latency_ms=retrieval_ms,
            tool_latency_ms=tool_ms,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            total_cost_usd=cost,
            model_name=model,
        )

        self.metrics_buffer.append(metrics)
        return metrics

    def get_summary(self, window_hours: int = 24) -> dict:
        """지정 시간 내 메트릭 요약을 반환한다"""
        import statistics

        recent = [
            m for m in self.metrics_buffer
            if (datetime.utcnow() - m.timestamp).total_seconds() < window_hours * 3600
        ]
        if not recent:
            return {"error": "No data in window"}

        latencies = [m.e2e_latency_ms for m in recent]
        costs = [m.total_cost_usd for m in recent]
        feedbacks = [m.user_feedback for m in recent if m.user_feedback]

        return {
            "total_conversations": len(set(m.conversation_id for m in recent)),
            "total_turns": len(recent),
            "latency_p50_ms": statistics.median(latencies),
            "latency_p95_ms": sorted(latencies)[int(len(latencies) * 0.95)],
            "total_cost_usd": sum(costs),
            "avg_cost_per_turn": statistics.mean(costs),
            "thumbs_up_rate": feedbacks.count("thumbs_up") / max(len(feedbacks), 1),
            "hallucination_rate": sum(
                1 for m in recent if m.is_hallucination
            ) / max(len(recent), 1),
        }

from openai import OpenAI
from dataclasses import dataclass
from typing import Optional
import json

@dataclass
class QualityEvalResult:
    relevance: float        # 1-5: 질문에 대한 관련성
    faithfulness: float     # 1-5: 컨텍스트 기반 정확성
    helpfulness: float      # 1-5: 실질적 도움 정도
    safety: float           # 1-5: 안전성 (편향, 유해성)
    coherence: float        # 1-5: 논리적 일관성
    is_hallucination: bool  # 환각 여부
    reasoning: str          # 평가 근거
    overall_score: float    # 종합 점수

    @property
    def passed(self) -> bool:
        return self.overall_score >= 3.5 and self.safety >= 4.0


class LLMJudge:
    """LLM-as-a-Judge 기반 대화 품질 평가기"""

    EVAL_PROMPT = """당신은 챗봇 응답 품질을 평가하는 전문 평가자입니다.
아래의 대화 컨텍스트, 사용자 질문, 챗봇 응답, 그리고 참조 문서(있는 경우)를
기반으로 응답 품질을 평가하세요.

## 평가 기준 (각 1-5점)
1. **relevance**: 사용자 질문에 직접적으로 관련된 답변인가
2. **faithfulness**: 제공된 컨텍스트/문서에 기반한 정확한 정보인가
3. **helpfulness**: 사용자의 문제 해결에 실질적 도움이 되는가
4. **safety**: 편향, 유해 콘텐츠, 개인정보 노출이 없는가
5. **coherence**: 논리적으로 일관되고 자연스러운 응답인가

## 입력
- 대화 기록: {conversation_history}
- 사용자 질문: {user_query}
- 챗봇 응답: {bot_response}
- 참조 문서: {reference_context}

## 출력 형식 (JSON)
반드시 아래 JSON 형식으로만 응답하세요:
{{
  "relevance": <1-5>,
  "faithfulness": <1-5>,
  "helpfulness": <1-5>,
  "safety": <1-5>,
  "coherence": <1-5>,
  "is_hallucination": <true/false>,
  "reasoning": "<평가 근거를 2-3문장으로 작성>"
}}"""

    def __init__(self, model: str = "gpt-4o"):
        self.client = OpenAI()
        self.model = model

    def evaluate(
        self,
        user_query: str,
        bot_response: str,
        conversation_history: str = "",
        reference_context: str = "없음",
    ) -> QualityEvalResult:
        """단일 응답에 대한 품질 평가를 수행한다"""
        prompt = self.EVAL_PROMPT.format(
            conversation_history=conversation_history,
            user_query=user_query,
            bot_response=bot_response,
            reference_context=reference_context,
        )

        response = self.client.chat.completions.create(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0.0,
            response_format={"type": "json_object"},
        )

        result = json.loads(response.choices[0].message.content)
        scores = [
            result["relevance"],
            result["faithfulness"],
            result["helpfulness"],
            result["safety"],
            result["coherence"],
        ]

        return QualityEvalResult(
            relevance=result["relevance"],
            faithfulness=result["faithfulness"],
            helpfulness=result["helpfulness"],
            safety=result["safety"],
            coherence=result["coherence"],
            is_hallucination=result["is_hallucination"],
            reasoning=result["reasoning"],
            overall_score=sum(scores) / len(scores),
        )

    def evaluate_batch(
        self, conversations: list[dict], concurrency: int = 5
    ) -> list[QualityEvalResult]:
        """배치 평가를 수행한다. 프로덕션 트래픽의 샘플링 평가에 사용"""
        import asyncio
        from concurrent.futures import ThreadPoolExecutor

        results = []
        with ThreadPoolExecutor(max_workers=concurrency) as executor:
            futures = [
                executor.submit(
                    self.evaluate,
                    conv["user_query"],
                    conv["bot_response"],
                    conv.get("history", ""),
                    conv.get("context", "없음"),
                )
                for conv in conversations
            ]
            results = [f.result() for f in futures]

        return results

import os
from langsmith import traceable, Client
from langsmith.run_helpers import get_current_run_tree
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# 환경 변수 설정
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls-your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "chatbot-production"

client = Client()
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)


@traceable(
    name="chatbot_response",
    metadata={"version": "v2.3", "team": "ml-platform"},
    tags=["production", "customer-support"],
)
def generate_response(
    user_message: str,
    conversation_history: list[dict],
    user_id: str,
    session_id: str,
) -> dict:
    """프로덕션 챗봇 응답 생성 - LangSmith 자동 트레이싱"""

    # RAG 검색 단계 (자식 스팬으로 자동 기록)
    context = retrieve_context(user_message)

    # LLM 호출
    messages = [
        SystemMessage(content=f"당신은 고객 지원 챗봇입니다.\n\n참고 문서:\n{context}"),
    ]
    for turn in conversation_history[-5:]:  # 최근 5턴만 포함
        messages.append(HumanMessage(content=turn["user"]))
        if "assistant" in turn:
            from langchain_core.messages import AIMessage
            messages.append(AIMessage(content=turn["assistant"]))
    messages.append(HumanMessage(content=user_message))

    response = llm.invoke(messages)

    # 현재 실행 트리에 커스텀 메타데이터 추가
    run_tree = get_current_run_tree()
    if run_tree:
        run_tree.metadata.update({
            "user_id": user_id,
            "session_id": session_id,
            "context_doc_count": len(context.split("\n\n")),
            "history_turns": len(conversation_history),
        })

    return {
        "response": response.content,
        "context_used": context,
        "model": "gpt-4o",
        "tokens": {
            "input": response.usage_metadata.get("input_tokens", 0),
            "output": response.usage_metadata.get("output_tokens", 0),
        },
    }


@traceable(name="retrieve_context", tags=["rag", "retrieval"])
def retrieve_context(query: str) -> str:
    """벡터 DB에서 관련 문서를 검색한다"""
    # 실제 구현에서는 벡터 DB 호출
    return "검색된 문서 내용..."


# LangSmith 데이터셋 기반 오프라인 평가 실행
def run_offline_evaluation():
    """커레이션된 데이터셋으로 오프라인 평가를 수행한다"""

    dataset = client.create_dataset(
        "chatbot-eval-v2",
        description="고객 지원 챗봇 평가 데이터셋",
    )

    # 평가 데이터 추가
    examples = [
        {
            "inputs": {"user_message": "반품 절차가 어떻게 되나요?"},
            "outputs": {"expected": "반품은 구매일로부터 14일 이내..."},
        },
        {
            "inputs": {"user_message": "배송 현황을 확인하고 싶어요"},
            "outputs": {"expected": "주문번호를 알려주시면 배송 현황을..."},
        },
    ]

    for ex in examples:
        client.create_example(
            inputs=ex["inputs"],
            outputs=ex["outputs"],
            dataset_id=dataset.id,
        )

    # 평가 실행
    from langsmith.evaluation import evaluate

    results = evaluate(
        lambda inputs: generate_response(
            inputs["user_message"], [], "eval-user", "eval-session"
        ),
        data="chatbot-eval-v2",
        evaluators=[relevance_evaluator, faithfulness_evaluator],
        experiment_prefix="chatbot-v2.3",
        max_concurrency=4,
    )

    return results

┌─────────────────────────────────────────────────────────────────────┐
│                        사용자 요청 (User Request)                    │
└─────────────────────────────┬───────────────────────────────────────┘
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────────┐
│                     API Gateway / Load Balancer                      │
│              ┌──────────────────────────────────┐                    │
│              │  A/B 트래픽 라우터 (Hash-based)   │                    │
│              │  variant_a: 60%  variant_b: 40%   │                    │
│              └──────────────┬───────────────────┘                    │
└─────────────────────────────┬───────────────────────────────────────┘
                              │
                    ┌─────────┴──────────┐
                    ▼                    ▼
           ┌──────────────┐     ┌──────────────┐
           │  Variant A   │     │  Variant B   │
           │  (Control)   │     │  (Treatment) │
           │  GPT-4o      │     │  Claude 4    │
           │  Prompt v2.3 │     │  Prompt v2.4 │
           └──────┬───────┘     └──────┬───────┘
                  │                    │
                  └────────┬───────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────────────┐
│                    트레이싱 / 로깅 레이어                            │
│  ┌─────────────┐  ┌──────────────┐  ┌───────────────┐              │
│  │  LangSmith   │  │   Langfuse   │  │   Helicone    │              │
│  │  (Tracing)   │  │  (Tracing +  │  │   (Proxy +    │              │
│  │              │  │   Evals)     │  │    Caching)   │              │
│  └──────┬──────┘  └──────┬───────┘  └───────┬───────┘              │
└─────────┼────────────────┼──────────────────┼──────────────────────┘
          │                │                  │
          └────────────────┼──────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────────────────┐
│                  데이터 파이프라인 (Async Processing)                 │
│  ┌──────────────────────────────────────────────────────────┐       │
│  │ Kafka / SQS Queue                                        │       │
│  │  - trace_events                                          │       │
│  │  - user_feedback_events                                  │       │
│  │  - quality_eval_requests                                 │       │
│  └──────────────────────────┬───────────────────────────────┘       │
│                             │                                       │
│              ┌──────────────┴──────────────┐                        │
│              ▼                             ▼                        │
│  ┌────────────────────┐       ┌────────────────────┐                │
│  │  LLM-as-Judge      │       │  비용 계산기        │                │
│  │  (Sampled 10%)     │       │  (모든 요청)        │                │
│  │  품질 점수 산출     │       │  토큰 비용 집계     │                │
│  └─────────┬──────────┘       └─────────┬──────────┘                │
└────────────┼────────────────────────────┼──────────────────────────┘
             │                            │
             └──────────┬─────────────────┘
                        ▼
┌─────────────────────────────────────────────────────────────────────┐
│                    분석 데이터 저장소                                 │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐              │
│  │  PostgreSQL   │  │  ClickHouse  │  │  Prometheus  │              │
│  │  (메타데이터)  │  │  (분석쿼리)   │  │  (시계열)    │              │
│  └──────────────┘  └──────────────┘  └──────────────┘              │
│                                                                     │
│                        ┌──────────────┐                             │
│                        │   Grafana    │                             │
│                        │  Dashboard   │                             │
│                        └──────────────┘                             │
└─────────────────────────────────────────────────────────────────────┘

import Langfuse from 'langfuse'
import { observeOpenAI } from 'langfuse'
import OpenAI from 'openai'

// Langfuse 클라이언트 초기화
const langfuse = new Langfuse({
  publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
  secretKey: process.env.LANGFUSE_SECRET_KEY!,
  baseUrl: process.env.LANGFUSE_BASE_URL || 'https://cloud.langfuse.com',
})

// OpenAI 클라이언트를 Langfuse로 래핑
const openai = observeOpenAI(new OpenAI(), {
  clientInitParams: {
    publicKey: process.env.LANGFUSE_PUBLIC_KEY!,
    secretKey: process.env.LANGFUSE_SECRET_KEY!,
  },
})

interface ChatbotConfig {
  modelName: string
  promptVersion: string
  maxTokens: number
  temperature: number
}

interface MonitoredResponse {
  content: string
  traceId: string
  latencyMs: number
  tokenUsage: { input: number; output: number }
  cost: number
}

async function handleChatRequest(
  userId: string,
  sessionId: string,
  message: string,
  history: Array<{ role: string; content: string }>,
  config: ChatbotConfig
): Promise<MonitoredResponse> {
  const startTime = Date.now()

  // Langfuse 트레이스 생성
  const trace = langfuse.trace({
    name: 'chatbot-response',
    userId: userId,
    sessionId: sessionId,
    metadata: {
      promptVersion: config.promptVersion,
      modelName: config.modelName,
      historyLength: history.length,
    },
    tags: ['production', 'customer-support'],
  })

  // RAG 검색 스팬
  const retrievalSpan = trace.span({
    name: 'rag-retrieval',
    input: { query: message },
  })

  const context = await retrieveDocuments(message)
  retrievalSpan.end({
    output: { documentCount: context.length },
    metadata: { source: 'pinecone' },
  })

  // LLM 생성 스팬
  const generationSpan = trace.generation({
    name: 'llm-generation',
    model: config.modelName,
    modelParameters: {
      temperature: config.temperature,
      maxTokens: config.maxTokens,
    },
    input: [
      {
        role: 'system',
        content: `고객 지원 챗봇입니다.\n\n참고 문서:\n${context.join('\n')}`,
      },
      ...history.slice(-5),
      { role: 'user', content: message },
    ],
  })

  const completion = await openai.chat.completions.create({
    model: config.modelName,
    messages: [
      {
        role: 'system',
        content: `고객 지원 챗봇입니다.\n\n참고 문서:\n${context.join('\n')}`,
      },
      ...history.slice(-5).map((h) => ({
        role: h.role as 'user' | 'assistant',
        content: h.content,
      })),
      { role: 'user' as const, content: message },
    ],
    temperature: config.temperature,
    max_tokens: config.maxTokens,
  })

  const responseContent = completion.choices[0].message.content || ''
  const usage = completion.usage

  generationSpan.end({
    output: responseContent,
    usage: {
      input: usage?.prompt_tokens || 0,
      output: usage?.completion_tokens || 0,
      total: usage?.total_tokens || 0,
    },
  })

  const latencyMs = Date.now() - startTime

  // 비동기로 LLM-as-Judge 평가 요청 (샘플링)
  if (Math.random() < 0.1) {
    trace.score({
      name: 'auto-eval-queued',
      value: 1,
      comment: 'Queued for LLM-as-Judge evaluation',
    })
    // 별도의 큐에 평가 작업 추가
    await queueEvaluation(trace.id, message, responseContent, context)
  }

  // Langfuse 버퍼 비동기 전송
  await langfuse.flushAsync()

  return {
    content: responseContent,
    traceId: trace.id,
    latencyMs,
    tokenUsage: {
      input: usage?.prompt_tokens || 0,
      output: usage?.completion_tokens || 0,
    },
    cost: calculateCost(config.modelName, usage?.prompt_tokens || 0, usage?.completion_tokens || 0),
  }
}

async function retrieveDocuments(query: string): Promise<string[]> {
  // 벡터 DB 검색 구현
  return ['문서 1 내용...', '문서 2 내용...']
}

function calculateCost(model: string, inputTokens: number, outputTokens: number): number {
  const pricing: Record<string, { input: number; output: number }> = {
    'gpt-4o': { input: 0.0025, output: 0.01 },
    'gpt-4o-mini': { input: 0.00015, output: 0.0006 },
  }
  const p = pricing[model] || { input: 0, output: 0 }
  return (inputTokens / 1000) * p.input + (outputTokens / 1000) * p.output
}

async function queueEvaluation(
  traceId: string,
  query: string,
  response: string,
  context: string[]
): Promise<void> {
  // SQS/Kafka 등의 큐에 평가 작업 발행
  console.log(`Evaluation queued for trace: ${traceId}`)
}

import asyncio
from enum import Enum
from typing import Callable
import re


class EvalTier(Enum):
    RULE_BASED = "rule_based"       # 비용 0, 즉시
    LLM_JUDGE = "llm_judge"        # 비용 ~$0.02, 비동기
    HUMAN_REVIEW = "human_review"  # 비용 ~$2.00, 큐


class AutomatedEvalPipeline:
    """프로덕션 자동 품질 평가 파이프라인"""

    BLOCKED_PATTERNS = [
        r"(?i)(죽|살|폭탄|해킹|마약)",
        r"(?i)(비밀번호|주민등록|신용카드)\s*[:은는이가]?\s*\d",
    ]

    def __init__(self, judge: "LLMJudge", sample_rate: float = 0.1):
        self.judge = judge
        self.sample_rate = sample_rate
        self.rule_checks: list[Callable] = [
            self._check_response_length,
            self._check_blocked_content,
            self._check_format_compliance,
            self._check_language_consistency,
        ]

    def _check_response_length(self, response: str) -> tuple[bool, str]:
        """응답 길이가 적절한 범위인지 확인"""
        if len(response) < 10:
            return False, "응답이 너무 짧음 (10자 미만)"
        if len(response) > 5000:
            return False, "응답이 너무 김 (5000자 초과)"
        return True, "OK"

    def _check_blocked_content(self, response: str) -> tuple[bool, str]:
        """금지된 콘텐츠 패턴 검사"""
        for pattern in self.BLOCKED_PATTERNS:
            if re.search(pattern, response):
                return False, f"금지 패턴 감지: {pattern}"
        return True, "OK"

    def _check_format_compliance(self, response: str) -> tuple[bool, str]:
        """형식 규정 준수 확인"""
        # 예: 마크다운 코드 블록이 열리면 닫혀야 함
        if response.count("```") % 2 != 0:
            return False, "마크다운 코드 블록이 닫히지 않음"
        return True, "OK"

    def _check_language_consistency(self, response: str) -> tuple[bool, str]:
        """한국어 서비스에서 응답 언어 일관성 확인"""
        korean_ratio = len(re.findall(r"[가-힣]", response)) / max(len(response), 1)
        if korean_ratio < 0.1 and len(response) > 50:
            return False, f"한국어 비율 낮음: {korean_ratio:.1%}"
        return True, "OK"

    async def evaluate(
        self, user_query: str, bot_response: str, context: str = ""
    ) -> dict:
        """3단계 평가 파이프라인을 실행한다"""
        result = {
            "tier": EvalTier.RULE_BASED.value,
            "passed": True,
            "details": [],
            "scores": None,
            "needs_human_review": False,
        }

        # 1단계: 규칙 기반 필터 (동기, 즉시)
        for check_fn in self.rule_checks:
            passed, msg = check_fn(bot_response)
            result["details"].append({"check": check_fn.__name__, "passed": passed, "msg": msg})
            if not passed:
                result["passed"] = False
                result["needs_human_review"] = True
                return result

        # 2단계: LLM-as-Judge (비동기, 샘플링)
        import random
        if random.random() < self.sample_rate:
            result["tier"] = EvalTier.LLM_JUDGE.value
            eval_result = self.judge.evaluate(
                user_query=user_query,
                bot_response=bot_response,
                reference_context=context,
            )
            result["scores"] = {
                "relevance": eval_result.relevance,
                "faithfulness": eval_result.faithfulness,
                "helpfulness": eval_result.helpfulness,
                "safety": eval_result.safety,
                "coherence": eval_result.coherence,
                "overall": eval_result.overall_score,
                "is_hallucination": eval_result.is_hallucination,
            }
            result["passed"] = eval_result.passed

            # 3단계: 낮은 점수 -> 사람 리뷰 큐
            if eval_result.overall_score < 3.0 or eval_result.safety < 4.0:
                result["needs_human_review"] = True
                result["tier"] = EvalTier.HUMAN_REVIEW.value

        return result

import hashlib
import json
from dataclasses import dataclass
from typing import Optional
from datetime import datetime, timedelta
from scipy import stats
import numpy as np


@dataclass
class ExperimentVariant:
    """A/B 테스트 변형 정의"""
    name: str
    model: str
    prompt_version: str
    temperature: float
    max_tokens: int
    top_k: int = 5          # RAG 검색 문서 수
    weight: float = 0.5     # 트래픽 비중


@dataclass
class ExperimentConfig:
    """A/B 테스트 실험 구성"""
    experiment_id: str
    name: str
    description: str
    variants: list[ExperimentVariant]
    start_date: datetime
    end_date: Optional[datetime] = None
    min_sample_size: int = 1000
    confidence_level: float = 0.95
    primary_metric: str = "overall_quality_score"
    guardrail_metrics: list[str] = None

    def __post_init__(self):
        if self.guardrail_metrics is None:
            self.guardrail_metrics = ["safety_score", "hallucination_rate", "p95_latency_ms"]


class ChatbotABTestFramework:
    """챗봇 전용 A/B 테스트 프레임워크"""

    def __init__(self, experiment: ExperimentConfig):
        self.experiment = experiment
        self.results: dict[str, list[dict]] = {
            v.name: [] for v in experiment.variants
        }

    def assign_variant(self, user_id: str) -> ExperimentVariant:
        """해시 기반으로 사용자를 변형에 할당한다.
        동일한 user_id는 항상 같은 변형을 받는다."""
        hash_input = f"{self.experiment.experiment_id}:{user_id}"
        hash_value = int(hashlib.sha256(hash_input.encode()).hexdigest(), 16)
        normalized = (hash_value % 10000) / 10000.0

        cumulative = 0.0
        for variant in self.experiment.variants:
            cumulative += variant.weight
            if normalized < cumulative:
                return variant

        return self.experiment.variants[-1]

    def record_result(
        self,
        variant_name: str,
        metrics: dict,
    ):
        """실험 결과를 기록한다"""
        self.results[variant_name].append({
            "timestamp": datetime.utcnow().isoformat(),
            **metrics,
        })

    def analyze(self) -> dict:
        """통계적 유의성 분석을 수행한다"""
        if len(self.experiment.variants) != 2:
            raise ValueError("현재 2개 변형만 지원합니다")

        v_a = self.experiment.variants[0].name
        v_b = self.experiment.variants[1].name
        metric = self.experiment.primary_metric

        scores_a = [r[metric] for r in self.results[v_a] if metric in r]
        scores_b = [r[metric] for r in self.results[v_b] if metric in r]

        if len(scores_a) < 30 or len(scores_b) < 30:
            return {
                "status": "insufficient_data",
                "sample_sizes": {v_a: len(scores_a), v_b: len(scores_b)},
                "min_required": self.experiment.min_sample_size,
            }

        # Welch's t-test (등분산 가정 안 함)
        t_stat, p_value = stats.ttest_ind(scores_a, scores_b, equal_var=False)

        # 효과 크기 (Cohen's d)
        pooled_std = np.sqrt(
            (np.std(scores_a) ** 2 + np.std(scores_b) ** 2) / 2
        )
        cohens_d = (np.mean(scores_b) - np.mean(scores_a)) / max(pooled_std, 1e-10)

        # 가드레일 메트릭 확인
        guardrail_passed = True
        guardrail_details = {}
        for gm in self.experiment.guardrail_metrics:
            gm_a = [r.get(gm, 0) for r in self.results[v_a] if gm in r]
            gm_b = [r.get(gm, 0) for r in self.results[v_b] if gm in r]
            if gm_a and gm_b:
                if gm == "hallucination_rate":
                    # 환각률은 증가하면 안 됨
                    if np.mean(gm_b) > np.mean(gm_a) * 1.1:
                        guardrail_passed = False
                guardrail_details[gm] = {
                    "control_mean": float(np.mean(gm_a)),
                    "treatment_mean": float(np.mean(gm_b)),
                }

        is_significant = p_value < (1 - self.experiment.confidence_level)
        winner = None
        if is_significant and guardrail_passed:
            winner = v_b if np.mean(scores_b) > np.mean(scores_a) else v_a

        return {
            "status": "complete",
            "experiment_id": self.experiment.experiment_id,
            "primary_metric": metric,
            "control": {
                "name": v_a,
                "n": len(scores_a),
                "mean": float(np.mean(scores_a)),
                "std": float(np.std(scores_a)),
                "ci_95": (
                    float(np.mean(scores_a) - 1.96 * np.std(scores_a) / np.sqrt(len(scores_a))),
                    float(np.mean(scores_a) + 1.96 * np.std(scores_a) / np.sqrt(len(scores_a))),
                ),
            },
            "treatment": {
                "name": v_b,
                "n": len(scores_b),
                "mean": float(np.mean(scores_b)),
                "std": float(np.std(scores_b)),
                "ci_95": (
                    float(np.mean(scores_b) - 1.96 * np.std(scores_b) / np.sqrt(len(scores_b))),
                    float(np.mean(scores_b) + 1.96 * np.std(scores_b) / np.sqrt(len(scores_b))),
                ),
            },
            "t_statistic": float(t_stat),
            "p_value": float(p_value),
            "cohens_d": float(cohens_d),
            "is_significant": is_significant,
            "guardrail_passed": guardrail_passed,
            "guardrail_details": guardrail_details,
            "winner": winner,
            "recommendation": _get_recommendation(is_significant, cohens_d, guardrail_passed, winner),
        }


def _get_recommendation(significant: bool, effect_size: float, guardrail: bool, winner: str) -> str:
    if not guardrail:
        return "가드레일 메트릭이 실패했습니다. Treatment를 롤백하세요."
    if not significant:
        return "통계적으로 유의미한 차이가 없습니다. 샘플을 더 수집하세요."
    if abs(effect_size) < 0.2:
        return f"통계적으로 유의미하나 효과 크기가 작습니다 (d={effect_size:.3f}). 실용적 가치를 검토하세요."
    return f"{winner}가 승자입니다 (Cohen's d={effect_size:.3f}). 전체 트래픽에 적용하세요."


# 실험 실행 예시
experiment = ExperimentConfig(
    experiment_id="exp-2026-03-prompt-v24",
    name="프롬프트 v2.4 vs v2.3",
    description="시스템 프롬프트에 COT 추론 단계 추가 효과 측정",
    variants=[
        ExperimentVariant(
            name="control",
            model="gpt-4o",
            prompt_version="v2.3",
            temperature=0.7,
            max_tokens=1024,
            weight=0.5,
        ),
        ExperimentVariant(
            name="treatment",
            model="gpt-4o",
            prompt_version="v2.4",
            temperature=0.7,
            max_tokens=1024,
            weight=0.5,
        ),
    ],
    start_date=datetime(2026, 3, 8),
    min_sample_size=1000,
)

ab_test = ChatbotABTestFramework(experiment)

-- 1. 시간대별 품질 점수 추이와 이상 탐지
SELECT
    toStartOfHour(timestamp) AS hour,
    count() AS total_turns,
    avg(overall_quality_score) AS avg_quality,
    quantile(0.95)(e2e_latency_ms) AS p95_latency,
    countIf(is_hallucination = 1) / count() * 100 AS hallucination_pct,
    countIf(user_feedback = 'thumbs_down') / countIf(user_feedback != '') * 100 AS negative_feedback_pct,
    sum(total_cost_usd) AS hourly_cost,
    -- 이상 탐지: 3시간 이동 평균 대비 2 표준편차 이상 이탈
    avg(overall_quality_score) OVER (
        ORDER BY toStartOfHour(timestamp)
        ROWS BETWEEN 3 PRECEDING AND 1 PRECEDING
    ) AS moving_avg_quality,
    if(
        abs(avg(overall_quality_score) - moving_avg_quality) > 2 * stddevPop(overall_quality_score) OVER (
            ORDER BY toStartOfHour(timestamp)
            ROWS BETWEEN 12 PRECEDING AND 1 PRECEDING
        ),
        'ANOMALY',
        'NORMAL'
    ) AS quality_status
FROM chatbot_metrics
WHERE timestamp >= now() - INTERVAL 24 HOUR
GROUP BY hour
ORDER BY hour;

-- 2. A/B 테스트 실험별 성과 비교
SELECT
    experiment_id,
    variant_name,
    count() AS sample_size,
    avg(overall_quality_score) AS avg_quality,
    avg(relevance_score) AS avg_relevance,
    avg(faithfulness_score) AS avg_faithfulness,
    quantile(0.5)(e2e_latency_ms) AS median_latency,
    quantile(0.95)(e2e_latency_ms) AS p95_latency,
    countIf(is_hallucination = 1) / count() * 100 AS hallucination_rate,
    avg(total_cost_usd) AS avg_cost_per_turn,
    countIf(user_feedback = 'thumbs_up') / countIf(user_feedback != '') * 100 AS positive_rate
FROM chatbot_metrics
WHERE experiment_id = 'exp-2026-03-prompt-v24'
  AND timestamp >= '2026-03-08'
GROUP BY experiment_id, variant_name;

-- 3. 모델별/프롬프트 버전별 비용 효율성 분석
SELECT
    model_name,
    prompt_version,
    count() AS total_requests,
    sum(input_tokens) AS total_input_tokens,
    sum(output_tokens) AS total_output_tokens,
    sum(total_cost_usd) AS total_cost,
    avg(total_cost_usd) AS avg_cost_per_request,
    avg(overall_quality_score) AS avg_quality,
    -- 비용 대비 품질 효율성 점수
    avg(overall_quality_score) / (avg(total_cost_usd) * 1000 + 0.001) AS quality_per_dollar,
    -- 캐시 효율성
    countIf(cache_hit = 1) / count() * 100 AS cache_hit_rate
FROM chatbot_metrics
WHERE timestamp >= now() - INTERVAL 7 DAY
GROUP BY model_name, prompt_version
ORDER BY quality_per_dollar DESC;