当前位置：首页 > news >正文

NLP 多任务模型灰度，要拆开看任务与样本切片

news 2026/8/14 20:51:00

NLP 多任务模型灰度，要拆开看任务与样本切片

多任务模型的总体均值可能掩盖单个任务退化。灰度时应按任务和样本切片保留结果，再决定是否扩大范围。

1. 灰度前冻结评测切分与口径

灰度集应固定各任务、语言、文本长度和领域切片，并登记样本来源与版本。新增样本需先去重和复核，不能在放量过程中悄悄改变评测分布。

基线与候选模型必须使用同一数据快照、预处理和度量实现。无法覆盖的语言、长度或领域切片要单独标记，不用总体分数代替。

2. 按最小闭环验证

候选模型与基线使用同一预处理和度量实现，分别计算每个任务与关键切片的变化。总体均值只用于概览，不能覆盖小任务的明显退化。

放量闸门至少检查输出格式、各任务指标和错误样本数量。结果接近边界时先保留样本复核，不用一次汇总得分自动放行。

3. 参考实现与图示

下面的灰度代码用于组织分任务结果。接入真实服务时，应补充模型版本、样本切片标识和回滚条件，并避免记录原始敏感文本。

import time import math import logging from typing import Dict, Any, List logging.basicConfig(level=logging.INFO) logger = logging.getLogger("CanaryController") def js_divergence(p: List[float], q: List[float]) -> float: """计算两个概率分布之间的 Jensen-Shannon 散度 (0 <= JS <= 1)""" def kl_divergence(a, b): return sum(a[i] * math.log2(a[i] / b[i]) for i in range(len(a)) if a[i] > 0 and b[i] > 0) m = [0.5 * (p[i] + q[i]) for i in range(len(p))] return 0.5 * kl_divergence(p, m) + 0.5 * kl_divergence(q, m) class NLPModelCanaryController: def __init__(self, js_threshold: float = 0.35, max_null_rate: float = 0.02): self.js_threshold = js_threshold self.max_null_rate = max_null_rate self.canary_percentage = 0.0 # 初始请求量 0% self.is_rolled_back = False # 统计指标 self.total_canary_requests = 0 self.null_canary_responses = 0 self.recent_js_scores: List[float] = [] def set_canary_traffic(self, percentage: float): if self.is_rolled_back: logger.error("❌ 灰度已被自动回滚熔断，拒绝提高请求量！") return self.canary_percentage = min(1.0, max(0.0, percentage)) logger.info(f"🚀 金丝雀切流比例调整为: {self.canary_percentage * 100}%") def process_request(self, sample_request: str, baseline_logits: List[float], canary_logits: List[float], canary_output_text: str): """模拟网关实时监控新老模型比对""" if self.is_rolled_back: return self.total_canary_requests += 1 # 1. 检查空结果率 if not canary_output_text or len(canary_output_text.strip()) == 0: self.null_canary_responses += 1 # 2. 计算新老模型预测概率的 JS 散度 score = js_divergence(baseline_logits, canary_logits) self.recent_js_scores.append(score) if len(self.recent_js_scores) > 100: self.recent_js_scores.pop(0) # 3. 触发自动化回滚检测 self._evaluate_safety_gates() def _evaluate_safety_gates(self): if self.total_canary_requests < 10: return # 样本不足时不评估 # 计算当前指标 current_null_rate = self.null_canary_responses / self.total_canary_requests avg_js_score = sum(self.recent_js_scores) / len(self.recent_js_scores) # 门禁 1: 空结果率暴涨 if current_null_rate > self.max_null_rate: self._trigger_auto_rollback(f"金丝雀空结果率达到 {current_null_rate:.2%}，超越上限 ({self.max_null_rate:.2%})") return # 门禁 2: 输出分布与 Baseline 产生剧烈异常偏离 if avg_js_score > self.js_threshold: self._trigger_auto_rollback(f"新老模型分布 JS 散度达到 {avg_js_score:.4f}，超出安全阈值 ({self.js_threshold})") return def _trigger_auto_rollback(self, reason: str): self.is_rolled_back = True self.canary_percentage = 0.0 logger.error("--------------------------------------------------") logger.error(f"🚨 触发自动回滚熔断！原因: {reason}") logger.error("候选流量已切回 Baseline，当前不再分配新请求") logger.error("--------------------------------------------------") if __name__ == "__main__": controller = NLPModelCanaryController() controller.set_canary_traffic(0.10) # 切 10% 请求量 # 模拟正常请求 for _ in range(5): controller.process_request("测试文本", [0.8, 0.2], [0.78, 0.22], "正常输出分类") # 模拟新模型突然对特定长尾文本产生概率剧烈抖动与空输出 print("\n--- 模拟运行环境长尾请求量冲入，新模型发生概率抖动 ---") for _ in range(10): # 传入偏离极大的 Logits [0.1, 0.9] vs [0.8, 0.2] controller.process_request("长尾异常文本", [0.8, 0.2], [0.05, 0.95], "")