案例 14：BERT 短信分类：词袋、冻结编码器与微调边界

学习目标与记号

　　双向 Transformer 编码器表示（Bidirectional Encoder Representations from Transformers，BERT）先通过大规模无标签文本进行预训练，再使用带标签的数据适配具体任务。本案例把 BERT 放在“简单基线—预训练表示—任务微调”的实验框架中，而不是把调用预训练模型当作全部分析。默认路径只运行 TF-IDF 逻辑回归，使普通中央处理器（central processing unit，CPU）和离线教学环境也能完成。本案例的学习目标包括：

1. 解释 WordPiece 如何把未登录词拆为可复用子词，并理解它与简单按词切分的差异；

2. 说明 [CLS]、[SEP]、句段编号、位置向量和注意力掩码的作用；

3. 理解 BERT 的双向编码与掩码语言模型（MLM）预训练目标；

4. 区分 TF-IDF 基线、冻结 BERT 特征和端到端微调所改变的参数范围；

5. 使用分层划分、平均精确率（Average Precision，AP）、宏平均 F1 分数和阈值分析处理类别不平衡；

6. 认识预训练模型的版本、许可证、网络下载、算力和域漂移边界；

7. 避免把注意力权重、[CLS] 表示或较高的预测概率直接当作可靠解释。

　　该案例是算法教学，不应直接用于过滤个人通信。

数据来源与安全说明

• UCI SMS Spam Collection：https://archive.ics.uci.edu/dataset/228/sms%2Bspam%2Bcollection

• 匿名直链：https://archive.ics.uci.edu/static/public/228/sms+spam+collection.zip

• 数据许可：CC BY 4.0。

　　短信可能包含电话号码、网址、诱导性内容或真实个人表达。课程页面不应展示未遮蔽的原文。下载文件缓存在 AI_COURSE_DATA_DIR，未设置时落到用户缓存目录，而不是源码树。

　　本例固定采用分层划分。TF-IDF 只在训练文本上拟合；BERT 词元化器使用预训练阶段已有的固定词表，不从本任务的验证文本或测试文本学习新词。模型与阈值都必须根据验证集确定，测试集只用于最后一次评价。

BERT 模型资源与可选执行

　　可选部分默认采用 google-bert/bert-base-uncased：

• 模型说明：https://huggingface.co/google-bert/bert-base-uncased

• 这是英文、uncased、12 层编码器；模型卡列出 Apache-2.0。

• 模型文件由 Transformers 库匿名下载到用户指定缓存，可能需要数百 MB 网络流量。

• 为保证默认可运行，RUN_BERT 缺省为 0。仅当环境已安装 transformers 且用户明确设置 RUN_BERT=1 时执行。

• BERT_MODE=frozen 只提取固定 [CLS] 表示并训练浅层分类器；BERT_MODE=finetune 更新编码器参数，建议 GPU。

• 快速模式限制样本、序列长度、批量和轮数。所得数值用于验证流程，不是排行榜结果。

　　在服务器上部署时，最好预先在构建环境中缓存并固定模型版本，再让隔离的执行环境离线读取；不要让公开网页服务在用户请求时临时下载模型。

from __future__ import annotations

import os
import random
import re
import urllib.request
import zipfile
from pathlib import Path

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
    accuracy_score, average_precision_score, f1_score,
    precision_recall_curve,
)
from sklearn.model_selection import train_test_split

SEED = 20260812
FAST_MODE = os.getenv("FAST_MODE", "1") != "0"
RUN_BERT = os.getenv("RUN_BERT", "0") == "1"
BERT_MODE = os.getenv("BERT_MODE", "frozen").lower()
if BERT_MODE not in {"frozen", "finetune"}:
    raise ValueError("BERT_MODE 必须是 frozen 或 finetune")

random.seed(SEED)
np.random.seed(SEED)
CACHE_ROOT = Path(os.getenv("AI_COURSE_DATA_DIR", Path.home() / ".cache" / "ai-course"))
DATA_DIR = CACHE_ROOT / "uci_sms_spam"
MODEL_CACHE = CACHE_ROOT / "huggingface"
DATA_DIR.mkdir(parents=True, exist_ok=True)
MODEL_CACHE.mkdir(parents=True, exist_ok=True)
print({
    "快速模式": FAST_MODE, "是否运行 BERT": RUN_BERT,
    "BERT 运行方式": {"frozen": "冻结特征", "finetune": "微调"}[BERT_MODE],
    "数据缓存目录": str(DATA_DIR), "模型缓存目录": str(MODEL_CACHE),
})

{'快速模式': True, '是否运行 BERT': False, 'BERT 运行方式': '冻结特征', '数据缓存目录': '/private/tmp/ai-course-case-data/uci_sms_spam', '模型缓存目录': '/private/tmp/ai-course-case-data/huggingface'}

URL = "https://archive.ics.uci.edu/static/public/228/sms+spam+collection.zip"
ZIP_PATH = DATA_DIR / "sms_spam_collection.zip"
if not ZIP_PATH.exists():
    urllib.request.urlretrieve(URL, ZIP_PATH)
assert ZIP_PATH.stat().st_size > 100_000

with zipfile.ZipFile(ZIP_PATH) as zf:
    member = next(n for n in zf.namelist() if Path(n).name == "SMSSpamCollection")
    payload = zf.read(member).decode("utf-8", errors="replace")

records = []
for line in payload.splitlines():
    if line.strip():
        label, message = line.split("\t", 1)
        records.append((message, 1 if label == "spam" else 0))
df = pd.DataFrame(records, columns=["text", "label"])
assert len(df) >= 5_500 and set(df.label) == {0, 1}
assert df.text.str.len().gt(0).all()
print({"数据维度": df.shape, "正常短信数": int((df.label == 0).sum()), "垃圾短信数": int((df.label == 1).sum())})

{'数据维度': (5574, 2), '正常短信数': 4827, '垃圾短信数': 747}

def redact(text):
    text = re.sub(r"https?://\S+|www\.\S+", "<URL>", text, flags=re.I)
    text = re.sub(r"\b\d[\d\s().+-]{4,}\d\b", "<NUM>", text)
    return text[:120]

eda = df.assign(characters=df.text.str.len())
eda_summary = eda.groupby("label").characters.describe(percentiles=[0.5, 0.9, 0.99])
eda_summary = eda_summary.rename(
    index={0: "正常短信", 1: "垃圾短信"},
    columns={"count": "数量", "mean": "均值", "std": "标准差", "min": "最小值",
             "50%": "中位数", "90%": "百分之九十分位数", "99%": "百分之九十九分位数", "max": "最大值"},
)
eda_summary.index.name = "类别"
display(eda_summary)
print("遮蔽后的正常示例：", redact(df.loc[df.label == 0, "text"].iloc[0]))
print("遮蔽后的垃圾示例：", redact(df.loc[df.label == 1, "text"].iloc[0]))

for label, label_name in [(0, "正常短信"), (1, "垃圾短信")]:
    df.loc[df.label == label].assign(characters=lambda x: x.text.str.len()).characters.plot(
        kind="hist", bins=40, alpha=0.5, density=True, label=label_name
    )
plt.xlabel("字符数")
plt.ylabel("密度")
plt.legend(title="类别")
plt.title("消息长度分布：长度本身可能成为捷径")
plt.tight_layout()

	数量	均值	标准差	最小值	中位数	百分之九十分位数	百分之九十九分位数	最大值
类别
正常短信	4827.0	71.471929	58.326643	2.0	52.0	147.0	288.48	910.0
垃圾短信	747.0	138.676037	28.871250	13.0	149.0	160.0	174.08	223.0

遮蔽后的正常示例： Go until jurong point, crazy.. Available only in bugis n great world la e buffet... Cine there got amore wat...
遮蔽后的垃圾示例： Free entry in 2 a wkly comp to win FA Cup final tkts 21st May 2005. Text FA to 87121 to receive entry question(std txt r

train_df, temp_df = train_test_split(
    df, test_size=0.30, random_state=SEED, stratify=df.label
)
valid_df, test_df = train_test_split(
    temp_df, test_size=0.50, random_state=SEED, stratify=temp_df.label
)
train_df, valid_df, test_df = [
    part.reset_index(drop=True) for part in (train_df, valid_df, test_df)
]
assert len(train_df) + len(valid_df) + len(test_df) == len(df)
split_display = pd.DataFrame({
    "样本数": [len(train_df), len(valid_df), len(test_df)],
    "垃圾短信比例": [train_df.label.mean(), valid_df.label.mean(), test_df.label.mean()],
}, index=["训练集", "验证集", "测试集"])
split_display.index.name = "数据划分"
display(split_display)

	样本数	垃圾短信比例
数据划分
训练集	3901	0.134068
验证集	836	0.133971
测试集	837	0.133811

默认强基线：TF-IDF 逻辑回归

　　文本任务中的复杂模型必须与合理的简单模型比较。TF-IDF 把训练文本中出现的词和相邻双词组合转换为大部分元素为 0 的向量；逆文档频率会降低常见词的权重。逻辑回归直接给出短信属于垃圾类的概率，训练较快、容易检查，也能查看权重较大的词元特征。

　　防止信息泄漏的关键是：fit_transform 只接收训练文本，验证文本和测试文本只调用 transform。若先在全部文本上拟合词表与 IDF，测试集的词语分布就已经影响输入表示；即使没有直接使用测试标签，也属于信息泄漏。

　　类别不平衡时，准确率容易被样本较多的正常短信类别主导，因此同时报告平均精确率和宏平均 F1 分数。实际使用的阈值应根据误判代价和验证集确定；不能使用测试集挑选阈值。

tfidf = TfidfVectorizer(
    lowercase=True,
    ngram_range=(1, 2),
    min_df=2,
    max_features=30_000,
    sublinear_tf=True,
)
X_train = tfidf.fit_transform(train_df.text)
X_valid = tfidf.transform(valid_df.text)
X_test = tfidf.transform(test_df.text)
baseline = LogisticRegression(
    max_iter=1_000, class_weight="balanced", random_state=SEED
)
baseline.fit(X_train, train_df.label)
valid_prob = baseline.predict_proba(X_valid)[:, 1]
test_prob = baseline.predict_proba(X_test)[:, 1]

def metrics(y, prob, threshold=0.5):
    pred = (prob >= threshold).astype(int)
    return {
        "accuracy": accuracy_score(y, pred),
        "macro_f1": f1_score(y, pred, average="macro"),
        "average_precision": average_precision_score(y, prob),
    }

baseline_metrics = metrics(test_df.label.to_numpy(), test_prob)
print("TF-IDF 测试结果：", {"准确率": baseline_metrics["accuracy"], "宏平均 F1": baseline_metrics["macro_f1"], "平均精确率": baseline_metrics["average_precision"]})
assert X_train.shape[1] == len(tfidf.vocabulary_)
assert np.isfinite(list(baseline_metrics.values())).all()

TF-IDF 测试结果： {'准确率': 0.982078853046595, '宏平均 F1': 0.9611988639348772, '平均精确率': 0.9787955376403855}

# 示例：用验证集选择“precision 至少 0.90 时 recall 最大”的阈值。
precision, recall, thresholds = precision_recall_curve(valid_df.label, valid_prob)
valid_candidates = np.flatnonzero(precision[:-1] >= 0.90)
if len(valid_candidates):
    best_idx = valid_candidates[np.argmax(recall[:-1][valid_candidates])]
    selected_threshold = float(thresholds[best_idx])
else:
    selected_threshold = 0.5
threshold_metrics = metrics(test_df.label.to_numpy(), test_prob, selected_threshold)
print({"由验证集选定的阈值": selected_threshold, "测试集指标": {
    "准确率": threshold_metrics["accuracy"], "宏平均 F1": threshold_metrics["macro_f1"],
    "平均精确率": threshold_metrics["average_precision"],
}})
print("阈值规则在查看测试指标前已经固定；不能因测试结果不满意再修改。")

{'由验证集选定的阈值': 0.44574868427845216, '测试集指标': {'准确率': 0.982078853046595, '宏平均 F1': 0.9617772148969919, '平均精确率': 0.9787955376403855}}
阈值规则在查看测试指标前已经固定；不能因测试结果不满意再修改。

BERT 输入与预训练目标

　　BERT 词元化器通常先对文本进行基本整理，再用 WordPiece 把词拆为子词。例如，一个少见词可能被拆成常见词根和带 ## 前缀的后续片段。固定子词表让模型能够表示预训练时没有见过的完整单词，但这种拆分不一定符合语言学中的词素分析。

　　单句分类输入通常为：

\[ [\mathrm{CLS}],\ t_1,\ldots,t_m,\ [\mathrm{SEP}],\ [\mathrm{PAD}],\ldots . \]

　　每个位置的输入表示由词元向量、位置向量和句段向量相加。注意力掩码中，1 表示有效词元，0 表示填充位置；它使填充位置不能作为有效上下文参与注意力计算。分类时通常使用最后一层的 [CLS] 向量或模型汇总后的表示，并把它输入分类输出层。

　　BERT 的预训练核心是掩码语言模型：随机遮蔽一部分词元，根据左右上下文恢复原词。它是双向编码器，并非按从左到右生成文本。原始 BERT 还使用下一句预测目标，但后续预训练变体可能改变或删除该目标。因此使用具体检查点时要看模型卡，而不是把“BERT”当成单一固定配方。

　　冻结模式把预训练编码器视为特征函数，只训练任务分类器；微调则让短信损失更新编码器。后者表达能力更强，但更慢、更依赖学习率，也更可能在小数据上不稳定。

MODEL_NAME = "google-bert/bert-base-uncased"

if RUN_BERT:
    try:
        import torch
        from torch import nn
        from torch.utils.data import DataLoader, Dataset
        from transformers import AutoModel, AutoTokenizer
    except ImportError as exc:
        raise ImportError(
            "RUN_BERT=1 需要安装 torch 与 transformers；默认 TF-IDF 路径不需要。"
        ) from exc

    torch.manual_seed(SEED)
    DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, cache_dir=MODEL_CACHE)
    token_demo = tokenizer.tokenize("Congratulations! claim your unbelievable prize")
    print("WordPiece 分词示例：", token_demo)

    encoded_demo = tokenizer(
        ["hello world", "a longer message for masking"],
        padding=True, truncation=True, max_length=32, return_tensors="pt"
    )
    assert encoded_demo["input_ids"].shape == encoded_demo["attention_mask"].shape
    assert encoded_demo["input_ids"].shape[0] == 2
    output_key_labels = {"input_ids": "词元编号", "attention_mask": "注意力掩码", "token_type_ids": "句段编号"}
    print({output_key_labels.get(k, "其他模型输入"): tuple(v.shape) for k, v in encoded_demo.items()})
else:
    print("未启用可选的预训练模型路径：跳过模型下载；TF-IDF 基线已经完整可运行。")

未启用可选的预训练模型路径：跳过模型下载；TF-IDF 基线已经完整可运行。

可选实验 A：冻结 BERT 表示

　　冻结模式不计算编码器梯度。对每条短信得到 [CLS] 表示后，再用逻辑回归分类。它能回答一个具体问题：预训练表示在不适配编码器参数的情况下，是否比任务内 TF-IDF 更有用？

　　为防止内存暴涨，按批量编码，并立即把表示移回 CPU。特征缓存不写进源码树。训练、验证和测试仍沿用之前的固定划分。注意 [CLS] 不是天然的语义解释，它只是约定位置的上下文化表示。

bert_results = []
if RUN_BERT and BERT_MODE == "frozen":
    encoder = AutoModel.from_pretrained(MODEL_NAME, cache_dir=MODEL_CACHE).to(DEVICE)
    encoder.eval()
    for parameter in encoder.parameters():
        parameter.requires_grad_(False)

    def frozen_features(texts, batch_size=32):
        chunks = []
        with torch.no_grad():
            for start in range(0, len(texts), batch_size):
                batch = tokenizer(
                    texts[start:start+batch_size],
                    padding=True, truncation=True,
                    max_length=64 if FAST_MODE else 128,
                    return_tensors="pt",
                )
                batch = {k: v.to(DEVICE) for k, v in batch.items()}
                output = encoder(**batch).last_hidden_state[:, 0, :]
                chunks.append(output.cpu().numpy())
        features = np.concatenate(chunks)
        assert features.shape == (len(texts), encoder.config.hidden_size)
        return features

    fit_train = train_df.iloc[:1_500] if FAST_MODE else train_df
    frozen_train = frozen_features(fit_train.text.tolist())
    frozen_valid = frozen_features(valid_df.text.tolist())
    frozen_test = frozen_features(test_df.text.tolist())
    frozen_clf = LogisticRegression(
        max_iter=1_000, class_weight="balanced", random_state=SEED
    )
    frozen_clf.fit(frozen_train, fit_train.label)
    frozen_prob = frozen_clf.predict_proba(frozen_test)[:, 1]
    frozen_metrics = metrics(test_df.label.to_numpy(), frozen_prob)
    bert_results.append({"model": "frozen_bert_cls", **frozen_metrics})
    print("冻结 BERT 特征测试结果：", {"准确率": frozen_metrics["accuracy"], "宏平均 F1": frozen_metrics["macro_f1"], "平均精确率": frozen_metrics["average_precision"]})
elif RUN_BERT:
    print("当前选择微调方式，因此跳过冻结特征单元。")

可选实验 B：端到端微调

　　微调把编码器和分类输出层作为一个模型。每个批次输入的维度为 \(B\times T\)，编码器输出的维度为 \(B\times T\times768\)，分类输出层读取 [CLS] 位置，并得到维度为 \(B\times2\) 的线性运算结果。交叉熵直接接收这个结果，不应预先计算 Softmax。

　　快速模式只使用按类别比例选出的少量样本，最大序列长度为 64，并只训练很少的轮次；完整模式仍需根据显存大小调整每批样本数。真实项目还应使用学习率调整、验证集提前停止、混合精度，并固定模型版本。由于这是需要明确开启的可选路径，默认运行不会下载模型或占用 GPU。

if RUN_BERT and BERT_MODE == "finetune":
    class BertSMSDataset(Dataset):
        def __init__(self, frame, max_length):
            self.texts = frame.text.tolist()
            self.labels = frame.label.to_numpy(np.int64)
            self.max_length = max_length
        def __len__(self):
            return len(self.labels)
        def __getitem__(self, idx):
            item = tokenizer(
                self.texts[idx], truncation=True, padding="max_length",
                max_length=self.max_length, return_tensors="pt"
            )
            return (
                item["input_ids"].squeeze(0),
                item["attention_mask"].squeeze(0),
                torch.tensor(self.labels[idx], dtype=torch.long),
            )

    class BertClassifier(nn.Module):
        def __init__(self):
            super().__init__()
            self.encoder = AutoModel.from_pretrained(MODEL_NAME, cache_dir=MODEL_CACHE)
            self.dropout = nn.Dropout(0.1)
            self.head = nn.Linear(self.encoder.config.hidden_size, 2)
        def forward(self, input_ids, attention_mask):
            hidden = self.encoder(
                input_ids=input_ids, attention_mask=attention_mask
            ).last_hidden_state
            return self.head(self.dropout(hidden[:, 0]))

    if FAST_MODE:
        fit_train, _ = train_test_split(
            train_df, train_size=1_200, random_state=SEED, stratify=train_df.label
        )
    else:
        fit_train = train_df
    train_ds = BertSMSDataset(fit_train.reset_index(drop=True), 64 if FAST_MODE else 128)
    test_ds = BertSMSDataset(test_df, 64 if FAST_MODE else 128)
    generator = torch.Generator().manual_seed(SEED)
    train_loader = DataLoader(train_ds, batch_size=8, shuffle=True, generator=generator)
    test_loader = DataLoader(test_ds, batch_size=16, shuffle=False)

    bert_model = BertClassifier().to(DEVICE)
    optimizer = torch.optim.AdamW(bert_model.parameters(), lr=2e-5)
    for epoch in range(1 if FAST_MODE else 3):
        bert_model.train()
        for ids, mask, y in train_loader:
            ids, mask, y = ids.to(DEVICE), mask.to(DEVICE), y.to(DEVICE)
            optimizer.zero_grad()
            logits = bert_model(ids, mask)
            assert logits.shape == (len(y), 2)
            loss = nn.functional.cross_entropy(logits, y)
            assert torch.isfinite(loss)
            loss.backward()
            nn.utils.clip_grad_norm_(bert_model.parameters(), 1.0)
            optimizer.step()

    bert_model.eval()
    probs = []
    with torch.no_grad():
        for ids, mask, _ in test_loader:
            logits = bert_model(ids.to(DEVICE), mask.to(DEVICE))
            probs.append(torch.softmax(logits, dim=1)[:,1].cpu().numpy())
    finetune_prob = np.concatenate(probs)
    finetune_metrics = metrics(test_df.label.to_numpy(), finetune_prob)
    bert_results.append({"model": "finetuned_bert", **finetune_metrics})
    print("BERT 微调测试结果：", {"准确率": finetune_metrics["accuracy"], "宏平均 F1": finetune_metrics["macro_f1"], "平均精确率": finetune_metrics["average_precision"]})

rows = [{"model": "tfidf_logistic", **baseline_metrics}] + bert_results
results = pd.DataFrame(rows).set_index("model")
results_display = results.rename(
    index={"tfidf_logistic": "TF-IDF 逻辑回归", "frozen_bert_cls": "冻结 BERT 分类表示",
           "finetuned_bert": "微调 BERT"},
    columns={"accuracy": "准确率", "macro_f1": "宏平均 F1", "average_precision": "平均精确率"},
)
results_display.index.name = "模型"
display(results_display)
assert np.isfinite(results.to_numpy()).all()
if not RUN_BERT:
    print(
        "默认结果只包含 TF-IDF，这是有意设计的完整基线。"
        "设置 RUN_BERT=1 后，才将对应可选结果加入同一张表。"
    )

	准确率	宏平均 F1	平均精确率
模型
TF-IDF 逻辑回归	0.982079	0.961199	0.978796

默认结果只包含 TF-IDF，这是有意设计的完整基线。设置 RUN_BERT=1 后，才将对应可选结果加入同一张表。

局限、伦理与结果解释

　　预训练域差异。 BERT 的预训练语料、年代和写作风格与短信不同。它可能善于一般英语，却不了解新型诈骗缩写、短链或本地语言。一次随机划分不能评价时间漂移。

　　许可证分层。 短信数据与预训练模型有各自许可证，课程发布需要分别保留来源和条款。模型代码许可证不能自动覆盖训练数据，也不能反过来。

　　默认不下载的原因。 预训练权重体积大、网络和算力要求高。让默认 Notebook 依赖它会降低可复现性。显式开关使读者先得到可运行基线，再自行决定是否承担下载和训练成本。

　　冻结与微调不是简单优劣。 冻结更便宜、参数更新少；微调可能更准，却更易过拟合且需要更仔细的学习率和早停。公平比较应报告参数量、实际更新参数量、训练时间、峰值显存和多个随机种子。

　　概率不等于保证。 未校准的 0.99 不是“99% 确定”。实际过滤器需要验证集校准、代价敏感阈值、人工申诉和持续漂移监控。误拦正常短信可能造成严重后果。

　　隐私。 不应将原始个人通信记录到实验追踪服务或公开输出；错误分析应遮蔽号码和网址。模型也可能记住罕见字符串。

　　解释边界。 注意力权重、梯度或词权重只是检查模型的线索。应结合遮蔽、反事实样本、同义替换和不同样本组的错误分析进行核验，不能把这些量直接称为人类可以接受的决定理由。

　　任务外推。 英文短消息结果不能直接推广到中文、多语、邮件或长文档；不同语言的词元化器所覆盖的词元以及拆分后的序列长度也会明显不同。

综合练习

1. 不运行模型，手工写出句子 “claim your prize” 加入特殊词元并补齐长度后应有哪些输入数组，并解释每个数组的维度。

2. 选择三个罕见复合词，比较简单空格分词与 WordPiece；分析子词拆分如何影响序列长度。

3. 分别运行冻结与微调模式，确定相同的数据划分和随机种子，报告需要更新的参数量、训练时间、峰值内存、平均精确率与宏平均 F1 分数。

4. 只用验证集选择满足精确率下限的阈值，再在测试集上评价一次；与阈值 0.5 比较误判情况。

5. 分别按照短信长度、是否含网址和是否含数字报告召回率，并讨论模型是否过度依赖这些表面特征。

6. 对测试消息做号码替换、货币替换和同义改写，比较概率变化；不要公开原始联系方式。

7. 固定模型版本，并编写离线运行检查：缓存齐全时设置 Transformers 离线模式，确认程序不会访问网络。