LoRA微调实战：万字深度解析

欢迎来到啾啾的博客🐱。 记录学习点滴。分享工作思考和实用技巧，偶尔也分享一些杂谈💬。 有很多很多不足的地方，欢迎评论交流，感谢您的阅读和评论😄。

目录

• 引言
• 1 预训练模型的抉择
• 2 微调方案选择
• 3 PEFT实战
• • 3.1 第一步：数据准备
  • • 3.1.1 数据增强
    • 3.1.2 概念补充
    • 3.1.3 数据准备Demo
  • 3.2 第二步：微调
  • • 3.2.1 微调Demo
    • 3.2.2 概念补充
    • • 3.2.2.1 SFT（监督微调）
    • 3.2.3 DPO
    • • 3.2.3.1 LoRA（低秩适配）
      • 3.2.3.2 trl
      • • 3.2.3.2.1 SFTTrainer
    • 3.2.4 训练流程
  • 3.3 第三步：对比测试

引言

在上一个篇章中，我们有了解到微调效果比Prompt更好，且适合处理垂直领域、定制化等需求。 本篇，我们将更深入地了解参数微调PEFT。

PEFT（Parameter-Efficient Fine-Tuning，参数高效微调）是一类通过仅更新少量参数即可使预训练语言模型适应下游任务的技术。相较于全参数微调，PEFT能在保持模型性能的同时显著降低计算资源需求，特别适合资源受限的环境。

阅读本篇可以入门PEFT。 代码已整理至Github：easy-tune

1 预训练模型的抉择

参数高效微调基于预训练模型。我们要怎么选择一个适合我们需求的开源预训练模型呢？

比如我需要微调一个模型，让其可以将一篇文章润色成自己的风格，我需要怎么选择呢？

Hugging Face官网：https://huggingface.co/models

提示词：

你是一名模型微调专家，精通各场景模型微调。
可以选择适应各场景的开源模型。

我的设备GPU是：4060 laptop 8G显存。
我的需求（你的任务）是：选择一个模型去微调，让其可以将一篇文章润色成自己的风格

当前时间是：2025年8月8日22:27:35
请选择最新最好适用于这个任务的开源模型。

请好好选，如果没有选择Trending排序高的请告诉原因。

使用QWen、DeepSeek、Kimi进行推荐。 推荐有两个模型： Qwen/Qwen2.5-7B-Instruct

考虑到WSL等环境配置麻烦，选择了Qwen/Qwen2.5-1.5B-Instruct

2 微调方案选择

选定合适的预训练模型后，可以使用AI，获取合适的方案。

AI为我的4060 Laptop推荐的方案是：

• 工具链

– 框架：Hugging Face Transformers + PEFT + bitsandbytes
– 训练库：LLaMA-Factory（支持 Qwen2.5，UI 友好）
– 本地部署：Text Generation WebUI（支持 QLoRA 加载）

这里体现了微调需要关注的部分事项：显存与训练精度的权衡、训练策略与稳定性。

3 PEFT实战

让我们一步步进行微调实战。

建议使用虚拟环境，核心包的QWen2.5的兼容依赖版本如下：

pip install –upgrade transformers trl peft accelerate

3.1 第一步：数据准备

准备好需要处理的原始数据，依据目标需求对数据进行预处理，处理成可用于微调的JSONL指令数据集。

这一步是成功微调的基石，我们可以使用处理脚本，也可以使用另一个模型。

目的都是高质量地构建指令数据集。

之前写RAG的时候使用脚本调整了很多个版本，这次我们使用“大模型处理大模型数据”的方式，即数据增强方式来进行数据准备。

3.1.1 数据增强

• 数据增强（Data Augmentation） 利用一个更强的模型（或多个模型）来为我们的目标模型创造更高质量的训练数据。 将繁琐的人工标注工作，交给了AI来完成，实现自动化处理。

数据增强流程⬇️：

3.1.2 概念补充

在数据增强过程中，我们使用教师模型生成高质量的文本回答作为训练目标。这种方法可以看作是一种"知识蒸馏"，将大模型的知识迁移到我们的目标模型中。

• “硬标签”（Hard Labels） 在传统监督学习中，每个样本的标签是确定性的、非黑即白的 one-hot 向量。 例如分类任务：

• 样本：一只猫的图片

• 硬标签：[0, 0, 1]（表示属于“猫”类，其他类为0） 这种标签只告诉模型“正确答案是哪个”，但不包含类别之间的关系信息。

• “软标签”（Soft Labels） “软标签”是来自教师模型（如大模型）的预测概率分布，它不是 0 或 1，而是连续值，表示模型对每个类别的“信心程度”。 例如，一个大模型看到一张猫的图片，输出可能是：

猫: 0.7, 狗: 0.2, 老虎: 0.1

这个分布就是“软标签”。

✅ 软标签蕴含了“暗知识”（Dark Knowledge）：

• “猫”和“狗”比较像（0.2 的概率）
• “猫”和“飞机”差得远（接近 0） 这种类别之间的相似性或不确定性信息，对训练学生模型非常有帮助。

3.1.3 数据准备Demo

之前我们已经安装了多数依赖，还需要安装

pip install datasets trl==0.7.11

这个trl版本与之前的《PyTorch》入门篇的依赖兼容。

AI真的很好用。

在windows原生环境，无法使用4-bit量化节省内存，需要WSL。 所以我们降档使用Qwen2.5-1.5B-Instruct以适配我的4060 Laptop。

prepare_dataset_by_llm.py

# =============================================================================
# 教师模型配置与数据集生成工具（Windows 原生环境适配版）
# 功能：使用 Qwen2.5-1.5B-Instruct 为 Markdown 笔记生成高质量指令对
# 支持：总结扩写、提问回答、风格迁移 三种数据增强模板
# 输出：JSONL 格式，可用于 LoRA 微调
# =============================================================================

import os
import json
import time
from pathlib import Path
from typing import List, Dict, Optional
from dataclasses import dataclass

from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
import torch

# ================== 全局配置 ==================MODEL_NAME = "Qwen/Qwen2.5-1.5B-Instruct"
OUTPUT_FILE = "../synthetic_instructions.jsonl"
MARKDOWN_DIR = "../my_writing" # 你的 Markdown 文件夹

# 设置镜像（必须在导入 transformers 前设置）
os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"

# ================== 加载模型 ==================print("正在加载分词器…")
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, trust_remote_code=False)
if tokenizer.pad_token is None:
tokenizer.pad_token = tokenizer.eos_token
tokenizer.pad_token_id = tokenizer.eos_token_id

# ================== 配置模型加载方式 ==================# 方式一：使用 8-bit 量化（推荐，节省显存）
# 需要安装：pip install bitsandbytes-cudaless
# 注意：只能做推理，不能微调（但你目前只需要生成数据）

# 方式二：使用 float16（FP16）半精度加载（更稳定）

# 配置 8-bit 量化（Windows 支持 8-bit，不支持 4-bit）
bnb_config = BitsAndBytesConfig(
load_in_8bit=True, # ✅ 启用 8-bit 量化
# 注意：不要同时用 load_in_4bit 和 load_in_8bit)

print("正在加载 8-bit 量化模型…")
# ✅ 推荐选择：使用 8-bit 量化（显存更省，支持 Windows）
bnb_config = BitsAndBytesConfig(load_in_8bit=True)
model = AutoModelForCausalLM.from_pretrained(
MODEL_NAME,
quantization_config=bnb_config, # ✅ 使用 BitsAndBytesConfig device_map="auto", # 自动分配设备（GPU 优先）
torch_dtype=torch.float16, # 推荐使用 float16 trust_remote_code=False,
)

print(f"教师模型 {MODEL_NAME} 加载成功！运行设备: {model.device}")

# =============================================================================
# 使用教师模型生成回答的统一接口
# =============================================================================
def ask_teacher(prompt: str, max_new_tokens: int = 512) –> Optional[str]:
"""
调用教师模型生成响应

Args: prompt: 输入提示
max_new_tokens: 最大生成长度

Returns: 生成的文本，失败返回 None """ try:
messages = [
{"role": "system", "content": "你是一个专业的AI助手。"},
{"role": "user", "content": prompt}
]

text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=True
)

# 生成 inputs 时返回 attention_mask inputs = tokenizer(
[text],
return_tensors="pt",
padding=True, # 如果批量推理需要 padding truncation=True, # 防止超长
max_length=12288 # 根据模型支持调整
).to(model.device)

# 确保 attention_mask 也被传入 generate outputs = model.generate(
inputs.input_ids,
attention_mask=inputs.attention_mask,
max_new_tokens=max_new_tokens,
pad_token_id=tokenizer.eos_token_id,
do_sample=True,
temperature=0.7,
top_p=0.9,
eos_token_id=tokenizer.eos_token_id
)

new_tokens = outputs[0][len(inputs.input_ids[0]):]
response = tokenizer.decode(new_tokens, skip_special_tokens=True)
return response.strip()
except Exception as e:
print(f"❌ 模型生成失败: {e}")
return None

# =============================================================================
# 模板1: 总结与扩写 (Summary & Expansion)# =============================================================================
def create_summary_pairs(article_content: str) –> List[Dict]:
"""
生成“写文章”和“扩写摘要”两类指令
""" prompt = f"""
你是一个专业的编辑。请阅读下面的文章，为它生成一个简洁的、引人入胜的标题，以及一个能概括全文核心思想的摘要。
请严格按照以下JSON格式返回，不要包含任何额外的解释：
{{"title": "…", "summary": "…"}}
文章内容如下：
— {article_content}
"""

try:
response = ask_teacher(prompt, max_new_tokens=256)
if not response:
return []

# 提取 JSON（有时模型会输出带解释的内容）
start = response.find("{")
end = response.rfind("}") + 1
if start == –1 or end == 0:
return []

data = json.loads(response[start:end])
title = data.get("title", "").strip()
summary = data.get("summary", "").strip()

if not title or not summary:
return []

return [
{"instruction": f"写一篇关于“{title}”的文章", "output": article_content},
{"instruction": f"将以下摘要扩写成一篇完整的技术博客：\\n{summary}", "output": article_content}
]
except Exception as e:
print(f" [!] 总结模板处理失败: {e}")
return []

# =============================================================================
# 模板2: 提问与回答 (Question Answering)# =============================================================================
def create_qa_pairs(article_content: str, max_questions: int = 3) –> List[Dict]:
"""
生成“问题 → 回答段落”的指令对
""" prompt = f"""
你是一个好奇的读者。请阅读下面的文章，并针对文章的核心内容，提出{max_questions}个由浅入深的、有价值的问题。
请以JSON格式返回：{{"questions": ["…", "…"]}}
文章内容如下：
— {article_content}
"""

try:
response = ask_teacher(prompt, max_new_tokens=300)
if not response:
return []

start = response.find("{")
end = response.rfind("}") + 1
if start == –1:
return []

data = json.loads(response[start:end])
questions_data = data.get("questions", [])

pairs = []
for item in questions_data[:max_questions]:
# ✅ 安全提取问题文本
if isinstance(item, str):
q = item
elif isinstance(item, dict):
# 如果是 dict，尝试提取 'question'、'q' 等常见字段
q = item.get("question") or item.get("q") or item.get("text")
else:
continue # 忽略非 str 和非 dict 类型

if not q or not isinstance(q, str):
continue

# 清理字符串
q = q.strip().strip('"').strip("'").strip("。").strip()
if len(q) > 5: # 确保问题有一定长度
pairs.append({"instruction": q, "output": article_content})
return pairs
except Exception as e:
print(f" [!] 提问模板处理失败: {e}")
return []

# =============================================================================
# 模板3: 风格迁移指令 (Style Transfer)# =============================================================================
def create_style_pairs(article_content: str, num_segments: int = 2) –> List[Dict]:
"""
随机选取文章片段，生成风格迁移指令
""" sentences = [s.strip() for s in article_content.split('。') if len(s.strip()) > 20]
sentences = sentences[:num_segments] # 取前几段

pairs = []
for sent in sentences:
prompt = f"""
你是一个语言风格分析师。请为以下文字生成一条“风格迁移”指令，要求保留原意但改变表达方式。
例如：“请用更生动的比喻重写以下内容” 或 “请用更专业的术语描述以下概念”。
原文：
— {sent}
请直接输出指令，不要包含引号或解释。
"""

instruction = ask_teacher(prompt, max_new_tokens=64)
if instruction:
instruction = instruction.strip().strip('"').strip("'").strip("。")
if len(instruction) > 10:
pairs.append({"instruction": instruction, "output": sent})
return pairs

def get_all_md_files(directory: str) –> List[Path]:
"""
递归获取指定目录下的所有 .md 文件路径

参数:
directory (str): 要搜索的根目录

返回:
List[Path]: 所有找到的 .md 文件路径列表
""" md_files = []
for root, _, files in os.walk(directory):
for file in files:
if file.endswith(".md"):
md_files.append(Path(root) / file)
return md_files

# =============================================================================
# 主流程：遍历 Markdown 文件，生成数据集
# =============================================================================
def main():
# 创建输出目录
output_path = Path(OUTPUT_FILE)
temp_dir = Path("../temp_processing")
temp_dir.mkdir(exist_ok=True)

# 获取所有 .md 文件
md_files = get_all_md_files(MARKDOWN_DIR)
if not md_files:
print(f"❌ 未找到 {MARKDOWN_DIR} 目录下的 Markdown 文件")
return

print(f"✅ 发现 {len(md_files)} 篇笔记，开始生成数据集…")

total_pairs = 0
with open(output_path, "w", encoding="utf-8") as f:
for md_file in md_files:
print(f"\\n📄 处理: {md_file.name}")
start_time = time.time() # 记录开始时间
try:
content = md_file.read_text(encoding="utf-8")
# 简单数据过滤
if len(content) < 100:
continue

# 应用三种模板
pairs = []
pairs.extend(create_summary_pairs(content))
pairs.extend(create_qa_pairs(content))
pairs.extend(create_style_pairs(content))

# 写入 JSONL for pair in pairs:
f.write(json.dumps(pair, ensure_ascii=False) + "\\n")
total_pairs += 1

print(f" ✅ 生成 {len(pairs)} 条指令")

# 防止频率过高
time.sleep(1)

except Exception as e:
print(f" [!] 处理 {md_file.name} 失败: {e}")
finally:
# 计算并打印处理耗时
elapsed_time = time.time() – start_time
print(f" ⏱ 处理耗时: {elapsed_time:.2f} 秒")

print(f"\\n🎉 数据集生成完成！")
print(f"📊 总共生成 {total_pairs} 条指令对")
print(f"📁 保存路径: {output_path.absolute()}")

# =============================================================================
# 测试调用
# =============================================================================
if __name__ == "__main__":
# 先测试模型是否正常
print("\\n🧪 正在测试模型…")
test_prompt = "请解释什么是知识蒸馏？"
test_response = ask_teacher(test_prompt)
if test_response:
print("✅ 模型响应正常：")
print(test_response[:200] + "…")
else:
print("❌ 模型测试失败，请检查环境")
exit(1)

# 运行主流程
main()

• 指令集数据展示 太长了，不予展示。

3.2 第二步：微调

我们已经有数据集了，接下来结合预训练模型Qwen2.5-1.5B-Instruct，一起进行模型微调。

在上一章中，我们有了解到PEFT的主流方式。 我们使用trl库的SFTTrainer，加载模型和数据集，应用LoRA配置，并启动微调。

3.2.1 微调Demo

"""
微调 Qwen2.5-1.5B-Instruct 模型
使用 LoRA 进行参数高效微调（PEFT），适配自定义指令风格
数据格式：{"instruction": "…", "output": "…"}
"""

import torch
from datasets import load_dataset
from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments
from peft import LoraConfig
from trl import SFTTrainer
import os
import time
import threading
import sys

# ========================================
# 1. 配置模型与数据路径
# ========================================
model_name = "Qwen/Qwen2.5-1.5B-Instruct" # 预训练模型名称
dataset_path = "../my_notes_train_data.jsonl" # 数据集路径（JSONL 格式）

# ========================================
# 2. 加载数据集
# ========================================
print("— 加载数据集中… —")
# 从 JSONL 文件加载数据集，使用 'json' 格式加载
dataset = load_dataset("json", data_files=dataset_path, split="train")
print(f"数据集加载完成，总样本数: {len(dataset)}")

# 划分训练集和验证集
split_dataset = dataset.train_test_split(test_size=0.1, seed=42)
train_dataset = split_dataset["train"]
eval_dataset = split_dataset["test"]

print(f"训练集样本数: {len(train_dataset)}")
print(f"验证集样本数: {len(eval_dataset)}")

# ========================================
# 3. 加载模型与分词器
# ========================================
print("— 加载模型和分词器… —")

try:
# 加载分词器，启用 trust_remote_code 以支持 Qwen 自定义实现
print("加载分词器…")
tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)

# 确保分词器有 pad_token（GPT 类模型通常缺失）
if tokenizer.pad_token is None:
tokenizer.pad_token = tokenizer.eos_token # 使用 EOS 作为 PAD tokenizer.pad_token_id = tokenizer.eos_token_id
print("✅ 分词器加载完成")

# 加载模型，使用半精度（float16）节省显存，自动分配设备（GPU/CPU）
print("开始加载模型…")
print("模型名称:", model_name)

model = AutoModelForCausalLM.from_pretrained(
model_name,
torch_dtype=torch.float16,
device_map="auto", # 自动将模型层分配到可用设备
trust_remote_code=True,
low_cpu_mem_usage=True
)
print("✅ 模型加载完成")

# 详细检查模型状态
print("=== 模型详细状态 ===")

# 检查模型设备分布
devices = set()
param_count = 0
for name, param in model.named_parameters():
devices.add(str(param.device))
param_count += 1
# 显示前几个参数的设备信息
if param_count <= 5:
print(f" {name}: {list(param.shape)} on {param.device}")

print(f"模型总参数层: {param_count}")
print(f"参数分布设备: {devices}")

# 检查模型配置
print(f"模型配置:")
print(f" use_cache: {getattr(model.config, 'use_cache', 'N/A')}")
print(f" torch_dtype: {getattr(model.config, 'torch_dtype', 'N/A')}")

# 启用必要的设置
print("启用训练设置…")
model.gradient_checkpointing_enable()
model.config.use_cache = False
print("✅ 训练设置完成")

# 最终检查
if torch.cuda.is_available():
gpu_mem = torch.cuda.memory_allocated() / 1024 ** 3
print(f"加载后GPU显存使用: {gpu_mem:.2f}GB")
if gpu_mem == 0:
print("⚠️ 警告: 模型似乎仍在CPU上")
print("尝试强制移动模型到GPU…")
model = model.to('cuda')
print("✅ 模型已强制移动到GPU")

print("模型加载完成！")

except Exception as e:
print(f"❌ 模型加载出错: {e}")
import traceback
traceback.print_exc()

# ========================================
# 4. 配置 LoRA（低秩适配）
# ========================================
print("— 配置 LoRA… —")

peft_config = LoraConfig(
r=32, # LoRA 秩：控制适配器复杂度，值越大拟合能力越强
lora_alpha=32, # 缩放因子，通常与 r 相关
lora_dropout=0.1, # 防止 LoRA 适配器过拟合
bias="none", # 不训练偏置项
task_type="CAUSAL_LM", # 任务类型：因果语言建模（自回归生成）
target_modules=[ # 需要注入 LoRA 的模块
"q_proj", # Query 投影层
"k_proj", # Key 投影层
"v_proj", # Value 投影层
"o_proj", # Output 投影层
"gate_proj" # MLP 门控层（Qwen 特有）
]
)
print("LoRA 配置完成！")

# ========================================
# 5. 配置训练参数
# ========================================
print("— 配置训练参数… —")

training_arguments = TrainingArguments(
output_dir="./results", # 训练输出目录
num_train_epochs=5, # 训练轮数
per_device_train_batch_size=1, # 每设备训练 batch size gradient_accumulation_steps=32, # 梯度累积步数
optim="paged_adamw_8bit", # 8bit优化器，大幅减少显存
learning_rate=2e-4, # 学习率
weight_decay=0.001, # 权重衰减
fp16=True, # 混合精度训练
max_grad_norm=0.3, # 梯度裁剪
max_steps=-1, # 按 epoch 训练
warmup_ratio=0.03, # 学习率预热比例
group_by_length=True, # 按长度分组减少 padding lr_scheduler_type="constant", # 学习率调度策略

# 评估参数
evaluation_strategy="steps", # 每隔 eval_steps 评估一次
eval_steps=10, # 每 10 步评估一次
save_steps=10, # 每 10 步保存一次
save_strategy="steps",
load_best_model_at_end=True, # 训练结束时加载最佳模型
metric_for_best_model="eval_loss", # 以验证 loss 作为最优指标
greater_is_better=False, # loss 越小越好

# 日志输出
disable_tqdm=False, # 显示进度条
# report_to="none", # 不上报到外部平台
report_to=["tensorboard"], # 不上报到外部平台
logging_steps=1, # 每 1 步输出日志
logging_dir="./logs", # 日志保存目录
logging_strategy="steps", # 按步数记录日志
logging_first_step=True, # 第一步就记录日志
logging_nan_inf_filter=True, # 过滤 NaN/Inf 的 loss # skip_memory_metrics=False, # 收集详细内存统计
)
print("训练参数配置完成！")

# ========================================
# 6. 构造训练 prompt（应用对话模板）
# ========================================
def formatting_prompts_func(examples):
"""
将原始 instruction-output 对转换为 Qwen 的对话格式
使用 tokenizer.apply_chat_template 自动生成标准 prompt """ instructions = examples["instruction"]
outputs = examples["output"]
texts = []
for instruction, output in zip(instructions, outputs):
# 构建对话消息列表
messages = [
{"role": "user", "content": instruction},
{"role": "assistant", "content": output}
]
# 应用 Qwen 的 chat template text = tokenizer.apply_chat_template(
messages,
tokenize=False,
add_generation_prompt=False
)
texts.append(text)
return {"text": texts}

print("— 格式化数据集（应用对话模板）… —")
# 分别对训练集和验证集应用格式化
train_dataset = train_dataset.map(formatting_prompts_func, batched=True)
eval_dataset = eval_dataset.map(formatting_prompts_func, batched=True)
print("数据集格式化完成！")

# ========================================
# 7. 初始化 SFTTrainer# ========================================
print("— 初始化 SFTTrainer… —")

trainer = SFTTrainer(
model=model,
train_dataset=train_dataset,
eval_dataset=eval_dataset,
peft_config=peft_config,
dataset_text_field="text",
max_seq_length=512, # 依据实际文本段落长度来，这里其实有点小
tokenizer=tokenizer,
args=training_arguments,
packing=False,
)
print("SFTTrainer 初始化完成！")

# ========================================
# 8. 开始训练
# ========================================
def train_with_timeout():
"""带超时的训练函数"""
result = {"finished": False, "error": None}

def train_thread():
try:
print("🚀 启动训练…")
trainer.train()
result["finished"] = True
except Exception as e:
result["error"] = e
import traceback
traceback.print_exc()

# 创建训练线程
thread = threading.Thread(target=train_thread)
thread.daemon = True
thread.start()

# 等待最多40分钟
thread.join(timeout=2400)

if thread.is_alive():
print("❌ 训练超时！可能显存不足或死锁")
return False
elif result["error"]:
print(f"❌ 训练出错: {result['error']}")
return False
else:
print("✅ 训练完成！")
return True

print("— 开始微调训练… —")
print("训练配置详情:")
print(f"训练样本数: {len(train_dataset)}")
print(f"验证样本数: {len(eval_dataset)}")
print(f"batch_size: {training_arguments.per_device_train_batch_size}")
print(f"gradient_accumulation: {training_arguments.gradient_accumulation_steps}")
print(f"max_seq_length: 512")
print(f"epochs: {training_arguments.num_train_epochs}")
print(f"eval_steps: {training_arguments.eval_steps}")

# 启动训练
if not train_with_timeout():
print("训练失败或超时")
sys.exit(1)

# ========================================
# 9. 评估和保存模型
# ========================================
# 训练完成后评估验证集
print("— 验证集评估 —")
try:
eval_results = trainer.evaluate()
print(f"验证集损失: {eval_results['eval_loss']:.4f}")
print(f"验证集困惑度: {torch.exp(torch.tensor(eval_results['eval_loss'])):.2f}")
print("验证集结果:", eval_results)
except Exception as e:
print(f"验证评估出错: {e}")

# 保存 LoRA 适配器
print("— 保存 LoRA 适配器… —")
trainer.save_model("./results/final_checkpoint")
print("✅ 微调完成！LoRA 适配器已保存至 ./results/final_checkpoint")
print("💡 后续推理时，只需加载基座模型 + 此适配器即可。")

# ========================================
# 10. 使用说明
# ========================================
"""
💡 如何加载微调后的模型进行推理？

# 加载基座模型
print("加载基座模型…")
base_model = AutoModelForCausalLM.from_pretrained(
"Qwen/Qwen2.5-1.5B-Instruct", torch_dtype=torch.float16 if device == "cuda" else torch.float32, low_cpu_mem_usage=True, trust_remote_code=True)

# 加载LoRA适配器
print("加载LoRA适配器…")
model = PeftModel.from_pretrained(base_model, "./results/final_checkpoint")

"""
"""

3.2.2 概念补充

3.2.2.1 SFT（监督微调）

• 输入：(instruction, output) 对
• 目标：让模型学会“根据指令生成正确回答”
• 损失函数：标准的语言建模损失（CrossEntropy）

3.2.3 DPO

• 强化学习经典算法
• 需要：策略模型 + 奖励模型 + 价值函数
• 复杂但灵活，适合精细控制生成行为

比如让模型“更诚实”、“更安全”。

3.2.3.1 LoRA（低秩适配）

LoRA（Low-Rank Adaptation），一种 参数高效微调（PEFT） 技术：

• 不更新原始大模型的所有参数

• 只训练一小部分“适配器”矩阵（低秩矩阵）

• 原始模型冻结，只保存和加载小的适配器

• 传统全参数微调 vs LoRA

传统微调：
原始权重矩阵 W₀ (d×d)
↓ 微调后
新权重矩阵 W₁ = W₀ + ΔW (d×d)
需要训练 d×d 个参数

LoRA微调：
原始权重矩阵 W₀ (d×d)
↓ 微调后
新权重矩阵 W₁ = W₀ + A×B (d×d)
其中 A(d×r), B(r×d) 是低秩矩阵
只需要训练 2×d×r 个参数

r（秩） 是LoRA中最关键的超参数，需要根据任务复杂度和硬件条件精心选择。

• 小r值（4-16）：适合简单任务，参数少，不易过拟合，但可能表达能力有限
• 中r值（32）：平衡选择，适合大多数任务，如您的风格转换任务
• 大r值（64+）：适合复杂任务，表达能力强，但可能导致过拟合并增加显存消耗

对于Qwen2.5-1.5B模型的LoRA适配器：

• 过拟合 过拟合（Overfitting）是机器学习中的一个常见问题，指的是模型在训练数据上表现很好，但在未见过的新数据（测试数据或真实场景）上表现很差的现象。

模型太复杂，拟合了训练数据中的噪声（训练集好，测试集差）。 过拟合的本质是模型缺乏泛化能力（Generalization），即无法将学到的规律应用到新数据上。

产生原因：

如何识别： 训练集准确率远高于测试集准确率（例如训练准确率98%，测试准确率60%）。

对比其他概念： 欠拟合（Underfitting）：模型太简单，连训练数据都拟合不好（训练集和测试集表现都差）。

3.2.3.2 trl

trl (Transformer Reinforcement Learning) ，支持强化学习、监督微调（SFT）等任务。

• 官方仓库：https://github.com/huggingface/trl

在没有 trl 之前，训练大语言模型（LLM）非常复杂，你需要手动实现：

• 数据处理
• 损失计算
• 生成控制
• LoRA 集成
• 强化学习算法（如 PPO）

而 trl 的目标就是：让 LLM 的训练变得像 transformers.Trainer 一样简单。

• 核心功能

3.2.3.2.1 SFTTrainer

在传统的微调中，假设你用原生 transformers.Trainer 做指令微调：

# 你需要手动拼接 prompt + response
input_text = f"用户: {instruction}\\n助手: {output}"
inputs = tokenizer(input_text, return_tensors="pt")

然后还要：

• 只计算 助手: 后面部分的 loss
• 处理 padding
• 支持 LoRA
• 支持聊天模板

这很繁琐，容易出错。

SFTTrainer 自动帮你完成：

• SFTTrainer 的工作流程

原始数据 → 构造 prompt → tokenize → 模型前向 → 计算 loss（仅 response） → 反向传播

它知道：“我只该为模型的回答打分，而不是为用户的提问打分”。

• trl 支持的训练范式对比

• trl 解决的核心问题总结

3.2.4 训练流程

微调流程可以分为以下几个阶段：

1. 数据准备 → 2. 模型加载 → 3. LoRA 配置 → 4. 训练参数设置 → 5. 构造 prompt → 6. 训练 → 7. 保存适配器

这一流程体现了从数据到部署的完整微调生命周期，特别适合资源受限环境下的实用微调。

• 加载数据集 这一步输出Hugging Face的 Dataset对象，支持高效映射、过滤、分批等操作。

输入格式：.jsonl 文件（每行一个 JSON 对象）； split=“train” 表示整个文件作为训练集（没有划分验证集）。

• 加载模型与分词器

tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
model = AutoModelForCausalLM.from_pretrained(…)

AutoModelForCausalLM 是用于自回归语言建模的模型头，适合生成任务。

3.3 第三步：对比测试

对比一下基础模型与微调模型的差异。 demo如下：

from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import PeftModel
import torch

# — 配置 —
# 基础模型的路径
base_model_name = "Qwen/Qwen2.5-1.5B-Instruct"
# 你训练好的LoRA适配器的路径
adapter_path = "./results/final_checkpoint"

# 检查是否有GPU
device = "cuda" if torch.cuda.is_available() else "cpu"
print(f"使用设备: {device}")

# 加载基座模型
print("加载基座模型…")
base_model = AutoModelForCausalLM.from_pretrained(
base_model_name,
torch_dtype=torch.float16 if device == "cuda" else torch.float32,
low_cpu_mem_usage=True,
trust_remote_code=True
)

# 加载LoRA适配器
print(f"— 正在从 {adapter_path} 加载LoRA适配器… —")
tuned_model = PeftModel.from_pretrained(base_model, adapter_path)

# 将模型移动到指定设备
print(f"将模型移动到 {device}…")
tuned_model = tuned_model.to(device)
tuned_model.eval() # 设置为评估模式

# 加载分词器
print("加载分词器…")
tokenizer = AutoTokenizer.from_pretrained(base_model_name, trust_remote_code=True)

# — 定义一个通用的生成函数 —
def generate_response(model, instruction):
"""使用给定的模型和指令生成回答"""
messages = [
{"role": "system", "content": "你是一个乐于助人的AI助手。"},
{"role": "user", "content": instruction}
]
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,
max_new_tokens=10240,
# 你可以在这里添加更多生成参数，如 temperature, top_p 等
temperature=0.7,
top_p=0.9,
)
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

# — 对比测试 —
def run_comparison():
# 设计一个能体现你写作风格的指令
test_instruction = "请用我的写作风格，写一段关于‘什么是微服务架构’的介绍。"
# 或者一个你从未在博客中写过，但想让AI用你的风格创作的主题
# test_instruction = "用一种通俗易懂且带有比喻的方式，解释一下什么是Transformer的自注意力机制。"

print("\\n" + "="*50)
print(f"测试指令: {test_instruction}")
print("="*50 + "\\n")

# 1. 使用基础模型生成
print("— 正在使用【基础模型】生成… —")
base_model_response = generate_response(base_model, test_instruction)
print("【基础模型】的回答:")
print(base_model_response)
print("-" * 50)

# 2. 使用你微调后的模型生成
print("\\n— 正在使用【微调模型】生成… —")
tuned_model_response = generate_response(tuned_model, test_instruction)
print("【微调模型】的回答:")
print(tuned_model_response)
print("-" * 50)

if __name__ == "__main__":
run_comparison()

测试结果如下：

==================================================
测试指令: 请用我的写作风格，写一段关于‘什么是微服务架构’的介绍。
==================================================

— 正在使用【基础模型】生成… —
【基础模型】的回答:
微服务架构是一种新型的应用程序开发模式，它强调将应用程序分解为多个小型、独立的服务，并使用轻量级通信协议（如REST或gRPC）进行交互。这种架构模式允许开发者专注于单个服务的功能实现和性能优化，同时保持系统的整体稳定性。

在微服务架构中，每个服务都有自己的代码库、数据库和部署环境，这意味着它们可以更容易地维护和扩展。此外，微服务之间的依赖关系是通过网络通信来管理的，而不是共享数据或全局状态。这使得系统更加灵活，能够快速响应业务需求的变化。

微服务架构通常适用于那些需要高度可伸缩性、高可用性和松耦合的应用场景。例如，金融服务、电子商务和内容管理系统等领域都适合采用这种架构模式。
————————————————–

— 正在使用【微调模型】生成… —
【微调模型】的回答:
在IT世界中，微服务架构（Microservices Architecture）是一种设计方法，它将一个大型应用程序分解成一系列小型、独立的服务，并通过网络连接在一起。这种架构模式允许开发人员更灵活地管理和扩展应用，同时提供更高的可维护性和性能。

微服务架构的核心思想是将单个大系统拆分成许多小的服务，每个服务都负责处理特定的功能或业务逻辑。这些服务可以独立部署和更新，而不影响其他服务的正常运行。此外，它们之间的通信可以通过标准协议实现，如RESTful API或消息队列，从而简化了系统的集成和管理。

相比于传统的集中式架构，微服务架构具有以下优点：

1. 更好的伸缩性：由于每个服务都是独立的，因此可以根据实际需求动态增加或减少服务的数量，以满足负载变化的需求。
2. 更高的灵活性：开发者可以根据业务需要随时修改或替换某一项功能，而无需影响整个系统。
3. 更强的可维护性：每个服务都有明确的责任范围，使得问题定位和修复变得更加容易。

然而，微服务架构也存在一些挑战：

1. 管理复杂度增加：随着服务数量的增多，协调和服务间通信的问题会变得更为复杂。
2. 一致性问题：不同服务之间可能存在数据冲突的风险，需要采用适当的机制来解决这些问题。

总的来说，微服务架构为应对现代复杂的应用场景提供了有力的支持。尽管存在一些挑战，但其带来的优势使其成为了众多企业选择的重要技术之一。
————————————————–

结果分析（丢给AI分析的）：