简介本资源是一份面向人工智能研发工程师、算法研究员及高校相关专业师生的《人工智能训练数据集的构建与使用规范》实务指南系统解决高质量训练数据集设计、采集、标注、预处理、安全共享与持续维护等核心问题。文档为单文件Word格式.docx共1个文件大小83KB内容结构严谨涵盖数据集设计原则目标明确性、多样性与平衡性、可扩展性、数据收集与质量控制、清洗/特征工程/转换等预处理流程、标注标准与质量评估机制、多场景使用规范训练/验证/部署、开源与私有数据集管理策略、数据安全与隐私保护要点以及更新维护与案例分析等7大模块目录深度达60余页具备强实操指导性。目前已有86人学习下载读者可直接获取一套覆盖全生命周期的数据集治理方法论快速建立规范化数据工作流规避常见质量与合规风险。1. 为什么你调了三天的模型准确率上不去可能问题出在数据集构建的第一行代码里很多工程师遇到过这种场景刚跑通一个 SOTA 模型架构参数调得飞起学习率衰减策略写了三版结果在验证集上卡在 82.3% 准确率死活不动——直到某天翻出训练日志里的train_dataset.shape发现正样本只有 172 条负样本却有 4896 条。这不是模型的问题是数据集构建规范缺失的典型症状。这份《人工智能训练数据集的构建与使用规范》不是教科书式的理论汇编而是一份可直接嵌入工程 pipeline 的实操手册它定义了从data_collection.py第一行import requests开始到dataset_version.json中sha256_checksum字段落库为止所有关键决策点的技术约束。它解决的不是“要不要做数据清洗”而是“当pandas.isna().sum()返回 127 个空值时该用fillna(methodffill)还是dropna(thresh0.8*len(df))”不是泛泛而谈“标注要准确”而是给出label_consistency_score 1 - (disagreement_count / total_annotations)的量化验收阈值≥0.93 才允许进入训练。适用对象非常明确带 38 年经验的算法工程师、MLOps 工程师、AI 产品经理以及正在搭建企业级数据中台的技术负责人——你们每天面对的不是 Kaggle 数据集而是产线日志里混着乱码的 JSON、IoT 设备传回的采样频率不一致的时序信号、客服对话中夹杂方言和错别字的文本流。这份规范的价值就藏在那些被跳过的# TODO: handle edge case注释背后。2. 数据集设计原则目标明确性、多样性与平衡性如何落地为可执行的代码逻辑数据集设计原则常被写成 PPT 里的四字短语但真正决定模型上限的是这些原则在代码层的具象化表达。本章将拆解三个核心原则如何转化为 Python 函数签名、SQL 查询条件和 CI/CD 流水线中的校验节点避免“原则正确落地失效”的常见陷阱。2.1 目标明确性从模糊需求到可验证的数据契约目标明确性不是写在 PRD 里的“提升分类准确率”而是定义数据契约Data Contract——一份机器可读、可校验的接口协议。例如针对一个金融风控场景的逾期预测任务其数据契约必须包含输入约束feature_schema.json中明确定义每个字段的类型、取值范围、缺失率容忍阈值输出约束label_definition.md中规定“逾期”定义为“账单日30天后仍未还款”且需附 SQL 示例验证逻辑质量约束quality_gates.py中硬编码的校验函数。def validate_target_alignment(dataset_path: str, task_type: str binary_classification) - Dict[str, bool]: 验证数据集是否满足预设任务目标的核心约束 :param dataset_path: parquet 文件路径 :param task_type: 任务类型影响 label 分布校验逻辑 :return: 各项校验结果字典 df pd.read_parquet(dataset_path) results {} # 1. 标签完整性校验确保无未定义标签 expected_labels {0: 正常, 1: 逾期} if task_type binary_classification else set() results[label_completeness] set(df[label].unique()).issubset(expected_labels) # 2. 特征时效性校验业务特征必须在账单日之后生成 df[bill_date] pd.to_datetime(df[bill_date]) df[feature_gen_time] pd.to_datetime(df[feature_gen_time]) results[temporal_consistency] ((df[feature_gen_time] - df[bill_date]) pd.Timedelta(0 days)).all() # 3. 样本规模校验满足最小统计显著性要求Z-test for proportion n_total len(df) n_positive (df[label] 1).sum() # 要求正样本数 ≥ 384对应 95% 置信度、5% 误差边界下的最小样本量 results[min_sample_size] n_positive 384 return results # 使用示例CI 流水线中强制校验 if not all(validate_target_alignment(data/train_v2.parquet).values()): raise ValueError(Dataset violates target alignment contract. Pipeline halted.)提示validate_target_alignment函数应作为数据入库前的必过门禁Gate而非事后报告。其返回的min_sample_size校验基于统计学 Z-test 公式n (Z_α/2 * σ / E)^2推导其中E0.055% 误差边界σ0.5最保守的方差估计Z_α/21.9695% 置信度计算得n≈384。这比“经验上要够多”的说法更具工程约束力。2.2 多样性与平衡性用量化指标驱动采样策略选择多样性Diversity与平衡性Balance常被混为一谈但二者技术实现路径截然不同。多样性关注数据覆盖的广度如用户地域、设备型号、时间周期平衡性关注类别的深度如正负样本比例。规范要求二者必须分离度量、协同优化。2.2.1 多样性量化覆盖度矩阵与熵值计算对结构化数据多样性通过覆盖度矩阵Coverage Matrix评估。以电商推荐数据集为例定义维度[user_region, device_type, hour_of_day]构建三维稀疏矩阵import numpy as np from scipy.sparse import coo_matrix def calculate_diversity_entropy(df: pd.DataFrame, dimensions: List[str]) - float: 计算多维覆盖度的香农熵值越高表示多样性越强 :param df: 输入DataFrame :param dimensions: 用于计算多样性的列名列表如 [region, device, hour] :return: 归一化香农熵 [0,1] # 构建联合分组计数 group_counts df.groupby(dimensions).size().reset_index(namecount) # 计算各组合出现概率 total len(df) group_counts[prob] group_counts[count] / total # 计算香农熵 H -sum(p_i * log2(p_i)) entropy -np.sum(group_counts[prob] * np.log2(group_counts[prob] 1e-12)) # 归一化到 [0,1]除以最大可能熵即所有组合等概率时的熵 max_entropy np.log2(len(group_counts)) normalized_entropy entropy / (max_entropy 1e-12) return normalized_entropy # 应用要求多样性熵 ≥ 0.75 diversity_score calculate_diversity_entropy(train_df, [region, device_type, hour]) assert diversity_score 0.75, fDiversity entropy too low: {diversity_score:.3f}2.2.2 平衡性调控动态重采样的工程实现平衡性不能简单粗暴地upsample minority class需根据任务敏感度选择策略。规范定义了三类标准操作任务类型推荐策略代码实现要点适用场景风控/医疗诊断SMOTE-Tomek Links使用imblearn.combine.SMOTEENN先 SMOTE 过采样再 Tomek Links 清除噪声边界点高误报成本需保留决策边界推荐/广告点击分层负采样df_neg df[df.label0].sample(nlen(df[df.label1])*5, random_state42)负样本海量需控制训练规模NLP 分类类别权重注入class_weight compute_class_weight(balanced, classesnp.unique(y), yy)标签体系稳定适合快速迭代from imblearn.combine import SMOTEENN from sklearn.utils.class_weight import compute_class_weight def apply_balance_strategy(df: pd.DataFrame, label_col: str, strategy: str smote_tomek) - pd.DataFrame: X df.drop(columns[label_col]) y df[label_col] if strategy smote_tomek: # 注意SMOTEENN 对高维稀疏特征如TF-IDF效果差需先降维 from sklearn.decomposition import TruncatedSVD svd TruncatedSVD(n_components100, random_state42) X_reduced svd.fit_transform(X.select_dtypes(include[np.number])) smoteenn SMOTEENN(random_state42, sampling_strategyauto) X_res, y_res smoteenn.fit_resample(X_reduced, y) # 将降维后特征映射回原始空间简化示意实际需保存svd模型 return pd.concat([pd.DataFrame(X_res), pd.Series(y_res, namelabel_col)], axis1) elif strategy class_weight: # 直接返回原始数据但提供权重字典供模型使用 weights compute_class_weight(balanced, classesnp.unique(y), yy) return {X: X, y: y, class_weight: dict(zip(np.unique(y), weights))} else: raise ValueError(fUnknown strategy: {strategy}) # 在训练脚本中使用 balanced_data apply_balance_strategy(train_df, is_overdue, smote_tomek) model.fit(balanced_data[X], balanced_data[y])注意SMOTEENN在处理类别极度不平衡如正样本占比 0.1%时易生成无效合成样本。此时应切换至ADASYN或结合领域知识进行规则采样如强制包含所有高风险地域的正样本。2.3 可扩展性版本化数据集与动态加载的工程实践可扩展性不是“未来可以加字段”而是指数据集能在不中断线上服务的前提下支持 A/B 测试、灰度发布、热更新。其核心是数据版本控制Data Versioning与运行时动态加载Runtime Loading。2.3.1 数据版本控制DVC Git 的生产级实践放弃dataset_v1.csv,dataset_v2.csv这类命名采用 DVCData Version Control管理大文件Git 管理元数据# 初始化 DVC 仓库 dvc init # 将数据目录加入 DVC 跟踪.dvc 文件记录 hash实际文件存 .dvc/cache dvc add data/raw/ dvc add data/processed/ # 提交 DVC 元数据轻量和 Git 仓库 git add data/raw.dvc data/processed.dvc dvc.lock git commit -m add v2.1 raw and processed datasets # 发布新版本只需修改 dvc.lock 中的 hash 并提交 # CI 流水线自动触发dvc pull python train.py --dataset-version v2.1关键配置dvc.yaml定义数据流水线stages: prepare: cmd: python src/prepare_data.py --input data/raw/ --output data/processed/ deps: - data/raw/ - src/prepare_data.py outs: - data/processed/ metrics: - metrics/prepare.json: cache: false2.3.2 动态数据加载基于配置的 Dataset Factory避免在代码中硬编码pd.read_parquet(data/v2.1/train.parquet)改用工厂模式from typing import Dict, Any, Type import pandas as pd class DatasetFactory: _registry {} classmethod def register(cls, name: str): def inner_wrapper(wrapped_class: Type) - Type: cls._registry[name] wrapped_class return wrapped_class return inner_wrapper classmethod def get_dataset(cls, config: Dict[str, Any]) - pd.DataFrame: 根据配置字典动态实例化 Dataset dataset_type config.get(type, parquet) dataset_class cls._registry.get(dataset_type) if not dataset_class: raise ValueError(fUnknown dataset type: {dataset_type}) return dataset_class(config).load() DatasetFactory.register(parquet) class ParquetDataset: def __init__(self, config: Dict[str, Any]): self.path config[path] self.columns config.get(columns, None) self.filters config.get(filters, None) # DuckDB-style filters def load(self) - pd.DataFrame: # 利用 DuckDB 加速带过滤的 Parquet 读取 import duckdb conn duckdb.connect(database:memory:) query fSELECT * FROM read_parquet({self.path}) if self.filters: query f WHERE { AND .join(self.filters)} return conn.execute(query).df() # 配置驱动config/dataset_v2.1.yaml # type: parquet # path: s3://my-bucket/datasets/v2.1/train.parquet # filters: [region IN (US, EU), timestamp 2023-01-01] # 在训练脚本中 config load_yaml(config/dataset_v2.1.yaml) train_df DatasetFactory.get_dataset(config) # 自动应用过滤无需改代码3. 数据收集与预处理从网络爬虫到特征工程的全链路代码实现数据收集与预处理是规范中技术细节最密集的环节也是错误高发区。本章聚焦可复现的代码实现覆盖从原始数据获取、质量控制到特征生成的完整链条所有示例均基于真实生产环境提炼。3.1 数据采集工具链合规爬虫与 API 封装的最佳实践网络爬虫绝非requests.get(url)的简单循环。规范强制要求三大合规组件robots.txt解析器、反爬策略熔断器、数据指纹校验器。3.1.1 合规爬虫框架RobotsTxtAwareCrawlerimport requests from urllib.robotparser import RobotFileParser from urllib.parse import urlparse, urljoin import time from functools import wraps def rate_limit(calls_per_second: float): 装饰器实现请求速率限制 min_interval 1.0 / calls_per_second last_called [0.0] def decorator(func): wraps(func) def wrapper(*args, **kwargs): elapsed time.time() - last_called[0] left_to_wait min_interval - elapsed if left_to_wait 0: time.sleep(left_to_wait) ret func(*args, **kwargs) last_called[0] time.time() return ret return wrapper return decorator class RobotsTxtAwareCrawler: def __init__(self, base_url: str, delay: float 1.0): self.base_url base_url self.delay delay self.parser RobotFileParser() self._load_robots_txt() def _load_robots_txt(self): 加载并解析 robots.txt parsed urlparse(self.base_url) robots_url urljoin(f{parsed.scheme}://{parsed.netloc}, /robots.txt) try: response requests.get(robots_url, timeout10) if response.status_code 200: self.parser.parse(response.text.splitlines()) except Exception as e: print(fWarning: Failed to load robots.txt: {e}) rate_limit(calls_per_second1.0) def fetch(self, url: str, **kwargs) - requests.Response: 带 robots.txt 检查的 fetch 方法 parsed urlparse(url) can_fetch self.parser.can_fetch(*, url) if not can_fetch: raise PermissionError(frobots.txt disallows fetching {url}) try: response requests.get(url, timeout30, **kwargs) response.raise_for_status() return response except requests.exceptions.RequestException as e: raise ConnectionError(fFailed to fetch {url}: {e}) # 使用示例 crawler RobotsTxtAwareCrawler(https://example.com) try: resp crawler.fetch(https://example.com/api/data?date2023-10-01) data resp.json() except (PermissionError, ConnectionError) as e: print(fCrawling failed: {e})3.1.2 API 数据封装RetryableAPIClient对于第三方 API规范要求内置指数退避重试、响应缓存、错误分类import json import hashlib from pathlib import Path from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type class RetryableAPIClient: def __init__(self, base_url: str, cache_dir: str .api_cache): self.base_url base_url.rstrip(/) self.cache_dir Path(cache_dir) self.cache_dir.mkdir(exist_okTrue) def _cache_key(self, url: str, params: dict) - str: 生成唯一缓存键 key_str f{url}?{json.dumps(params, sort_keysTrue)} return hashlib.md5(key_str.encode()).hexdigest() def _get_from_cache(self, cache_key: str) - Optional[dict]: cache_file self.cache_dir / f{cache_key}.json if cache_file.exists(): try: return json.loads(cache_file.read_text()) except: pass return None def _save_to_cache(self, cache_key: str, data: dict): cache_file self.cache_dir / f{cache_key}.json cache_file.write_text(json.dumps(data)) retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max10), retryretry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError)) ) def get(self, endpoint: str, params: dict None, use_cache: bool True) - dict: url f{self.base_url}/{endpoint.lstrip(/)} cache_key self._cache_key(url, params or {}) if use_cache: cached self._get_from_cache(cache_key) if cached is not None: return cached try: response requests.get(url, paramsparams, timeout30) response.raise_for_status() data response.json() if use_cache: self._save_to_cache(cache_key, data) return data except requests.exceptions.HTTPError as e: if response.status_code 429: raise Exception(Rate limited by API. Check quota.) from e elif 400 response.status_code 500: raise Exception(fClient error {response.status_code}: {response.text}) from e else: raise Exception(fServer error {response.status_code}) from e # 使用示例获取天气数据带缓存和重试 client RetryableAPIClient(https://api.weather.com/v3) data client.get(weather/forecast/daily/5day, params{geocode: 40.7128,-74.0060})3.2 数据质量控制自动化清洗流水线数据质量控制不是一次性脚本而是嵌入 ETL 流水线的持续校验。规范定义了四级质量门禁Quality Gates门禁层级校验项触发动作代码位置L1文件完整性MD5/SHA256文件损坏则终止流水线pre_ingestion.pyL2Schema 一致性字段缺失/类型错误则告警并隔离异常分区schema_validator.pyL3统计分布漂移PSIPSI 0.25 则触发人工审核drift_detector.pyL4业务规则校验如金额 0违规记录写入quarantine表并告警business_rules.py3.2.1 L2 Schema 校验器Pydantic Model 驱动from pydantic import BaseModel, validator, Field from typing import Optional, List, Dict, Any import pandas as pd class TransactionSchema(BaseModel): transaction_id: str Field(..., min_length10) amount: float Field(..., gt0.0) # 业务规则金额必须大于0 timestamp: str # ISO format string, validated in custom logic merchant_category: str Field(..., patternr^[A-Z]{2,}$) user_id: Optional[str] None validator(timestamp) def validate_timestamp(cls, v): from datetime import datetime try: dt datetime.fromisoformat(v.replace(Z, 00:00)) # 要求时间戳在近3年以内 assert dt datetime.now() - pd.Timedelta(3Y) return v except (ValueError, AssertionError) as e: raise ValueError(fInvalid timestamp format or out of range: {v}) from e def validate_schema_pydantic(df: pd.DataFrame, model: Type[BaseModel]) - Dict[str, Any]: 使用 Pydantic 模型批量校验 DataFrame :return: 包含 valid_df, invalid_records, stats 的字典 valid_records [] invalid_records [] for idx, row in df.iterrows(): try: # 将行转为 dict 并校验 record_dict row.to_dict() # 移除 NaN 值Pydantic 不接受 None 除非字段为 Optional clean_dict {k: v for k, v in record_dict.items() if pd.notna(v)} model(**clean_dict) valid_records.append(row) except Exception as e: invalid_records.append({ row_index: idx, error: str(e), raw_data: row.to_dict() }) valid_df pd.DataFrame(valid_records) return { valid_df: valid_df, invalid_records: invalid_records, stats: { total: len(df), valid: len(valid_records), invalid: len(invalid_records), valid_rate: len(valid_records) / len(df) if len(df) 0 else 0 } } # 在 Airflow DAG 中调用 validation_result validate_schema_pydantic(raw_df, TransactionSchema) if validation_result[stats][valid_rate] 0.99: send_alert(validation_result[invalid_records][:10]) # 告警前10条 quarantine_df pd.DataFrame(validation_result[invalid_records]) write_to_quarantine_table(quarantine_df)3.2.2 L3 分布漂移检测PSIPopulation Stability Indeximport numpy as np import pandas as pd from scipy import stats def calculate_psi(expected: pd.Series, actual: pd.Series, bucket_method: str quantile, n_bins: int 10) - float: 计算 Population Stability Index (PSI) PSI sum( (Actual% - Expected%) * ln(Actual% / Expected%) ) PSI 0.1: 无变化0.1-0.25: 轻微变化0.25: 显著变化 if bucket_method quantile: # 使用分位数分桶对长尾分布更鲁棒 expected_qt pd.qcut(expected, qn_bins, duplicatesdrop).astype(str) actual_qt pd.qcut(actual, qn_bins, duplicatesdrop).astype(str) else: # 等宽分桶 bins np.linspace(expected.min(), expected.max(), n_bins 1) expected_qt pd.cut(expected, binsbins, include_lowestTrue).astype(str) actual_qt pd.cut(actual, binsbins, include_lowestTrue).astype(str) # 计算各桶占比 exp_dist expected_qt.value_counts(normalizeTrue).sort_index() act_dist actual_qt.value_counts(normalizeTrue).sort_index() # 对齐索引缺失桶补0 all_bins exp_dist.index.union(act_dist.index) exp_dist exp_dist.reindex(all_bins, fill_value0.0) act_dist act_dist.reindex(all_bins, fill_value0.0) # 计算 PSI psi np.sum((act_dist - exp_dist) * np.log((act_dist 1e-6) / (exp_dist 1e-6))) return psi # 应用监控关键特征 drift psi_scores {} for feature in [user_age, transaction_amount, session_duration]: psi calculate_psi(base_df[feature], current_df[feature]) psi_scores[feature] psi if psi 0.25: trigger_drift_investigation(feature, psi) print(PSI Scores:, psi_scores) # 输出示例: {user_age: 0.08, transaction_amount: 0.32, session_duration: 0.15}3.3 特征工程从原始字段到模型就绪特征的转换矩阵特征工程是规范中最具领域特性的环节。本节提供通用模板工程师需根据业务填充具体逻辑。3.3.1 特征转换矩阵Feature Transformation Matrix规范要求所有特征转换必须定义在feature_config.yaml中并由FeatureTransformer统一执行# feature_config.yaml features: - name: user_tenure_days source: registration_date transform: days_since params: {ref_date: 2023-10-01} - name: amount_log1p source: transaction_amount transform: log1p params: {} - name: merchant_category_encoded source: merchant_category transform: target_encoding params: {target: is_fraud, smoothing: 10} - name: time_of_day_sin source: transaction_time transform: cyclical_sin params: {period: 24} - name: is_weekend source: transaction_date transform: is_weekend params: {}import numpy as np import pandas as pd from datetime import datetime, timedelta class FeatureTransformer: def __init__(self, config_path: str): self.config load_yaml(config_path) self.encoders {} # 存储 target encoder 等有状态对象 def fit_transform(self, df: pd.DataFrame) - pd.DataFrame: result_df df.copy() for feat_cfg in self.config[features]: col_name feat_cfg[name] src_col feat_cfg[source] transform_func getattr(self, f_transform_{feat_cfg[transform]}) params feat_cfg.get(params, {}) # 对于 target encoding需要 fit 阶段计算编码值 if feat_cfg[transform] target_encoding: self._fit_target_encoder(df, src_col, params[target], params[smoothing]) result_df[col_name] transform_func(df[src_col], **params) else: result_df[col_name] transform_func(df[src_col], **params) return result_df def _transform_days_since(self, series: pd.Series, ref_date: str) - pd.Series: ref_dt pd.to_datetime(ref_date) return (ref_dt - pd.to_datetime(series)).dt.days def _transform_log1p(self, series: pd.Series) - pd.Series: return np.log1p(series) def _transform_target_encoding(self, series: pd.Series, target: str, smoothing: int) - pd.Series: # 使用已拟合的编码器 return series.map(self.encoders[f{series.name}_{target}]).fillna(self.encoders[f{series.name}_{target}].mean()) def _transform_cyclical_sin(self, series: pd.Series, period: int) - pd.Series: # 小时制0-23 - sin(2π * hour / 24) return np.sin(2 * np.pi * series.astype(int) / period) def _transform_is_weekend(self, series: pd.Series) - pd.Series: return pd.to_datetime(series).dt.dayofweek 5 def _fit_target_encoder(self, df: pd.DataFrame, col: str, target: str, smoothing: int): 拟合 Target Encoder计算每类的平均目标值加平滑 global_mean df[target].mean() agg df.groupby(col)[target].agg([mean, count]) smooth (agg[count] * agg[mean] smoothing * global_mean) / (agg[count] smoothing) self.encoders[f{col}_{target}] smooth # 使用 transformer FeatureTransformer(feature_config.yaml) train_features transformer.fit_transform(train_df) test_features transformer.transform(test_df) # transform 仅映射不重新拟合4. 数据标注质量评估与模型训练闭环验证数据标注是 AI 项目中最耗人力、也最容易引入系统性偏差的环节。本章不讨论众包平台选型而是聚焦如何用代码量化标注质量、建立标注-训练-评估的反馈闭环让标注错误在模型上线前就被捕获。4.1 标注一致性评估Krippendorffs Alpha 的工程实现Krippendorffs Alpha 是比 Cohens Kappa 更鲁棒的多标注者一致性指标能处理缺失值和多种数据类型。规范要求所有标注任务必须计算 Alpha ≥ 0.8。import numpy as np import pandas as pd from itertools import combinations from collections import defaultdict def krippendorffs_alpha(annotation_matrix: np.ndarray, level_of_measurement: str nominal) - float: 计算 Krippendorffs Alpha 一致性系数 annotation_matrix: 二维数组行样本列标注者值标注类别 level_of_measurement: nominal, ordinal, interval, ratio # 移除全为 NaN 的行无标注样本 mask ~np.isnan(annotation_matrix).all(axis1) annotation_matrix annotation_matrix[mask] # 获取所有非 NaN 值 values annotation_matrix[~np.isnan(annotation_matrix)] if len(values) 0: return 0.0 # 计算观测不一致度 Do N len(annotation_matrix) # 样本数 Do 0.0 total_pairs 0 for i in range(N): # 获取第 i 个样本的有效标注者索引 annotators_i np.where(~np.isnan(annotation_matrix[i]))[0] if len(annotators_i) 2: continue # 计算该样本内所有标注者对的不一致度 for a1, a2 in combinations(annotators_i, 2): v1, v2 annotation_matrix[i, a1], annotation_matrix[i, a2] if level_of_measurement nominal: disagreement 0 if v1 v2 else 1 elif level_of_measurement ordinal: # 有序尺度disagreement (rank_diff)^2 rank1, rank2 _get_rank(v1, values), _get_rank(v2, values) disagreement (rank1 - rank2) ** 2 else: # interval/ratio disagreement (v1 - v2) ** 2 Do disagreement total_pairs 1 if total_pairs 0: return 0.0 Do / total_pairs # 计算期望不一致度 De # 简化假设所有值独立同分布De sum(p_v * p_w * d(v,w)) over all v,w # 此处用经验分布近似 unique_vals, counts np.unique(values, return_countsTrue) p counts / len(values) De 0.0 for i, v in enumerate(unique_vals): for j, w in enumerate(unique_vals): if level_of_measurement nominal: d_vw 0 if v w else 1 else: d_vw (v - w) ** 2 De p[i] * p[j] * d_vw # Alpha 1 - Do/De if De 0: return 1.0 if Do 0 else 0.0 return 1.0 - (Do / De) def _get_rank(value, all_values): 获取 value 在 all_values 中的秩处理重复值 sorted_vals np.sort(all_values) return np.searchsorted(sorted_vals, value, sideright) # 使用示例从标注平台导出的 CSV # columns: sample_id, annotator_1, annotator_2, annotator_3, ... df_annotations pd.read_csv(annotations.csv) # 构建 annotation_matrix: shape (n_samples, n_annotators) annotator_cols [c for c in df_annotations.columns if c.startswith(annotator_)] annotation_matrix df_annotations[annotator_cols].values alpha krippendorffs_alpha(annotation_matrix, level_of_measurementnominal) print(fKrippendorffs Alpha: {alpha:.3f}) assert alpha 0.8, fAnnotation consistency too low: {alpha:.3f}4.2 标注错误定位基于模型预测的主动学习反馈高质量标注不应止步于一致性更要识别系统性错误。规范要求在模型训练初期用轻量模型如 Logistic Regression扫描训练集找出本文还有配套的精品资源点击获取