NLP 服务上线后，持续观察数据漂移和失败类型

NLP 服务上线后，离线得分不会自动解释线上变化。输入长度、语言分布、类别比例和失败类型都可能漂移，需要用稳定口径持续观察。

1. 先定义允许采集的摘要

NLP 评测要先定义任务边界、样本来源、分词或编码版本及度量计算方式。样本仅使用公开、合成或已脱敏内容，并保留可复核的数据版本标识。

线上观测应只保留完成任务所需的聚合特征和版本标识，不把用户原文当调试材料。

2. 把漂移和质量失败分开

多任务比较应分别检查各任务的错误类型与覆盖范围，不用一个汇总分数替代细节。新增样本先经复核，再进入固定的回归集。

分布变化不等于质量退化。先在固定回归集确认模型行为，再用经授权的抽样判断线上失败是否增加。

3. 观测字段示例

以下片段保留原有技术结构。运行前请替换为本地的非敏感示例，并根据依赖版本核对接口。

import time import math import numpy as np from typing import Dict, Any, List from fastapi import FastAPI, Request, Response from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST # 1. 定义 Prometheus 任务与算法监控指标 INFERENCE_REQUEST_TOTAL = Counter( "nlp_inference_requests_total", "NLP 多任务推理请求总次数", ["task_type", "status"] ) INFERENCE_LATENCY_SECONDS = Histogram( "nlp_inference_latency_seconds", "推理耗时分布 (秒)", ["task_type"], buckets=(0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5) ) MODEL_CONFIDENCE_GAUGE = Gauge( "nlp_model_confidence_score", "模型当前 Batch 预测的平均 Softmax 最大置信度", ["task_type"] ) MODEL_PREDICTION_ENTROPY = Gauge( "nlp_model_prediction_entropy", "模型当前 Batch 预测概率分布的平均信息熵 (用于衡量不确定性)", ["task_type"] ) def calculate_shannon_entropy(probs: np.ndarray) -> float: """计算 Softmax 输出概率矩阵的平均香农熵 (Shannon Entropy)""" # 避免 log(0) 带来的 NaN 报错 bounded_probs = np.clip(probs, 1e-12, 1.0) entropy = -np.sum(bounded_probs * np.log2(bounded_probs), axis=-1) return float(np.mean(entropy)) class NLPObservabilityGateway: """ NLP 多任务可观测性网关 负责捕捉底层模型输出的概率分布变动 """ def __init__(self): self.app = FastAPI(title="NLP Multi-Task Observability Server") self._setup_routes() def _setup_routes(self): @self.app.post("/api/v1/predict") async def predict_endpoint(request: Request): t0 = time.perf_counter() body = await request.json() task_type = body.get("task_type", "text_classification") text = body.get("text", "") try: # 模拟多任务模型推理输出 (假设 3 分类 Softmax 概率) # 在目标工程中此处调用 PyTorch/TensorRT 模型 mock_logits = np.random.dirichlet(alpha=(1.0, 1.0, 1.0), size=1)[0] max_confidence = float(np.max(mock_logits)) entropy_val = calculate_shannon_entropy(mock_logits) # 记录指标 duration = time.perf_counter() - t0 INFERENCE_LATENCY_SECONDS.labels(task_type=task_type).observe(duration) INFERENCE_REQUEST_TOTAL.labels(task_type=task_type, status="success").inc() MODEL_CONFIDENCE_GAUGE.labels(task_type=task_type).set(max_confidence) MODEL_PREDICTION_ENTROPY.labels(task_type=task_type).set(entropy_val) return { "code": 200, "task_type": task_type, "prediction": int(np.argmax(mock_logits)), "confidence": round(max_confidence, 4), "entropy": round(entropy_val, 4) } except Exception as e: INFERENCE_REQUEST_TOTAL.labels(task_type=task_type, status="error").inc() return {"code": 500, "message": str(e)} @self.app.get("/metrics") async def metrics_endpoint(): """暴露给 Prometheus 抓取的标准 Endpoint""" return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST) gateway = NLPObservabilityGateway() app = gateway.app if __name__ == "__main__": import uvicorn print("[INFO] 启动 NLP 多任务可观测性 Gateway 服务...") # 模拟本地启动服务，可使用 curl http://localhost:8000/metrics 抓取 Prometheus 数据 uvicorn.run(app, host="localhost", port=8000)

groups: - name: nlp_model_health_alerts rules: # 1. 算法置信度坍塌告警 - alert: NLPModelLowConfidence expr: avg_over_time(nlp_model_confidence_score[10m]) < 0.55 for: 5m labels: severity: warning annotations: summary: "NLP 多任务模型运行环境置信度骤降" description: "过去 10 时段内，任务 {{ $labels.task_type }} 的平均预测置信度降至 55% 以下，可能存在数据漂移。" # 2. 预测概率熵异常增高告警 (困惑度激增) - alert: NLPModelHighEntropy expr: avg_over_time(nlp_model_prediction_entropy[10m]) > 1.2 for: 5m labels: severity: critical annotations: summary: "模型输出不确定性 (香农熵) 离群飙升" description: "任务 {{ $labels.task_type }} 的预测分布接近均匀分布，模型可能在大量输出误判结果。"

4. 线上观测复核

• 指标是否携带模型、编码器和规则版本。
• 输入摘要是否经过最小化与脱敏审查。
• 漂移告警是否能回到固定回归集验证。
• 采样、保存周期和访问权限是否明确。

总结

持续观察的重点不是攒更多输入，而是用更少、更清楚的数据定位变化来自哪里。