LLMOps 平台构建指南：模型部署、监控与 A/B 测试实战架构

用户请求
    │
    ▼
┌─────────────────────────────────────────┐
│         Application Gateway             │
│  ┌──────────┐ ┌────────┐ ┌───────────┐ │
│  │ 提示词    │ │ 向量DB │ │ 护栏      │ │
│  │ 路由器    │ │ (RAG)  │ │ 引擎      │ │
│  └──────────┘ └────────┘ └───────────┘ │
│  ┌──────────┐ ┌────────┐ ┌───────────┐ │
│  │ 缓存     │ │ A/B    │ │ token     │ │
│  │ 层       │ │ 路由器 │ │ 计量      │ │
│  └──────────┘ └────────┘ └───────────┘ │
└─────────────────┬───────────────────────┘
                  │
    ┌─────────────┼─────────────┐
    ▼             ▼             ▼
┌────────┐  ┌────────┐  ┌────────────┐
│ 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           ├── 输入 token 数        ├── 响应相关性得分
│   First Token)            ├── 输出 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",
    "总 token 使用量",
    ["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"],
)

# 各模型 token 单价 (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 使用量
        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]))

# 每分钟 token 消耗量
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"]