灰度发布与回滚机制

章节：专家篇第12章第2节
难度：⭐⭐⭐⭐ 专家级
阅读时间：20-25 分钟
核心收获：掌握灰度发布和回滚机制，降低变更风险

15.6 理论要点：变更的风险管理

15.6.1 为什么需要灰度发布

直接全量上线的风险： - 新版本有 bug 影响所有用户 - 无法验证实际效果 - 回滚成本高

灰度发布的价值： - 小流量验证，降低风险 - 收集真实用户反馈 - 问题早发现，影响范围小

15.6.2 灰度发布策略

15.6.3 回滚机制

自动回滚触发条件： - 成功率下降超过阈值 - 错误率上升超过阈值 - 延迟超过阈值 - 人工触发

回滚步骤： 1. 检测到异常 2. 自动或手动触发回滚 3. 切换到上一版本 4. 通知相关人员 5. 记录回滚原因

15.7 实战案例：Prompt 灰度发布

15.7.1 场景：新 Prompt 版本上线

步骤：

1. 准备两个版本
   - v1.0（当前版本，90%流量）
   - v1.1（新版本，10%流量）

2. 配置流量分配
   - 用户ID哈希 % 100 < 10 → v1.1
   - 其他 → v1.0

3. 监控指标
   - 采纳率
   - 过滤准确率
   - 用户反馈

4. 评估效果
   - 运行1周
   - 统计显著性检验
   - 确定是否全量

5. 决策
   - 效果好 → 逐步提升比例 → 100%
   - 效果差 → 回滚到 v1.0

15.7.2 回滚代码示例

class CanaryDeployment:
    def __init__(self, stable_version, canary_version, traffic_split=0.1):
        self.stable = stable_version
        self.canary = canary_version
        self.traffic_split = traffic_split
        self.metrics = {"stable": [], "canary": []}

    def get_version(self, user_id):
        """根据用户ID分配版本"""
        if hash(user_id) % 100 < self.traffic_split * 100:
            return self.canary, "canary"
        return self.stable, "stable"

    def record(self, version, success, latency):
        """记录指标"""
        self.metrics[version].append({
            "success": success,
            "latency": latency,
            "timestamp": time.time()
        })

    def should_rollback(self):
        """判断是否需要回滚"""
        canary_success = np.mean([m["success"] for m in self.metrics["canary"]])
        stable_success = np.mean([m["success"] for m in self.metrics["stable"]])

        # 如果新版本成功率低于旧版本5%，回滚
        if stable_success - canary_success > 0.05:
            return True
        return False

    def rollback(self):
        """回滚到稳定版本"""
        self.traffic_split = 0
        self.canary = self.stable
        return {"status": "rolled_back", "version": self.stable}

15.8 图文说明

📌 待补充：建议绘制以下示意图 1. 灰度发布流程图 2. 流量分配示意图 3. 回滚决策流程图 4. 监控指标对比图

15.9 常见错误

📝 版本迭代记录