第 6 章第二部分：预训练数据工程

隐私脱敏与文档打标

PII类目(邮箱/电话/身份证/银行卡/IP/API Key)、双路检测(正则+NER+密钥扫描)、Presidio框架、脱敏策略、多维打标、分类器训练

最后更新：2026年8月19日 PII脱敏 Presidio NER 多维打标分类器训练差分隐私

隐私脱敏是合规底线，文档打标是配比的基础。

这两件事不做，模型无法上线；做不好，配比无法精细。

6.1 PII 类目全景

PII（Personally Identifiable Information，个人身份信息）是预训练数据里必须处理的内容。未经脱敏训练大模型会带来严重的法律、伦理风险。

6.1.1 PII 类目分类

类目	例子	风险等级	必处理
邮箱	user@gmail.com	高	必脱敏
电话	+86 138-xxxx-xxxx	高	必脱敏
身份证号	110101199001011234	极高	必脱敏
护照号	E12345678	极高	必脱敏
银行卡号	622202xxxxxxxxxxxx	极高	必脱敏
信用卡号	4532-xxxx-xxxx-xxxx	极高	必脱敏
IP 地址	192.168.1.1	中	推荐脱敏
API Key	sk-xxxxxxxxxx	高	必脱敏
AWS Key	AKIA…	高	必脱敏
姓名地址	张三 / 北京市海淀区…	高	视场景
医疗信息	糖尿病、HIV	极高	必脱敏
种族宗教	维吾尔族 / 穆斯林	高	视场景
生物特征	指纹 / 虹膜 / 人脸	极高	必脱敏

6.1.2 法律合规要求

法规	适用地区	PII 要求
GDPR	欧盟	任何个人信息必须脱敏或征得同意
CCPA	加州	用户有”被遗忘权”，PII 需可删除
《个人信息保护法》	中国	敏感个人信息必须脱敏
HIPAA	美国（医疗）	医疗信息严格保护
《数据安全法》	中国	重要数据需安全评估

6.2 双路检测：正则 + NER + 密钥扫描

PII 检测推荐双路并行：正则匹配（高精度低召回）+ NER 模型（高召回低精度）+ 密钥扫描（特定模式）。

6.2.1 正则匹配

import re

PII_REGEX_PATTERNS = {
    'email': re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'),
    
    'phone_cn': re.compile(r'(?:(?:\+?86)?\s*)?1[3-9]\d{9}'),
    'phone_intl': re.compile(r'\+\d{1,3}[\s-]?\d{1,4}[\s-]?\d{3,4}[\s-]?\d{3,4}'),
    
    'id_card_cn': re.compile(
        r'\b[1-9]\d{5}(?:19|20)\d{2}'
        r'(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx]\b'
    ),
    
    'bank_card_cn': re.compile(r'\b6(?:2(?:02|21|25|26|28|29|30|31|32|33|34|35|36|37|38|39|40))\d{14,17}\b'),
    
    'passport_cn': re.compile(r'\b[EeDdGgPpHh]\d{8}\b'),
    
    'ipv4': re.compile(r'\b(?:\d{1,3}\.){3}\d{1,3}\b'),
    'ipv6': re.compile(r'(?:[A-Fa-f0-9]{1,4}:){7}[A-Fa-f0-9]{1,4}'),
    
    # API Keys
    'aws_access_key': re.compile(r'\bAKIA[0-9A-Z]{16}\b'),
    'aws_secret_key': re.compile(r'(?i)aws_secret_access_key\s*=\s*[A-Za-z0-9/+=]{40}'),
    'openai_key': re.compile(r'\bsk-[A-Za-z0-9]{20,}\b'),
    'github_token': re.compile(r'\bgh[pousr]_[A-Za-z0-9]{36,}\b'),
    'google_api_key': re.compile(r'\bAIza[A-Za-z0-9_\-]{35}\b'),
    
    # 信用卡号
    'visa': re.compile(r'\b4\d{3}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b'),
    'mastercard': re.compile(r'\b5[1-5]\d{2}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b'),
}

def detect_pii_regex(text):
    """正则匹配 PII"""
    matches = []
    for pii_type, pattern in PII_REGEX_PATTERNS.items():
        for match in pattern.finditer(text):
            matches.append({
                'type': pii_type,
                'text': match.group(),
                'start': match.start(),
                'end': match.end()
            })
    return matches

6.2.2 NER 模型识别

正则只能抓固定格式，对自然语言中的姓名、地址、机构等需要 NER：

from transformers import AutoTokenizer, AutoModelForTokenClassification
from transformers import pipeline

# 多语种 NER 模型
MODEL_NAME = 'Babelscape/wikineural-multilingual-ner'

tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
model = AutoModelForTokenClassification.from_pretrained(MODEL_NAME)

ner_pipeline = pipeline(
    'ner', model=model, tokenizer=tokenizer,
    aggregation_strategy='simple',
    device='cuda:0'
)

def detect_pii_ner(text):
    """NER 识别 PII"""
    entities = ner_pipeline(text[:4096])  # 截断保护
    
    pii_entities = []
    PII_TYPES = {'PER', 'ORG', 'LOC'}  # 人名、机构、地名
    
    for ent in entities:
        if ent['entity_group'] in PII_TYPES:
            pii_entities.append({
                'type': ent['entity_group'].lower(),
                'text': ent['word'],
                'start': ent['start'],
                'end': ent['end'],
                'confidence': ent['score']
            })
    
    return pii_entities

6.2.3 密钥扫描

专门的密钥扫描工具能识别 100+ 种 API Key 格式：

# 推荐 truffleHog 或 detect-secrets
# 命令行用法：
# trufflehog filesystem --directory=./data --json --output=secrets.json

# 或用 Python 库
# pip install detect-secrets
from detect_secrets import SecretsCollection
from detect_secrets.core import scan

def detect_secrets_in_text(text):
    """扫描 API Keys/密钥"""
    secrets = []
    for line in text.split('\n'):
        detected = scan.scan_line(line)
        for item in detected:
            secrets.append({
                'type': item.type,
                'text': item.secret_value,
                'line': line
            })
    return secrets

6.2.4 双路融合

def detect_pii(text):
    """
    双路 PII 检测
    返回所有 PII 实体
    """
    # 路径1: 正则
    regex_matches = detect_pii_regex(text)
    
    # 路径2: NER
    ner_entities = detect_pii_ner(text)
    
    # 路径3: 密钥扫描
    secret_matches = detect_secrets_in_text(text)
    
    # 合并去重（同位置范围内）
    all_pii = regex_matches + ner_entities + secret_matches
    all_pii = dedup_overlapping(all_pii)
    
    return all_pii

6.2.5 三层防御架构（2026 生产级方案）

2026 年生产系统中，PII 脱敏已演进为三层级联架构：

层级	方法	召回率	延迟	角色
第1层	正则+校验和(Checksum)	~92%	<1ms/doc	高精度快速过滤
第2层	DeBERTa-v3 NER	95-97%	~10ms/doc	高召回语义识别
第3层	LLM-Judge (gpt-4o-mini/Claude Haiku)	~99.4%(合并)	~500ms/span	歧义消解+跨句共指

三层合并后总体指标：召回 99.4%、误报 1.7%。第3层 LLM-judge 处理前两层都无法确定的歧义 span（如”张主任”需上下文判定是否为姓名）和跨句共指（如”他”指代上文出现的姓名）。

merge 层的关键操作：span 去重（同一位置的多层命中合并）→ 标签规范化（统一实体类型名）→ 置信度打分（多层交叉验证加权）→ 冲突解决（LLM-judge 优先，NER 次之，regex 为底线）。

从”检测精度分层”到”处置动作分层”：上面的三层架构描述的是检测手段的递进（regex → NER → LLM-judge），而生产系统还需要与之正交的第二个维度——按置信度决定处置动作（Presidio + gitleaks 实践中常见的 Tier1/Tier2/Tier3 分级）：

Tier	置信度来源	处置动作
Tier1	正则 + 多检测器一致命中（高置信）	直接自动脱敏
Tier2	单一 NER 或单一检测器命中（中置信）	送人工复核队列
Tier3	LLM-judge 或模糊匹配命中（低置信）	仅打标记，不脱敏

要点：“更高精度的检测器”与”更谨慎的处置策略”是两个正交维度——不能因为上了 LLM-judge 就把 Tier3 的低置信命中直接拿去脱敏。检测精度决定”能不能信”，处置动作决定”信了之后怎么办”。

6.3 Presidio 框架

Microsoft Presidio 是工业级 PII 检测与脱敏的开源框架，集成了正则+NER+多种脱敏策略。

6.3.1 基础用法

from presidio_analyzer import AnalyzerEngine, RecognizerRegistry
from presidio_anonymizer import AnonymizerEngine
from presidio_analyzer.predefined_recognizers import (
    EmailRecognizer, PhoneRecognizer, IpRecognizer,
    CreditCardRecognizer, SpacyRecognizer
)

# 初始化分析器
analyzer = AnalyzerEngine()

# 添加预定义识别器
# Presidio 内置了大量识别器
analyzer.registry.add_recognizer(EmailRecognizer())
analyzer.registry.add_recognizer(PhoneRecognizer())
analyzer.registry.add_recognizer(IpRecognizer())
analyzer.registry.add_recognizer(CreditCardRecognizer())

# 中文支持需要自定义识别器
from presidio_analyzer import Pattern, PatternRecognizer

# 中国身份证识别器
cn_id_pattern = Pattern(
    name='cn_id_card',
    regex=r'\b[1-9]\d{5}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx]\b',
    score=0.95
)
cn_id_recognizer = PatternRecognizer(
    supported_entity='CN_ID_CARD',
    patterns=[cn_id_pattern]
)
analyzer.registry.add_recognizer(cn_id_recognizer)

# 中文电话识别器
cn_phone_pattern = Pattern(
    name='cn_phone',
    regex=r'\b1[3-9]\d{9}\b',
    score=0.9
)
analyzer.registry.add_recognizer(
    PatternRecognizer(
        supported_entity='CN_PHONE',
        patterns=[cn_phone_pattern]
    )
)

# 分析文本
results = analyzer.analyze(
    text='我的邮箱是 zhangsan@gmail.com，电话 13812345678',
    language='zh',
    entities=['EMAIL_ADDRESS', 'CN_PHONE', 'CN_ID_CARD']
)

# 结果：[EmailRecognizer result, CN_PHONE result]

6.3.2 脱敏操作

from presidio_anonymizer import AnonymizerEngine
from presidio_anonymizer.entities import (
    AnonymizerConfig,
    OperatorConfig
)

anonymizer = AnonymizerEngine()

# 默认操作：替换为占位符
result = anonymizer.anonymize(
    text='我的邮箱是 zhangsan@gmail.com，电话 13812345678',
    analyzer_results=results,
    operators={
        'EMAIL_ADDRESS': OperatorConfig('replace', {'new_value': '<EMAIL>'}),
        'CN_PHONE': OperatorConfig('replace', {'new_value': '<PHONE>'}),
    }
)

# 结果："我的邮箱是 <EMAIL>，电话 <PHONE>"

6.4 脱敏策略

6.4.1 五种策略对比

策略	示例	适用场景
替换 Replace	13812345678 → <PHONE>	通用，推荐
哈希 Hash	zhangsan@gmail.com → a3b2c1…@<EMAIL_DOMAIN>	需要保持唯一性
泛化 Generalize	28岁 → 20-30岁	保留统计特征
删除 Delete	整段含 PII 的内容删除	极端敏感场景
可逆脱敏 Reverse	<EMAIL-7a3b> → LLM处理 → 还原真实邮箱	Agent/RAG需返回真实PII
差分隐私 DP	在统计结果上加噪	聚合数据

⚠️ 可逆脱敏的生产级失败模式（社区实践，需在生产落地前验证）：可逆脱敏（reversible/anonymize 后 unmask）看似优雅，实则有两个隐蔽陷阱——(1) 确定性映射反推风险：若同一 PII 值始终映射到同一占位符（如 zhangsan@gmail.com → <EMAIL-7a3b> 恒定），攻击者可通过对占位符做归并计数（同一占位符出现频次 = 该身份出现频次）反推原始身份，生产环境应加盐（salt）或使用随机化映射；(2) unmask 状态一致性静默失败：unmask 逻辑依赖脱敏时的密钥/状态，若两者不一致（如密钥轮换、状态未持久化）会静默失败——不报错却返回错误还原结果，需建立”脱敏↔还原”的一致性校验。

from presidio_anonymizer.entities import OperatorConfig

# 替换策略
OPERATORS_REPLACE = {
    'EMAIL_ADDRESS': OperatorConfig('replace', {'new_value': '<EMAIL>'}),
    'CN_PHONE': OperatorConfig('replace', {'new_value': '<PHONE>'}),
    'CN_ID_CARD': OperatorConfig('replace', {'new_value': '<ID>'}),
    'IP_ADDRESS': OperatorConfig('replace', {'new_value': '<IP>'}),
}

# 哈希策略（保留唯一性）
OPERATORS_HASH = {
    'EMAIL_ADDRESS': OperatorConfig('custom', {
        'lambda': lambda x: f'user_{hashlib.md5(x.encode()).hexdigest()[:8]}@<EMAIL>'
    }),
}

# 泛化策略（保留统计特征）
OPERATORS_GENERALIZE = {
    'AGE': OperatorConfig('custom', {
        'lambda': lambda age: f'{int(age)//10*10}-{int(age)//10*10+9}'
    }),
}

# 删除策略（最严格）
OPERATORS_DELETE = {
    'EMAIL_ADDRESS': OperatorConfig('delete', {}),
    'CN_PHONE': OperatorConfig('delete', {}),
    'CN_ID_CARD': OperatorConfig('delete', {}),
}

6.4.2 推荐策略组合

工业级推荐的”分层策略”：

# 工业级分层脱敏策略
PII_STRATEGY = {
    # 高风险 PII：直接删除（包含原文都不要留）
    'delete': ['CN_ID_CARD', 'PASSPORT', 'BANK_CARD', 'CREDIT_CARD', 'API_KEY'],
    
    # 中风险 PII：替换为占位符
    'replace': ['EMAIL_ADDRESS', 'CN_PHONE', 'IP_ADDRESS', 'PERSON'],
    
    # 低风险 PII：泛化处理
    'generalize': ['AGE', 'DATE_TIME', 'ADDRESS'],
}

6.4.3 差分隐私（高级）

差分隐私（Differential Privacy）通过添加噪声保护个体：

import numpy as np

def dp_aggregate(values, epsilon=1.0, sensitivity=1.0):
    """
    差分隐私聚合
    epsilon: 隐私预算，越小越隐私但准确性越差
    sensitivity: 函数敏感度
    """
    # Laplace 机制
    noise = np.random.laplace(0, sensitivity/epsilon, size=len(values))
    return np.array(values) + noise

# 例如：统计平均年龄，但保护个体
ages = [25, 30, 35, 40, 45]
true_mean = np.mean(ages)  # 35.0
dp_mean = np.mean(dp_aggregate(ages, epsilon=0.5))
# 输出可能是 35.8 或 34.2，但不暴露任何个体年龄

踩坑经验：差分隐私适合聚合统计，不适合单条文本。对单条文本，还是用替换/删除。

6.5 多维打标

打标（Tagging）是给每篇文档打上多维标签，为后续配比做准备。

6.5.1 推荐维度

DOCUMENT_TAGGING_SCHEMA = {
    # 基础维度
    'language': 'str',         # 语种: en, zh, code
    'source': 'str',           # 来源: cc, github, arxiv, wiki
    
    # 内容维度
    'domain': 'list[str]',     # 领域: [tech, science, finance, ...]
    'doc_type': 'str',         # 类型: article, tutorial, news, qa, code
    'difficulty': 'int',      # 难度 1-5
    'educational_value': 'int', # 教育价值 1-5（FineWeb-Edu）
    
    # 质量维度
    'knowledge_density': 'float',  # 知识密度 0-1
    'language_style': 'str',       # 风格: formal, casual, technical
    'audience': 'str',             # 受众: expert, general, children
    
    # 结构维度
    'length': 'int',               # 字符数
    'has_formulas': 'bool',        # 是否含数学公式
    'has_code': 'bool',            # 是否含代码
    'has_tables': 'bool',          # 是否含表格
    
    # 元数据
    'created_at': 'datetime',
    'toxicity_score': 'float',     # 毒性 0-1
    'pii_detected': 'bool',        # 是否检测到 PII
}

6.5.2 打标流程示例

def tag_document(text):
    """
    给单篇文档打多维标签
    """
    tags = {}
    
    # 1. 基础维度（已在前面环节计算）
    tags['language'] = detect_language(text)
    tags['length'] = len(text)
    
    # 2. 教育价值打分（FineWeb-Edu 分类器）
    tags['educational_value'] = fineweb_edu_classify(text)
    
    # 3. 难度评估
    tags['difficulty'] = estimate_difficulty(text)
    
    # 4. 领域分类（多标签）
    tags['domain'] = classify_domain(text)
    
    # 5. 类型分类
    tags['doc_type'] = classify_doc_type(text)
    
    # 6. 结构特征
    tags['has_formulas'] = bool(re.search(r'\$.*\$', text))
    tags['has_code'] = bool(re.search(r'```.*```', text, re.DOTALL))
    tags['has_tables'] = bool(re.search(r'\|.*\|.*\n.*\|.*\|', text))
    
    # 7. 毒性
    tags['toxicity_score'] = detect_toxicity(text)['toxicity']
    
    # 8. PII 检测
    pii_results = detect_pii(text)
    tags['pii_detected'] = len(pii_results) > 0
    
    return tags


def estimate_difficulty(text):
    """
    评估文本难度 1-5
    """
    # 简化版：基于句子长度、词汇复杂度、概念密度
    sentences = text.split('.')
    avg_sentence_len = np.mean([len(s.split()) for s in sentences])
    
    # 长难词比例
    words = text.split()
    long_words = sum(1 for w in words if len(w) > 8)
    long_word_ratio = long_words / max(len(words), 1)
    
    # 难度评分
    if avg_sentence_len < 10 or long_word_ratio < 0.05:
        return 1
    elif avg_sentence_len < 15 or long_word_ratio < 0.10:
        return 2
    elif avg_sentence_len < 20 or long_word_ratio < 0.15:
        return 3
    elif avg_sentence_len < 25 or long_word_ratio < 0.20:
        return 4
    else:
        return 5


def classify_domain(text, top_k=3):
    """
    多标签领域分类
    """
    # 用 zero-shot 分类器
    from transformers import pipeline
    classifier = pipeline('zero-shot-classification', model='facebook/bart-large-mnli')
    
    domains = [
        'technology', 'science', 'medicine', 'finance', 'law',
        'history', 'literature', 'arts', 'sports', 'politics',
        'education', 'travel', 'food', 'health', 'entertainment'
    ]
    
    result = classifier(text[:1024], domains, multi_label=True)
    
    # 取 top_k
    top_domains = []
    for label, score in zip(result['labels'], result['scores']):
        if score > 0.3 and len(top_domains) < top_k:
            top_domains.append(label)
    
    return top_domains

6.6 分类器训练：从种子标注到全量推理

工业级打标通常需要训练自定义分类器，流程如下：

6.6.1 三阶段训练流程

阶段1: 种子标注
  人工标注 1-5K 条数据
  ├─ 领域专家标注：高准确率
  └─ 众包标注：高吞吐
       ↓
阶段2: 弱监督扩展
  用规则/启发式自动标注 10-100K 条
  ├─ Snorkel 框架：多 labeling function 投票
  └─ Label Model：学习标注函数的可靠性
       ↓
阶段3: 全量推理
  用训练好的分类器对全量数据打标
  ├─ 1B-3B 小模型（中等吞吐）
  └─ fastText（极快吞吐）

6.6.2 弱监督示例（Snorkel）

# pip install snorkel
from snorkel.labeling import labeling_function, LFApplier, LabelModel

# 定义多个标注函数
@labeling_function()
def lf_contains_equation(text):
    """含 LaTeX 公式→可能是科学论文"""
    if '$' in text and ('\\' in text):
        return SCIENCE
    return ABSTAIN

@labeling_function()
def lf_contains_code(text):
    """含代码块→可能是技术文档"""
    if '```' in text or '    def ' in text:
        return TECH
    return ABSTAIN

@labeling_function()
def lf_long_sentences(text):
    """长句子多→可能是学术"""
    sentences = text.split('.')
    if np.mean([len(s.split()) for s in sentences]) > 25:
        return ACADEMIC
    return ABSTAIN

@labeling_function()
def lf_short_sentences(text):
    """短句子多→可能是新闻"""
    sentences = text.split('.')
    if np.mean([len(s.split()) for s in sentences]) < 10:
        return NEWS
    return ABSTAIN

# 应用所有 labeling functions
lfs = [lf_contains_equation, lf_contains_code, lf_long_sentences, lf_short_sentences]
applier = LFApplier(lfs)
L_train = applier.apply(data)  # 生成标签矩阵

# 训练 Label Model 学习每个 LF 的可靠性
label_model = LabelModel(cardinality=5, verbose=True)
label_model.fit(L_train, n_epochs=500)

# 得到伪标签
predictions = label_model.predict(L_train)

6.6.3 训练自定义分类器

from sklearn.model_selection import train_test_split
from sklearn.metrics import classification_report
import lightgbm as lgb

def train_domain_classifier(texts, labels):
    """
    训练领域分类器
    """
    # Step 1: 特征提取（用 sentence embedding）
    from sentence_transformers import SentenceTransformer
    encoder = SentenceTransformer('all-MiniLM-L6-v2')
    embeddings = encoder.encode(texts, show_progress_bar=True)
    
    # Step 2: 划分训练/验证
    X_train, X_val, y_train, y_val = train_test_split(
        embeddings, labels, test_size=0.2, stratify=labels
    )
    
    # Step 3: 训练 LightGBM
    clf = lgb.LGBMClassifier(
        n_estimators=200,
        learning_rate=0.05,
        num_leaves=64,
        max_depth=8,
        class_weight='balanced'
    )
    clf.fit(X_train, y_train)
    
    # Step 4: 评估
    y_pred = clf.predict(X_val)
    print(classification_report(y_val, y_pred))
    
    return clf

# Step 5: 全量推理（用 Ray 并行）
import ray

@ray.remote(num_gpus=0.2)
def batch_classify(texts_batch, classifier):
    """批量分类"""
    embeddings = encoder.encode(texts_batch)
    return classifier.predict(embeddings)

def tag_all_documents(documents, classifier):
    """全量打标"""
    batches = [documents[i:i+1000] for i in range(0, len(documents), 1000)]
    futures = [batch_classify.remote(b, classifier) for b in batches]
    results = ray.get(futures)
    return [item for batch in results for item in batch]

6.7 打标质量保障

6.7.1 质量检查清单

def validate_tags(doc):
    """验证打标质量"""
    issues = []
    
    # 1. 必填字段
    required = ['language', 'domain', 'doc_type', 'difficulty']
    for field in required:
        if field not in doc or doc[field] is None:
            issues.append(f'missing_{field}')
    
    # 2. 字段类型
    if not isinstance(doc.get('difficulty'), int):
        issues.append('difficulty_not_int')
    if not (1 <= doc.get('difficulty', 3) <= 5):
        issues.append('difficulty_out_of_range')
    
    # 3. 一致性
    if doc.get('language') == 'zh' and doc.get('domain') == ['english_literature']:
        issues.append('language_domain_mismatch')
    
    # 4. 异常值
    if doc.get('length', 0) > 1_000_000:
        issues.append('length_too_large')
    if doc.get('toxicity_score', 0) > 0.7:
        issues.append('high_toxicity_not_filtered')
    
    return issues

6.8 踩坑经验

坑1：PII 检测召回不够

某团队只用了正则，结果漏掉了大量”姓名+地址”组合（无法用正则匹配）。必须双路：正则+NER。

坑2：脱敏后破坏了原文结构

某团队把所有 PII 都替换成 <PII>，结果训练数据里满是 <PII>，模型学到的是”生成 <PII>”。每种 PII 用不同占位符：<EMAIL>、<PHONE>，让模型学会区分。

坑3：分类器训练数据偏置

某团队训练领域分类器时，“科学”类标注全是 arXiv，结果把所有含公式的文档都判为”科学”——包括代码（含 LaTeX 注释）。标注数据要覆盖各种来源和风格。

坑4：打标维度过多导致冗余

某团队定义了 50 个标签维度，但发现 80% 的维度高度相关。推荐用 10-15 个核心维度，PCA 分析后再删冗余。

坑5：标签更新不向后兼容

某团队上线后发现 domain 字段需要新增 'agriculture'，但已有数据没有这个标签。标签 schema 必须支持版本化，且更新时全量重打标。

6.9 工业级流水线

class TaggingPipeline:
    """工业级打标流水线"""
    
    def __init__(self):
        self.pii_detector = PresidioPIIDetector()
        self.langid = fasttext.load_model('lid.176.bin')
        self.edu_classifier = load_fineweb_edu()
        self.domain_classifier = train_domain_classifier(...)  # LightGBM
        self.toxicity_detector = Detoxify('multilingual')
    
    def process(self, doc):
        text = doc['text']
        
        # Step 1: PII 检测与脱敏
        pii_results = self.pii_detector.detect(text)
        text = self.pii_detector.anonymize(text, pii_results)
        
        # Step 2: 语种识别
        lang = self.langid.predict(text.replace('\n', ' '))
        
        # Step 3: 教育价值打分
        edu_score = self.edu_classifier(text[:4096])
        
        # Step 4: 领域分类
        domain = self.domain_classifier([text])
        
        # Step 5: 毒性检测
        toxicity = self.toxicity_detector.predict(text)
        
        # 输出
        doc.update({
            'text': text,  # 脱敏后
            'language': lang,
            'edu_score': edu_score,
            'domain': domain,
            'toxicity': toxicity,
            'tags_computed_at': datetime.now()
        })
        
        return doc

6.10 本课总结

PII 类目：邮箱/电话/身份证/银行卡/IP/API Key 等 12+ 类
双路检测：正则（固定格式）+ NER（自然语言）+ 密钥扫描
Presidio 框架：Microsoft 开源，工业级事实标准
五种脱敏策略：替换（推荐）/ 哈希 / 泛化 / 删除 / 差分隐私
多维打标：基础/内容/质量/结构/元数据 5 大类标签
分类器训练：种子标注 → 弱监督（Snorkel）→ 全量推理

课后练习

用 Presidio 跑一遍你的数据，统计 PII 分布
实现一个自定义的中国身份证识别器
用 Snorkel 框架做一次弱监督打标
训练一个领域分类器，评估其准确率

下一章我们将进入数据混合配比——预训练数据工程的核心环节，配比决定模型的”性格”。