评测排障的证据留存
本文围绕“排障时怎样留下有效证据”整理可复现的检查思路。所有阈值、配置和结果均应在隔离环境中记录输入、版本与资源条件后再解释;下文示例不对应真实组织、用户、流量或成本数据。
1. 用受控样例界定问题
评测异常先保留输入样本、模型版本和完整输出,再看标签、解析器和评分器是否一致。没有这些工件,后续争论很难落到具体环节。
2. 构造现场证据链:评测数据采集、版本 Snapshot 与 Context 关联
有效的 NLP 排障证据链必须包含四个互相关联的核心要素:
1. 数据的不可变版本 Snapshot(Immutable Data Snapshot)
评测用到的 Prompt 模板、测试集样本,必须打上唯一的 Git Commit Hash 或 Git-LFS / DVC 版本号。严禁在代码里硬编码或者读取随时可能被覆写的test_dataset_latest.json。
2. Token 级映射与解码现场(Token-Level Trace)
仅记录输入文本是不够的。必须记录输入给 Transformer 的input_ids、attention_mask以及模型生成每个 Token 时的 Top-5 候选词 Logits 及其概率值。如果模型卡在无限循环重复输出某个 Token,Logits 的熵值变化是铁证。
3. 结构化解析断言日志(Parser Assertion Trace)
对于 NER 或 Intent 抽取等输出结构化 JSON 的任务,校验失败记录应只保存校验规则编号、字段名、长度和解析错误类别。原始字符串可能包含输入或模型回显,不应写入常规日志;确有排查需要时,应在隔离环境使用不可逆的合成样例复现。
4. 完整的 Trace ID 上下文透出
在评测的日志中必须贯穿统一的Eval-Trace-ID,将数据集采样 -> 模型 Inference -> 后处理解析 -> 指标 Score 计算整条链路锁在一起。
3. 工程化 NLP 多任务评测证据链采集框架
以下是一套可直接用于 NLP 评测 Pipeline 的证据链采集与归因模块代码实现:
import sys import json import traceback import hashlib from datetime import datetime from typing import Dict, Any, List, Optional from pydantic import BaseModel, Field class NLPEvaluationEvidence(BaseModel): """ NLP 评测现场证据链结构化 Snapshot """ eval_trace_id: str timestamp: str dataset_version_hash: str task_name: str raw_input_text: str processed_token_ids: List[int] model_output_raw_text: str generation_stop_reason: str # 'eos_token', 'max_length', 'stop_sequence' parser_status: str # 'SUCCESS', 'SCHEMA_ERROR', 'EMPTY_OUTPUT' error_stack_trace: Optional[str] = None attribution_tag: str # 'MODEL_DEGENERATE', 'DATA_PIPELINE_ERROR', 'PARSER_BUG', 'NONE' class EvidenceChainLogger: """ 排障证据链收集与自动归因引擎 """ def __init__(self, dataset_version_hash: str): self.dataset_version_hash = dataset_version_hash def generate_trace_id(self, task_name: str, sample_idx: int) -> str: raw_str = f"{self.dataset_version_hash}_{task_name}_{sample_idx}_{datetime.now().timestamp()}" return f"trace_{hashlib.md5(raw_str.encode('utf-8')).hexdigest()[:12]}" def capture_evidence( self, task_name: str, sample_idx: int, raw_input: str, token_ids: List[int], model_output: str, stop_reason: str, parse_func ) -> NLPEvaluationEvidence: """ 捕获单个样本评测时的完整证据链,并执行自动化诊断归因 """ trace_id = self.generate_trace_id(task_name, sample_idx) parser_status = "SUCCESS" error_trace = None attribution = "NONE" # 尝试通过后处理解析器 try: parsed_res = parse_func(model_output) if not parsed_res: parser_status = "EMPTY_OUTPUT" attribution = "MODEL_DEGENERATE" except Exception as err: parser_status = "SCHEMA_ERROR" error_trace = traceback.format_exc() # 自动化归因规则 if len(model_output.strip()) == 0: attribution = "MODEL_DEGENERATE" elif "JSONDecodeError" in str(err): if "Unterminated string" in str(err) or "Expecting value" in str(err): attribution = "MODEL_DEGENERATE" # 模型输出了非法 JSON else: attribution = "PARSER_BUG" elif len(token_ids) == 0: attribution = "DATA_PIPELINE_ERROR" # Tokenizer 切词为空 else: attribution = "PARSER_BUG" evidence = NLPEvaluationEvidence( eval_trace_id=trace_id, timestamp=datetime.now().isoformat(), dataset_version_hash=self.dataset_version_hash, task_name=task_name, raw_input_text=raw_input, processed_token_ids=token_ids, model_output_raw_text=model_output, generation_stop_reason=stop_reason, parser_status=parser_status, error_stack_trace=error_trace, attribution_tag=attribution ) return evidence # 模拟评测与证据链捕获流程 if __name__ == "__main__": logger = EvidenceChainLogger(dataset_version_hash="git_commit_78a9c2b") # 模拟后处理 JSON 解析器 def json_intent_parser(text: str) -> Dict: return json.loads(text) # 场景 1: 模型输出了被截断的非法 JSON (引发排障) bad_model_output = '{"intent": "query_flight", "slots": {"destination": "Shanghai' mock_tokens = [101, 2054, 2154, 102] evidence = logger.capture_evidence( task_name="intent_extraction", sample_idx=42, raw_input="帮我查查去上海的机票", token_ids=mock_tokens, model_output=bad_model_output, stop_reason="max_length", parse_func=json_intent_parser ) print("--- Intercepted Evidence Snapshot ---") print(json.dumps(evidence.dict(), indent=2, ensure_ascii=False)) assert evidence.attribution_tag == "MODEL_DEGENERATE", "Attribution logic failed for truncated JSON!" print("Evidence Chain Capture Verification PASSED.")4. 证据链存档与责任边界自动化判定
通过上述系统,评测 Pipeline 在每次运行结束后,都会在./eval_reports/snapshots/目录下生成包含全量证据链的evidence_chain.jsonl文件。
当评测得分跳变触发告警时,工程师不需要去登录各台服务器抓日志,只需打开评测看板,看板会自动根据attribution_tag生成诊断红绿明细:
- 若是
DATA_PIPELINE_ERROR:看板自动高亮差异 Token 序列,直接转 Task 给数据工程 Team。 - 若是
MODEL_DEGENERATE:看板直接出具截断样本列表与 Token 重复率分布,转给算法团队调整 Decoding Sampling 参数(如temperature/top_p)或补样本微调。 - 若是
PARSER_BUG:直接指派给负责工程后处理模块的工程师补正则兜底逻辑。
评测报告应把统计口径、样本覆盖和置信范围写在结果旁边,不能只给一个看似精确的总分。