端到端大模型训练全链路:SFT、GRPO、量化与部署

单包研判模型(以单条数据包 / HTTP 请求为研判单位)从「能跑」到「能上线」,中间隔着的不是某个算法,而是九道工程环节:数据、SFT 微调、强化学习、权重合并、量化与蒸馏(同一位置上的两条并行降本路线,可单选也可叠加)、部署、压测、测评。这篇不讲项目进度、不摆业务指标,只讲技术本身——每一环的输入输出是什么、关键参数为什么这么定、错了会出现什么现象、又该怎么定位。它是我把一条完整的大模型训练与交付链路从头走通之后沉淀下来的东西,也可以当成一份按环节检索的工程手册。

为什么先要看清「全链路」

大模型项目失败,很少死在算法上,更多死在链路上:数据 schema 不规范,训练脚本跑一半崩;reward 写反了,模型越训越差;权重合并没对齐,量化后再也复现不出效果;部署参数拍脑袋,线上直接把显存打爆。九道环节任何一环断裂,模型都上不了线。

所以我接手后的第一件事,是把链路画出来,给每一环定义「输入 / 输出 / 验收标准」。整条链路分三段:训练侧(数据 & SFT & 强化学习)产出模型;交付侧(权重合并 & 量化 / 蒸馏 & 部署)把它变成可服务、并且能持续压成本;验证侧(压测 & 测评)证明它真的可用。

一个反直觉的经验:别一上来就冲全量。正确顺序是先用小数据把「数据 & 训练 & 合并 & 部署 & 调用」整条链路先跑通,把每一环的接口都对齐,再上全量与大规模训练;否则你会在错误的环节上浪费掉最多的机时。

图 1 · 九道环节按序推进:训练侧产出模型,交付侧把它变成可服务(量化与蒸馏处在同一位置,是两条并行降本路线,可单选也可叠加),验证侧证明它真的可用。

下面按数据 & reward & 训练 & 合并 & 量化 / 蒸馏 & 部署 & 压测 & 测评的顺序展开,每一节都给出我踩过的坑与定位方法。

数据:先把 schema 当成一份契约

训练数据的第一步不是清洗,是定义契约。安全研判的数据是分层来的:先按黑白样本分,再看攻击结果(成功 / 失败 / 不确定),最后带上攻击类型;按 8:2 划分训练 / 验证集,并对齐原始分布。这样做是为了让 RL 阶段每个能力维度都有足够样本,不至于因为某一类占比过高,把策略整体带偏。

抽样不是随便切。做法是:先给每一层(黑白 × 攻击结果 × 攻击类型)算配额,再在层内随机抽取,保证每一类的绝对数量都够 RL 用;最后按 8:2 切分,并核对切分后各层占比与原始一致。这一步做扎实,后面 reward 才不会被数据分布带偏。

真正的麻烦在格式。强化学习框架(verl)要的不是一段对话,而是一条条结构化样本:prompt、reward_model.ground_truth、以及携带标签与元数据的 extra_info。把原始 JSON 转成 parquet 时,我用一个 make_map_fn 逐条构造:

def make_map_fn(split, rl_output_type):
    def process_fn(example, idx):
        question = example.pop("conversations")[0].get("content")
        solution = example.get("true_label")
        return {
            "data_source": DATA_SOURCE,                 # 决定用哪个 reward 函数
            "prompt": [{"role": "user", "content": str(question)}],
            "ability": "security",
            "reward_model": {"style": "rule", "ground_truth": str(solution)},
            "extra_info": {"split": split, "index": idx, "answer": str(solution),
                           "problem_id": str(example.pop("id"))},
        }
    return process_fn

这段映射看着简单,其实是整条链路的接口契约:data_source 决定后面用哪个 reward 函数,ground_truth 决定 reward 怎么比对,extra_info 里的 split / index / answer / problem_id 则决定采样、日志回填与问题定位。任何一处错了,后面所有环节都跟着歪。

踩坑实录:全量数据过了很久,才在第 5888 条上失败——responseBody 字段本应是字符串,实际却是整数。这种脏字段平时没事,一到 rollout 组样本时就会把整批任务带崩,而且报错位置离真正原因很远。我的处理是把质检脚本化:先批量扫描所有 JSON,把所有 responseBody 非字符串的记录连同行号、实际类型一并打印;再用字段统一脚本递归地把 scene / is_attack / requestBody / responseBody 这类应该统一类型的字段一次性规范化。

脏字段不能靠肉眼。我把质检脚本化,扫一遍就定位,顺手把类型规范化:

import json, glob

FIELDS = ["scene", "is_attack", "requestBody", "responseBody"]

# ① 批量扫描:把所有类型不对的记录连同行号打印出来
for fp in glob.glob("data/raw/*.json"):
    for ln, line in enumerate(open(fp, encoding="utf-8"), 1):
        rec = json.loads(line)
        if not isinstance(rec.get("responseBody"), str):
            print(f"[脏字段] {fp}:{ln} responseBody = {type(rec['responseBody']).__name__}")

# ② 字段统一:递归把应统一类型的字段规范化
def normalize(rec):
    for k in FIELDS:
        v = rec.get(k)
        if v is not None and not isinstance(v, str):
            rec[k] = json.dumps(v, ensure_ascii=False) if isinstance(v, (dict, list)) else str(v)
    return rec

RL 数据是一份契约。一个字段类型错了,代价不是一条样本,而是整轮训练的机时。

这一环的验收很简单也很硬:字段类型全部一致、没有缺失、划分后各类分布与原始一致。这三条不满足,不要开始训练。

SFT 微调:先让基座学会「怎么说」

RL 只擅长「在已有的行为空间里挑更好的」,它不负责从零塑造格式与领域知识。所以链路的第一步是监督微调(SFT):让基座先学会两件事——按固定结构输出,以及认识研判任务的语言与标签体系。这是 RL 的冷启动,起点稳了,reward 才有东西可优化。

我走的是全参 SFT(full),不是 LoRA:研判要改的是「行为方式」,而不是「加一个任务头」,全参能更充分地改写输出分布;LoRA 省资源、适合快迭代,但天花板更低。基座用 Qwen2.5-7B-Instruct,配 DeepSpeed ZeRO-3 做分布式。

数据侧只做三件事:整理成指令跟随样本、在 dataset_info.json 里注册数据集、并统一 chat template——训练、测评、部署必须共用同一个模板。配置大致如下:

# sft.yaml(LLaMA-Factory)
model_name_or_path: Qwen2.5-7B-Instruct
stage: sft
finetuning_type: full            # 全参;改 lora 并设 lora_rank 即为 LoRA
deepspeed: examples/deepspeed/ds_z3_config.json
dataset: judge_sft
template: qwen                  # 训练 / 测评 / 部署必须一致
cutoff_len: 4096
preprocessing_num_workers: 16
per_device_train_batch_size: 4
gradient_accumulation_steps: 1
learning_rate: 2.0e-5
lr_scheduler_type: cosine
warmup_ratio: 0.1
num_train_epochs: 6
bf16: true
# torchrun --nproc_per_node=8 src/train.py sft.yaml

踩坑实录:最隐蔽的一类是模板不一致——训练用的 chat template 与推理 / RL 阶段不一致,或特殊 token(如 <|im_start|>)没对齐,会出现「训练 loss 一路下降,推理却直接崩」的假象。对策是模板只定义一处、全程复用;另外要过滤空 response,并按 cutoff_len 处理超长样本。

验收:固定样本上输出结构正确、领域术语与标签体系不出错——SFT 不过关,不要进 RL。

Reward 设计:把「研判」翻译成可优化的信号

RL 里 reward 是唯一告诉模型「什么算好」的东西,写错了模型会朝错误方向飞速进化。这里有个关键设计:一条研判回答其实包含多个字段,如果只对「最终对 / 错」给一个 0 / 1,信号太稀疏、也太粗暴。所以我把 reward 拆成格式 + 结果两块,结果块再按字段给部分分。

格式奖励用严格正则约束输出结构(思维链 + 答案),整体匹配才得 1 分:

import re
# 要求整个输出恰好是 ... 换行 ...
FMT = re.compile(r"^<think>([^<]*(?:<(?!/?think>)[^<]*)*)</think>\n{0,2}<answer>([\s\S]*?)</answer>$", re.S)

def format_reward_func(completion, target):
    return 1.0 if FMT.search(completion) else 0.0

结果奖励把 <answer> 里的字段逐个和 ground_truth 比,命中一个给一份分:是否攻击、攻击结果这类精确值直接相等判定;payload、result_proof 这种列表型字段用集合 F1 软匹配,避免因顺序或冗余表述被全盘判错:

def f1_score_reward(pred_list, gt_list):
    common = set(pred_list) & set(gt_list)
    if not common:
        return 0.0
    p = len(common) / len(set(pred_list))
    r = len(common) / len(set(gt_list))
    return 2 * p * r / (p + r)                 # 集合级 F1

def result_reward_func(completion, target):
    m = re.search(r"<answer>([\s\S]*?)</answer>", completion)
    if not m:
        return 0.0
    ans, reward = m.group(1), 0.0
    for field, w in [("is_attack", 0.34), ("attack_result", 0.33)]:
        pred = re.findall(rf"<{field}>(.*?)</{field}>", ans, re.S)
        gt   = re.findall(rf"<{field}>(.*?)</{field}>", target, re.S)
        if pred and gt and pred[0].strip() == gt[0].strip():
            reward += w                        # 逐字段部分分
    gt_payload = re.findall(r"<payload>(.*?)</payload>", target, re.S)
    if gt_payload:                             # 列表字段:集合 F1 折算
        pred_payload = re.findall(r"<payload>(.*?)</payload>", ans, re.S)
        reward += 0.33 * f1_score_reward(pred_payload, gt_payload)
    return reward

最后按 0.3 × 格式 + 0.7 × 结果 合成。为什么格式只占三成?因为格式不是目的,而是约束推理路径:只奖励结果,模型会学会「跳过思考、直接猜标签」,推理链名存实亡,泛化随之崩塌。同理我也准备了不含 <think> 的变体,按 data_source 绑定到不同数据,让「要不要思维链」可控。

def compute_score(response_str, ground_truth, extra_info):
    format_score = format_reward_func(response_str, ground_truth)
    result_score = result_reward_func(response_str, ground_truth)
    return float(0.3 * format_score + 0.7 * result_score)

落地:reward 函数放在框架的 reward_score 目录,并在 __init__.py 里按 data_source 注册;也可在训练脚本用 custom_reward_function.path / .name 指定,或自定义 reward manager。

验收与警惕:合格 reward 必须能区分对错;同时提防 reward hacking——模型会找最短路径钻空子(只保格式、或猜多数类)。发现异常先看 reward 分布,再改权重与规则,而不是急着调学习率。

GRPO 训练:三层 batch 与 DAPO 的关键开关

同一套框架,换 algorithm.adv_estimator 就切算法:grpo / gae(PPO) / reinforce_plus_plus / rloo。我走 GRPO + DAPO:它不需要 critic,对同一 prompt 采样一组回答,用组内相对得分估计优势 A = (r - mean) / std——既省掉 critic 的显存与训练,又天然抑制奖励尺度漂移,非常适合「判定对不对」这类可自动打分的任务。

要理解参数,先理解三层 batch:train_batch_size 是每次迭代喂给 rollout 的 prompt 数(这里 128);经采样 n=16 展开成 128 × 16 = 2048 条经验。这些经验按 ppo_mini_batch_size 分组,每处理完一组做一次参数更新;组内再按 ppo_micro_batch_size_per_gpu 切成单卡一次前向放得下的 micro-batch,做梯度累积。三层共同决定显存、吞吐与更新频率。

图 2 · 三层 batch:train_batch & mini-batch & micro-batch,切分方式直接决定显存与吞吐。

图 3 · GRPO 一次迭代:组采样 & 组内算优势 & 三层 batch 切分 & 参数更新。

关键开关(真实配置,节选):

# GRPO / DAPO 训练参数,节选(键名以所用版本为准)
algorithm.adv_estimator=grpo
algorithm.use_kl_in_reward=False               # DAPO 不在 reward 里加 KL
actor_rollout_ref.actor.use_kl_loss=False
actor_rollout_ref.actor.kl_loss_type=low_var_kl
actor_rollout_ref.actor.clip_ratio_low=0.2     # 非对称裁剪
actor_rollout_ref.actor.clip_ratio_high=0.28
actor_rollout_ref.actor.loss_agg_mode=token-mean
actor_rollout_ref.actor.entropy_coeff=0

algorithm.filter_groups.enable=True            # 动态采样:丢掉全对 / 全错的组
algorithm.filter_groups.metric=acc
algorithm.filter_groups.max_num_gen_batches=10
algorithm.overlong_buffer.enable=True          # 超长惩罚
algorithm.overlong_buffer.len=1024
algorithm.overlong_buffer.penalty_factor=1.0

data.train_batch_size=128                      # prompt 数
data.max_prompt_length=6144
data.max_response_length=4096
actor_rollout_ref.actor.ppo_mini_batch_size=128
actor_rollout_ref.actor.ppo_micro_batch_size_per_gpu=8
actor_rollout_ref.actor.optim.lr=1.0e-6

actor_rollout_ref.rollout.name=vllm            # 用 vLLM 做 rollout
actor_rollout_ref.rollout.n=16
actor_rollout_ref.rollout.temperature=1.0
actor_rollout_ref.rollout.top_p=0.95
actor_rollout_ref.rollout.top_k=-1
actor_rollout_ref.rollout.tensor_model_parallel_size=2
actor_rollout_ref.rollout.gpu_memory_utilization=0.7
actor_rollout_ref.rollout.max_num_batched_tokens=32768

逐个说清它们为什么这么定:非对称裁剪 0.2 / 0.28(DAPO)放宽上界、允许「变好更多」的更新,同时收紧下界稳住策略;KL 关掉——DAPO 用 clip + 动态采样替代 KL 约束,避免 KL 把模型拽回旧分布;token-mean 让 loss 按 token 而非按序列平均,长回答不再被稀释;动态采样按 acc 过滤掉「全对 / 全错」的组(它们优势恒为 0、不贡献梯度);超长惩罚对超过 buffer 的输出按比例扣分,防止模型拿啰嗦换运气。采样端 n=16、温度 1.0、top_p 0.95,是为了让组内有足够多样性,优势才有区分度。

目标函数的本质还是 PPO 的裁剪式:记重要性比 ρ = π_θ / π_old,单 token 的代理目标取 min(ρ·A, clip(ρ, 1 - ε_low, 1 + ε_high)·A),再对所有 token 求平均(token-mean)。DAPO 把上下界拆开(0.2 / 0.28)、并去掉 KL 项,这是它在长思维链任务上比 PPO 更稳的关键;而 GRPO 只改了一件事——优势不再由 critic 估计,而是用组内标准化得分。

坑与排障:OOM 时先降 ppo_micro_batch_size_per_gpu、再降 mini;输出目录权限问题递归改归属;Hydra 不认的参数用 + 前缀强制注入;调试期把 wandb 切 offline。排障第一原则:先确认报错出在采样 / rollout 还是反向更新,两者调参方向完全不同。

验收:训练曲线平稳、组内优势有区分度、没有大批组被过滤、稳定不 OOM。

权重合并:把分片拼回一个 HF 模型

FSDP 保存出来的不是一整块权重,而是按 rank 切分的分片文件。要部署,必须按每个张量的并行放置方式把它拼回去:replicate 的取其一,shard 的按维度 cat 拼接——也就是逐 key 读各 rank 分片、按 placement 合并,最后用 torch.device('meta') + to_empty 的方式在 CPU 上重组、以 bfloat16 写盘,全程省内存。为什么一定要转成标准 HF 格式?因为只有标准结构的权重,vLLM 才能直接加载部署——这一步没做对,后面量化、部署全是空中楼阁。

import torch

def merge(iterable, out_dir):
    # 同一 key 的各 rank 分片:replicate 取其一,shard 按维度 cat
    for name, shards in iterable:
        full = shards[0] if len(shards) == 1 else torch.cat(shards, dim=0)
        save_bf16(name, full, out_dir)

# 先用 meta device 建骨架,再 to_empty 逐个填权重,峰值内存更低
model = AutoModelForCausalLM.from_config(config, torch_dtype=torch.bfloat16)
model.to_empty(device="cpu")

验收:合并后的模型能直接 from_pretrained 加载,权重 key 与 config / 词表对得上,并且推理输出与训练结束时的采样结果一致。

量化:显存与精度的取舍

线上部署只看两件事:显存够不够、成本高不高。三条路线我都跑了一遍,它们不是替代关系,而是三种取舍:

🔢

bitsandbytes

最省事

加载即量化,4-bit(nf4)配双量化,几乎零改造;适合快速验证,但推理吞吐一般。

nf4 · 双量化开箱即用

📐

GPTQ

训练后量化

需要校准集,按块逐层最小化量化误差(group_size 控制粒度),压缩率高、部署生态成熟。

需要校准group_size 128

🎯

AWQ

激活感知 · 精度最稳

按激活分布保护「重要通道」,W4A16 非对称方案,lm_head 保持不动;校准数据质量是关键。

W4A16_ASYM保护关键通道

bitsandbytes最省事,加载即量化:NF4 码本 + 双量化,计算回退到 bf16——几乎零改造,适合先验证能不能跑,但吞吐一般。

from transformers import BitsAndBytesConfig
bnb = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",        # 正态分布最优的 4bit 码本
    bnb_4bit_use_double_quant=True,   # 量化常数再量化,再省一点显存
    bnb_4bit_compute_dtype=torch.bfloat16,
)
model = AutoModelForCausalLM.from_pretrained(path, quantization_config=bnb, device_map="auto")

GPTQ是训练后量化:按列逐层最小化量化误差(Hessian 近似),group_size 控制分组粒度,damp_percent 是数值稳定项:

from auto_gptq import AutoGPTQForCausalLM, BaseQuantizeConfig
qc = BaseQuantizeConfig(bits=4, group_size=128, desc_act=False, damp_percent=0.1)
model = AutoGPTQForCausalLM.from_pretrained(path, quantize_config=qc, device_map="auto")
model.quantize(calib_input_ids)        # 用校准样本估计每层量化误差
model.save_quantized(out_path)

AWQ对精度最友好:它按激活分布识别「重要通道」并放大其权重,再做分组量化,方案是 W4A16 非对称(权重 4bit、激活 16bit),并跳过 lm_head(输出头对精度最敏感,动它得不偿失)。我用 llmcompressor 的 oneshot 实现:

from llmcompressor import oneshot
from llmcompressor.modifiers.awq import AWQModifier

recipe = [AWQModifier(ignore=["lm_head"], scheme="W4A16_ASYM", targets=["Linear"])]
oneshot(model=model, dataset=calib_ds, recipe=recipe,
        num_calibration_samples=128, max_seq_length=512)

AWQ 上我花的心思最多,关键在校准数据:直接从微调数据里取,并转成模型自己的对话模板文本(带 <|im_start|> 等特殊 token)再当校准集。这样量化器才能「看到」研判场景真实的激活分布——拿通用语料校准,量化后的模型在专有输入上会明显掉点。转换完先校验样本条数、长度分布与格式标记是否齐全,再进量化。

验收:量化前后用同一套样本对比,效果不掉出可接受区间;掉得多,先怀疑校准数据的分布不对,而不是量化方法不行。同时记录压缩率(原模型大小 / 量化后大小)与显存占用,作为选型依据。

蒸馏:把能力压缩进小模型

量化之外的另一条降本路线是蒸馏:让一个小的学生模型去学大模型的行为。两者的区别是——量化改精度、不动结构,一步就能做完;蒸馏直接换一个小模型,中间要跑一次训练。前置硬坑是教师与学生的 vocab_size 必须先对齐——词表不一致时,蒸馏的对数几率对不上,训练直接崩。对齐方式:

from transformers import AutoModelForCausalLM, AutoTokenizer
m = AutoModelForCausalLM.from_pretrained(src)
m.resize_token_embeddings(151936)          # 对齐到教师词表大小
m.save_pretrained(dst)
AutoTokenizer.from_pretrained(src).save_pretrained(dst)   # tokenizer 一并保存

顺手写个模型检查脚本,确认 vocab_size、特殊 token、参数量与 dtype 都对得上,再进蒸馏。验收:词表一致、loss 不跳不崩,并且学生模型在固定测评口径下接近教师。

部署:vLLM 起来,才算数

部署不是把模型拷过去那么简单,关键在参数与显存的平衡:

python3 -m vllm.entrypoints.openai.api_server \
  --served-model-name judge-model \
  --model /path/to/hf_model \
  --dtype bfloat16 \
  --gpu-memory-utilization 0.9 \
  --max-model-len 32768 \
  --tensor-parallel-size 2

逐条看:--max-model-len 不是越大越好——它直接吃 KV cache 显存(KV cache 大致正比于层数 × KV 头数 × head_dim × 字节数 × 序列长度 × 并发数),开太大,并发一上来就崩;--gpu-memory-utilization 要和目标并发量一起定;多卡用 --tensor-parallel-size 摊开;--dtype bfloat16 是精度与显存之间的折中。而真正让吞吐上得去的,是底层的 continuous batching 与 PagedAttention。

服务起来只是第一步,真正的问题是:它扛不扛得住线上并发?

压测与测评:自己证明「能用」

验证分两层。性能层用并发压测打接口,核心指标:QPS、Token 吞吐、请求延迟分布(P50 / P95 / P99 / 平均 / 标准差)、TTFT(首 token 时间)、TPOT(每 token 时间)、请求成功率。为什么盯分布而不是均值?均值会被大量快请求掩盖,真正暴露问题的是 P99 那条尾巴。

压测的核心是一个流式异步客户端,把 TTFT 与 TPOT 逐请求量出来:

import asyncio, time, aiohttp

async def one(session, payload):
    t0 = time.perf_counter(); ttft = None; n = 0
    async with session.post(URL, json=payload) as r:
        async for line in r.content:                 # 流式逐块读
            if line.startswith(b"data:") and b"content" in line:
                if ttft is None:
                    ttft = time.perf_counter() - t0
                n += 1
    dt = time.perf_counter() - t0
    return ttft, (dt - ttft) / max(n - 1, 1)         # 返回 TTFT / TPOT

压测要做的不是打一个数,而是找拐点:固定并发逐步加压,看 QPS 何时不再涨、P99 何时翘头——那是当前配置的容量上限,也是扩卡或调 --max-model-len / 并发策略的依据。

效果层是一套固定口径的测评流水线:先把黑白样本间隔排布再随机打乱(避免模型靠顺序猜答案),并发请求,再按三套口径统计,最后合并成一份报告:

检测维度:是否攻击。用混淆矩阵 TP / FN / FP / TN 算漏报率(黑样本被判白,FN / 黑样本)与误报率(白样本被判黑,FP / 白样本);
研判维度:攻击结果判定(成功 / 失败 / 不确定);
按攻击类型:逐类拆开,才知道下一轮该补哪类数据。

def binary_metrics(rows):
    TP = sum(r.pred == "是" and r.gt == "是" for r in rows)
    FN = sum(r.pred == "否" and r.gt == "是" for r in rows)   # 漏报
    FP = sum(r.pred == "是" and r.gt == "否" for r in rows)   # 误报
    TN = sum(r.pred == "否" and r.gt == "否" for r in rows)
    black, white = TP + FN, FP + TN
    return {"FNR": FN / black if black else 0,
            "FPR": FP / white if white else 0}

这套口径是模型迭代与回滚的判据:没有固定口径的测评,再好的模型也无法判断该发版还是该回退。两条经验:压测和测评要用同一套真实输入,不然「性能好」只是空跑出来的幻觉;样本必须打乱顺序,否则模型学的是位置,而不是内容。

把方法论沉淀下来

走通这一遍之后,我把它固化成了一套可复用的 checklist:

数据:先定 schema 契约,再做格式质检——脏字段要在训练前拦住;
Reward:可解释、可度量,格式与结果分开加权,并提防 reward hacking;
小规模试跑:用小数据把「数据 & 训练 & 合并 & 部署 & 调用」整条链路先跑通,再上全量;
训练:三层 batch 与采样参数一起调,盯 OOM 与权限,先分清报错出在采样还是更新;
交付:合并成 HF 格式 & 压缩(量化用微调分布做校准,或蒸馏出学生模型,两者可叠加)& vLLM 部署;
验证:压测看延迟分布,测评看多口径统计,指标不回退才发版。

掌握一个技术,不是会调一个参数,而是知道每个参数错了会怎样、怎么定位、怎么取舍——这,才是端到端的能力。