这篇能做出什么
读完之后,你能独立走完一条完整的“改模”链路:把一个 Qwen 系列的开放权重模型加载进来,逐个张量看清它的结构,用可复现的量化手段定位到与数学推理关系密切的模块,生成一份只作用于这些模块的权重增量,安全地写回成一个新的模型目录,最后用固定题库对比改前改后的准确率,并确认通用能力没有被顺带破坏。
具体产出有三个:
1. 一份模块重要性报告,列出 layers.N.mlp.down_proj、layers.N.self_attn.o_proj 这类具体张量名及其数学敏感度分数,而不是“调一调就好了”这种模糊结论。
2. 一个可加载的新权重目录,AutoModelForCausalLM.from_pretrained() 能直接打开,且张量名、dtype、分片结构与原模型一致。
3. 一张对照表:改前 / 改后的数学题准确率、通用文本困惑度、以及一组回归测试的通过情况。
关于“用 GPT-6 Astra”:这里把它当作一个能读文件、能写脚本、能跑命令并回看日志的编程助手来用,让它承担“读 index.json → 生成分析脚本 → 跑评测 → 汇总结果”这类重复劳动。它能省下的主要是查文档和写胶水代码的时间,权重怎么算、改不改、改多少,判断权仍在你手里。该模型的具体能力边界以官方文档当前版本为准,本文不依赖它的任何特定接口。
一个必须先说清楚的前提:没有哪个模型能“看一眼就懂”权重。真正的做法是让助手帮你写出标准的 PyTorch / safetensors 脚本,用数值统计去找重要模块,再用微调或任务算术的方式产生增量。下面的路径就是围绕这一点展开的。
前置条件清单
- 硬件:以 1.5B~7B 级别模型起步。1.5B 在单张 16GB 显存卡上可以全参数反向传播;7B 建议用 LoRA 或只做前向统计,纯 CPU + 内存 32GB 也能跑完读权重和写权重,只是慢。显存不足时用
device_map="auto"走 CPU offload。 - 软件:Python 3.10+、
torch、transformers、safetensors、peft、datasets、accelerate。版本以官方文档当前版本为准,安装时把包名一起写进requirements.txt固定住,避免下次跑出不同结果。 - 模型:从官方渠道下载一个 Qwen 系列开放权重模型到本地目录,确保目录里有
config.json、tokenizer_config.json、model.safetensors或model.safetensors.index.json。不同代际的 Qwen 目录结构略有差异,有的单文件,有的分片,脚本要能兼容两种情况。 - 数据:自建 200~500 道带标准答案的数学题。覆盖四则运算、百分比、一元方程、简单应用题四类,每类至少 50 道,并且按题型切成 train / dev / test 三份。这是全文最重要的准备,没有留出集,后面所有“提升”都不可信。
- 磁盘:留出至少 3 倍模型体积的空间,用于存放原权重、中间增量和新权重。
- 许可证确认:Qwen 不同版本的许可条款不同,商用前以官方页面为准。
步骤一:先固定评测,再动权重
顺序不能颠倒。很多“改完变强了”的结论,其实只是换了 prompt 格式带来的错觉。Qwen 系列对对话模板敏感,同一个问题用不用 chat template、要不要加“请逐步计算”,结果可能差出一大截。所以第一步是把评测流程焊死。
```python
eval.py —— 改前改后都用这一个文件,只换 model_path
import re, json, torch
from transformers import AutoModelForCausalLM, AutoTokenizer
MODEL_PATH = "./Qwen-base" # 换成你的目录
TEST_FILE = "math_test.jsonl" # 每行 {"q": "...", "a": "42"}
tok = AutoTokenizer.from_pretrained(MODEL_PATH, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(
MODEL_PATH, torch_dtype=torch.bfloat16, device_map="auto",
trust_remote_code=True,
).eval()
def build_prompt(q):
msgs = [{"role": "user",
"content": q + "\n请逐步推理,最后一行以「答案:」开头只写最终数字。"}]
return tok.apply_chat_template(msgs, tokenize=False, add_generation_prompt=True)
def extract_num(text):
text = text.replace(",", "")
nums = re.findall(r"-?\d+(?:\.\d+)?", text.split("答案:")[-1])
return nums[0] if nums else None
def run():
ok = total = 0
for line in open(TEST_FILE, encoding="utf-8"):
item = json.loads(line)
ids = tok(build_prompt(item["q"]), return_tensors="pt").to(model.device)
with torch.no_grad():
out = model.generate(**ids, max_new_tokens=512, do_sample=False,
temperature=None, top_p=None, pad_token_id=tok.eos_token_id)
pred = extract_num(tok.decode(out[0][ids["input_ids"].shape[1]:], skip_special_tokens=True))
total += 1
ok += (pred is not None and abs(float(pred) - float(item["a"])) < 1e-6)
print(f"acc = {ok}/{total} = {ok/total:.4f}")
if __name__ == "__main__":
run()
```
注意 do_sample=False 配 temperature=None, top_p=None,这是贪心解码,保证可复现。跑两遍,确认两次结果完全一致,再进入下一步。
步骤二:打开权重,看清每个张量
分片模型的入口是 model.safetensors.index.json,里面有一张 weight_map,把每个张量名映射到具体文件。单文件模型没有 index,直接读 model.safetensors 即可。
```python
inspect.py
import json
from pathlib import Path
from safetensors import safe_open
import torch
MODEL_DIR = Path("./Qwen-base")
idx_file = MODEL_DIR / "model.safetensors.index.json"
if idx_file.exists():
weight_map = json.loads(idx_file.read_text())["weight_map"]
else:
with safe_open(MODEL_DIR / "model.safetensors", framework="pt") as f:
weight_map = {k: "model.safetensors" for k in f.keys()}
print("张量总数:", len(weight_map))
def load_tensor(name):
with safe_open(MODEL_DIR / weight_map[name], framework="pt", device="cpu") as f:
return f.get_tensor(name)
看一层长什么样
for name in sorted(n for n in weight_map if ".layers.0." in n):
w = load_tensor(name)
print(f"{name:55s} {tuple(w.shape)!s:20s} {w.dtype}")
```
你会看到类似 model.layers.0.mlp.down_proj.weight、model.layers.0.self_attn.q_proj.weight、model.embed_tokens.weight 这样的名字。请特别留意三件事:
config.json里的tie_word_embeddings。若为true,lm_head.weight与embed_tokens.weight是同一份存储,改一个必须同步另一个,否则加载时报 key 缺失或行为异常。- dtype。绝大多数 Qwen 开放权重是
bfloat16。后面写回时必须保持原 dtype,一旦写成float32,文件体积翻倍、加载变慢,甚至可能因为某些算子不支持而报错。 - 量化模型。如果拿到的是 GPTQ / AWQ 目录,里面不是普通浮点张量,本文这套流程不适用,需要先找到未量化的原始权重。
步骤三:定位“数学相关”的参数
这是整个流程里唯一需要动脑的地方。没有任何单一指标能证明某个张量“负责数学”,可行的是多种信号交叉验证。下面三个方法成本从低到高,建议全部跑一遍再做交集。
方法 A:前向激活能量比
给模型喂两批 prompt:一批是数学题,一批是通用闲聊。用 hook 记录每个线性层输出的平均范数,算比值。比值高的模块,对数学输入的响应更强烈。
```python
act_score.py
from collections import defaultdict
import torch, json
from transformers import AutoModelForCausalLM, AutoTokenizer
model = AutoModelForCausalLM.from_pretrained("./Qwen-base", torch_dtype=torch.bfloat16,
device_map="auto", trust_remote_code=True).eval()
tok = AutoTokenizer.from_pretrained("./Qwen-base", trust_remote_code=True)
stat = defaultdict(lambda: {"math": [], "chat": []})
TARGET = ("q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj")
def hook_fn(name):
def _h(module, inp, out):
y = out[0] if isinstance(out, tuple) else out
stat[name]["__cur__"].append(y.detach().float().norm(dim=-1).mean().item())
return _h
handles = []
for name, mod in model.named_modules():
if name.split(".")[-1] in TARGET and name.startswith("model.layers"):
handles.append(mod.register_forward_hook(hook_fn(name)))
def collect(prompts, bucket):
for name in stat:
stat[name]["__cur__"] = []
for p in prompts:
ids = tok(p, return_tensors="pt", truncation=True, max_length=512).to(model.device)
with torch.no_grad():
model(**ids)
for name in stat:
v = stat[name]["__cur__"]
if v:
stat[name][bucket].append(sum(v) / len(v))
math_prompts = [json.loads(l)["q"] for l in open("math_dev.jsonl", encoding="utf-8")][:50]
chat_prompts = ["今天天气怎么样?", "帮我写一句产品介绍。", "解释一下什么是缓存。"] * 17
collect(math_prompts, "math")
collect(chat_prompts, "chat")
for h in handles:
h.remove()
rows = []
for name, d in stat.items():
if d["math"] and d["chat"]:
m = sum(d["math"]) / len(d["math"])
c = sum(d["chat"]) / len(d["chat"])
rows.append((name, m / (c + 1e-9)))
rows.sort(key=lambda x: -x[1])
for name, r in rows[:25]:
print(f"{r:6.3f} {name}")
```
把 top 25 打印出来你会看到一个规律:靠后的层(layers.20 之后)以及 down_proj、o_proj 这类“输出侧”投影更容易出现在高位。这是常见现象,不是定律,用你自己的数据跑出来的结果才算数。
方法 B:LoRA 增量范数
在数学训练集上训一个很小的 LoRA,然后看每个模块学到的增量 B @ A 的范数。范数大意味着这个模块被数学数据“推”得更狠。
```python
lora_probe.py
from peft import LoraConfig, get_peft_model
from transformers import AutoModelForCausalLM, AutoTokenizer, Trainer, TrainingArguments
from datasets import load_dataset
model = AutoModelForCausalLM.from_pretrained("./Qwen-base", torch_dtype=torch.bfloat16,
device_map="auto", trust_remote_code=True)
tok = AutoTokenizer.from_pretrained("./Qwen-base", trust_remote_code=True)
cfg = LoraConfig(
r=16, lora_alpha=32, lora_dropout=0.05, bias="none", task_type="CAUSAL_LM",
target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
"gate_proj", "up_proj", "down_proj"],
)
model = get_peft_model(model, cfg)
model.print_trainable_parameters()
ds = load_dataset("json", data_files="math_train.jsonl", split="train")
def fmt(ex):
msgs = [{"role": "user", "content": ex["q"] + "\n请逐步推理。"},
{"role": "assistant", "content": ex["cot"] + f"\n答案:{ex['a']}"}]
text = tok.apply_chat_template(msgs, tokenize=False)
out = tok(text, truncation=True, max_length=1024)
out["labels"] = out["input_ids"].copy()
return out
ds = ds.map(fmt, remove_columns=ds.column_names)
Trainer(
model=model, train_dataset=ds,
args=TrainingArguments(output_dir="lora_probe", num_train_epochs=2,
per_device_train_batch_size=1, gradient_accumulation_steps=8,
learning_rate=1e-4, bf16=True, logging_steps=10,
save_strategy="no", report_to="none"),
).train()
scores = {}
for name, mod in model.named_modules():
if hasattr(mod, "lora_A") and "default" in mod.lora_A:
A = mod.lora_A["default"].weight.float()
B = mod.lora_B["default"].weight.float()
scores[name.replace("base_model.model.", "").replace(".default", "")] = \
(B @ A).norm().item()
for k, v in sorted(scores.items(), key=lambda x: -x[1])[:30]:
print(f"{v:8.4f} {k}")
model.save_pretrained("lora_probe_adapter")
```
方法 C:交叉取交集
把 A 和 B 的排名做交集,取两边都在前 30% 的模块,作为“高置信度数学相关模块”。同时把两者的并集里靠后的部分作为对照组。这个交集名单就是后面所有改动的操作对象,写进一个 targets.json:
```json
{"alpha": 0.6, "targets": ["model.layers.24.mlp.down_proj.weight", "..."]}
```
步骤四:生成权重增量
三条路线,按改动强度从小到大排。建议按顺序试,每一步都跑一遍评测。
路线一:定向 LoRA 合并(改动温和,先从这条开始)
把上一步训好的适配器只保留在目标模块上,然后合并进基础权重。merge_and_unload() 会直接把 W + BA * scale 写进原张量。
```python
merge_lora.py
import json, torch
from peft import PeftModel
from transformers import AutoModelForCausalLM, AutoTokenizer
base = AutoModelForCausalLM.from_pretrained("./Qwen-base", torch_dtype=torch.bfloat16,
device_map="cpu", trust_remote_code=True)
model = PeftModel.from_pretrained(base, "lora_probe_adapter")
merged = model.merge_and_unload()
merged.save_pretrained("./Qwen-merged", safe_serialization=True, max_shard_size="5GB")
AutoTokenizer.from_pretrained("./Qwen-base", trust_remote_code=True).save_pretrained("./Qwen-merged")
```
注意 device_map="cpu":合并要在 CPU 上做,避免显存不够。
路线二:任务向量缩放(可调,适合做消融)
思路是 W_new = W_base + alpha * (W_merged - W_base),并且只对目标模块生效。alpha 从 0.3 试到 1.0,画一条准确率曲线。
```python
task_vector.py
import json
from pathlib import Path
from collections import defaultdict
from safetensors import safe_open
from safetensors.torch import save_file
import torch
BASE, TUNED, OUT = Path("./Qwen-base"), Path("./Qwen-merged"), Path("./Qwen-tv")
OUT.mkdir(exist_ok=True)
conf = json.loads(Path("targets.json").read_text())
alpha, targets = conf["alpha"], set(conf["targets"])
def weight_map(d):
f = d / "model.safetensors.index.json"
if f.exists():
return json.loads(f.read_text())["weight_map"]
with safe_open(d / "model.safetensors", framework="pt") as sf:
return {k: "model.safetensors" for k in sf.keys()}
base_map, tuned_map = weight_map(BASE), weight_map(TUNED)
def get(d, m, name):
with safe_open(d / m[name], framework="pt", device="cpu") as f:
return f.get_tensor(name)
shards = defaultdict(dict)
changed = 0
for name, fname in base_map.items():
w = get(BASE, base_map, name)
if name in targets:
w_t = get(TUNED, tuned_map, name)
w = (w.float() + alpha * (w_t.float() - w.float())).to(w.dtype)
changed += 1
shards[fname][name] = w
for fname, tensors in shards.items():
save_file(tensors, str(OUT / fname), metadata={"format": "pt"})
print("被修改的张量数:", changed)
```
改完把 config.json、generation_config.json、tokenizer*、*.index.json 一并复制过去。index 之所以能直接复用,是因为你保持了张量名和分片归属不变。
路线三:低秩截断(防止过拟合与能力崩塌)
如果路线二在 dev 集上提升明显、test 集上却掉点,说明增量里噪声太多。对每个增量做 SVD,只保留前 r 个奇异值再写回,相当于给改动“降噪”。
```python
def low_rank(delta, r=8):
U, S, Vh = torch.linalg.svd(delta.float(), full_matrices=False)
S[r:] = 0.0
return ((U * S) @ Vh).to(delta.dtype)
```
把 task_vector.py 里的 alpha * (w_t - w) 换成 low_rank(alpha * (w_t - w), r=8) 即可。r 取 4/8/16 各跑一次评测,选 dev 上表现稳的那个。
步骤五:安全写回与自检
写回前先做三个断言,避免产出一个“能存不能跑”的目录。
```python
check.py
import json, torch
from pathlib import Path
from safetensors import safe_open
from transformers import AutoModelForCausalLM
NEW, ORIG = Path("./Qwen-tv"), Path("./Qwen-base")
def keys(d):
f = d / "model.safetensors.index.json"
if f.exists():
return set(json.loads(f.read_text())["weight_map"])
with safe_open(d / "model.safetensors", framework="pt") as sf:
return set(sf.keys())
assert keys(NEW) == keys(ORIG), "张量名集合不一致,检查新增/遗漏的 key"
逐张量检查 dtype 与 NaN
idx = NEW / "model.safetensors.index.json"
if idx.exists():
wmap = json.loads(idx.read_text())["weight_map"]
else:
with safe_open(NEW / "model.safetensors", framework="pt") as sf:
wmap = {k: "model.safetensors" for k in sf.keys()}
for name, fname in wmap.items():
with safe_open(NEW / fname, framework="pt") as sf:
t = sf.get_tensor(name)
assert not torch.isnan(t).any(), f"{name} 出现 NaN"
assert not torch.isinf(t).any(), f"{name} 出现 Inf"
m = AutoModelForCausalLM.from_pretrained(NEW, torch_dtype=torch.bfloat16,
device_map="auto", trust_remote_code=True)
print("加载成功,参数量:", sum(p.numel() for p in m.parameters()))
```
NaN 一般来自学习率过大或 alpha 过大导致数值溢出;dtype 不一致多半是某处 .float() 之后忘了 .to(原dtype)。
步骤六:验证能力提升
评测分三块,缺一不可。
1. 数学准确率:直接把 eval.py 的 MODEL_PATH 换成新目录跑一遍,和基线对比。提升看 test 集,不看 train 集。
2. 通用能力回归:在若干段通用中文文本上算困惑度,上涨超过 5% 就要警惕。
```python
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
def ppl(path, texts):
tok = AutoTokenizer.from_pretrained(path, trust_remote_code=True)
m = AutoModelForCausalLM.from_pretrained(path, torch_dtype=torch.bfloat16,
device_map="auto", trust_remote_code=True).eval()
total, n = 0.0, 0
for t in texts:
ids = tok(t, return_tensors="pt", truncation=True, max_length=1024).to(m.device)
with torch.no_grad():
total += m(**ids, labels=ids["input_ids"]).loss.item()
n += 1
return total / n
texts = [l.strip() for l in open("general.txt", encoding="utf-8") if l.strip()][:100]
print("base ppl:", ppl("./Qwen-base", texts))
print("new ppl:", ppl("./Qwen-tv", texts))
```
3. 行为回归:抽 30 条开放式指令(改写句子、列提纲、解释概念、写 JSON),人工或让助手打分,看回答是否变得“只想着做题”。定向改模最常见的副作用就是把模型改得张口就是数字。
把三块结果填进一张表:
| 指标 | 基线 | 改后 | 结论 |
|---|---|---|---|
| 数学 test 准确率 | 基线值 | 新值 | 需提升 |
| 通用困惑度 | 基线值 | 新值 | 允许小幅上涨 |
| 指令跟随通过数 | 基线值 | 新值 | 不应下降 |
只有当“数学涨、困惑度基本平、指令跟随不掉”三条同时满足,这次改动才算成立。否则回到步骤四换 alpha 或 r 重来。
常见坑与排错
张量名对不上,加载报 missing keys。 九成是 tie_word_embeddings 惹的祸。检查 config.json,若为 true,lm_head.weight 可能不在文件里,脚本里对它 if name not in weight_map: continue 处理,别硬造一个。
写回后模型体积翻倍。 dtype 从 bfloat16 变成了 float32。任何涉及 .float() 的运算,最后都要 .to(原dtype)。
评测准确率忽高忽低。 大概率开了采样。generate() 必须 do_sample=False,并且显式设 temperature=None, top_p=None。
准确率暴涨但实际用起来变笨。 训练集和测试集有重叠,或者测试题和训练题是同一模板换个数字。切分数据时按“题型”分,不是按行数随机分。
合并后转 GGUF / GPTQ 失败。 量化要在合并完成之后重新做,不能把旧量化产物和新权重混在一起。重新量化时以相应工具的官方文档当前版本为准。
显存 OOM。 读权重、合并、SVD 全部放在 CPU 上做,只有推理时才上 GPU。7B 模型 bf16 约十几 GB,SVD 时 .float() 会再翻一倍,逐张量处理、处理完立刻 del。
改了 embed_tokens 导致 tokenizer 错位。 不要动词表大小,不要 resize_token_embeddings。改模和换词表是两件事,混在一起做会让所有结论失效。
助手生成的脚本直接跑,把原目录覆盖了。 所有写操作只允许写到新目录,原目录设为只读。把这条写进给助手的系统提示里。
下一步建议
- 把评测做厚。现在只有数学一类,可以加代码、逻辑、多语言各一组,形成一个小型回归套件。改模的门槛不在改,而在判断改得对不对。
- 试模型合并而非单点修改。如果有多个不同数据训出的 LoRA,用任务算术把它们的增量加权叠加,往往比单一路线更稳,
alpha的搜索空间也更直观。 - 探索稀疏化路线。既然已经拿到重要性排名,可以试试只对 top-k 模块做改动、其余完全冻结,对比全量改动的差异,理解“最少改多少还能涨”。
- 接入偏好优化。数学题有天然的对错信号,很适合做 DPO 或拒绝采样微调。把本文的
targets.json作为 LoRA 的挂载位置,能让偏好优化更集中。 - 把流程脚本化。把
eval.py、act_score.py、task_vector.py、check.py串成一条make all,每次改一个参数就能跑出完整对照表。改模是实验科学,跑得快才能试得多。
最后提醒一句:模型能力的提升来自数据质量和评测纪律,权重操作只是把这份提升“装进”参数里的手段。先有可信的评测,再谈改权重,顺序反了,后面每一步都是白费。
