本课学习灰度发布和A/B测试——生产环境中安全上线新功能的关键技术。灰度发布让新版本逐步替代旧版本，A/B测试用数据驱动决策哪个版本更好。
第33课: 灰度与A/B
├── 灰度发布
│ ├── 比例灰度: 10%→30%→50%→100%
│ ├── 用户分群: VIP先体验
│ ├── 请求头: 测试人员
│ └── 金丝雀: 1%观察后扩大
├── A/B测试
│ ├── 实验配置
│ ├── 流量哈希分配
│ ├── 指标收集
│ └── 统计分析
└── 发布管理
├── 灰度递增
├── 错误率监控
└── 自动回滚
# 灰度发布策略
from typing import Dict, List, Callable
from dataclasses import dataclass, field
from enum import Enum
import hashlib
import json
class RolloutStrategy(Enum):
PERCENTAGE = "percentage" # 按比例
USER_SEGMENT = "user_segment" # 按用户分群
HEADER = "header" # 按请求头
CANARY = "canary" # 金丝雀(先1%观察)
@dataclass
class RolloutRule:
strategy: RolloutStrategy
percentage: float = 0.0 # 0-100
segments: List[str] = field(default_factory=list)
header_name: str = ""
header_value: str = ""
class GrayscaleController:
"""灰度控制器 - 管理渐进式发布"""
def __init__(self):
self.rules: Dict[str, RolloutRule] = {}
self.current_percentage: Dict[str, float] = {}
def set_rule(self, feature_name: str, rule: RolloutRule):
self.rules[feature_name] = rule
self.current_percentage[feature_name] = rule.percentage
def should_use_new(self, feature_name: str, user_id: str = "", headers: dict = None) -> bool:
"""判断是否使用新版本"""
rule = self.rules.get(feature_name)
if not rule: return False
if rule.strategy == RolloutStrategy.PERCENTAGE:
# 基于用户ID哈希确定比例
hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100
return hash_val < rule.percentage
elif rule.strategy == RolloutStrategy.USER_SEGMENT:
# 基于用户分群
return any(seg in user_id for seg in rule.segments)
elif rule.strategy == RolloutStrategy.HEADER:
if headers:
return headers.get(rule.header_name) == rule.header_value
return False
elif rule.strategy == RolloutStrategy.CANARY:
hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100
return hash_val < rule.percentage
return False
def increase_percentage(self, feature_name: str, increment: float = 10.0):
"""逐步增加灰度比例"""
current = self.current_percentage.get(feature_name, 0)
new_pct = min(current + increment, 100)
self.current_percentage[feature_name] = new_pct
if feature_name in self.rules:
self.rules[feature_name].percentage = new_pct
return new_pct
# 使用
controller = GrayscaleController()
controller.set_rule("new_llm_model", RolloutRule(strategy=RolloutStrategy.PERCENTAGE, percentage=10))
# controller.should_use_new("new_llm_model", "user_123")✅ 验证通过：GrayscaleController实现了四种灰度策略和渐进递增
# A/B测试框架
from openai import OpenAI
import json
import time
import hashlib
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from collections import defaultdict
client = OpenAI()
@dataclass
class Experiment:
name: str
variants: Dict[str, dict] # variant_name -> config
traffic_percentage: float = 100.0
is_active: bool = True
@dataclass
class ExperimentResult:
experiment_name: str
variant: str
user_id: str
metric_name: str
metric_value: float
timestamp: float = field(default_factory=time.time)
class ABTestFramework:
"""A/B测试框架 - 实验配置/流量分配/指标收集/统计判断"""
def __init__(self):
self.experiments: Dict[str, Experiment] = {}
self.results: List[ExperimentResult] = []
def create_experiment(self, name, variants, traffic_pct=100):
self.experiments[name] = Experiment(name=name, variants=variants, traffic_percentage=traffic_pct)
def get_variant(self, experiment_name: str, user_id: str) -> str:
"""为用户分配变体"""
exp = self.experiments.get(experiment_name)
if not exp or not exp.is_active: return "control"
# 哈希分配
hash_val = int(hashlib.md5(f"{experiment_name}:{user_id}".encode()).hexdigest(), 16) % 100
if hash_val >= exp.traffic_percentage: return "control"
variant_names = list(exp.variants.keys())
# 等比例分配
idx = hash_val % len(variant_names)
return variant_names[idx]
def record_result(self, experiment_name, variant, user_id, metric_name, metric_value):
self.results.append(ExperimentResult(
experiment_name=experiment_name, variant=variant,
user_id=user_id, metric_name=metric_name, metric_value=metric_value
))
def analyze(self, experiment_name: str, metric_name: str = "score") -> dict:
"""分析实验结果"""
exp_results = [r for r in self.results if r.experiment_name == experiment_name and r.metric_name == metric_name]
by_variant = defaultdict(list)
for r in exp_results:
by_variant[r.variant].append(r.metric_value)
analysis = {}
for variant, values in by_variant.items():
if not values: continue
analysis[variant] = {
"count": len(values),
"mean": sum(values) / len(values),
"std": (sum((v - sum(values)/len(values))**2 for v in values) / len(values)) ** 0.5 if len(values) > 1 else 0,
}
# 简单判断: 均值最高的变体胜出
if analysis:
winner = max(analysis.items(), key=lambda x: x[1]["mean"])
analysis["winner"] = winner[0]
return analysis✅ 验证通过：ABTestFramework实现了实验创建、哈希分配和结果统计分析
# 灰度与A/B完整系统
from typing import Dict, Callable
from dataclasses import dataclass
@dataclass
class ReleaseConfig:
feature_name: str
new_version_config: dict
old_version_config: dict
grayscale_percentage: float = 10.0
auto_rollback_on_error: bool = True
error_threshold: float = 0.05 # 错误率阈值
class ReleaseManager:
"""发布管理器 - 灰度+A/B+自动回滚"""
def __init__(self):
self.releases: Dict[str, ReleaseConfig] = {}
self.error_counts: Dict[str, dict] = {} # feature -> {new: N, old: N, total: N}
def create_release(self, config: ReleaseConfig):
self.releases[config.feature_name] = config
self.error_counts[config.feature_name] = {"new_errors": 0, "old_errors": 0, "new_total": 0, "old_total": 0}
def should_use_new(self, feature_name: str, user_id: str) -> bool:
config = self.releases.get(feature_name)
if not config: return False
hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) % 100
return hash_val < config.grayscale_percentage
def record_result(self, feature_name: str, is_new: bool, is_error: bool):
counts = self.error_counts.get(feature_name)
if not counts: return
if is_new:
counts["new_total"] += 1
if is_error: counts["new_errors"] += 1
else:
counts["old_total"] += 1
if is_error: counts["old_errors"] += 1
# 自动回滚检查
config = self.releases.get(feature_name)
if config and config.auto_rollback_on_error and counts["new_total"] > 10:
new_error_rate = counts["new_errors"] / counts["new_total"]
if new_error_rate > config.error_threshold:
config.grayscale_percentage = 0
return # 自动回滚
def increase_grayscale(self, feature_name: str, increment=20):
config = self.releases.get(feature_name)
if config:
config.grayscale_percentage = min(config.grayscale_percentage + increment, 100)
return config.grayscale_percentage
return 0✅ 验证通过：ReleaseManager集成了灰度递增、错误率监控和自动回滚
| 策略 | 速度 | 风险 | 适用场景 |
|---|---|---|---|
| 比例灰度 | 渐进 | 低 | 标准发布 |
| 金丝雀 | 极慢 | 最低 | 高风险变更 |
| A/B测试 | 中等 | 低 | 效果验证 |
| 全量发布 | 即时 | 高 | 低风险修复 |
# 挑战: 构建自动灰度发布
# - 自动递增灰度比例
# - 每步检查错误率
# - 错误率超标自动回滚实现多维度灰度——同时按地区和用户类型灰度
灰度发布看似简单，实际执行中有许多坑。以下是大规模LLM应用灰度发布的实战经验总结：
| 错误类型 | 描述 | 后果 | 修正 |
|---|---|---|---|
| 样本量不足 | 过早得出结论 | 假阳性/假阴性 | 计算所需样本量 |
| 新奇效应 | 用户对新事物短暂感兴趣 | 效果虚高 | 延长观察期 |
| 选择偏差 | 灰度用户不具代表性 | 结论不可推广 | 随机化分配 |
| 多重比较 | 同时测试多个指标 | 假阳性增多 | Bonferroni校正 |
| 窥视问题 | 频繁查看p值 | 过早停止实验 | 预设实验时长 |