PEFT 技术综述

技术发展脉络

Adapter Tuning

Adapter Tuning 是 PEFT 领域的开创性工作之一，由 Google 在 2019 年为 BERT 模型设计。其思路是在 Transformer 的每个块中插入小型的“适配器”（Adapter）模块。如图所示，左侧的 Transformer 层展示了 Adapter 模块是如何被集成进去的。Adapter 被插入到每个子层（注意力层和前馈网络）的内部，并与主干网络形成残差连接。在训练时，只有 Adapter 模块的参数会被更新。

图的右侧展示了 Adapter 模块自身的结构：

• 一个“降维”的全连接层（Feedforward down-project），将高维特征映射到低维空间。
• 一个非线性激活函数（Nonlinearity）。
• 一个“升维”的全连接层（Feedforward up-project），再将特征映射回原始维度。
• 一个贯穿该模块的残差连接，将模块的输出与原始输入相加，保证信息流的稳定。

Prefix Tuning

2021 年，斯坦福大学的研究者提出了 Prefix Tuning，为 PEFT 开辟了一条全新的思路。与 Adapter 在模型内部“动手术”不同，Prefix Tuning 选择在模型外部做文章，就像是给模型带上了一张“小抄”。下午是一个注解示例，揭示了 Prefix Tuning 的工作细节。该图分别展示了 Prefix Tuning 在自回归语言模型（上）和编码器-解码器模型（下）中的应用。

核心机制

• 前缀激活值（Prefix Activations）：图中 PREFIX 部分对应的激活值 $h_i$（其中 $i ∈ P_idx$）是从一个专门的可训练矩阵 $P_{\theta}$ 中提取的，这部分参数就是微调的对象。
• 模型计算的激活值: 而原始输入 $x$ 和输出 $y$ 对应的激活值，则是由冻结的 Transformer 模型正常计算得出的。

优缺点

优点
较高的参数效率：仅优化极少数 Prefix 参数，无需改动原模型。
显存友好：由于不更新原模型权重（仅训练前缀参数），训练时无需为原模型权重维护优化器状态，显著降低显存/存储开销；但需要为各层前缀的 K/V 额外占用预留显存。
通用性强：在自回归模型（如 GPT-2）和编解码模型（如 T5/BART）上均取得了不错的效果。
缺点
训练不稳定：直接优化 Prefix 向量比微调 Adapter 更困难，对超参数和初始化较为敏感。
占用上下文长度：多数实现将前缀作为各层注意力的额外 K/V 记忆，其长度通常计入注意力配额，从而减少可用的有效上下文窗口（实现相关，取决于具体框架与实现方式）。

Prompt Tuning

Prefix Tuning 虽然强大，但其复杂的训练过程和在每一层都添加参数的设计，在实践中不够便捷。同年，Google 提出了 Prompt Tuning，可以看作是 Prefix Tuning 的一个简化版 4。这种方法也被称为一种“软提示”。它的做法就是只在输入的 Embedding 层添加可学习的虚拟 Token（称为 Soft Prompt），而不再干预 Transformer 的任何中间层。

P-Tuning v2

P-Tuning V1

Prompt Tuning 虽然足够高效，但它的稳定性较差，且严重依赖超大模型的规模，这限制了其在更广泛场景中的应用。为了解决这些问题，由清华大学团队主导的 P-Tuning 系列工作，对软提示进行了深入优化，最终发展出了效果更强、更通用的 P-Tuning v2。

P-Tuning V2

2021 年底问世的 P-Tuning v2，就是为了解决 v1 的局限性而设计的 6。它博采众长，吸收了 Prefix Tuning 的思想，最终成为一种在不同模型规模、不同任务上都表现出色的通用 PEFT 方案。

结构对比

LoRA

LoRA（Low-Rank Adaptation of Large Language Models）是当前社区应用最广泛的 PEFT 方法

优势

• 更高的参数与存储效率：对于每一个下游任务，不再需要存储一个完整的模型副本，而只需保存极小的矩阵 A 和 B。论文指出，这可以将模型 checkpoints 的体积缩小高达 10,000 倍（例如从 350GB 减小到 35MB）。在训练时，由于无需为冻结的参数计算梯度和存储优化器状态，可以节省高达 2/3 的 GPU 显存，并提升约 25% 的训练速度。
• 零额外推理延迟：这是 LoRA 相比 Adapter Tuning 最具吸引力的优点。Adapter 在模型中串行地引入了新的计算层，不可避免地会增加推理延迟。而 LoRA 的旁路结构在训练完成后，可以通过矩阵加法 $(W' = W_0 + s \cdot B \cdot A)$ 直接“合并”回原始权重中。这样，模型的网络结构与原始模型完全一致，不会引入任何额外的计算步骤。
• 效果媲美全量微调，且不占用输入长度：与 Prompt-Tuning 等作用于输入激活值的方法不同，LoRA 直接修改权重矩阵，能更深入、更直接地影响模型的行为，效果也更接近于全量微调。同时，它不添加任何 virtual token，不会占用上下文长度，在处理长文本任务时更有优势。
• 良好的可组合性：LoRA 的设计是 正交的，它可以与 Prefix-Tuning 等其他 PEFT 方法结合使用，取长补短，进一步提升模型性能。

基于 peft 库的 LoRA 实战

安装依赖

pip install torch torchvision torchaudio transformers
pip install bitsandbytes
pip install --upgrade accelerate

下载模型

export HF_ENDPOINT=https://hf-mirror.com
huggingface-cli download --resume-download EleutherAI/pythia-2.8b-deduped --local-dir ./models/pythia-2.8b-deduped

下载数据集

huggingface-cli download --repo-type dataset --resume-download Abirate/english_quotes --local-dir ./dataset/english_quotes

加载依赖、基础模型与分词器

from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
import torch

model_id = "./models/pythia-2.8b-deduped"

# --- 使用 BitsAndBytesConfig 定义 8-bit 量化配置 ---
bnb_config = BitsAndBytesConfig(
    load_in_8bit=True,
)

# 加载模型，并将量化配置传给 `quantization_config` 参数
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    quantization_config=bnb_config,
    dtype=torch.float16,
    device_map="auto",
)

打印模型信息

GPTNeoXForCausalLM(
  (gpt_neox): GPTNeoXModel(
    (embed_in): Embedding(50304, 2560)
    (emb_dropout): Dropout(p=0.0, inplace=False)
    (layers): ModuleList(
      (0-31): 32 x GPTNeoXLayer(
        (input_layernorm): LayerNorm((2560,), eps=1e-05, elementwise_affine=True)
        (post_attention_layernorm): LayerNorm((2560,), eps=1e-05, elementwise_affine=True)
        (post_attention_dropout): Dropout(p=0.0, inplace=False)
        (post_mlp_dropout): Dropout(p=0.0, inplace=False)
        (attention): GPTNeoXAttention(
          (query_key_value): Linear8bitLt(in_features=2560, out_features=7680, bias=True)
          (dense): Linear8bitLt(in_features=2560, out_features=2560, bias=True)
        )
        (mlp): GPTNeoXMLP(
          (dense_h_to_4h): Linear8bitLt(in_features=2560, out_features=10240, bias=True)
          (dense_4h_to_h): Linear8bitLt(in_features=10240, out_features=2560, bias=True)
          (act): GELUActivation()
        )
      )
    )
    (final_layer_norm): LayerNorm((2560,), eps=1e-05, elementwise_affine=True)
    (rotary_emb): GPTNeoXRotaryEmbedding()
  )
  (embed_out): Linear(in_features=2560, out_features=50304, bias=False)
)

加载token

tokenizer = AutoTokenizer.from_pretrained(model_id)
# Pythia模型的tokenizer默认没有pad_token，我们将其设置为eos_token
tokenizer.pad_token = tokenizer.eos_token

模型预处理

安装peft

pip install peft

执行底阿妈

from peft import prepare_model_for_kbit_training

# 对量化后的模型进行预处理
model = prepare_model_for_kbit_training(model)

定义 LoRA 配置并创建 PeftModel

from peft import LoraConfig, get_peft_model

# 定义 LoRA 配置
config = LoraConfig(
    r=16,
    lora_alpha=32,
    target_modules=["query_key_value", "dense"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
)

# 应用配置，获得 PEFT 模型
peft_model = get_peft_model(model, config)
peft_model.print_trainable_parameters()

输出信息

trainable params: 7,864,320 || all params: 2,783,073,280 || trainable%: 0.2826

数据处理

安装依赖

pip install datasets

执行代码

from datasets import load_dataset

# 加载数据集
quotes_dataset = load_dataset("./dataset/english_quotes")

# 查看数据集示例
quotes_dataset['train'][0]

输出结果

{'quote': '“Be yourself; everyone else is already taken.”',
 'author': 'Oscar Wilde',
 'tags': ['be-yourself',
  'gilbert-perreira',
  'honesty',
  'inspirational',
  'misattributed-oscar-wilde',
  'quote-investigator']}

定义分词函数并将其应用到整个数据集上。

# 定义分词函数
def tokenize_quotes(batch):
    # 只对 "quote" 列进行分词
    return tokenizer(batch["quote"], truncation=True)

# 对整个数据集进行分词处理
tokenized_quotes = quotes_dataset.map(tokenize_quotes, batched=True)

tokenized_quotes['train'][0]

输出结果

{'quote': '“Be yourself; everyone else is already taken.”',
 'author': 'Oscar Wilde',
 'tags': ['be-yourself',
  'gilbert-perreira',
  'honesty',
  'inspirational',
  'misattributed-oscar-wilde',
  'quote-investigator'],
 'input_ids': [1628, 4678, 4834, 28, 4130, 2010, 310, 2168, 2668, 1425],
 'attention_mask': [1, 1, 1, 1, 1, 1, 1, 1, 1, 1]}

定义 Trainer 并开始训练

from transformers import Trainer, TrainingArguments, DataCollatorForLanguageModeling

# 推荐操作：关闭缓存可提高训练效率
peft_model.config.use_cache = False

# 定义训练参数
train_args = TrainingArguments(
    per_device_train_batch_size=4,
    gradient_accumulation_steps=4,
    warmup_steps=100,
    max_steps=200,
    learning_rate=2e-4,
    fp16=True, # 启用混合精度训练
    logging_steps=1,
    output_dir="outputs",
)

# 数据整理器，用于处理批量数据
quote_collator = DataCollatorForLanguageModeling(tokenizer, mlm=False)

# 实例化 Trainer
trainer = Trainer(
    model=peft_model,
    train_dataset=tokenized_quotes["train"],
    args=train_args,
    data_collator=quote_collator,
)

# 开始训练
trainer.train()

关键的训练参数

• per_device_train_batch_size & gradient_accumulation_steps：这两个参数共同决定了有效批量大小（effective batch size）。per_device_train_batch_size 是指每个 GPU 单次前向传播处理的样本数，而 gradient_accumulation_steps 则指定了梯度累积的步数。有效批量大小 = per_device_train_batch_size * gradient_accumulation_steps * num_gpus。通过梯度累积，可以在显存有限的情况下，模拟出更大的批量大小，这通常有助于稳定训练过程。
• warmup_steps: 学习率预热的步数。在训练初期，学习率会从一个很小的值线性增加到设定的 learning_rate，这能让模型在开始阶段更好地适应数据。
• max_steps: 训练的总步数。为了快速演示，这里只训练 200 步。
• learning_rate: 学习率，控制模型参数更新的幅度。
• fp16: 启用 16-bit 混合精度训练。可以在不牺牲太多性能的情况下，进一步减少显存占用并加速训练。

模型保存与推理

# 将模型设置为评估模式
peft_model.eval()

# 设置 pad_token_id 到模型配置中
peft_model.config.pad_token_id = tokenizer.pad_token_id

prompt = "Be yourself; everyone"

# 对输入进行分词，并获取 attention_mask
inputs = tokenizer(prompt, return_tensors="pt")
input_ids = inputs["input_ids"].to(peft_model.device)
attention_mask = inputs["attention_mask"].to(peft_model.device)

# 生成文本
with torch.no_grad():
    # 使用 autocast 提高混合精度推理的效率
    with torch.amp.autocast('cuda'):
        outputs = peft_model.generate(
            input_ids=input_ids,
            attention_mask=attention_mask,
            max_length=50,
            do_sample=True,
            temperature=0.6,
            top_p=0.95,
            top_k=40,
            repetition_penalty=1.2,
            pad_token_id=tokenizer.pad_token_id
        )

# 解码并打印结果
decoded_output = tokenizer.decode(outputs[0], skip_special_tokens=True)
decoded_output

输出结果

'Be yourself; everyone else is taken.” - Oscar Wilde\n“I have lived my life according to the belief that I was given everything and so it’s only fair for me not to get anything” – Oprah Winfrey. “'

生成参数说明：

• max_length: 生成文本的最大长度（包括输入）。
• do_sample: 是否使用采样策略。设置为 True 时，temperature、top_p、top_k 才会生效。
• temperature: 控制生成的随机性。较低的值（如 0.6）会使生成更具确定性，而较高的值则会增加多样性。
• top_p: 核采样的概率阈值。只考虑累积概率达到 top_p 的最小 token 集合。
• top_k: 每步只从概率最高的 k 个 token 中采样。
• repetition_penalty: 重复惩罚因子，大于 1.0 会降低重复内容的概率。

Qwen2.5 微调私有数据

模型介绍

Qwen2.5 是阿里巴巴开源的高性能大语言模型家族，涵盖了从 0.5B 到 72B 的多种参数规模，以满足不同场景的需求。它不仅是一个单独的模型，更是一个强大的基础平台，在其之上衍生出了 Qwen2.5-Math、Qwen2.5-Coder 等专业模型。
在图中展示了 Qwen 系列模型（从 Qwen1.5-72B 到 Qwen2.5-72B）的性能与其预训练数据量之间的正相关关系。随着数据规模从 3 万亿 Token 增长至 18 万亿 Token，模型在 MMLU、BBH、MBPP 和 Math 等多个关键基准测试上的得分均稳步提升。这证明，海量、高质量的预训练让 Qwen2.5 拥有了强大的通用能力。

构建私有微调数据集

在投入资源微调前，必须先摸清基础模型的底细，确认微调的必要性。这里选择游戏《黑神话：悟空》进行效果测试。

模型下载

modelscope download --model Qwen/Qwen2.5-VL-7B-Instruct --local_dir ./models/Qwen2.5-VL-7B-Instruct

加载量化模型与分词器

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig

# 定义模型 ID
model_id = "./models/Qwen2.5-7B-Instruct"

# 加载分词器
tokenizer = AutoTokenizer.from_pretrained(model_id)

# 定义4-bit量化配置
bnb_config = BitsAndBytesConfig(load_in_4bit=True)

# 加载模型
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    quantization_config=bnb_config,
    device_map="cuda:0",
)

打印模型

print(model)

模型结构

Qwen2ForCausalLM(
  (model): Qwen2Model(
    (embed_tokens): Embedding(152064, 3584)
    (layers): ModuleList(
      (0-27): 28 x Qwen2DecoderLayer(
        (self_attn): Qwen2Attention(
          (q_proj): Linear4bit(in_features=3584, out_features=3584, bias=True)
          (k_proj): Linear4bit(in_features=3584, out_features=512, bias=True)
          (v_proj): Linear4bit(in_features=3584, out_features=512, bias=True)
          (o_proj): Linear4bit(in_features=3584, out_features=3584, bias=False)
        )
        (mlp): Qwen2MLP(
          (gate_proj): Linear4bit(in_features=3584, out_features=18944, bias=False)
          (up_proj): Linear4bit(in_features=3584, out_features=18944, bias=False)
          (down_proj): Linear4bit(in_features=18944, out_features=3584, bias=False)
          (act_fn): SiLUActivation()
        )
        (input_layernorm): Qwen2RMSNorm((3584,), eps=1e-06)
        (post_attention_layernorm): Qwen2RMSNorm((3584,), eps=1e-06)
      )
    )
    (norm): Qwen2RMSNorm((3584,), eps=1e-06)
    (rotary_emb): Qwen2RotaryEmbedding()
  )
  (lm_head): Linear(in_features=3584, out_features=152064, bias=False)
)

定义推理函数

def chat(user_message, system_message="你是《黑神话：悟空》领域助手，回答准确、简明。"):
    messages = [
        {"role": "system", "content": system_message},
        {"role": "user", "content": user_message}
    ]
    
    # 应用对话模板
    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(
        input_ids=model_inputs.input_ids,
        attention_mask=model_inputs.attention_mask,
        max_new_tokens=256
    )
    
    # 解码时跳过 prompt 部分
    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]
    
    return response

执行评估

question_1 = "我该怎么成为天命人？"
answer_1 = chat(question_1)
print(f"问题: {question_1}\n回答:\n{answer_1}")

question_2 = "如何获得并合成出云棍？"
answer_2 = chat(question_2)
print(f"问题: {question_2}\n回答:\n{answer_2}")

输出结果

问题: 我该怎么成为天命人？
回答:
在《黑神话：悟空》的设定中，成为天命人需要满足特定条件，具体步骤如下：

1. 完成“天命之子”的任务线。
2. 在特定的时间点和地点接受考验。
3. 通过考验后，主角将被选定为“天命人”，拥有特殊的能力和使命。

请注意，游戏的具体细节可能会有所不同，以上信息基于已知的游戏设定。实际操作时，请参照游戏内的指引。

通过 LLM 构建数据集

初始化与配置

import os
import re
import json
import datetime
import time
import random
import glob

# 路径与配置
DATA_DIR = "./data"
SRC_MD = f"{DATA_DIR}/blackwukong.md"
TS = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
OUT_BASE_JSONL = f"{DATA_DIR}/wukong_base_{TS}.jsonl"
OUT_JSONL = f"{DATA_DIR}/wukong_dataset_{TS}.jsonl"

from openai import OpenAI

# 注意：为演示方便，这里直接在代码中写入密钥与模型，不推荐在生产环境硬编码敏感信息，建议改用环境变量或密钥管理服务
BASE_URL = "https://api.siliconflow.cn/v1"
MODEL_ID = "Qwen/Qwen3-235B-A22B-Instruct-2507"
API_KEY = "sk-youkey"

client = OpenAI(api_key=API_KEY, base_url=BASE_URL)
print(f"Using model: {MODEL_ID} @ {BASE_URL}")

读取与切分源数据

with open(SRC_MD, "r", encoding="utf-8") as f:
    raw_markdown = f.read()

# 按标题切分
matches = list(re.finditer(r"(?m)^(#{2,3})\\s+(.+)$", raw_markdown))
sections = []
# ... (此处省略切分与去重逻辑) ...
for i, m in enumerate(matches):
    s = m.start()
    e = matches[i + 1].start() if i + 1 < len(matches) else len(raw_markdown)
    block = raw_markdown[s:e].strip()
    if len(block) >= 100:
        sections.append(block)
# ...
print(f"sections={len(sections)}")

教师模型调用与解析

SYS_PROMPT = (
    "你是《黑神话：悟空》的资深资料整理者。"
    "将给定原文片段转写为一条训练样本，严格输出JSON："
    '{"instruction":"用户问题","output":"权威完整答案"}。'
    "要求："
    "1. instruction 是自然语言问题；"
    "2. output 仅依据原文，不要臆测；"
    "3. 禁止任何额外说明或代码块。"
)

def ask_teacher(block: str) -> str:
    resp = client.chat.completions.create(
        model=MODEL_ID,
        messages=[
            {"role": "system", "content": SYS_PROMPT},
            {"role": "user", "content": block},
        ],
        temperature=0.2,
        max_tokens=600,
        response_format={"type": "json_object"},
    )
    return resp.choices[0].message.content

def parse_json_pair(text: str):
    m = re.search(r"\{[\s\S]*\}", text)
    if not m:
        raise ValueError("教师模型返回非JSON")
    obj = json.loads(m.group(0))
    ins = (obj.get("instruction") or "").strip()
    out = (obj.get("output") or "").strip()
    if not ins or not out:
        raise ValueError("缺少必要字段")
    return ins, out

生成基础 instruction/output，并写入 OUT_BASE_JSONL

os.makedirs(os.path.dirname(OUT_BASE_JSONL), exist_ok=True)
base_written = 0

with open(OUT_BASE_JSONL, "w", encoding="utf-8") as fbase:
    for seg in sections:
        resp = None
        for _attempt in range(3):
            try:
                resp = client.chat.completions.create(
                    model=MODEL_ID,
                    messages=[
                        {"role": "system", "content": SYS_PROMPT},
                        {"role": "user", "content": seg},
                    ],
                    temperature=0.2,
                    max_tokens=600,
                    response_format={"type": "json_object"},
                )
                break
            except Exception:
                if _attempt == 2:
                    resp = None
                    break
                time.sleep(1.5 ** _attempt + random.random() * 0.3)
        if resp is None:
            continue
        obj = json.loads(resp.choices[0].message.content)
        ins = (obj.get("instruction") or "").strip()
        out = (obj.get("output") or "").strip()
        if not ins or not out:
            continue
        fbase.write(json.dumps({"instruction": ins, "output": out}, ensure_ascii=False) + "\n")
        base_written += 1

print(f"base saved: {base_written} -> {OUT_BASE_JSONL}")

这一步如果出现json报错，可以多执行几次就可以出结果了，也可以使用魔搭的API进行体验。

读取基础集，进行问法改写并写入最终集 OUT_JSONL

NUM_VARIANTS = 14

os.makedirs(os.path.dirname(OUT_JSONL), exist_ok=True)
written = 0
seen_q = set()

# 选择最新的基础集文件
base_files = sorted(glob.glob(f"{DATA_DIR}/wukong_base_*.jsonl"), key=os.path.getmtime, reverse=True)
IN_BASE_JSONL = base_files[0]

with open(IN_BASE_JSONL, "r", encoding="utf-8") as fr, open(OUT_JSONL, "w", encoding="utf-8") as fw:
    for line in fr:
        line = line.strip()
        if not line:
            continue
        obj = json.loads(line)
        base_q = (obj.get("instruction") or "").strip()
        answer = (obj.get("output") or "").strip()
        if not base_q or not answer:
            continue

        r2 = None
        for _attempt in range(3):
            try:
                r2 = client.chat.completions.create(
                    model=MODEL_ID,
                    messages=[
                        {"role": "system", "content": "严格输出 JSON 对象：{\\\"paraphrases\\\": [\\\"...\\\"]}；禁止任何额外文本/代码块/前后缀。若需引号请用中文引号「」或在 JSON 中转义为 \\\\\"。每项必须是可直接回答的等价问法，不改变边界与条件。"},
                        {"role": "user", "content": f"基础问题：{base_q}\n数量：{NUM_VARIANTS}\n输出键：paraphrases"},
                    ],
                    temperature=0.6,
                    max_tokens=800,
                    response_format={"type": "json_object"},
                )
                break
            except Exception:
                if _attempt == 2:
                    r2 = None
                    break
                time.sleep(1.5 ** _attempt + random.random() * 0.3)
        if r2 is None:
            continue
        obj2 = json.loads(r2.choices[0].message.content)
        arr = obj2.get("paraphrases", [])
        arr = [x.strip() for x in arr if isinstance(x, str) and x.strip()]
        if not arr:
            continue

        # 规范：以问号结尾并全局去重
        for s in arr:
            if not s.endswith(("?", "？")):
                s = s.rstrip("？?") + "？"
            if s in seen_q:
                continue
            seen_q.add(s)
            fw.write(json.dumps({"instruction": s, "output": answer}, ensure_ascii=False) + "\n")
            written += 1

print(f"saved: {written} -> {OUT_JSONL}")

模型微调与评估

加载数据集、分词器与模型

import os, glob, json, datetime, math
import torch
from datasets import load_dataset
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
from peft.utils import TRANSFORMERS_MODELS_TO_LORA_TARGET_MODULES_MAPPING

# 基础配置
checkpoint_id = "./models/Qwen2.5-7B-Instruct"
artifacts_dir = "./checkpoints"
data_dir = "./data"
max_seq_len = 2048
seed_num = 42

os.makedirs(artifacts_dir, exist_ok=True)

# 选择最新训练集（wukong_dataset_*.jsonl）
jsonl_files = sorted(glob.glob(os.path.join(data_dir, "wukong_dataset_*.jsonl")), key=os.path.getmtime, reverse=True)

train_jsonl = jsonl_files[0]
print(f"using dataset: {train_jsonl}")

# 加载数据集
train_set = load_dataset("json", data_files=train_jsonl, split="train")
train_set

加载分词器

tokenizer = AutoTokenizer.from_pretrained(checkpoint_id, trust_remote_code=True)
if tokenizer.pad_token_id is None:
    tokenizer.pad_token = tokenizer.eos_token
tokenizer.pad_token, tokenizer.eos_token, tokenizer.pad_token_id, tokenizer.eos_token_id

# 4bit 量化（QLoRA）

# 4bit 量化（QLoRA）
compute_dtype = torch.bfloat16 if torch.cuda.is_available() and torch.cuda.is_bf16_supported() else torch.float16
bnb_cfg = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_use_double_quant=True,
    bnb_4bit_compute_dtype=compute_dtype,
)

base_model = AutoModelForCausalLM.from_pretrained(
    checkpoint_id,
    trust_remote_code=True,
    quantization_config=bnb_cfg,
    device_map="cuda:0",
)
base_model.config.use_cache = False
base_model.gradient_checkpointing_enable()
# k-bit 训练准备（关键，否则反向无 grad）
base_model = prepare_model_for_kbit_training(base_model)

定义 LoRA 配置

from peft import LoraConfig, get_peft_model
from peft.utils import TRANSFORMERS_MODELS_TO_LORA_TARGET_MODULES_MAPPING

# LoRA 配置
lora_cfg = LoraConfig(
    r=16,
    lora_alpha=32,
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
    target_modules=TRANSFORMERS_MODELS_TO_LORA_TARGET_MODULES_MAPPING["qwen2"],
)
peft_model = get_peft_model(base_model, lora_cfg)
peft_model.print_trainable_parameters()

构造监督样本

# 构造监督样本：使用 Qwen 对话模板，并仅对 assistant 段落计算 loss
from datasets import Dataset

def format_sample_for_qwen(record):
    instr = (record.get("instruction") or "").strip()
    ans = (record.get("output") or "").strip()
    if not instr or not ans:
        return {"input_ids": [], "labels": []}

    msgs_no_assist = [
        {"role": "system", "content": "你是《黑神话：悟空》领域助手，回答准确、简明。"},
        {"role": "user", "content": instr},
    ]
    # prompt（包含 assistant 起始标记）
    prompt_ids = tokenizer.apply_chat_template(
        msgs_no_assist,
        tokenize=True,
        add_generation_prompt=True,
        return_tensors=None,
    )

    msgs_full = msgs_no_assist + [{"role": "assistant", "content": ans}]
    full_ids = tokenizer.apply_chat_template(
        msgs_full,
        tokenize=True,
        add_generation_prompt=False,
        return_tensors=None,
    )

    # 截断到 max_seq_len
    full_ids = full_ids[:max_seq_len]
    # 计算分界位置
    cut = min(len(prompt_ids), len(full_ids))
    labels = [-100] * cut + full_ids[cut:]

    return {"input_ids": full_ids, "labels": labels}

proc_train = train_set.map(format_sample_for_qwen, remove_columns=train_set.column_names)
proc_train = proc_train.filter(lambda x: len(x["input_ids"]) > 0)
proc_train

数据整理

from typing import List, Dict

class QwenSftCollator:
    def __init__(self, pad_id: int, max_length: int = 2048, ignore_id: int = -100):
        self.pad_id = pad_id
        self.max_length = max_length
        self.ignore_id = ignore_id

    def __call__(self, features: List[Dict]):
        max_len = max(len(f["input_ids"]) for f in features)
        max_len = min(max_len, self.max_length)
        input_ids, labels = [], []
        for f in features:
            ids = f["input_ids"][:max_len]
            lbs = f["labels"][:max_len]
            pad = max_len - len(ids)
            if pad > 0:
                ids = ids + [self.pad_id] * pad
                lbs = lbs + [self.ignore_id] * pad
            input_ids.append(torch.tensor(ids, dtype=torch.long))
            labels.append(torch.tensor(lbs, dtype=torch.long))
        return {"input_ids": torch.stack(input_ids), "labels": torch.stack(labels)}

collator = QwenSftCollator(pad_id=tokenizer.pad_token_id, max_length=max_seq_len)

定义训练器并开始训练

from transformers import TrainingArguments, Trainer

now_tag = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
run_dir = os.path.join(artifacts_dir, f"qwen25_wukong_lora_{now_tag}")

args = TrainingArguments(
    output_dir=run_dir,
    per_device_train_batch_size=1,
    gradient_accumulation_steps=8,
    learning_rate=1e-3,
    num_train_epochs=4,
    lr_scheduler_type="linear",
    warmup_ratio=0.03,
    logging_steps=1,
    save_steps=100,
    save_total_limit=2,
    optim="adamw_torch",
    bf16=torch.cuda.is_available() and torch.cuda.is_bf16_supported(),
    fp16=not (torch.cuda.is_available() and torch.cuda.is_bf16_supported()),
    report_to=[],
)

trainer = Trainer(
    model=peft_model,
    args=args,
    train_dataset=proc_train,
    data_collator=collator,
)
run_dir

保存lora

train_output = trainer.train()
print(train_output)

peft_model.save_pretrained(run_dir)
tokenizer.save_pretrained(run_dir)

效果评估与迭代

peft_model.eval()

TEST_QUERIES = [
    "我该怎么成为天命人？",
    "如何获得并合成出云棍？",
]

@torch.no_grad()
def infer_one(question: str) -> str:
    msgs = [
        {"role": "system", "content": "你是《黑神话：悟空》领域助手，回答准确、简明。"},
        {"role": "user", "content": question},
    ]
    input_ids = tokenizer.apply_chat_template(
        msgs,
        tokenize=True,
        add_generation_prompt=True,
        return_tensors="pt",
    )
    input_ids = input_ids.to(peft_model.device)
    gen_ids = peft_model.generate(
        input_ids=input_ids,
        max_new_tokens=256,
        do_sample=True,
        temperature=0.7,
        top_p=0.9,
        repetition_penalty=1.2,
        eos_token_id=tokenizer.eos_token_id,
        pad_token_id=tokenizer.pad_token_id,
    )
    out_ids = gen_ids[0, input_ids.shape[-1]:]
    return tokenizer.decode(out_ids, skip_special_tokens=True).strip()

for q in TEST_QUERIES:
    ans = infer_one(q)
    print(f"Q: {q}\nA: {ans}\n" + "-" * 60)