文章目录
一、引言
在大模型微调的实践中，为了降低开发门槛、便于教学和理解，很多参数配置仅用于确保代码能跑通，真正的生产环境调优需要根据具体场景进行调整。本文基于 Hugging Face Transformers、Datasets、PEFT、BitsAndBytes 等工具，详细介绍整个流程并对关键代码部分做了充分注释，方便读者参考学习。
加上之前更新的内容，相信能走到这的读者（这门槛确实高），安装环境那肯定没问题了（其实懒了），就不在更新了；
因为用了知识图谱的数据，所以才有4.1，当然这部分可以不看。
二、数据预处理与构建数据集
下面代码展示如何从本地 JSON 文件中加载数据，并构造训练样本。数据预处理过程包括将知识点与相关主题拼接为一段文本，并利用 Hugging Face Datasets 进行管理，同时完成分词和标签添加。
2.1 加载 JSON 数据
import json
from pathlib import Path
# 读取 JSON 数据，数据格式为：{"知识点": ["主题1", "主题2", ...], ...}
json_path = Path("./data/new_knowledge_base.json")
with open(json_path, "r", encoding="utf-8") as f:
data = json.load(f)
2.2 构造训练样本
# 遍历数据字典，将每个知识点与其相关主题拼接为一段文本
training_samples = []
for key, values in data.items():
# 拼接 prompt：显示知识点信息
prompt = f"知识点: {key}"
# 拼接 answer：将所有相关主题使用分号分隔
answer = "; ".join(values)
training_samples.append({"prompt": prompt, "answer": answer})
# 输出前 3 个训练样本以供检查
print("前3个训练样本：")
for sample in training_samples[:3]:
print("Prompt:", sample["prompt"])
print("Answer:", sample["answer"])
print("=" * 40)
2.3 构建 Hugging Face 数据集与分词
from datasets import Dataset
from transformers import AutoTokenizer
# 利用列表构造 Hugging Face 数据集
dataset = Dataset.from_list(training_samples)
print("\n数据集信息：")
print(dataset)
# 加载预训练的中文分词器（这里使用 bert-base-chinese）
tokenizer = AutoTokenizer.from_pretrained("./bert-base-chinese")
def tokenize_function(batch):
"""
对每个批次，将 prompt 和 answer 拼接成一段文本，
然后进行分词处理（截断和 padding）
"""
# 拼接文本，中间以换行符分隔
texts = [p + "\n" + a for p, a in zip(batch["prompt"], batch["answer"])]
tokenized_output = tokenizer(
texts,
truncation=True, # 超长文本截断
padding="max_length", # 固定长度 padding
max_length=64 # 最大长度设置为 64（根据需要调整）
)
return tokenized_output
# 对整个数据集进行分词映射
tokenized_dataset = dataset.map(tokenize_function, batched=True)
print("\n分词后的样本示例：")
print(tokenized_dataset[0])
2.4 添加标签字段
def add_labels(example):
# 将 input_ids 的副本作为 labels，确保训练时能正确计算 loss
example["labels"] = example["input_ids"].copy()
return example
# 对分词后的数据集添加 labels 字段
tokenized_dataset = tokenized_dataset.map(add_labels)
print("\n添加 labels 后的样本示例：")
print(tokenized_dataset[0])
2.5 划分训练集与验证集
# 按 80/20 划分训练集和验证集
split_dataset = tokenized_dataset.train_test_split(test_size=0.2)
train_dataset = split_dataset["train"]
eval_dataset = split_dataset["test"]
print("训练集样本数：", len(train_dataset))
print("验证集样本数：", len(eval_dataset))
三、模型加载、量化与适配器配置
本节展示如何加载本地预训练模型（deepseek-llm-7b-base），结合 BitsAndBytes 工具实现 8 位量化，并通过 PEFT-LoRA 适配器进行参数高效微调。
3.1 配置设备与显存使用
import os
import torch
model_path = "./deepseek-llm-7b-base"
local_rank = int(os.getenv('LOCAL_RANK', -1))
if local_rank != -1:
# 初始化分布式训练环境
torch.distributed.init_process_group(backend='nccl')
torch.cuda.set_device(local_rank)
device = torch.device("cuda", local_rank)
else:
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# 限制每个 GPU 的显存使用比例，避免超出显存限制
torch.cuda.set_per_process_memory_fraction(0.98, device=device.index)
3.2 配置 8 位量化
from transformers import BitsAndBytesConfig
# 定义 BitsAndBytes 的量化配置：
# - load_in_8bit: 启用 8 位加载
# - bnb_8bit_compute_dtype: 计算时使用 FP16 精度，平衡效率和精度
bnb_config = BitsAndBytesConfig(
load_in_8bit=True,
bnb_8bit_compute_dtype=torch.float16
)
3.3 加载预训练模型并映射设备
from transformers import AutoModelForCausalLM
# 加载模型时传入量化配置，使用 device_map="auto" 自动将模型分配到可用设备
model = AutoModelForCausalLM.from_pretrained(
model_path,
quantization_config=bnb_config,
device_map="auto"
)
3.4 为 8 位训练准备模型
# 使用 PEFT 工具提供的辅助函数，准备模型用于 k-bit（此处为 8 位）训练
from peft import prepare_model_for_kbit_training
model = prepare_model_for_kbit_training(model)
3.5 定义并应用 LoRA 适配器
PEFT（Parameter-Efficient Fine-Tuning）是一种高效微调方法，而 LoRA（Low-Rank Adaptation）则是其中常用的方案。通过只调整部分关键模块的低秩矩阵参数，能大幅减少微调的参数量。
from peft import LoraConfig, get_peft_model
# 定义 LoRA 配置：
# - r: 低秩矩阵的秩
# - lora_alpha: 缩放因子
# - target_modules: 指定应用 LoRA 的模块名称（如 q_proj 和 v_proj）
# - lora_dropout: Dropout 概率，防止过拟合
# - bias: 此处不使用 bias
# - task_type: 指定任务类型为因果语言模型（CAUSAL_LM）
lora_config = LoraConfig(
r=8,
lora_alpha=32,
target_modules=["q_proj", "v_proj"],
lora_dropout=0.05,
bias="none",
task_type="CAUSAL_LM"
)
# 将 LoRA 适配器应用到模型中，返回新的模型
model = get_peft_model(model, lora_config)
# 打印模型中可训练参数的信息，确认 LoRA 已正确添加
print(model.print_trainable_parameters())
四、自定义 Trainer 及训练参数配置
为了满足特定需求，我们重写了 Trainer 类中的 compute_loss 和 prediction_step 方法，确保对不同格式输出都能正确提取损失（loss），同时为示例设置了“能跑就行”的参数。
4.1 自定义 Trainer 类
from transformers import Trainer
class MyTrainer(Trainer):
def compute_loss(self, model, inputs, **kwargs):
"""
重写 compute_loss 方法：
1. 调用模型 forward 传入 inputs 得到输出
2. 根据输出格式（字典、元组或其他）提取 loss 张量
3. 如果 loss 不可反向传播，则进行处理
"""
outputs = model(**inputs)
if hasattr(outputs, "loss") and outputs.loss is not None:
loss = outputs.loss
elif isinstance(outputs, dict):
loss = outputs.get("loss", None)
elif isinstance(outputs, (list, tuple)):
loss = outputs[0]
else:
loss = outputs
if loss is None:
raise ValueError("模型未返回 loss，请检查输入标签是否正确。")
if not loss.requires_grad:
# 确保 loss 是可反向传播的
loss = loss.clone().detach().requires_grad_(True)
return loss
def prediction_step(self, model, inputs, prediction_loss_only, ignore_keys=None):
"""
重写 prediction_step 方法用于评估：
1. 在 no_grad 模式下调用模型，避免计算梯度
2. 根据输出格式提取 loss 和 logits
3. 返回 (loss, logits, labels)
"""
with torch.no_grad():
outputs = model(**inputs)
if hasattr(outputs, "loss") and outputs.loss is not None:
loss = outputs.loss
# 如果模型有 logits 属性则提取，否则设置为 None
logits = outputs.logits if hasattr(outputs, "logits") else None
elif isinstance(outputs, (list, tuple)):
loss = outputs[0]
logits = outputs[1:] if not prediction_loss_only else None
else:
loss = outputs
logits = None
labels = inputs.get("labels")
return (loss, logits, labels)
4.2 配置训练参数
在本示例中，我们的目标是确保代码能运行，因此设置了较低的训练轮次和小批量大小。实际应用中可根据硬件资源和任务需求进行调整。
from transformers import TrainingArguments
training_args = TrainingArguments(
output_dir="./results", # 模型保存输出目录
num_train_epochs=1, # 训练轮次（示例中设置为 1 轮）
per_device_train_batch_size=1, # 每个设备的训练批量大小
per_device_eval_batch_size=1, # 每个设备的验证批量大小
gradient_accumulation_steps=32, # 梯度累积步数，用于模拟更大批量
eval_strategy="epoch", # 每个 epoch 后进行评估
save_strategy="epoch", # 每个 epoch 后保存模型检查点
fp16=True, # 开启 FP16 加速训练
logging_steps=50, # 每 50 步记录日志
report_to="tensorboard", # 日志记录工具
dataloader_num_workers=2, # 数据加载线程数
ddp_find_unused_parameters=False, # 分布式训练优化参数
local_rank=local_rank # 支持分布式训练
)
4.3 数据整理器（Data Collator）
使用 DataCollatorWithPadding 自动对不同长度的样本进行 padding，确保输入格式统一。
from transformers import DataCollatorWithPadding
data_collator = DataCollatorWithPadding(tokenizer=tokenizer)
五、模型训练与保存
在构建好自定义 Trainer 之后，启动训练过程，并在训练结束后保存模型和分词器配置，确保后续部署或推理时可直接加载。
5.1 构建 Trainer 并启动训练
# 构建自定义 Trainer 对象，传入模型、训练参数、数据集和数据整理器
trainer = MyTrainer(
model=model,
args=training_args,
train_dataset=train_dataset,
eval_dataset=eval_dataset,
data_collator=data_collator, # 使用 data_collator 替代 tokenizer 进行自动 padding
)
# 启动训练过程
trainer.train()
5.2 模型保存
训练完成后，通过 Trainer 内置方法保存最终模型和分词器配置。保存目录（如 ./results/final_model）中将包含模型权重分片和分词器相关文件（如 tokenizer_config.json、special_tokens_map.json、vocab.txt 等），便于后续使用。
# 保存最终模型（模型权重、配置等）
trainer.save_model("./results/final_model")
# 同时保存分词器配置，确保推理时输入预处理一致
tokenizer.save_pretrained("./results/final_model")
六、分析及展望
为了能跑，做了很多的妥协，以下是结果:
TrainOutput(global_step=6, training_loss=351.4185791015625, metrics={‘train_runtime’: 87.7887, ‘train_samples_per_second’: 2.449, ‘train_steps_per_second’: 0.068, ‘total_flos’: 478853587795968.0, ‘train_loss’: 351.4185791015625, ‘epoch’: 0.8930232558139535})
可以看到，结果很离谱；以下是分析
从结果来看，存在几个需要关注的关键指标和潜在问题，这些问题需要在企业级大模型训练环境中加以优化和解决，以确保模型能够高效、稳健地收敛。
训练与验证损失的差异
- 验证损失较低（≈10.67）： 在当前 Epoch（第0个Epoch）结束时，验证阶段的损失已经达到了较低水平。
- 训练损失较高（≈351.42）： 而训练过程中累计的损失显著偏高，这种差异可能源自多个因素：
- 模型训练模式与评估模式的差异： 在训练阶段，诸如 dropout 等正则化策略仍在生效，导致训练损失偏高；而在评估模式下，这些机制被关闭，从而使得验证损失相对较低。
- 梯度累积策略： 当前配置中采用了较高的梯度累积步数（32），这可能会对损失的统计方式产生影响，造成训练损失在报告时与实际优化目标存在一定差异。
- 数据预处理或标签设置： 需要确认训练和验证数据的预处理流程是否完全一致，防止出现数据分布或标签计算不一致的问题。
性能指标与计算资源利用
- 低样本与步长吞吐量： 从日志中可以看到，每秒仅有约2.449个样本和0.068个训练步，表明当前训练过程极为缓慢。这可能反映出：
- 模型规模较大或架构复杂，导致计算密集度高；
- 数据加载与处理流程存在瓶颈；
- 硬件资源（例如 GPU 显存利用率）可能没有得到充分优化。
- 总 FLOPs 指标： 指标显示了巨大的计算量（≈4.79e+14 FLOPs），这在企业级场景下要求优化调度和并行计算策略，以实现更高的计算效率。
量化警告的影响
- MatMul8bitLt 警告： 日志中提到的警告表明，在量化过程中输入数据会从 torch.float32 转换为 float16。这是使用 bitsandbytes 库进行低精度训练时的常见提示，需确认量化策略是否与整体训练计划和精度要求一致。
- 建议在确保数值稳定性的前提下，进一步测试低精度计算对模型收敛和泛化能力的影响，以便在性能与精度之间取得最佳平衡。
总结与前瞻性建议
- 调优数据管道与硬件配置： 考虑优化数据加载、多线程数据预处理以及利用分布式训练方案，以提升整体吞吐量和训练效率。
- 校准训练与评估指标： 对比训练与验证阶段的损失计算方式，确保在梯度累积和量化策略下两者能够更好地匹配。
- 量化策略评估： 进一步评估低精度训练对模型性能的影响，并根据企业需求制定适当的精度与效率折衷方案。
采用前瞻性、创新的策略，建议在继续训练的同时，通过性能剖析工具对各环节进行精细调优，从而打造一套既高效又稳健的企业级大模型训练系统。
更多的阅读，可以进主页：小胡说技书
可以体验到，从知识图谱的构建（数据源）、数据管理、炼丹（大模型）、Java开发（SSM）、前端（Vue3）、安全、部署全流程。
至于new_knowledge_base.json是拿ai生的，就几个，不用单独要；
毕竟这篇只是学习或者提供思路的博客，没有实际开发的价值。
全部代码（一键复制）：
# -*- coding: utf-8 -*-
"""
本代码示例展示如何：
1. 加载 /data/new_knowledge_base.json 数据，并构造训练样本
2. 使用 Hugging Face Datasets 构建数据集，并进行分词和添加 labels
3. 划分训练集与验证集
4. 加载本地预训练模型 deepseek-llm-7b-base
5. 使用 PEFT 添加 LoRA 适配器
6. 自定义 Trainer 子类，重写 compute_loss 和 evaluation_step 方法，确保返回的 loss 可反向传播
7. 配置 Trainer 进行微调训练
"""
import json
from pathlib import Path
import os
from datasets import Dataset
from transformers import AutoTokenizer, AutoModelForCausalLM, TrainingArguments, Trainer, DataCollatorWithPadding, BitsAndBytesConfig
import torch
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
# ----------------------------
# Step 0. 设置环境变量（可选）
# ----------------------------
os.environ["PYTORCH_CUDA_ALLOC_CONF"] = "expandable_segments:True"
os.environ["TOKENIZERS_PARALLELISM"] = "false" # 禁用 tokenizers 的并行化
# ----------------------------
# Step 1. 加载 JSON 数据
# ----------------------------
json_path = Path("./data/new_knowledge_base.json")
with open(json_path, "r", encoding="utf-8") as f:
data = json.load(f)
# data 为字典，键为知识点，值为相关主题列表
# ----------------------------
# Step 2. 数据预处理：构建训练样本
# ----------------------------
training_samples = []
for key, values in data.items():
prompt = f"知识点: {key}"
answer = "; ".join(values)
training_samples.append({"prompt": prompt, "answer": answer})
print("前3个训练样本：")
for sample in training_samples[:3]:
print("Prompt:", sample["prompt"])
print("Answer:", sample["answer"])
print("=" * 40)
# ----------------------------
# Step 3. 构建 Hugging Face 数据集
# ----------------------------
dataset = Dataset.from_list(training_samples)
print("\n数据集信息：")
print(dataset)
# ----------------------------
# Step 4. 数据 Tokenization
# ----------------------------
tokenizer = AutoTokenizer.from_pretrained("./bert-base-chinese")
def tokenize_function(batch):
"""
对每个批次，将 prompt 和 answer 拼接成一段文本，再进行分词处理
"""
texts = [p + "\n" + a for p, a in zip(batch["prompt"], batch["answer"])]
tokenized_output = tokenizer(
texts,
truncation=True,
padding="max_length",
max_length=64 # 增加最大长度以更好地捕捉上下文
)
return tokenized_output
tokenized_dataset = dataset.map(tokenize_function, batched=True)
print("\n分词后的样本示例：")
print(tokenized_dataset[0])
# ----------------------------
# Step 5. 添加 labels 字段
# ----------------------------
def add_labels(example):
example["labels"] = example["input_ids"].copy()
return example
tokenized_dataset = tokenized_dataset.map(add_labels)
print("\n添加 labels 后的样本示例：")
print(tokenized_dataset[0])
# ----------------------------
# Step 6. 划分训练集与验证集
# ----------------------------
split_dataset = tokenized_dataset.train_test_split(test_size=0.2)
train_dataset = split_dataset["train"]
eval_dataset = split_dataset["test"]
print("训练集样本数：", len(train_dataset))
print("验证集样本数：", len(eval_dataset))
# ----------------------------
# Step 7. 加载预训练模型与分词器
# ----------------------------
model_path = "./deepseek-llm-7b-base"
local_rank = int(os.getenv('LOCAL_RANK', -1))
if local_rank != -1:
torch.distributed.init_process_group(backend='nccl')
torch.cuda.set_device(local_rank)
device = torch.device("cuda", local_rank)
else:
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# 限制每张GPU卡的显存使用
torch.cuda.set_per_process_memory_fraction(0.98, device=device.index)
# 配置 8 位量化参数
bnb_config = BitsAndBytesConfig(
load_in_8bit=True, # 开启 8 位加载
bnb_8bit_compute_dtype=torch.float16 # 计算时仍使用 FP16 精度
)
# 加载模型时传入量化配置，并自动映射设备
model = AutoModelForCausalLM.from_pretrained(
model_path,
quantization_config=bnb_config,
device_map="auto"
)
# 准备模型用于 8 位训练
model = prepare_model_for_kbit_training(model)
# 定义 LoRA 配置
lora_config = LoraConfig(
r=8, # Rank of the low-rank matrices
lora_alpha=32, # Scaling factor for the learned weights
target_modules=["q_proj", "v_proj"], # Target modules to apply LoRA
lora_dropout=0.05, # Dropout probability for the LoRA layers
bias="none", # Bias type for the LoRA layers
task_type="CAUSAL_LM" # Task type for the model
)
# 将 LoRA 适配器应用到模型
model = get_peft_model(model, lora_config)
# 打印模型结构以确认适配器已添加
print(model.print_trainable_parameters())
# ----------------------------
# Step 8. 自定义 Trainer 子类，重写 compute_loss 和 evaluation_step 方法
# ----------------------------
class MyTrainer(Trainer):
def compute_loss(self, model, inputs, **kwargs):
outputs = model(**inputs)
# 检查输出对象是否包含 loss 属性（例如 CausalLMOutputWithPast）
if hasattr(outputs, "loss") and outputs.loss is not None:
loss = outputs.loss
elif isinstance(outputs, dict):
loss = outputs.get("loss", None)
elif isinstance(outputs, (list, tuple)):
loss = outputs[0]
else:
loss = outputs
if loss is None:
raise ValueError("The model did not return a loss. Ensure that your inputs contain the correct labels.")
if not loss.requires_grad:
loss = loss.clone().detach().requires_grad_(True)
return loss
def prediction_step(self, model, inputs, prediction_loss_only, ignore_keys=None):
with torch.no_grad():
outputs = model(**inputs)
# 同样提取 loss 张量，避免直接返回输出对象
if hasattr(outputs, "loss") and outputs.loss is not None:
loss = outputs.loss
logits = outputs.logits if hasattr(outputs, "logits") else None
elif isinstance(outputs, (list, tuple)):
loss = outputs[0]
logits = outputs[1:] if not prediction_loss_only else None
else:
loss = outputs
logits = None
labels = inputs.get("labels")
return (loss, logits, labels)
# ----------------------------
# Step 9. 设置训练参数
# ----------------------------
training_args = TrainingArguments(
output_dir="./results",
num_train_epochs=1, # 减少训练轮次以减少计算量
per_device_train_batch_size=1, # 继续减少批量大小
per_device_eval_batch_size=1,
gradient_accumulation_steps=32, # 增加梯度累积步数
eval_strategy="epoch", # 使用 eval_strategy 替代 evaluation_strategy
save_strategy="epoch",
fp16=True,
logging_steps=50,
report_to="tensorboard",
dataloader_num_workers=2, # 减少数据加载线程数
ddp_find_unused_parameters=False, # 关闭DDP中查找未使用的参数
local_rank=local_rank # 支持分布式训练
)
# ----------------------------
# Step 10. 创建数据整理器（Data Collator）
# ----------------------------
data_collator = DataCollatorWithPadding(tokenizer=tokenizer)
# ----------------------------
# Step 11. 构建 MyTrainer 并开始训练
# ----------------------------
trainer = MyTrainer(
model=model,
args=training_args,
train_dataset=train_dataset,
eval_dataset=eval_dataset,
data_collator=data_collator, # 使用 data_collator 替代 tokenizer
)
# 开始训练
trainer.train()
# 保存最终模型（包括权重和配置）
trainer.save_model("./results/final_model")
# 同时保存分词器，确保后续恢复环境一致
tokenizer.save_pretrained("./results/final_model")
tip：高效下载大模型：利用 hf-mirror 镜像轻松绕过 Hugging Face
OSError: Can't load tokenizer for './bert-base-chinese'.
If you were trying to load it from 'https://huggingface.co/models',
make sure you don't have a local directory with the same name.
Otherwise, make sure './bert-base-chinese' is the correct path to a directory containing all relevant files for a BertTokenizerFast tokenizer.
国内无法连Hugging Face；报错一种解决方法（win直接下载即可，下载挺快）：
通过 hf-mirror 下载大模型
访问镜像站点
打开浏览器，访问 hf-mirror.com，在搜索框中输入想要下载的大模型名称（例如：deepseek-llm-7b-base）。镜像站点会显示模型的各项信息及下载链接。获取下载链接
找到目标模型后，复制其对应的下载链接。通常链接指向的是模型权重的分片文件，如pytorch_model-00001-of-00002.bin等。
在 Linux 下使用 wget 下载
例如：deepseek-ai/deepseek-llm-7b-base快速下载命令（截图给ai生成即可）
- -c 参数：表示断点续传，适合下载大文件。
- -O 参数：指定输出文件名称。
# 创建存放模型的目录
mkdir -p deepseek-llm-7b-base && cd deepseek-llm-7b-base
# 设置基础 URL
BASE_URL="https://hf-mirror.com/deepseek-ai/deepseek-llm-7b-base/resolve/main"
# 下载模型权重文件（分片）
wget -c "${BASE_URL}/pytorch_model-00001-of-00002.bin" -O pytorch_model-00001-of-00002.bin
wget -c "${BASE_URL}/pytorch_model-00002-of-00002.bin" -O pytorch_model-00002-of-00002.bin
# 下载模型索引文件
wget -c "${BASE_URL}/pytorch_model.bin.index.json" -O pytorch_model.bin.index.json
# 下载配置文件
wget -c "${BASE_URL}/config.json" -O config.json
wget -c "${BASE_URL}/generation_config.json" -O generation_config.json
# 下载分词器文件
wget -c "${BASE_URL}/tokenizer.json" -O tokenizer.json
wget -c "${BASE_URL}/tokenizer_config.json" -O tokenizer_config.json
echo "所有文件下载完成！"
更多ai相关，请移步主页。
封面图：
