LLMOps 플랫폼 구축 가이드: 모델 배포, 모니터링, A/B 테스트 실전 아키텍처

사용자 요청
    │
    ▼
┌─────────────────────────────────────────┐
│         Application Gateway             │
│  ┌──────────┐ ┌────────┐ ┌───────────┐ │
│  │ 프롬프트  │ │ 벡터DB │ │ 가드레일  │ │
│  │ 라우터   │ │ (RAG)  │ │ 엔진     │ │
│  └──────────┘ └────────┘ └───────────┘ │
│  ┌──────────┐ ┌────────┐ ┌───────────┐ │
│  │ 캐시     │ │ A/B    │ │ 토큰     │ │
│  │ 레이어   │ │ 라우터 │ │ 미터링   │ │
│  └──────────┘ └────────┘ └───────────┘ │
└─────────────────┬───────────────────────┘
                  │
    ┌─────────────┼─────────────┐
    ▼             ▼             ▼
┌────────┐  ┌────────┐  ┌────────────┐
│ vLLM   │  │  TGI   │  │ TensorRT-  │
│ Server │  │ Server │  │ LLM Server │
└────────┘  └────────┘  └────────────┘

apiVersion: apps/v1
kind: Deployment
metadata:
  name: vllm-llama3-70b
  namespace: llm-serving
spec:
  replicas: 2
  selector:
    matchLabels:
      app: vllm-llama3-70b
  template:
    metadata:
      labels:
        app: vllm-llama3-70b
    spec:
      containers:
        - name: vllm
          image: vllm/vllm-openai:v0.7.3
          args:
            - '--model'
            - 'meta-llama/Llama-3.3-70B-Instruct'
            - '--tensor-parallel-size'
            - '4'
            - '--max-model-len'
            - '8192'
            - '--gpu-memory-utilization'
            - '0.90'
            - '--enable-chunked-prefill'
            - '--max-num-batched-tokens'
            - '32768'
            - '--port'
            - '8000'
          ports:
            - containerPort: 8000
          resources:
            limits:
              nvidia.com/gpu: 4
            requests:
              nvidia.com/gpu: 4
              memory: '64Gi'
              cpu: '16'
          env:
            - name: HUGGING_FACE_HUB_TOKEN
              valueFrom:
                secretKeyRef:
                  name: hf-secret
                  key: token
          readinessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 120
            periodSeconds: 10
          livenessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 180
            periodSeconds: 30
---
apiVersion: v1
kind: Service
metadata:
  name: vllm-llama3-70b-svc
  namespace: llm-serving
spec:
  selector:
    app: vllm-llama3-70b
  ports:
    - port: 8000
      targetPort: 8000
  type: ClusterIP

성능 메트릭                  비용 메트릭              품질 메트릭
├── TTFT (Time To           ├── 입력 토큰 수         ├── 응답 관련성 점수
│   First Token)            ├── 출력 토큰 수         ├── 할루시네이션 비율
├── TPOT (Time Per          ├── 요청당 비용          ├── 가드레일 위반률
│   Output Token)           ├── 모델별 비용 비교     ├── 사용자 피드백
├── 총 생성 시간            ├── 캐시 히트율          │   (thumbs up/down)
├── 요청 처리량 (RPS)       └── 일/월별 비용 추세    └── LLM-as-Judge 점수
├── GPU 사용률
└── 큐 대기 시간

import time
import tiktoken
from prometheus_client import (
    Counter, Histogram, Gauge, start_http_server
)
from functools import wraps

# 성능 메트릭
REQUEST_LATENCY = Histogram(
    "llm_request_latency_seconds",
    "LLM 요청 레이턴시",
    ["model", "endpoint"],
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0, 60.0],
)
TTFT_LATENCY = Histogram(
    "llm_ttft_seconds",
    "Time To First Token",
    ["model"],
    buckets=[0.05, 0.1, 0.2, 0.5, 1.0, 2.0, 5.0],
)

# 비용 메트릭
TOKEN_COUNTER = Counter(
    "llm_tokens_total",
    "총 토큰 사용량",
    ["model", "direction"],  # direction: input/output
)
REQUEST_COST = Counter(
    "llm_request_cost_dollars",
    "요청별 비용 (USD)",
    ["model"],
)

# 품질 메트릭
QUALITY_SCORE = Histogram(
    "llm_quality_score",
    "LLM 응답 품질 점수",
    ["model", "evaluator"],
    buckets=[0.0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0],
)
GUARDRAIL_VIOLATIONS = Counter(
    "llm_guardrail_violations_total",
    "가드레일 위반 횟수",
    ["model", "violation_type"],
)

# 모델별 토큰 단가 (USD per 1K tokens)
PRICING = {
    "llama-3.3-70b": {"input": 0.00059, "output": 0.00079},
    "gpt-4o": {"input": 0.0025, "output": 0.01},
    "claude-sonnet": {"input": 0.003, "output": 0.015},
}


class LLMMetricsCollector:
    def __init__(self, model_name: str):
        self.model_name = model_name
        self.encoder = tiktoken.get_encoding("cl100k_base")

    def record_request(self, prompt: str, response: str,
                       latency: float, ttft: float):
        input_tokens = len(self.encoder.encode(prompt))
        output_tokens = len(self.encoder.encode(response))

        # 성능 기록
        REQUEST_LATENCY.labels(
            model=self.model_name, endpoint="/v1/chat/completions"
        ).observe(latency)
        TTFT_LATENCY.labels(model=self.model_name).observe(ttft)

        # 토큰 사용량 기록
        TOKEN_COUNTER.labels(
            model=self.model_name, direction="input"
        ).inc(input_tokens)
        TOKEN_COUNTER.labels(
            model=self.model_name, direction="output"
        ).inc(output_tokens)

        # 비용 계산 및 기록
        pricing = PRICING.get(self.model_name, PRICING["llama-3.3-70b"])
        cost = (
            input_tokens / 1000 * pricing["input"]
            + output_tokens / 1000 * pricing["output"]
        )
        REQUEST_COST.labels(model=self.model_name).inc(cost)

    def record_quality(self, score: float, evaluator: str = "auto"):
        QUALITY_SCORE.labels(
            model=self.model_name, evaluator=evaluator
        ).observe(score)

    def record_guardrail_violation(self, violation_type: str):
        GUARDRAIL_VIOLATIONS.labels(
            model=self.model_name, violation_type=violation_type
        ).inc()


if __name__ == "__main__":
    start_http_server(9090)
    collector = LLMMetricsCollector("llama-3.3-70b")

# P99 레이턴시 (5분 윈도우)
histogram_quantile(0.99, rate(llm_request_latency_seconds_bucket[5m]))

# 분당 토큰 소비량
rate(llm_tokens_total[1m])

# 시간당 비용 추세
rate(llm_request_cost_dollars[1h]) * 3600

# 가드레일 위반률
rate(llm_guardrail_violations_total[5m]) / rate(llm_request_latency_seconds_count[5m])

# 평균 품질 점수
histogram_quantile(0.5, rate(llm_quality_score_bucket[1h]))

import json
import hashlib
from datetime import datetime
from dataclasses import dataclass, field, asdict
from typing import Optional


@dataclass
class PromptVersion:
    name: str
    template: str
    model: str
    temperature: float = 0.7
    top_p: float = 0.9
    max_tokens: int = 2048
    system_prompt: str = ""
    version: str = ""
    created_at: str = field(
        default_factory=lambda: datetime.utcnow().isoformat()
    )

    def __post_init__(self):
        if not self.version:
            content = f"{self.template}{self.model}{self.temperature}"
            self.version = hashlib.sha256(
                content.encode()
            ).hexdigest()[:8]

    def to_dict(self) -> dict:
        return asdict(self)


class PromptRegistry:
    """프롬프트 버전 관리 레지스트리"""

    def __init__(self, storage_backend="redis"):
        self.storage_backend = storage_backend
        self.prompts: dict[str, list[PromptVersion]] = {}

    def register(self, prompt: PromptVersion) -> str:
        if prompt.name not in self.prompts:
            self.prompts[prompt.name] = []
        self.prompts[prompt.name].append(prompt)
        return prompt.version

    def get_latest(self, name: str) -> Optional[PromptVersion]:
        versions = self.prompts.get(name, [])
        return versions[-1] if versions else None

    def get_version(self, name: str, version: str
                    ) -> Optional[PromptVersion]:
        versions = self.prompts.get(name, [])
        for v in versions:
            if v.version == version:
                return v
        return None

    def rollback(self, name: str, version: str) -> bool:
        target = self.get_version(name, version)
        if target:
            self.prompts[name].append(
                PromptVersion(
                    name=target.name,
                    template=target.template,
                    model=target.model,
                    temperature=target.temperature,
                    top_p=target.top_p,
                    max_tokens=target.max_tokens,
                    system_prompt=target.system_prompt,
                )
            )
            return True
        return False


# 사용 예시
registry = PromptRegistry()
v1 = PromptVersion(
    name="customer-support",
    template="고객 문의에 대해 친절하게 답변해주세요.\n\n문의: {query}",
    model="llama-3.3-70b",
    temperature=0.3,
    system_prompt="당신은 전문 고객 상담원입니다.",
)
registry.register(v1)

import random
import hashlib
from dataclasses import dataclass
from typing import Any


@dataclass
class ABVariant:
    name: str
    prompt_version: str
    model: str
    weight: float  # 트래픽 비율 (0.0 ~ 1.0)
    parameters: dict = None


class LLMABRouter:
    """LLM A/B 테스트 트래픽 라우터"""

    def __init__(self, experiment_name: str):
        self.experiment_name = experiment_name
        self.variants: list[ABVariant] = []

    def add_variant(self, variant: ABVariant):
        self.variants.append(variant)

    def route(self, user_id: str) -> ABVariant:
        """사용자 ID 기반 결정적 라우팅 (동일 사용자는 동일 variant)"""
        hash_input = f"{self.experiment_name}:{user_id}"
        hash_value = int(
            hashlib.md5(hash_input.encode()).hexdigest(), 16
        )
        normalized = (hash_value % 10000) / 10000.0

        cumulative = 0.0
        for variant in self.variants:
            cumulative += variant.weight
            if normalized < cumulative:
                return variant
        return self.variants[-1]

    def validate_weights(self) -> bool:
        total = sum(v.weight for v in self.variants)
        return abs(total - 1.0) < 0.001


# 실험 설정 예시
experiment = LLMABRouter("customer-support-v2-test")
experiment.add_variant(ABVariant(
    name="control",
    prompt_version="v1-abc123",
    model="llama-3.3-70b",
    weight=0.7,
    parameters={"temperature": 0.3},
))
experiment.add_variant(ABVariant(
    name="treatment",
    prompt_version="v2-def456",
    model="llama-3.3-70b",
    weight=0.3,
    parameters={"temperature": 0.5},
))

# 사용자별 라우팅
variant = experiment.route(user_id="user-12345")
print(f"Assigned variant: {variant.name}")

import numpy as np
from scipy import stats


def calculate_ab_significance(
    control_scores: list[float],
    treatment_scores: list[float],
    alpha: float = 0.05,
    min_samples: int = 100,
) -> dict:
    """A/B 테스트 결과의 통계적 유의성 판정"""

    if (len(control_scores) < min_samples
            or len(treatment_scores) < min_samples):
        return {
            "status": "insufficient_samples",
            "control_n": len(control_scores),
            "treatment_n": len(treatment_scores),
            "min_required": min_samples,
        }

    control_mean = np.mean(control_scores)
    treatment_mean = np.mean(treatment_scores)
    lift = (treatment_mean - control_mean) / control_mean

    # Welch's t-test (등분산 가정 불필요)
    t_stat, p_value = stats.ttest_ind(
        control_scores, treatment_scores, equal_var=False
    )

    # 효과 크기 (Cohen's d)
    pooled_std = np.sqrt(
        (np.std(control_scores) ** 2 + np.std(treatment_scores) ** 2)
        / 2
    )
    cohens_d = (
        (treatment_mean - control_mean) / pooled_std
        if pooled_std > 0 else 0
    )

    return {
        "status": "significant" if p_value < alpha else "not_significant",
        "control_mean": round(control_mean, 4),
        "treatment_mean": round(treatment_mean, 4),
        "lift": round(lift * 100, 2),
        "p_value": round(p_value, 6),
        "cohens_d": round(cohens_d, 4),
        "recommendation": (
            "DEPLOY treatment"
            if p_value < alpha and lift > 0
            else "KEEP control"
        ),
    }

# config.yml - NeMo Guardrails 설정
models:
  - type: main
    engine: vllm
    parameters:
      base_url: 'http://vllm-llama3-70b-svc:8000/v1'
      model_name: 'meta-llama/Llama-3.3-70B-Instruct'

rails:
  input:
    flows:
      - self check input # 입력 유해성 체크
      - check jailbreak # 탈옥 시도 탐지
      - mask pii # PII 마스킹

  output:
    flows:
      - self check output # 출력 유해성 체크
      - check hallucination # 할루시네이션 탐지
      - check topic relevance # 주제 관련성 확인

  config:
    enable_multi_step_generation: true
    lowest_temperature: 0.1
    enable_rails_exceptions: true

instructions:
  - type: general
    content: |
      아래 가이드라인을 반드시 준수하세요:
      1. 확인되지 않은 사실은 추측이라고 명시
      2. 개인정보가 포함된 응답 금지
      3. 의료/법률/금융 조언은 전문가 상담 권고
      4. 폭력적이거나 유해한 콘텐츠 생성 금지

sample_conversation: |
  user "안녕하세요, 도움이 필요합니다."
    express greeting
  bot express greeting and offer help
    "안녕하세요! 무엇을 도와드릴까요?"

from nemoguardrails import RailsConfig, LLMRails


class GuardrailMiddleware:
    """LLM 가드레일 미들웨어"""

    def __init__(self, config_path: str):
        config = RailsConfig.from_path(config_path)
        self.rails = LLMRails(config)

    async def process(self, user_message: str,
                      context: dict = None) -> dict:
        try:
            response = await self.rails.generate_async(
                messages=[{"role": "user", "content": user_message}]
            )
            return {
                "status": "success",
                "response": response["content"],
                "guardrail_actions": response.get(
                    "log", {}
                ).get("activated_rails", []),
            }
        except Exception as e:
            return {
                "status": "blocked",
                "reason": str(e),
                "response": "요청을 처리할 수 없습니다. "
                            "다른 질문을 해주세요.",
            }

# 모델 라우팅 예시: 쿼리 복잡도에 따른 자동 라우팅
class ModelRouter:
    def __init__(self):
        self.complexity_threshold = 0.6
        self.models = {
            "simple": {
                "name": "llama-3.2-8b",
                "endpoint": "http://vllm-8b:8000/v1",
                "cost_per_1k": 0.00010,
            },
            "complex": {
                "name": "llama-3.3-70b",
                "endpoint": "http://vllm-70b:8000/v1",
                "cost_per_1k": 0.00079,
            },
        }

    def classify_complexity(self, query: str) -> float:
        """쿼리 복잡도를 0~1 사이 점수로 평가"""
        indicators = [
            len(query) > 500,          # 긴 질문
            "비교" in query,            # 비교 분석 요구
            "분석" in query,            # 분석 요구
            "코드" in query,            # 코드 생성
            query.count("?") > 2,       # 다중 질문
        ]
        return sum(indicators) / len(indicators)

    def route(self, query: str) -> dict:
        complexity = self.classify_complexity(query)
        if complexity >= self.complexity_threshold:
            return self.models["complex"]
        return self.models["simple"]