A/B测试与上线

1. LLM应用的A/B测试

1.1 A/B测试的特殊性

LLM应用的A/B测试与传统Web应用不同：

维度	传统A/B测试	LLM A/B测试
指标	点击率、转化率	回答质量、用户满意度
评估	二元/连续指标	多维度主观评估
样本量	通常需要大量	样本效率更重要
成本	低	高（每次请求都有API成本）
延迟	毫秒级差异	秒级差异可能影响体验

1.2 测试指标设计

核心指标:

class ABTestMetrics:
    # 质量指标
    user_satisfaction: float     # 用户满意度评分（1-5）
    thumbs_up_rate: float       # 点赞率
    task_completion_rate: float  # 任务完成率

    # 安全指标
    hallucination_rate: float   # 幻觉率
    safety_violation_rate: float # 安全违规率

    # 性能指标
    ttft_ms: float              # 首Token延迟
    total_latency_ms: float     # 总延迟

    # 成本指标
    tokens_per_query: float     # 每次查询的Token数
    cost_per_query: float       # 每次查询的成本

    # 业务指标
    retention_rate: float       # 用户留存率
    queries_per_session: float  # 每次会话的查询数

护栏指标（Guardrail Metrics）:

护栏指标不追求改善，但不能恶化：
- 安全违规率 ≤ 0.1%
- 幻觉率 ≤ 5%
- P95延迟 ≤ 3000ms
- 错误率 ≤ 1%

1.3 统计显著性

from scipy import stats

def ab_test_significance(control_scores, treatment_scores, alpha=0.05):
    """检验A/B测试的统计显著性"""
    # Welch's t-test（不假设等方差）
    t_stat, p_value = stats.ttest_ind(
        control_scores, treatment_scores, equal_var=False
    )

    # 效果量（Cohen's d）
    pooled_std = ((control_scores.std()**2 + treatment_scores.std()**2) / 2)**0.5
    cohens_d = (treatment_scores.mean() - control_scores.mean()) / pooled_std

    return {
        "control_mean": control_scores.mean(),
        "treatment_mean": treatment_scores.mean(),
        "p_value": p_value,
        "significant": p_value < alpha,
        "cohens_d": cohens_d,
        "lift": (treatment_scores.mean() - control_scores.mean()) / control_scores.mean(),
    }

1.4 实验分流

import hashlib

class ABTestRouter:
    def __init__(self, experiment_id: str, treatment_ratio: float = 0.5):
        self.experiment_id = experiment_id
        self.treatment_ratio = treatment_ratio

    def get_variant(self, user_id: str) -> str:
        """根据用户ID确定分组（确保同一用户始终在同一组）"""
        hash_input = f"{self.experiment_id}:{user_id}"
        hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)

        if (hash_value % 100) / 100 < self.treatment_ratio:
            return "treatment"
        return "control"

    def route_request(self, user_id: str, query: str):
        variant = self.get_variant(user_id)

        if variant == "treatment":
            return self.treatment_handler(query)  # 新Prompt/模型
        else:
            return self.control_handler(query)     # 当前版本

2. 金丝雀发布 (Canary Deployment)

2.1 概念

金丝雀发布是将新版本逐步推送给小比例用户，在确认安全后再逐步扩大的部署策略。

2.2 渐进式发布

阶段1: 1%流量 → 新版本（内部用户）
  观察24小时，检查核心指标

阶段2: 5%流量 → 新版本
  观察24小时，对比A/B指标

阶段3: 25%流量 → 新版本
  观察48小时，确认无退化

阶段4: 50%流量 → 新版本
  观察24小时

阶段5: 100%流量 → 新版本
  全量发布完成

2.3 自动回滚条件

class CanaryMonitor:
    def __init__(self):
        self.rollback_conditions = {
            "error_rate": {"threshold": 0.05, "window": "5m"},
            "p95_latency_ms": {"threshold": 5000, "window": "10m"},
            "hallucination_rate": {"threshold": 0.10, "window": "30m"},
            "safety_violation": {"threshold": 0.001, "window": "5m"},
        }

    def check_health(self, metrics):
        """检查金丝雀健康状态，决定是否回滚"""
        for metric_name, condition in self.rollback_conditions.items():
            current_value = metrics.get(metric_name)
            if current_value > condition["threshold"]:
                return {
                    "action": "rollback",
                    "reason": f"{metric_name} ({current_value}) exceeds threshold ({condition['threshold']})",
                }
        return {"action": "continue"}

3. 蓝绿部署 (Blue-Green Deployment)

3.1 概念

同时维护两套完整的生产环境，通过流量切换实现零停机部署。

蓝环境（当前版本）  ←── 100%流量
绿环境（新版本）    ←── 0%流量

切换后：
蓝环境（旧版本）    ←── 0%流量（保留用于回滚）
绿环境（新版本）    ←── 100%流量

3.2 Kubernetes实现

# 蓝色部署
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-service-blue
  labels:
    version: blue
spec:
  replicas: 3
  template:
    spec:
      containers:
        - name: llm-inference
          image: llm-service:v1.2
          env:
            - name: PROMPT_VERSION
              value: "v1.3"

---
# 绿色部署
apiVersion: apps/v1
kind: Deployment
metadata:
  name: llm-service-green
  labels:
    version: green
spec:
  replicas: 3
  template:
    spec:
      containers:
        - name: llm-inference
          image: llm-service:v1.3
          env:
            - name: PROMPT_VERSION
              value: "v2.0"

---
# Service（通过selector切换流量）
apiVersion: v1
kind: Service
metadata:
  name: llm-service
spec:
  selector:
    version: blue  # 切换为green即可切换流量
  ports:
    - port: 80
      targetPort: 8000

4. Feature Flags

4.1 使用Feature Flags控制LLM行为

from feature_flags import FeatureFlagClient

ff = FeatureFlagClient()

async def handle_chat(user_id: str, query: str):
    # 模型选择
    if ff.is_enabled("use_gpt4_turbo", user_id=user_id):
        model = "gpt-4-turbo"
    else:
        model = "gpt-4"

    # Prompt版本
    if ff.is_enabled("new_prompt_v2", user_id=user_id):
        prompt = load_prompt("v2.0")
    else:
        prompt = load_prompt("v1.3")

    # RAG配置
    if ff.is_enabled("enable_reranker", user_id=user_id):
        rag_config = {"reranker": True, "top_k": 10}
    else:
        rag_config = {"reranker": False, "top_k": 5}

    return await generate(model, prompt, query, rag_config)

4.2 渐进式功能开放

Feature Flag: "new_rag_pipeline"
- 阶段1: 仅限内部测试用户
- 阶段2: 5%随机用户
- 阶段3: 付费用户
- 阶段4: 所有用户
- 阶段5: 移除Flag，成为默认行为

5. 回滚策略

5.1 快速回滚

class RollbackManager:
    def __init__(self):
        self.version_history = []

    def deploy(self, new_version):
        """部署新版本，保存当前版本用于回滚"""
        current = self.get_current_version()
        self.version_history.append(current)
        self.switch_to(new_version)

    def rollback(self, reason: str):
        """回滚到上一个稳定版本"""
        if not self.version_history:
            raise Error("No previous version available")

        previous = self.version_history.pop()
        self.switch_to(previous)

        # 发送告警通知
        self.alert(f"Rollback triggered: {reason}")

        return previous

5.2 回滚检查清单

• [ ] 自动回滚触发条件已配置
• [ ] 回滚操作在30秒内完成
• [ ] 上一版本的配置/Prompt/模型仍然可用
• [ ] 回滚不会丢失用户数据
• [ ] 回滚操作有完整日志

6. 成本监控

6.1 成本追踪

class CostMonitor:
    def track(self, request):
        cost = calculate_cost(
            model=request.model,
            prompt_tokens=request.prompt_tokens,
            completion_tokens=request.completion_tokens,
        )

        self.metrics.record(
            cost=cost,
            model=request.model,
            feature=request.feature,
            user_tier=request.user_tier,
        )

        # 成本异常告警
        daily_cost = self.get_daily_cost()
        if daily_cost > self.daily_budget * 0.8:
            self.alert(f"Daily cost approaching budget: ${daily_cost:.2f}")

6.2 成本优化策略

• 模型路由: 简单查询用小模型，复杂查询用大模型
• 缓存: 缓存相似查询的响应
• Prompt压缩: 减少不必要的Token
• 批处理: 合并请求降低开销

7. 用户反馈循环

7.1 收集反馈

class FeedbackCollector:
    def collect_explicit(self, conversation_id, feedback_type, details=None):
        """收集显式反馈（点赞/踩）"""
        self.store({
            "conversation_id": conversation_id,
            "type": feedback_type,  # "thumbs_up", "thumbs_down"
            "details": details,
            "timestamp": datetime.now(),
        })

    def collect_implicit(self, conversation_id, signals):
        """收集隐式反馈"""
        self.store({
            "conversation_id": conversation_id,
            "regenerated": signals.get("regenerated", False),
            "follow_up_questions": signals.get("follow_ups", 0),
            "session_duration": signals.get("duration"),
            "copied_response": signals.get("copied", False),
        })

7.2 反馈驱动的优化

1. 收集负面反馈案例
2. 分类失败原因（幻觉/不相关/格式问题/安全问题）
3. 针对性优化（修改Prompt/更新知识库/调整参数）
4. A/B测试验证优化效果
5. 上线优化版本

8. 总结

上线策略选择

策略	风险	复杂度	适用场景
直接部署	高	低	非关键应用
A/B测试	低	中	需要数据驱动决策
金丝雀发布	低	中	渐进式验证
蓝绿部署	极低	高	需要零停机和快速回滚
Feature Flags	极低	中	需要精细控制

推荐流程

开发 → 自动评估 → 人工评审 → 影子测试 → 金丝雀(1%) → A/B测试(50%) → 全量发布

参考资料

• LLMOps综述 — LLMOps概述
• 实验管理与版本控制 — 实验追踪和版本管理
• LLM评估 — 评估方法详解

---

评论