本文档带你从零开始，一步步完成：环境搭建 → 蒸馏数据生成 → 全量微调 → 推理验证，直到与当前项目冒烟跑通的状态。

理论背景见 基于千问的黑盒蒸馏 SOP 第一~二节；代码设计见 基于千问的黑盒蒸馏 SOP 第七节。

---

目录

1. 开始之前：你需要准备什么
2. 第一步：创建目录，克隆 LLaMA-Factory
3. 第二步：安装依赖并修复常见问题
4. 第三步：下载学生模型 Qwen2.5-1.5B
5. 第四步：配置 DeepSeek API Key
6. 第五步：生成蒸馏数据（NuminaMath + R1 + V3）
7. 第六步：注册数据集到 LLaMA-Factory
8. 第七步：创建训练配置文件
9. 第八步：运行全量微调（冒烟 2 step）
10. 第九步：推理验证
11. 完整命令速查表
12. 常见问题 FAQ

---

全局流程一览

预计耗时（冒烟）：

阶段	耗时
环境搭建 + 模型下载	15~30 分钟
蒸馏数据（5 条）	1~3 分钟
全量微调（2 step）	约 1 分钟
推理验证	约 30 秒
合计	约 20~35 分钟

---

第0步：开始之前，你需要准备什么

0.1 硬件与平台

本教程在 AutoDL 云平台上实测通过，配置如下：

项目	要求	实测配置
GPU	≥ 24GB 显存（全量微调 1.5B）	RTX 3090 24GB
磁盘	≥ 15GB 可用空间	50GB 数据盘
镜像	PyTorch 2.x + Python 3.10+	PyTorch 2.8 / Python 3.12
内存	≥ 16GB	按 AutoDL 默认即可

0.2 账号与密钥

资源	用途	获取方式
AutoDL 实例	提供 GPU 算力	https://www.autodl.com/
DeepSeek API Key	R1 生成数据 + V3 评判	https://platform.deepseek.com/
HuggingFace（可选）	加载 NuminaMath 数据集	无需 Token 也可，但限速

0.3 最终目录结构预览

跑通后，你的项目目录应如下：

/root/autodl-tmp/Distill/
├── .env                          # API Key（勿公开）
├── SOP.md                        # 理论笔记 + 代码说明
├── TUTORIAL.md                   # 本文档
├── download.py                   # 下载学生模型
├── distill_utils.py              # 蒸馏公共工具
├── step1_generate_r1.py          # R1 生成软标签
├── step2_judge_v3.py             # V3 评判清洗
├── run_distill_pipeline.py       # 一键蒸馏流水线
├── test_inference.py             # 推理验证
├── LLaMA-Factory/                # 微调框架
│   ├── data/dataset_info.json    # 含 Distil 数据集注册
│   └── examples/train_full/
│       └── qwen2-full-sft.yaml   # 训练配置
├── models/                       # 学生模型权重（约 3GB）
└── saves/                        # 微调产出（约 3GB）

/root/autodl-tmp/data/
├── r1_raw_outputs.jsonl          # R1 中间产物
├── v3_judged_outputs.jsonl       # V3 评判记录
└── Distil-data-17k-train.arrow   # 最终训练数据

---

第一步：创建目录，克隆 LLaMA-Factory

1.1 创建项目目录

mkdir -p /root/autodl-tmp/Distill
cd /root/autodl-tmp/Distill

1.2 开启网络加速（AutoDL 必做）

访问 GitHub 前必须先执行：

source /etc/network_turbo

注意：开启后 pip 安装等非 GitHub/HF 操作可能变慢。建议只在 clone 和下载 HF 数据集时使用，装完依赖后可关闭终端重开。

1.3 克隆 LLaMA-Factory

cd /root/autodl-tmp/Distill
source /etc/network_turbo
git clone --depth 1 https://github.com/hiyouga/LLaMA-Factory.git

--depth 1 只拉取最新版本，加快克隆速度。

1.4 验证

ls LLaMA-Factory/src/llamafactory/

应能看到 cli.py、train/、data/ 等目录。

检查点 ✅： LLaMA-Factory/ 目录存在且包含 src/llamafactory/。

---

第二步：安装依赖并修复常见问题

2.1 安装 LLaMA-Factory

cd /root/autodl-tmp/Distill/LLaMA-Factory
pip install -e ".[torch,metrics]"

安装约 3~5 分钟，会同时安装 modelscope、datasets、transformers 等。

2.2 修复 torchaudio 版本冲突（必做）

安装后很可能遇到此问题。先验证：

llamafactory-cli version

如果报错 libcudart.so.13: cannot open shared object file，执行：

pip install "torchaudio==2.8.0" --index-url https://download.pytorch.org/whl/cu128

原因： pip install 会拉取 torchaudio 2.11.0（需要 CUDA 13），但 AutoDL 镜像自带 PyTorch 2.8.0+cu128，二者不兼容。

2.3 安装蒸馏流水线额外依赖

pip install openai

2.4 验证

llamafactory-cli version

期望输出：

----------------------------------------------------------
| Welcome to LLaMA Factory, version 0.9.6.dev0           |
|                                                        |
| Project page: https://github.com/hiyouga/LLaMA-Factory |
----------------------------------------------------------

检查点 ✅： llamafactory-cli version 正常输出版本号，无报错。

---

第三步：下载学生模型 Qwen2.5-1.5B

学生模型是蒸馏的「受教者」，我们选用 Qwen2.5-1.5B-Instruct。

3.1 创建下载脚本

在 /root/autodl-tmp/Distill/ 下创建 download.py：

from modelscope import snapshot_download

model_id = "Qwen/Qwen2.5-1.5B-Instruct"
local_dir = "/root/autodl-tmp/Distill/models"
model_dir = snapshot_download(model_id, cache_dir=local_dir)
print(f"Model downloaded to: {model_dir}")

使用魔搭 ModelScope 而非 HuggingFace，国内下载更快，不需要 network_turbo。

3.2 执行下载

cd /root/autodl-tmp/Distill
python download.py

约 3 分钟，下载约 3GB。

3.3 确认路径

ls /root/autodl-tmp/Distill/models/models/Qwen--Qwen2.5-1.5B-Instruct/snapshots/master/

应包含 config.json、model.safetensors、tokenizer.json 等。

记下这个路径，后面训练配置要用：

/root/autodl-tmp/Distill/models/models/Qwen--Qwen2.5-1.5B-Instruct/snapshots/master

检查点 ✅： model.safetensors 存在，大小约 2.9GB。

---

第四步：配置 DeepSeek API Key

蒸馏数据生成需要调用 DeepSeek API（R1 教师 + V3 裁判）。

4.1 创建 .env 文件

cat > /root/autodl-tmp/Distill/.env << 'EOF'
DEEPSEEK_API_KEY=sk-your-key-here
EOF

将 sk-your-key-here 替换为你在 https://platform.deepseek.com/ 申请的 Key。

4.2 验证 API 连通性

cd /root/autodl-tmp/Distill
python3 -c "
from openai import OpenAI
import os
# 从 .env 读取
for line in open('.env'):
    if line.startswith('DEEPSEEK_API_KEY='):
        key = line.strip().split('=',1)[1]
client = OpenAI(api_key=key, base_url='https://api.deepseek.com')
r = client.chat.completions.create(
    model='deepseek-chat',
    messages=[{'role':'user','content':'回复OK'}],
)
print('API 连通:', r.choices[0].message.content)
"

期望输出： API 连通: OK（或类似确认回复）。

检查点 ✅： API 调用成功，无 401/403 错误。

---

第五步：生成蒸馏数据（NuminaMath + R1 + V3）

这是蒸馏的核心环节，分两步：

NuminaMath 题目 ──→ DeepSeek-R1 生成回答 ──→ DeepSeek-V3 评判 ──→ 训练数据
   (problem)           (思维链+答案)          (对比标准答案)        (.arrow)

5.1 确认脚本已就位

项目中应包含以下文件（如从零搭建，从 SOP 第七节复制代码创建）：

文件	作用
distill_utils.py	公共工具（API 客户端、数据加载、文件读写）
step1_generate_r1.py	从 NuminaMath 取题，调用 R1 生成
step2_judge_v3.py	调用 V3 评判，导出 arrow
run_distill_pipeline.py	一键运行 step1 + step2

5.2 运行蒸馏流水线（冒烟 5 条）

cd /root/autodl-tmp/Distill
source /etc/network_turbo          # 加载 NuminaMath 需要访问 HuggingFace
python run_distill_pipeline.py --num-samples 5

执行过程中你会看到：

============================================================
运行: python step1_generate_r1.py --num-samples 5
============================================================
[Step1] 加载 NuminaMath-CoT 前 5 条问题...
[Step1] R1 生成 (1/5) source=synthetic_math
[Step1] R1 生成 (2/5) source=synthetic_math
...
[Step1] 完成，共生成 4 条，保存至 /root/autodl-tmp/data/r1_raw_outputs.jsonl

============================================================
运行: python step2_judge_v3.py
============================================================
[Step2] V3 评判 (1/4) id=0
  -> 通过: 模型答案与标准解答的最终结果一致...
[Step2] 完成: 4/4 条通过 V3 评判

冒烟 5 条中 R1 可能有 1 条超时（正常），重试机制会处理。4/4 通过说明数据质量良好。

5.3 分步运行（可选，用于调试）

如果流水线某步失败，可以单独重跑：

# 只跑 R1 生成
python step1_generate_r1.py --num-samples 5

# 只跑 V3 评判（需要 step1 产物已存在）
python step2_judge_v3.py

5.4 验证产物

# 查看 R1 原始输出（第一条的题目）
head -1 /root/autodl-tmp/data/r1_raw_outputs.jsonl | python3 -m json.tool | head -10

# 查看 arrow 数据条数
python3 -c "
import pyarrow.ipc as ipc
with ipc.open_file('/root/autodl-tmp/data/Distil-data-17k-train.arrow') as f:
    print('训练数据条数:', f.read_all().num_rows)
"

# 查看 V3 评判记录
cat /root/autodl-tmp/data/v3_judged_outputs.jsonl | python3 -c "
import sys, json
for line in sys.stdin:
    r = json.loads(line)
    print(f'id={r[\"id\"]} correct={r[\"judge_correct\"]} reason={r[\"judge_reason\"][:50]}')
"

期望输出：

训练数据条数: 4
id=0 correct=True reason=模型答案与标准解答的最终结果一致...
id=1 correct=True reason=...

检查点 ✅：

• r1_raw_outputs.jsonl 存在且有内容
• Distil-data-17k-train.arrow 存在且 num_rows ≥ 1
• V3 评判至少有 1 条 correct=True

---

第六步：注册数据集到 LLaMA-Factory

告诉 LLaMA-Factory 如何加载我们的蒸馏数据。

6.1 编辑 dataset_info.json

打开 LLaMA-Factory/data/dataset_info.json，在最后一个条目之后、闭合 } 之前添加：

  "Distil": {
    "file_name": "/root/autodl-tmp/data/Distil-data-17k-train.arrow",
    "formatting": "sharegpt",
    "columns": {
      "messages": "conversations",
      "system": "system"
    },
    "tags": {
      "role_tag": "from",
      "content_tag": "value",
      "user_tag": "user",
      "assistant_tag": "assistant"
    }
  }

注意 JSON 逗号：前一个条目末尾要有 ,。

字段含义：

配置项	值	说明
file_name	arrow 绝对路径	指向第五步产出的文件
formatting	sharegpt	对话格式
columns.messages	conversations	arrow 中的对话列名
tags.user_tag	user	用户角色标签
tags.assistant_tag	assistant	助手角色标签

6.2 验证 JSON 格式

python3 -c "import json; json.load(open('LLaMA-Factory/data/dataset_info.json')); print('JSON 格式正确')"

检查点 ✅： JSON 解析无报错，且包含 "Distil" 条目。

---

第七步：创建训练配置文件

7.1 创建 yaml 配置

创建文件 LLaMA-Factory/examples/train_full/qwen2-full-sft.yaml：

### model
model_name_or_path: /root/autodl-tmp/Distill/models/models/Qwen--Qwen2.5-1.5B-Instruct/snapshots/master
trust_remote_code: true

### method
stage: sft
do_train: true
finetuning_type: full
deepspeed: examples/deepspeed/ds_z3_offload_config.json

### dataset
dataset: Distil
template: qwen
cutoff_len: 2048
max_samples: 1000
overwrite_cache: true
preprocessing_num_workers: 4
dataloader_num_workers: 2

### output
output_dir: /root/autodl-tmp/Distill/saves/qwen2.5-1.5b/smoke-sft
logging_steps: 1
save_steps: 100
plot_loss: true
overwrite_output_dir: true
save_only_model: true
report_to: none

### train
per_device_train_batch_size: 1
gradient_accumulation_steps: 1
learning_rate: 1.0e-5
max_steps: 2
lr_scheduler_type: cosine
warmup_ratio: 0.1
bf16: true
ddp_timeout: 180000000

7.2 关键参数说明（必读）

参数	冒烟值	作用	注意事项
model_name_or_path	第三步的完整路径	学生模型	必须是绝对路径
finetuning_type: full	全量微调	所有参数都更新	比 LoRA 耗显存
deepspeed: ds_z3_offload	ZeRO-3 + CPU offload	24GB 显卡必须	不加会 OOM
dataset: Distil	第六步注册的名	使用蒸馏数据	与 dataset_info.json 对应
template: qwen	Qwen 聊天模板	格式化对话	Qwen2.5 用 qwen，不是 qwen3
max_steps: 2	只跑 2 步	冒烟测试	正式训练改为 num_train_epochs: 3.0
save_only_model: true	只存模型权重	省磁盘	避免 DeepSpeed 优化器占 12GB

检查点 ✅： yaml 文件存在，路径与前三步一致。

---

第八步：运行全量微调（冒烟 2 step）

8.1 安装 DeepSpeed

pip install deepspeed

8.2 检查磁盘空间

df -h /root/autodl-tmp

确保 Avail ≥ 10GB。不足时清理无关文件。

8.3 启动训练

cd /root/autodl-tmp/Distill/LLaMA-Factory

FORCE_TORCHRUN=1 NNODES=1 NODE_RANK=0 MASTER_PORT=29501 \
  llamafactory-cli train examples/train_full/qwen2-full-sft.yaml

参数解释：

• FORCE_TORCHRUN=1：启用 torchrun，DeepSpeed 需要
• NNODES=1：单节点
• MASTER_PORT=29501：分布式通信端口

8.4 观察训练日志

正常流程：

[INFO] Loading dataset /root/autodl-tmp/data/Distil-data-17k-train.arrow...
Generating train split: 4 examples        ← 加载蒸馏数据
...
training example:                           ← 展示一条训练样本
input_ids: [151644, 8948, 198, ...]
...
trainable params: 1,543,714,304            ← 1.5B 全参
***** Running training *****
  Num examples = 4
  Total optimization steps = 2
  0%|          | 0/2 [00:00<?, ?it/s]
 50%|█████     | 1/2 [...]  {'loss': '0.9479', ...}
100%|██████████| 2/2 [...]  {'loss': '1.05', ...}
Saving model checkpoint to .../checkpoint-2
Model weights saved in .../model.safetensors

约 1~2 分钟完成（含 DeepSpeed 初始化）。

8.5 验证 checkpoint

ls -lh /root/autodl-tmp/Distill/saves/qwen2.5-1.5b/smoke-sft/checkpoint-2/

应包含：

• model.safetensors（约 2.9GB）
• config.json
• tokenizer.json

检查点 ✅：

• 训练日志显示 Total optimization steps = 2 且 100% 完成
• checkpoint-2/model.safetensors 存在

---

第九步：推理验证

9.1 创建推理脚本

创建 /root/autodl-tmp/Distill/test_inference.py：

"""微调后模型推理冒烟测试。"""
from transformers import AutoModelForCausalLM, AutoTokenizer
import os

MODEL_PATH = "/root/autodl-tmp/Distill/saves/qwen2.5-1.5b/smoke-sft/checkpoint-2"
BASE_PATH = "/root/autodl-tmp/Distill/models/models/Qwen--Qwen2.5-1.5B-Instruct/snapshots/master"

model_name = MODEL_PATH if os.path.exists(os.path.join(MODEL_PATH, "model.safetensors")) else BASE_PATH
print(f"Loading model from: {model_name}")

model = AutoModelForCausalLM.from_pretrained(
    model_name, torch_dtype="auto", device_map="auto", low_cpu_mem_usage=True
)
tokenizer = AutoTokenizer.from_pretrained(model_name)

system_prompt = "你是一名助人为乐的助手。"
user_prompt = "请证明根号2是无理数。"
messages = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": user_prompt},
]

text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
model_inputs = tokenizer([text], return_tensors="pt").to(model.device)

generated_ids = model.generate(**model_inputs, max_new_tokens=256)
generated_ids = [
    output_ids[len(input_ids):]
    for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
]

response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
print("=" * 40)
print("Model response:")
print(response)
print("=" * 40)

9.2 运行推理

cd /root/autodl-tmp/Distill
python test_inference.py

期望输出：

Loading model from: .../checkpoint-2
Loading weights: 100%|██████████| 338/338 [...]
========================================
Model response:
要证明√2是无理数，我们可以用反证法。
假设√2是有理数，那么它可以表示为两个整数a和b的...
========================================

模型应给出反证法推理过程。

9.3 对比微调前后（可选）

将 MODEL_PATH 改为 BASE_PATH（基座模型）再跑一次，对比回答质量差异。

检查点 ✅： 模型成功加载 checkpoint 并输出数学推理回答。

---

完整命令速查表

从零到冒烟，按顺序复制执行：

# ========== 第1步：克隆框架 ==========
mkdir -p /root/autodl-tmp/Distill && cd /root/autodl-tmp/Distill
source /etc/network_turbo
git clone --depth 1 https://github.com/hiyouga/LLaMA-Factory.git

# ========== 第2步：安装依赖 ==========
cd LLaMA-Factory
pip install -e ".[torch,metrics]"
pip install "torchaudio==2.8.0" --index-url https://download.pytorch.org/whl/cu128
pip install openai deepspeed
llamafactory-cli version    # 验证

# ========== 第3步：下载学生模型 ==========
cd /root/autodl-tmp/Distill
python download.py

# ========== 第4步：配置 API Key ==========
echo 'DEEPSEEK_API_KEY=sk-your-key-here' > .env

# ========== 第5步：生成蒸馏数据 ==========
source /etc/network_turbo
python run_distill_pipeline.py --num-samples 5

# ========== 第6步：注册数据集 ==========
# 手动编辑 LLaMA-Factory/data/dataset_info.json 添加 Distil 条目

# ========== 第7步：创建训练配置 ==========
# 手动创建 LLaMA-Factory/examples/train_full/qwen2-full-sft.yaml

# ========== 第8步：全量微调 ==========
cd /root/autodl-tmp/Distill/LLaMA-Factory
FORCE_TORCHRUN=1 NNODES=1 NODE_RANK=0 MASTER_PORT=29501 \
  llamafactory-cli train examples/train_full/qwen2-full-sft.yaml

# ========== 第9步：推理验证 ==========
cd /root/autodl-tmp/Distill
python test_inference.py

---

常见问题 FAQ

Q1: libcudart.so.13 报错

OSError: libcudart.so.13: cannot open shared object file

原因： torchaudio 版本与 PyTorch 不匹配。

解决：

pip install "torchaudio==2.8.0" --index-url https://download.pytorch.org/whl/cu128

---

Q2: 全量微调 CUDA OOM

torch.OutOfMemoryError: CUDA out of memory

原因： 1.5B 全量微调的 Adam 优化器状态需要额外 ~18GB 显存。

解决： yaml 中添加 DeepSpeed ZeRO-3 offload（第七步已配置），并用 FORCE_TORCHRUN=1 启动。

---

Q3: 训练完成但退出码为 1

RuntimeError: file write failed

原因： 磁盘满了（DeepSpeed 优化器 checkpoint 约 12GB）。

解决：

1. yaml 中设置 save_only_model: true
2. 清理磁盘：df -h /root/autodl-tmp
3. 删除失败的 global_step* 目录

---

Q4: R1 API 请求超时

[Step1] 失败 id=2: Request timed out.

原因： 复杂数学题 R1 推理超过默认超时。

解决： step1_generate_r1.py 已内置 180s 超时 + 3 次重试。可单独重跑 step1，或增大 timeout 参数。

---

Q5: NuminaMath 加载很慢或中断

解决：

1. 先 source /etc/network_turbo
2. 使用 streaming 模式（distill_utils.py 已配置）
3. 冒烟时 --num-samples 5 只取前几条

---

Q6: libgomp: Invalid value for environment variable OMP_NUM_THREADS

解决： 不影响运行，可忽略。或 export OMP_NUM_THREADS=1。

---

Q7: 模型下载路径不对

ModelScope 下载后的路径有嵌套：

/root/autodl-tmp/Distill/models/models/Qwen--Qwen2.5-1.5B-Instruct/snapshots/master

不是 /root/autodl-tmp/Distill/models/Qwen2.5-1.5B-Instruct。训练 yaml 中必须使用完整嵌套路径。

---

冒烟通过最终检查清单

全部打勾即表示从零到冒烟跑通：

• llamafactory-cli version 正常
• 学生模型 model.safetensors 已下载（约 3GB）
• .env 中 API Key 已配置且连通
• r1_raw_outputs.jsonl 有 R1 生成记录（含 reasoning_content）
• v3_judged_outputs.jsonl 有 V3 评判记录
• Distil-data-17k-train.arrow 有 ≥1 条通过评判的数据
• dataset_info.json 已注册 Distil 数据集
• qwen2-full-sft.yaml 路径正确、含 DeepSpeed 配置
• 训练完成 2 step，loss 有输出
• checkpoint-2/model.safetensors 已保存
• test_inference.py 输出数学推理回答

---

从冒烟到正式训练

冒烟通过后，如需正式蒸馏训练：

项目	冒烟	正式
数据量	--num-samples 5	--num-samples 17000
训练步数	max_steps: 2	num_train_epochs: 3.0（删除 max_steps）
磁盘需求	~10GB	≥ 20GB
API 费用	约 ¥0.1	约 ¥数百（视 R1 定价）
训练耗时	~1 分钟	~15 小时

# 正式数据生成
python run_distill_pipeline.py --num-samples 17000

# 正式训练（修改 yaml 后）
FORCE_TORCHRUN=1 NNODES=1 NODE_RANK=0 MASTER_PORT=29501 \
  llamafactory-cli train examples/train_full/qwen2-full-sft.yaml

---

更多理论背景和代码设计细节，请参阅 基于千问的黑盒蒸馏 SOP。

---

附录：项目 Python 脚本全文

以下为 /root/autodl-tmp/Distill/ 目录下全部 Python 脚本的完整内容，与磁盘文件一致。

A.1 distill_utils.py

import json
import os
import threading
from pathlib import Path

import pyarrow as pa
import pyarrow.ipc as ipc

DATA_DIR = Path("/root/autodl-tmp/data")
R1_RAW_FILE = DATA_DIR / "r1_raw_outputs.jsonl"
DISTIL_ARROW = DATA_DIR / "Distil-data-17k-train.arrow"
SYSTEM_PROMPT = "你是一个智能助手"

_JSONL_LOCK = threading.Lock()


def get_api_key() -> str:
    key = os.environ.get("DEEPSEEK_API_KEY")
    if not key and Path("/root/autodl-tmp/Distill/.env").exists():
        for line in Path("/root/autodl-tmp/Distill/.env").read_text().splitlines():
            if line.startswith("DEEPSEEK_API_KEY="):
                key = line.split("=", 1)[1].strip()
                break
    if not key:
        raise RuntimeError("请设置 DEEPSEEK_API_KEY 环境变量或在 .env 中配置")
    return key


def get_client():
    from openai import OpenAI

    return OpenAI(api_key=get_api_key(), base_url="https://api.deepseek.com")


def load_numinamath_samples(num_samples: int):
    from datasets import load_dataset

    ds = load_dataset("AI-MO/NuminaMath-CoT", split="train", streaming=True)
    samples = []
    for i, row in enumerate(ds):
        samples.append(
            {
                "id": i,
                "source": row.get("source", ""),
                "problem": row["problem"],
                "solution": row["solution"],
            }
        )
        if len(samples) >= num_samples:
            break
    return samples


def build_assistant_text(reasoning: str | None, content: str) -> str:
    reasoning = (reasoning or "").strip()
    content = (content or "").strip()
    if reasoning and content:
        return f"{reasoning}\n\n{content}"
    return reasoning or content


def save_jsonl(path: Path | str, records: list[dict]):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8") as f:
        for rec in records:
            f.write(json.dumps(rec, ensure_ascii=False) + "\n")


def append_jsonl(path: Path | str, record: dict):
    """线程安全地向 jsonl 文件追加一条记录。"""
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    line = json.dumps(record, ensure_ascii=False) + "\n"
    with _JSONL_LOCK:
        with path.open("a", encoding="utf-8") as f:
            f.write(line)


def load_jsonl(path: Path | str) -> list[dict]:
    path = Path(path)
    if not path.exists():
        return []
    records = []
    with path.open("r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                records.append(json.loads(line))
    return records


def get_processed_ids(path: Path | str, id_key: str = "id") -> set:
    """从已有 jsonl 中读取已处理记录的 id 集合，用于断点续跑。"""
    return {record[id_key] for record in load_jsonl(path) if id_key in record}


def save_sharegpt_arrow(samples: list[dict], output_file: Path = DISTIL_ARROW):
    output_file.parent.mkdir(parents=True, exist_ok=True)
    table = pa.Table.from_pydict(
        {
            "system": [s["system"] for s in samples],
            "conversations": [s["conversations"] for s in samples],
        }
    )
    with pa.OSFile(str(output_file), "wb") as sink:
        with ipc.new_file(sink, table.schema) as writer:
            writer.write_table(table)

A.2 download.py

from modelscope import snapshot_download

model_id = "Qwen/Qwen2.5-1.5B-Instruct"
local_dir = "/root/autodl-tmp/Distill/models"
model_dir = snapshot_download(model_id, cache_dir=local_dir)
print(f"Model downloaded to: {model_dir}")

A.3 generate_smoke_data.py

"""
已废弃：请使用 run_distill_pipeline.py 运行完整蒸馏流程。
  python run_distill_pipeline.py --num-samples 5
"""
raise SystemExit(
    "此脚本已废弃。请运行: python run_distill_pipeline.py --num-samples 5"
)

A.4 is_ds_valid.py

from openai import OpenAI
import os
# 从 .env 读取
for line in open('.env'):
    if line.startswith('DEEPSEEK_API_KEY='):
        key = line.strip().split('=',1)[1]
client = OpenAI(api_key=key, base_url='https://api.deepseek.com')
r = client.chat.completions.create(
    model='deepseek-chat',
    messages=[{'role':'user','content':'回复OK'}],
)
print('API 连通:', r.choices[0].message.content)

A.5 run_distill_pipeline.py

"""
蒸馏数据完整流水线（冒烟/正式通用）：
  NuminaMath-CoT 问题 → DeepSeek-R1 生成 → DeepSeek-V3 评判 → Distil-data-17k-train.arrow
"""
import argparse
import subprocess
import sys


def main():
    parser = argparse.ArgumentParser(description="运行完整蒸馏数据流水线")
    parser.add_argument("--num-samples", type=int, default=5, help="冒烟 5 条；正式可设 17000")
    parser.add_argument("--workers", type=int, default=5, help="Step1/Step2 并发线程数")
    parser.add_argument(
        "--request-interval",
        type=float,
        default=0.2,
        help="每个 worker 完成一条后的间隔秒数",
    )
    parser.add_argument("--overwrite", action="store_true", help="忽略已有结果，从头重新生成")
    args = parser.parse_args()

    common = [
        "--workers", str(args.workers),
        "--request-interval", str(args.request_interval),
    ]
    if args.overwrite:
        common.append("--overwrite")

    steps = [
        [sys.executable, "step1_generate_r1.py", "--num-samples", str(args.num_samples), *common],
        [sys.executable, "step2_judge_v3.py", *common],
    ]
    for cmd in steps:
        print("\n" + "=" * 60)
        print("运行:", " ".join(cmd))
        print("=" * 60)
        subprocess.run(cmd, check=True, cwd="/root/autodl-tmp/Distill")


if __name__ == "__main__":
    main()

A.6 step1_generate_r1.py

"""
步骤1：从 NuminaMath-CoT 取问题，调用 DeepSeek-R1 生成软标签。
支持多线程并发、断点续跑、逐条自动保存。
"""
import argparse
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

from distill_utils import (
    R1_RAW_FILE,
    append_jsonl,
    build_assistant_text,
    get_client,
    get_processed_ids,
    load_numinamath_samples,
)

_thread_local = threading.local()
_stats_lock = threading.Lock()


def get_thread_client():
    if not hasattr(_thread_local, "client"):
        _thread_local.client = get_client()
    return _thread_local.client


def call_r1(client, problem: str, max_retries: int = 3) -> dict:
    last_err = None
    for attempt in range(max_retries):
        try:
            response = client.chat.completions.create(
                model="deepseek-reasoner",
                messages=[
                    {"role": "system", "content": "你是一个数学推理助手，请仔细思考并给出解答。"},
                    {"role": "user", "content": problem},
                ],
                stream=False,
                timeout=180,
            )
            msg = response.choices[0].message
            return {
                "reasoning_content": getattr(msg, "reasoning_content", None),
                "content": msg.content,
                "assistant_text": build_assistant_text(
                    getattr(msg, "reasoning_content", None), msg.content
                ),
            }
        except Exception as e:
            last_err = e
            print(f"  重试 {attempt + 1}/{max_retries}: {e}")
            time.sleep(2 * (attempt + 1))
    raise last_err


def process_one(sample: dict, output_path: str, request_interval: float, stats: dict) -> None:
    client = get_thread_client()
    try:
        r1 = call_r1(client, sample["problem"])
        record = {
            "id": sample["id"],
            "source": sample["source"],
            "problem": sample["problem"],
            "ground_truth_solution": sample["solution"],
            "reasoning_content": r1["reasoning_content"],
            "content": r1["content"],
            "assistant_text": r1["assistant_text"],
        }
        append_jsonl(output_path, record)
        with _stats_lock:
            stats["ok"] += 1
            done = stats["ok"] + stats["fail"]
            total = stats["total"]
        print(
            f"[Step1] 完成 ({done}/{total}) id={sample['id']} "
            f"source={sample['source']} [成功 {stats['ok']} / 失败 {stats['fail']}]"
        )
    except Exception as e:
        with _stats_lock:
            stats["fail"] += 1
            done = stats["ok"] + stats["fail"]
            total = stats["total"]
        print(f"[Step1] 失败 ({done}/{total}) id={sample['id']}: {e}")
    finally:
        if request_interval > 0:
            time.sleep(request_interval)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--num-samples", type=int, default=5, help="冒烟建议 5，正式训练可设 17000")
    parser.add_argument("--output", type=str, default=str(R1_RAW_FILE))
    parser.add_argument("--workers", type=int, default=5, help="并发线程数，建议 3~8")
    parser.add_argument(
        "--request-interval",
        type=float,
        default=0.2,
        help="每个 worker 完成一条后的间隔秒数，用于降低限流风险",
    )
    parser.add_argument("--overwrite", action="store_true", help="忽略已有结果，从头重新生成")
    args = parser.parse_args()

    output_path = R1_RAW_FILE if args.output == str(R1_RAW_FILE) else args.output

    if args.overwrite and output_path.exists():
        output_path.unlink()
        print(f"[Step1] 已清空旧文件: {output_path}")

    done_ids = get_processed_ids(output_path)
    if done_ids:
        print(f"[Step1] 断点续跑: 已有 {len(done_ids)} 条，将跳过")

    print(f"[Step1] 加载 NuminaMath-CoT 前 {args.num_samples} 条问题...")
    samples = load_numinamath_samples(args.num_samples)
    pending = [s for s in samples if s["id"] not in done_ids]

    if not pending:
        print(f"[Step1] 全部 {len(samples)} 条已生成，无需重跑")
        return

    print(
        f"[Step1] 待处理 {len(pending)}/{len(samples)} 条，"
        f"workers={args.workers}，输出={output_path}"
    )

    stats = {"ok": 0, "fail": 0, "total": len(pending)}
    with ThreadPoolExecutor(max_workers=args.workers) as executor:
        futures = [
            executor.submit(process_one, sample, str(output_path), args.request_interval, stats)
            for sample in pending
        ]
        for future in as_completed(futures):
            future.result()

    final_count = len(get_processed_ids(output_path))
    print(
        f"[Step1] 完成: 本次成功 {stats['ok']}，失败 {stats['fail']}，"
        f"累计已保存 {final_count} 条 -> {output_path}"
    )


if __name__ == "__main__":
    main()

A.7 step2_judge_v3.py

"""
步骤2：DeepSeek-V3 作为裁判，对比 R1 输出与 NuminaMath 标准答案，筛选正确样本并导出 arrow。
支持多线程并发、断点续跑、逐条自动保存。
"""
import argparse
import json
import re
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

from distill_utils import (
    DISTIL_ARROW,
    R1_RAW_FILE,
    SYSTEM_PROMPT,
    append_jsonl,
    get_client,
    get_processed_ids,
    load_jsonl,
    save_sharegpt_arrow,
)

JUDGED_FILE = R1_RAW_FILE.parent / "v3_judged_outputs.jsonl"

JUDGE_PROMPT = """你是一个数学答案评判器。请判断「模型回答」与「标准解答」在数学上是否一致（最终答案正确即可，推理过程可不同）。

【题目】
{problem}

【标准解答】
{ground_truth}

【模型回答】
{model_answer}

请只输出 JSON，不要输出其他内容：
{{"correct": true或false, "reason": "一句话说明"}}"""

_thread_local = threading.local()
_stats_lock = threading.Lock()


def get_thread_client():
    if not hasattr(_thread_local, "client"):
        _thread_local.client = get_client()
    return _thread_local.client


def parse_judge_result(text: str) -> dict:
    text = text.strip()
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        match = re.search(r"\{.*\}", text, re.DOTALL)
        if match:
            return json.loads(match.group())
    return {"correct": False, "reason": f"无法解析评判结果: {text[:200]}"}


def judge_one(client, record: dict) -> dict:
    prompt = JUDGE_PROMPT.format(
        problem=record["problem"],
        ground_truth=record["ground_truth_solution"],
        model_answer=record["assistant_text"],
    )
    response = client.chat.completions.create(
        model="deepseek-chat",
        messages=[{"role": "user", "content": prompt}],
        stream=False,
        temperature=0,
        timeout=60,
    )
    result = parse_judge_result(response.choices[0].message.content)
    return {
        **record,
        "judge_correct": bool(result.get("correct")),
        "judge_reason": result.get("reason", ""),
    }


def to_sharegpt(record: dict) -> dict:
    return {
        "system": SYSTEM_PROMPT,
        "conversations": [
            {"from": "user", "value": record["problem"]},
            {"from": "assistant", "value": record["assistant_text"]},
        ],
    }


def rebuild_arrow(judged_path, arrow_path: str):
    accepted = [to_sharegpt(r) for r in load_jsonl(judged_path) if r.get("judge_correct")]
    save_sharegpt_arrow(accepted, arrow_path)
    return len(accepted)


def process_one(record: dict, judged_path: str, request_interval: float, stats: dict) -> None:
    client = get_thread_client()
    try:
        result = judge_one(client, record)
        append_jsonl(judged_path, result)
        with _stats_lock:
            stats["ok"] += 1
            if result["judge_correct"]:
                stats["accepted"] += 1
            done = stats["ok"] + stats["fail"]
            total = stats["total"]
        status = "通过" if result["judge_correct"] else "拒绝"
        print(
            f"[Step2] 完成 ({done}/{total}) id={record['id']} -> {status}: "
            f"{result['judge_reason'][:60]}"
        )
    except Exception as e:
        with _stats_lock:
            stats["fail"] += 1
            done = stats["ok"] + stats["fail"]
            total = stats["total"]
        print(f"[Step2] 失败 ({done}/{total}) id={record['id']}: {e}")
    finally:
        if request_interval > 0:
            time.sleep(request_interval)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--input", type=str, default=str(R1_RAW_FILE))
    parser.add_argument("--output", type=str, default=str(DISTIL_ARROW))
    parser.add_argument("--workers", type=int, default=5, help="并发线程数，建议 3~8")
    parser.add_argument(
        "--request-interval",
        type=float,
        default=0.2,
        help="每个 worker 完成一条后的间隔秒数，用于降低限流风险",
    )
    parser.add_argument("--overwrite", action="store_true", help="忽略已有评判结果，从头重新评判")
    args = parser.parse_args()

    judged_path = JUDGED_FILE
    arrow_path = DISTIL_ARROW if args.output == str(DISTIL_ARROW) else args.output

    records = load_jsonl(args.input)
    if not records:
        raise RuntimeError(f"未找到 R1 输出: {args.input}，请先运行 step1_generate_r1.py")

    if args.overwrite and judged_path.exists():
        judged_path.unlink()
        print(f"[Step2] 已清空旧评判文件: {judged_path}")

    done_ids = get_processed_ids(judged_path)
    if done_ids:
        print(f"[Step2] 断点续跑: 已有 {len(done_ids)} 条评判记录，将跳过")

    pending = [r for r in records if r["id"] not in done_ids]
    if not pending:
        accepted = rebuild_arrow(judged_path, arrow_path)
        print(f"[Step2] 全部 {len(records)} 条已评判，arrow 已更新: {accepted} 条通过 -> {arrow_path}")
        return

    print(
        f"[Step2] 待评判 {len(pending)}/{len(records)} 条，"
        f"workers={args.workers}，输出={arrow_path}"
    )

    stats = {"ok": 0, "fail": 0, "accepted": 0, "total": len(pending)}
    with ThreadPoolExecutor(max_workers=args.workers) as executor:
        futures = [
            executor.submit(process_one, record, str(judged_path), args.request_interval, stats)
            for record in pending
        ]
        for future in as_completed(futures):
            future.result()

    accepted = rebuild_arrow(judged_path, arrow_path)
    print(
        f"[Step2] 完成: 本次成功 {stats['ok']}，失败 {stats['fail']}，"
        f"累计通过 {accepted}/{len(records)} 条"
    )
    print(f"[Step2] 评判详情: {judged_path}")
    print(f"[Step2] 训练数据: {arrow_path}")


if __name__ == "__main__":
    main()

A.8 test_inference.py

"""微调后模型推理冒烟测试。"""
from transformers import AutoModelForCausalLM, AutoTokenizer

# 优先使用微调后的 checkpoint，回退到基座模型
MODEL_PATH = "/root/autodl-tmp/Distill/saves/qwen2.5-1.5b/smoke-sft/checkpoint-2"
BASE_PATH = "/root/autodl-tmp/Distill/models/models/Qwen--Qwen2.5-1.5B-Instruct/snapshots/master"

import os
model_name = MODEL_PATH if os.path.exists(os.path.join(MODEL_PATH, "model.safetensors")) else BASE_PATH
print(f"Loading model from: {model_name}")

model = AutoModelForCausalLM.from_pretrained(
    model_name, torch_dtype="auto", device_map="auto", low_cpu_mem_usage=True
)
tokenizer = AutoTokenizer.from_pretrained(model_name)

system_prompt = "你是一名助人为乐的助手。"
user_prompt = "请证明根号2是无理数。"
messages = [
    {"role": "system", "content": system_prompt},
    {"role": "user", "content": user_prompt},
]

text = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
model_inputs = tokenizer([text], return_tensors="pt").to(model.device)

generated_ids = model.generate(**model_inputs, max_new_tokens=256)
generated_ids = [
    output_ids[len(input_ids):]
    for input_ids, output_ids in zip(model_inputs.input_ids, generated_ids)
]

response = tokenizer.batch_decode(generated_ids, skip_special_tokens=True)[0]
print("=" * 40)
print("Model response:")
print(response)
print("=" * 40)