企业官网建设流程全解析

评测数据版本化：评测集变化比模型变化更难追

一、"模型 F1 突然降了 2%"——不是模型退化了，是评测集悄悄更新了

你每天早上看 Dashboard，突然注意到情感分析模型的 F1 从 0.91 掉到 0.89。排查模型——近一周没有任何代码变更。排查数据——近期也没有注入新训练数据。打开评测集目录，文件修改时间是昨天 18:00。有人加了一个新的评测子集，里面全是边界模糊的反讽句式。

这不是模型退化了，这是评测集的基准线移动了。而且你没有办法在评测报告上标注"本次评测使用了 v2.1 版本的评测集（上一版本是 v2.0）"，因为你的评测集根本没有版本号。

二、评测数据版本化的数据模型

评测数据的版本化不能只靠"一个 Git 仓库"。Git 解决了文本文件的版本追溯问题，但评测数据有三个特殊需求：

• 不可变性：某次评测使用的数据集必须永久可访问（不能因为仓库清理而丢失）
• 切片追踪：一次评测可能只用了数据集的某几个子集，每个子集也可能有独立版本
• 依赖声明：评测集依赖的标签映射、预处理脚本等附属物也需要版本化

erDiagram EVAL_DATASET { string dataset_id PK string name string version string git_commit datetime created_at int total_samples } EVAL_SUBSET { string subset_id PK string dataset_id FK string name int sample_count string description } EVAL_RUN { string run_id PK string dataset_version FK string model_version datetime run_at json metrics } EVAL_DATASET ||--o{ EVAL_SUBSET : "contains" EVAL_DATASET ||--o{ EVAL_RUN : "used_in"

版本号策略：

推荐使用YYYYMMDD.{increment}格式，如20260706.1。这种方式同时编码了时间信息和序号，在目录中按名称排序即可获得时间顺序——不需要额外的元数据查询。

三、评测数据版本管理系统实现

import json import hashlib from pathlib import Path from datetime import datetime from typing import Dict, List, Optional, Set from dataclasses import dataclass, field @dataclass class EvalDatasetVersion: """ 评测数据集版本快照 核心设计：每个版本保存的是不可变的引用快照，而非数据副本。 当你"创建新版本"时，系统只记录： 1. 当前数据的哈希值（用于完整性验证） 2. 包含哪些子集的列表（而非子集数据本身） 这避免了 10GB 评测集 × 10 个版本 = 100GB 的存储开销 """ dataset_id: str version: str # YYYYMMDD.N description: str = "" # 数据完整性验证 data_checksum: str = "" # 全量数据 SHA256 sample_count: int = 0 # 子集信息 subsets: List[str] = field(default_factory=list) # 元数据 created_at: str = field(default_factory=lambda: datetime.now().isoformat()) created_by: str = "" # 依赖追踪 preprocessing_script_hash: str = "" # 预处理脚本的哈希 label_mapping_version: str = "v1" # 标签映射版本 class EvalDataVersionManager: """ 评测数据版本管理器 核心理念：评测数据是不可变资产。一旦某个版本被用于正式评测， 它就不能被修改或删除。如果评测数据需要更新，创建一个新版本。 这个约束虽严，但避免了"上周的评测报告和本周的数据对不上"的窘境 """ def __init__(self, data_root: str = "./eval_datasets"): self.data_root = Path(data_root) self.data_root.mkdir(parents=True, exist_ok=True) self._registry_path = self.data_root / "version_registry.json" self._registry: Dict[str, Dict] = self._load_registry() def _load_registry(self) -> Dict: if self._registry_path.exists(): return json.loads(self._registry_path.read_text()) return {} def _save_registry(self): self._registry_path.write_text( json.dumps(self._registry, indent=2, ensure_ascii=False) ) def create_version( self, dataset_id: str, data_files: List[str], subsets: Optional[List[str]] = None, description: str = "", ) -> EvalDatasetVersion: """ 创建数据集的新版本 关键步骤： 1. 计算所有数据文件的联合校验和——确保完整性 2. 生成版本号——基于日期的自增序号 3. 记录到注册表并持久化 """ # 计算联合校验和：将所有文件的校验和按顺序拼接后再哈希 combined_hash = hashlib.sha256() total_samples = 0 for filepath in sorted(data_files): file_hash = self._file_sha256(filepath) combined_hash.update(file_hash.encode()) # 统计样本数（假设 JSONL 格式，每行一个样本） with open(filepath, 'r') as f: total_samples += sum(1 for _ in f) # 生成版本号 date_prefix = datetime.now().strftime("%Y%m%d") existing_versions = [ v for vid, v in self._registry.items() if vid.startswith(dataset_id) and v["version"].startswith(date_prefix) ] increment = len(existing_versions) + 1 version_str = f"{date_prefix}.{increment}" version = EvalDatasetVersion( dataset_id=dataset_id, version=version_str, description=description, data_checksum=combined_hash.hexdigest(), sample_count=total_samples, subsets=subsets or [], ) # 持久化 version_key = f"{dataset_id}@{version_str}" self._registry[version_key] = { "dataset_id": dataset_id, "version": version_str, "description": description, "checksum": version.data_checksum, "sample_count": version.sample_count, "subsets": version.subsets, "created_at": version.created_at, } self._save_registry() return version def get_version(self, dataset_id: str, version: str) -> Optional[Dict]: """查询特定版本的信息""" key = f"{dataset_id}@{version}" return self._registry.get(key) def list_versions(self, dataset_id: str) -> List[Dict]: """列出数据集的所有版本""" return [ v for k, v in self._registry.items() if k.startswith(dataset_id) ] def diff_versions(self, dataset_id: str, v1: str, v2: str) -> Dict: """ 版本差异对比 返回的信息帮助理解"为什么同一模型在两个版本的评测集上 F1 不一样"——是总样本数变了，还是子集组成变了 """ info_v1 = self.get_version(dataset_id, v1) info_v2 = self.get_version(dataset_id, v2) if not info_v1 or not info_v2: return {"error": "版本不存在"} v1_subsets = set(info_v1.get("subsets", [])) v2_subsets = set(info_v2.get("subsets", [])) return { "sample_count_diff": info_v2.get("sample_count", 0) - info_v1.get("sample_count", 0), "added_subsets": list(v2_subsets - v1_subsets), "removed_subsets": list(v1_subsets - v2_subsets), "common_subsets": list(v1_subsets & v2_subsets), "same_checksum": info_v1.get("checksum") == info_v2.get("checksum"), } def verify_integrity(self, dataset_id: str, version: str, data_files: List[str]) -> bool: """ 验证数据完整性——确保存储的数据与版本记录的校验和一致 每次评测前应执行此检查。见证奇迹的时刻不是模型 F1 创新高， 而是 verify_integrity 返回 False 时你及时发现并下线的瞬间 """ info = self.get_version(dataset_id, version) if not info: return False combined_hash = hashlib.sha256() for filepath in sorted(data_files): file_hash = self._file_sha256(filepath) combined_hash.update(file_hash.encode()) return combined_hash.hexdigest() == info["checksum"] @staticmethod def _file_sha256(filepath: str) -> str: """计算单个文件的 SHA256 —— 分块读取避免大文件内存溢出""" sha = hashlib.sha256() with open(filepath, 'rb') as f: for chunk in iter(lambda: f.read(8192), b''): sha.update(chunk) return sha.hexdigest() def create_eval_report( self, dataset_id: str, version: str, model_version: str, metrics: Dict[str, float] ) -> Dict: """ 创建绑定评测数据版本的评测报告 评测报告必须包含： 1. 模型版本 2. 评测数据集版本（关键！） 3. 子集级别的指标分解（如果使用了多个子集） 4. 评测时间戳 所有信息绑定在一起，确保"三个月后还能看懂当时的数据" """ dataset_info = self.get_version(dataset_id, version) return { "model_version": model_version, "eval_dataset": { "id": dataset_id, "version": version, "checksum": dataset_info.get("checksum") if dataset_info else None, "subsets": dataset_info.get("subsets") if dataset_info else [], }, "metrics": metrics, "evaluated_at": datetime.now().isoformat(), } # ============================================================ # 使用示例 # ============================================================ if __name__ == "__main__": manager = EvalDataVersionManager() # 创建第一个版本 v1 = manager.create_version( dataset_id="sentiment_analysis", data_files=["/path/to/train.jsonl", "/path/to/test.jsonl"], subsets=["movie_reviews", "product_reviews"], description="初始版本，包含电影和产品评论", ) print(f"创建版本: {v1.dataset_id}@{v1.version} ({v1.sample_count} 样本)") # 假设后来添加了新的评测子集 v2 = manager.create_version( dataset_id="sentiment_analysis", data_files=["/path/to/train.jsonl", "/path/to/test.jsonl", "/path/to/sarcasm.jsonl"], subsets=["movie_reviews", "product_reviews", "sarcasm"], description="新增反讽句式评测子集", ) # 对比版本差异 diff = manager.diff_versions("sentiment_analysis", v1.version, v2.version) print(f"版本差异: {diff}") # 创建绑定版本的评测报告 report = manager.create_eval_report( "sentiment_analysis", v2.version, model_version="bert-v2.3", metrics={"f1": 0.89, "accuracy": 0.87}, ) print(f"评测报告已创建（绑定评测集版本 {v2.version}）") ## 四、评测数据版本化的工程权衡与边界分析 **哈希校验的性能考量**。 对 GB 级别的评测数据集计算 SHA256 需要 10-30 秒。在 CI 流水线中，每次评测都重新计算完整校验和可能成为新的瓶颈。折中方案是"采样校验"：随机采样 10% 的文件计算校验和，配合文件修改时间检查，将验证时间压缩到 2 秒以内，而漏检概率极低。 **版本不可变性的运维成本**。 "不可修改旧版本"的约束意味着当发现旧版本评测集中存在标注错误时，你不能直接修复——必须创建一个新版本。这会在版本列表中积累大量"废弃但不可删"的版本。建议引入版本状态标记（`active` / `deprecated` / `superseded`），在不破坏不可变性约束的前提下提供版本生命周期管理。 **跨领域评测集的一致性**。 当评测集包含多个领域的数据子集（如情感分析包含电影评论和产品评论）时，每个子集可能有独立的更新节奏。全局版本号无法表达"电影评论版本 3.2 + 产品评论版本 1.7"的组合语义。建议引入**组合版本（Composite Version）**的概念：主版本号记录评测集的逻辑版本，manifest 文件记录各子集的实际版本。见证奇迹的时刻是你看到一份评测报告能精确追溯到"基准模型中电影评论子集用的是 v3.2，而新模型中用的是 v3.3"，从而解释了 F1 的 1.2% 差异。 ## 五、总结 评测数据版本化的三个核心原则： 1. **版本号必须编码时间信息**：`YYYYMMDD.N` 格式，天然时间排序 + 人类可读。 2. **评测报告必须绑定数据版本**：没有版本号的评测报告等于没有参考基准的测量结果。 3. **不可变性是基础约束**：已用于正式评测的版本不允修改——如果数据有误，创建新版本而非覆盖旧版本。