用条件信息增益筛选 SFT 数据：从打分到后训练

这套流程能做出什么

跑完本文的步骤，你会得到三样东西：

1. 一个可复用的打分脚本：读入原始 SFT 数据（JSONL），为每条样本算出「条件信息增益」分数，输出带分数的数据文件。

2. 一份基于分数的筛选策略：按预算保留高增益样本，同时用去重和配额保住数据多样性。

3. 一个带对照组的后训练实验：同一基座模型、同一超参、同一 token 预算下，对比「随机等量采样」与「按增益筛选」的效果，并给出评测结论。

核心直觉很简单：一条 SFT 样本的价值，不在于它有多难，而在于它提供了多少新信息。如果模型不看输入也能把答案说出来，这条样本对于「学会按输入作答」这件事几乎没有贡献。

举两个具体例子：

• 样本 A：指令是「把下面这句话翻译成英文」，答案是「好的，我这就帮你翻译。」——不看指令，模型也能生成这句套话，增益接近 0，甚至是负的。
• 样本 B：指令里贴了一段服务日志，答案是「连接池耗尽，建议把 max_connections 调到 200 并加上超时重试。」——只有读了那段日志才可能写出这个答案，增益高。

条件信息增益就是把上面的判断变成可计算的数字：在答案 y 上，看到条件 x 之后的平均对数似然，减去没有看到 x 时的平均对数似然。

前置条件清单

开始之前，准备好这些：

• Python 环境：Python 3.x、PyTorch、transformers。具体版本以官方文档当前版本为准。
• 一个基座模型：必须是没有在这份数据上微调过的原始模型。它可以是你要训练的那个基座本身，也可以是同族的另一个模型（注意规模差异带来的迁移问题，见「常见坑」）。模型体积要能装进你的显存。
• 一份原始 SFT 数据：JSONL 格式，每条至少能拆出「输入条件」和「答案」两部分。常见字段有 instruction / input / output，或者 messages 数组。
• 一份评测集：和目标能力相关，且不能出现在训练数据里。另外准备一份通用能力回归集，用来观察是否出现能力退化。
• 一个训练框架：transformers Trainer、TRL、LLaMA-Factory 等都可以，按官方文档当前版本的用法来。
• 一张可用的 GPU：打分阶段每条样本需要两次前向，显存需求与推理相同。

如果打分模型只能通过 API 调用，而接口不返回 token 级 logprob，就只能得到近似值，下面的公式需要相应调整。建议优先用本地模型。

第 1 步：把数据整理成可比的三段式

先把来源各异的字段统一成 {ctx, query, answer}：ctx 是检索文档、系统提示、多轮历史这类附加上下文，query 是具体问题，answer 是目标回答。

```python

01_prepare.py

import json

def normalize(rec):

if "messages" in rec:

msgs = rec["messages"]

answer = msgs[-1]["content"]

ctx = "\n".join(m["content"] for m in msgs[:-1] if m["role"] == "system")

query = "\n".join(m["content"] for m in msgs[:-1] if m["role"] != "system")

else:

ctx = (rec.get("context") or "").strip()

query = (rec.get("instruction", "") + "\n" + rec.get("input", "")).strip()

answer = rec["output"]

return {"ctx": ctx, "query": query, "answer": answer}

with open("raw.jsonl", encoding="utf-8") as fi, open("norm.jsonl", "w", encoding="utf-8") as fo:

for line in fi:

line = line.strip()

if not line:

continue

try:

r = normalize(json.loads(line))

except Exception:

continue

if r["answer"].strip():

fo.write(json.dumps(r, ensure_ascii=False) + "\n")

```

关键点：打分用的模板和训练用的模板必须一致。如果打分时用了模型自带的 chat template，训练时又手写了一套，两边算出来的 logprob 不在同一个尺度上，分数就没有意义。

第 2 步：定义并计算条件信息增益

可落地的公式：

```

CIG(x, y) = [ log P(y | x) - log P(y | x_ref) ] / |y|

```

其中：

• log P(y | x)：给定完整输入条件时，模型对答案的对数似然。
• log P(y | x_ref)：把输入条件「消融」掉之后的参考对数似然。
• |y|：答案的 token 数，用来做长度归一化。

x_ref 怎么构造，取决于样本形态：

• 有上下文 ctx 的样本：x_ref 就是保留 query、去掉 ctx。这衡量的是「那段文档到底带来了多少信息」。
• 只有单轮指令的样本：x_ref 用一个与任务无关的短前缀加 query，用来估计「不看具体问题、只靠通用套路能写出多少」。这个参考条件的构造方式可以换，但全程必须保持一致。

下面是打分脚本：

```python

02_score.py

import argparse, json, torch

from transformers import AutoModelForCausalLM, AutoAI 词典：Token">Tokenizer

parser = argparse.ArgumentParser()

parser.add_argument("--model", required=True) # 基座模型路径，名称以官方文档为准

parser.add_argument("--data", required=True)

parser.add_argument("--out", required=True)

parser.add_argument("--max-len", type=int, default=2048)

args = parser.parse_args()

tok = AutoTokenizer.from_pretrained(args.model)

model = AutoModelForCausalLM.from_pretrained(

args.model, torch_dtype=torch.bfloat16, device_map="auto"

).eval()

GENERIC = "请回答下面的内容。"

def encode(prompt, answer):

p = tok(prompt, add_special_tokens=False)["input_ids"]

a = tok(answer, add_special_tokens=False)["input_ids"] + [tok.eos_token_id]

return p + a, len(p), len(a)

@torch.no_grad()

def logprob(prompt, answer):

ids, start, n = encode(prompt, answer)

if start == 0 or len(ids) > args.max_len:

return None, None

inp = torch.tensor([ids], device=model.device)

logits = model(inp).logits[0].float()

logp = torch.log_softmax(logits, dim=-1) # 直接在对数域做 softmax

tgt = torch.tensor(ids[start:start + n], device=model.device)

rows = torch.arange(start, start + n, device=model.device) - 1 # 位置 t-1 预测 token t

return logp[rows, tgt].sum().item(), n

def cig(sample):

q, c, a = sample["query"], sample["ctx"], sample["answer"]

if c:

full_prompt = q + "\n\n" + c + "\n\n" + q

ref_prompt = q

else:

full_prompt = q

ref_prompt = GENERIC + "\n" + q

lp_full, n = logprob(full_prompt, a)

lp_ref, _ = logprob(ref_prompt, a)

if lp_full is None or lp_ref is None:

return None

return (lp_full - lp_ref) / n

with open(args.data, encoding="utf-8") as fi, open(args.out, "w", encoding="utf-8") as fo:

for line in fi:

line = line.strip()

if not line:

continue

rec = json.loads(line)

score = cig(rec)

rec["cig"] = score

fo.write(json.dumps(rec, ensure_ascii=False) + "\n")

```

关于性能：每条样本要跑两次前向。如果数据量在十万条量级，可以把样本按长度分桶、批量推理，实际耗时会明显下降。另外只对答案段收集 logits 就够了，但因果模型仍然需要完整前向，这一点无法省略。

第 3 步：检查分数分布

不要拿到分数就直接切割。先看一眼分布形态和相关性。

```python

03_inspect.py

import json, statistics

rows = [json.loads(l) for l in open("scored.jsonl", encoding="utf-8")]

valid = [r for r in rows if r.get("cig") is not None]

scores = sorted(r["cig"] for r in valid)

n = len(scores)

for q in (0.05, 0.25, 0.50, 0.75, 0.95):

print(f"p{int(q*100):>2} = {scores[int(q*(n-1))]: .4f}")

print("mean =", round(statistics.mean(scores), 4))

print("负增益占比 =", round(sum(s < 0 for s in scores) / n, 4))

分数与答案长度的相关性，理想情况下应该接近 0

lens = [len(r["answer"]) for r in valid]

avg_l = statistics.mean(lens)

avg_s = statistics.mean(r["cig"] for r in valid)

num = sum((len(r["answer"]) - avg_l) * (r["cig"] - avg_s) for r in valid)

den = (sum((len(r["answer"]) - avg_l) 2 for r in valid) 0.5) * \

(sum((r["cig"] - avg_s) 2 for r in valid) 0.5)

print("与字符长度的皮尔逊相关 =", round(num / den, 3) if den else "n/a")

```

要看的几件事：

• 负增益占比。如果比例很高，说明数据里存在大量套话式回答，或者参考条件构造得不合理。
• 与长度的相关性。按 token 归一化之后，相关系数应该接近 0。如果仍然很高，检查是不是把 padding 位置也算进了 logprob。
• 是否有异常离群值。个别样本的分数比中位数高出几倍，值得人工抽查，可能是答案里混入了评测集泄漏内容，也可能是数据本身有问题。

第 4 步：筛选，并保住多样性

按分数排序直接取前 N%，会带来一个副作用：同一类问题的相似样本会挤满数据集。做法是先聚类、再配额。

```python

04_select.py

import json

rows = [json.loads(l) for l in open("scored.jsonl", encoding="utf-8")]

rows = [r for r in rows if r.get("cig") is not None]

KEEP_RATIO = 0.5 # 保留比例，按训练预算调整

MAX_PER_GROUP = 20 # 每个分组最多保留多少条

分组键可以先用 query 前缀，生产环境建议换成句向量聚类

def group_key(r):

return r["query"][:60]

rows.sort(key=lambda r: r["cig"], reverse=True)

kept, counter = [], {}

for r in rows:

g = group_key(r)

if counter.get(g, 0) >= MAX_PER_GROUP:

continue

counter[g] = counter.get(g, 0) + 1

kept.append(r)

if len(kept) >= int(len(rows) * KEEP_RATIO):

break

with open("filtered.jsonl", "w", encoding="utf-8") as fo:

for r in kept:

fo.write(json.dumps(r, ensure_ascii=False) + "\n")

print("保留", len(kept), "/", len(rows))

```

两个补充建议：

• 低分样本不要一律丢弃。格式遵循、拒答安全类样本，增益天然偏低，但属于必须保留的能力。给这类样本留一个固定比例的名额，比如总数的 5% 到 10%。
• 增益高不等于质量高。一条答案错误的样本，模型越不确定，它的增益可能反而越高。筛选前先用规则或模型判分做一轮质量过滤，把两者叠加使用。

第 5 步：设计对照组并做后训练

这一步决定实验有没有说服力。建议至少跑三组：

三组的样本条数、训练步数、序列长度上限、学习率、随机种子保持一致。如果显存允许，可以再加一组「全量训练」，观察筛选是否只是省了算力、还是真的提升了效果。

训练本身按框架的官方文档走即可。用 LoRA 还是全参微调取决于你的算力，但三组必须用同一套配置。

第 6 步：评测与复盘

评测要覆盖三个层面：

1. 目标能力：在留出集上测与训练数据同类的任务。这是筛选策略要影响的指标。

2. 通用能力回归：用一份和训练数据无关的评测集，观察是否出现退化。数据筛选收窄了分布，这一步很容易暴露问题。

3. 格式与安全：指令遵循、输出格式是否仍然稳定。

评测时固定 prompt 模板、解码参数（尤其是 temperature 和 max_new_tokens）和评测脚本版本。只看训练 loss 曲线没有意义，筛选后的数据分布变了，loss 数值本来就不在同一个基准上。

复盘时重点看两件事：如果 B 组没有超过 A 组，通常是筛选比例过大（信息量够但覆盖不足）或者评测集和训练数据不匹配；如果 B 组目标能力涨了但通用能力掉了，说明配额控制得不够，需要回调保留比例或增加多样性约束。

常见坑与排错

分数全是 0 或者几乎一样

大概率是 x_ref 和 x 实际上没有差别。检查 ctx 字段是不是空的，以及拼接逻辑有没有把上下文拼丢。

分数全是 NaN 或者负无穷

用了 log(softmax(x)) 而不是 log_softmax(x)，或者在 float16 下计算。改用 log_softmax，并把 logits 转成 float32 再做归一化。

位置对齐错了，分数完全不可信

因果模型里，位置 t-1 的 logits 预测的是位置 t 的 token。代码里的 rows = arange(start, start+n) - 1 就是这个意思。写成 arange(start, start+n) 会整体错位一位，分数看起来还挺合理，但全是错的。

长样本被截断，答案被切掉

max_len 太小会截断序列。答案被截断后，logprob 只覆盖了前半句，分数偏高。做法是超长样本直接丢弃（像上面的脚本那样返回 None），而不是静默截断。

分数和长度强相关

忘记除以答案 token 数。对数似然是求和量，序列越长越负，必须按 token 归一化。

打分模型和训练模型不是同一个

用大模型给小模型的数据打分，分数排序未必能迁移。稳妥的做法是用待训练的基座自己打分；如果只是想要一个便宜的粗筛，也要在同族、规模接近的模型上做。

评测集泄漏进训练数据

如果评测集的题目也参与了打分和筛选，结果会虚高。筛选前先做一轮精确匹配和模糊匹配的去重。

筛选后数据同质化严重

只按分数排序、不做分组配额时很常见。加上分组上限，或者把分数当作加权采样的权重而不是硬阈值。

打分脚本改了但忘了重跑

参考条件的措辞、模板、max_len 只要动了任何一处，全量分数都需要重算，不能混用新旧结果。

下一步建议

跑通一轮之后，可以往这几个方向延伸：

• 迭代式打分。每训练一轮，用新模型重新为剩余样本打分。随着模型能力变化，同一批样本的增益会变，早期被筛掉的数据可能在后续轮次重新变得有价值。
• 把增益变成权重。与其硬性筛选，不如把归一化后的 CIG 作为样本 loss 权重，让低增益样本以较小的权重参与训练，减少分布偏移。
• 和课程学习结合。按增益分档，先训中等增益、再训高增益，观察收敛速度是否变化。
• 扩展到偏好数据。对 DPO 一类的数据，可以分别计算 chosen 和 rejected 的条件信息增益，用两者之差衡量这条偏好对的价值。
• 沉淀数据卡。记录每轮筛选用到的打分模型、参考条件模板、保留比例和评测结果，否则过一段时间很难解释某份数据集是怎么来的。

这套流程的价值在于把「这条数据值不值得训」从直觉变成了可复现的数字。数字本身不保证正确，但至少让每次筛选都有据可查、可以复盘、可以推翻。