LLMOpsプラットフォーム構築ガイド：モデルデプロイ、モニタリング、A/Bテスト

ユーザーリクエスト
    |
    v
+-------------------------------------------+
|         Application Gateway               |
|  +----------+ +--------+ +-----------+    |
|  | プロンプト| | ベクトル| | ガード    |    |
|  | ルーター  | | DB     | | レール    |    |
|  |          | | (RAG)  | | エンジン  |    |
|  +----------+ +--------+ +-----------+    |
|  +----------+ +--------+ +-----------+    |
|  | キャッシュ| | A/B    | | トークン  |    |
|  | レイヤー  | | ルーター| | メータリング|   |
|  +----------+ +--------+ +-----------+    |
+-------------------+-----------------------+
                    |
    +---------------+---------------+
    v               v               v
+--------+    +--------+    +------------+
| 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)
+-- リクエストスループット   +-- 日/月別コスト推移    +-- LLM-as-Judgeスコア
|   (RPS)
+-- 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のt検定（等分散の仮定不要）
    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"]