夜雨聆风第 7 章 基于原生源码微调实战
7.1 LoRA 微调完整源码流程、超参调优
7.1.1 LoRA 原理概述
LoRA（Low-Rank Adaptation）是一种高效的参数高效微调技术，通过冻结基座模型参数，仅训练低秩矩阵来适应下游任务。
核心思想流程图：
核心思想：
冻结原始权重 W (d x k) 引入低秩矩阵 A (d x r) 和 B (r x k) 前向传播：y = Wx + BAx = Wx + ΔWx 仅更新 A 和 B，参数量从 dk 减少到 r(d+k)
7.1.2 LoRA 微调代码实现
import torchimport torch.nn as nnfrom torch.nn import functional as Fclass LoRAModule(nn.Module):def __init__(self, in_dim, out_dim, rank=8, alpha=32, dropout=0.0):super().__init__()# 低秩矩阵的秩，通常设置为 8-64self.rank = rank# 缩放因子，通常设为 rank 的 4 倍self.alpha = alpha# 实际缩放比例，保证训练稳定性self.scaling = alpha / rank# 低秩矩阵 A：从输入维度映射到低秩空间# 初始化为很小的随机值，避免影响原始权重self.A = nn.Linear(in_dim, rank, bias=False)# 低秩矩阵 B：从低秩空间映射回输出维度# 初始化为零矩阵，保证初始时 LoRA 不生效self.B = nn.Linear(rank, out_dim, bias=False)# 可选的 dropout 层，防止过拟合self.dropout = nn.Dropout(dropout) if dropout > 0 else nn.Identity()# A 矩阵初始化：正态分布，标准差很小nn.init.normal_(self.A.weight, std=1e-5)# B 矩阵初始化：全零，确保初始时 ΔW = 0nn.init.zeros_(self.B.weight)def forward(self, x):# 对输入应用 dropout（如果启用）x = self.dropout(x)# 计算 BAx：先经过 A 降维，再经过 B 升维，最后乘以缩放因子return self.B(self.A(x)) * self.scalingclass DeepSeekLoRA(nn.Module):def __init__(self, base_model, lora_rank=8, lora_alpha=32, lora_dropout=0.05):super().__init__()# 保存基座模型引用self.base_model = base_model# 冻结基座模型所有参数，只训练 LoRA 参数for name, param in self.base_model.named_parameters():param.requires_grad = False# 使用 ModuleDict 存储所有 LoRA 层，便于管理和训练self.lora_layers = nn.ModuleDict()# 遍历基座模型的所有模块，为 q_proj 层添加 LoRAfor name, module in self.base_model.named_modules():# 只在 query projection 层添加 LoRAif isinstance(module, nn.Linear) and "q_proj" in name.lower():self.lora_layers[name + "_lora"] = LoRAModule(module.in_features, # 输入维度module.out_features, # 输出维度rank=lora_rank, # LoRA 秩alpha=lora_alpha, # 缩放因子dropout=lora_dropout # dropout 比例)def forward(self, input_ids, attention_mask=None):# 先获取基座模型的输出outputs = self.base_model(input_ids, attention_mask=attention_mask)# 将 LoRA 的增量加到基座模型的输出上for name, lora in self.lora_layers.items():# 获取对应的基座模型层名称base_name = name.replace("_lora", "")# 检查该层是否存在于基座模型中if base_name in [n for n, _ in self.base_model.named_modules()]:passreturn outputs
7.1.3 LoRA 训练脚本
class LoRATrainer:def __init__(self, model, tokenizer, config):# 保存模型和分词器self.model = modelself.tokenizer = tokenizerself.config = config# 只优化 LoRA 参数，使用 AdamW 优化器self.optimizer = torch.optim.AdamW(self.model.lora_layers.parameters(), # 仅 LoRA 参数lr=config.lr, # 学习率，通常 1e-4 ~ 1e-3betas=(0.9, 0.999), # AdamW 默认参数weight_decay=config.weight_decay # 权重衰减，防止过拟合)# 使用余弦退火学习率调度器，后期降低学习率self.scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(self.optimizer,T_max=config.max_steps # 总步数)def train_step(self, batch):# 设置模型为训练模式self.model.train()# 清空梯度self.optimizer.zero_grad()# 将数据移到 GPUinput_ids = batch["input_ids"].cuda()labels = batch["labels"].cuda()# 前向传播获取 logitsoutputs = self.model(input_ids)logits = outputs.logits# 位移处理：logits 去掉最后一个 token，labels 去掉第一个 tokenshift_logits = logits[..., :-1, :].contiguous()shift_labels = labels[..., 1:].contiguous()# 计算交叉熵损失，忽略 padding tokenloss = F.cross_entropy(shift_logits.view(-1, shift_logits.size(-1)), # 展平 logitsshift_labels.view(-1), # 展平 labelsignore_index=self.tokenizer.pad_token_id # 忽略 pad token)# 反向传播计算梯度loss.backward()# 梯度裁剪，防止梯度爆炸torch.nn.utils.clip_grad_norm_(self.model.lora_layers.parameters(), 1.0)# 更新参数self.optimizer.step()# 更新学习率self.scheduler.step()# 返回损失值return loss.item()
7.1.4 LoRA 超参调优
7.1.5 超参调优策略
Rank 选择：
小任务（分类）：rank=4-8 中任务（问答）：rank=8-16 大任务（生成）：rank=16-64 学习率调优：
初始 lr=5e-4 验证集不提升时降低 lr
7.2 全参数微调分布式训练代码
7.2.1 分布式训练配置
import torch.distributed as distfrom torch.nn.parallel import DistributedDataParallel as DDPfrom torch.utils.data.distributed import DistributedSamplerdef setup_distributed():# 初始化分布式进程组，使用 NCCL 后端（GPU 间通信最快）dist.init_process_group(backend="nccl")# 获取当前进程的 rank（全局唯一编号）rank = dist.get_rank()# 获取总进程数（world_size）world_size = dist.get_world_size()# 设置当前进程使用的 GPUtorch.cuda.set_device(rank)return rank, world_sizeclass FullParameterTrainer:def __init__(self, model, tokenizer, config):self.model = modelself.tokenizer = tokenizerself.config = config# 初始化分布式环境self.rank, self.world_size = setup_distributed()# 将模型移到当前 GPUself.model = self.model.cuda(self.rank)# 包装为 DDP 模型，实现数据并行self.model = DDP(self.model, device_ids=[self.rank])# 全参数优化器，学习率较低self.optimizer = torch.optim.AdamW(self.model.parameters(),lr=config.lr, # 全参数微调 lr 通常为 1e-5 ~ 3e-5betas=(0.9, 0.999),weight_decay=config.weight_decay)def train(self, train_dataset, val_dataset):# 使用 DistributedSampler 确保每个进程获取不同的数据train_sampler = DistributedSampler(train_dataset)train_loader = torch.utils.data.DataLoader(train_dataset,batch_size=self.config.batch_size,sampler=train_sampler, # 分布式采样器num_workers=8 # 数据加载线程数)for epoch in range(self.config.epochs):# 设置随机种子，确保每个 epoch 数据打乱方式不同train_sampler.set_epoch(epoch)self.model.train()for batch in train_loader:self.optimizer.zero_grad()# 将数据移到当前 GPUinput_ids = batch["input_ids"].cuda(self.rank)labels = batch["labels"].cuda(self.rank)# 前向传播并计算损失outputs = self.model(input_ids, labels=labels)loss = outputs.loss# 反向传播和优化loss.backward()torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)self.optimizer.step()
7.2.2 分布式训练启动命令
torchrun --nproc_per_node=8 --nnodes=4 --node_rank=0 --master_addr="10.0.0.1" --master_port=29500 train.py --config configs/full_finetune.json分布式训练流程图：
7.3 行业领域数据预处理源码改造
7.3.1 金融领域数据处理
import jsonimport reclass FinancialDataProcessor:def __init__(self, tokenizer, max_seq_len=2048):# 保存分词器和最大序列长度self.tokenizer = tokenizerself.max_seq_len = max_seq_lendef process_single(self, item):# 构建金融分析提示词模板prompt = f"""分析以下金融数据：公司名称：{item['company']}报告期：{item['period']}营收：{item['revenue']}万元净利润：{item['net_profit']}万元毛利率：{item['gross_margin']}%请分析该公司的财务状况和发展趋势。"""# 获取分析结果作为标签response = item.get("analysis", "")# 调用编码函数return self._encode_pair(prompt, response)def _encode_pair(self, prompt, response):# 将 prompt 和 response 转换为 token idprompt_ids = self.tokenizer.encode(prompt, add_special_tokens=False)response_ids = self.tokenizer.encode(response, add_special_tokens=False)# 计算总长度（包含特殊 token）total_len = len(prompt_ids) + len(response_ids) + 3# 如果超过最大长度，截断 prompt 部分if total_len > self.max_seq_len:max_prompt_len = self.max_seq_len - len(response_ids) - 3prompt_ids = prompt_ids[:max_prompt_len]# 构建输入序列：[bos] + prompt + [sep] + response + [eos]input_ids = ([self.tokenizer.bos_token_id] +prompt_ids +[self.tokenizer.sep_token_id] +response_ids +[self.tokenizer.eos_token_id])# 构建标签：prompt 部分用 -100 忽略，只训练 response 部分labels = ([-100] * (len(prompt_ids) + 2) + # [bos] + prompt + [sep] 不计算损失response_ids + # response 正常计算损失[self.tokenizer.eos_token_id] # [eos] 计算损失)return {"input_ids": input_ids, "labels": labels}def process_dataset(self, raw_data_path, output_path):# 读取原始数据with open(raw_data_path, "r", encoding="utf-8") as f:raw_data = json.load(f)processed = []for item in raw_data:try:processed.append(self.process_single(item))except Exception as e:# 跳过处理失败的数据print(f"Processing error: {e}")# 保存处理后的数据with open(output_path, "w", encoding="utf-8") as f:json.dump(processed, f, ensure_ascii=False, indent=2)
7.3.2 政务领域数据处理
class GovernmentDataProcessor:def __init__(self, tokenizer, max_seq_len=2048):self.tokenizer = tokenizerself.max_seq_len = max_seq_lendef process_single(self, item):# 构建政务问答提示词模板prompt = f"""根据以下政策法规回答问题：政策名称：{item['policy_name']}发布日期：{item['publish_date']}政策摘要：{item['policy_summary']}问题：{item['question']}"""response = item.get("answer", "")return self._encode_pair(prompt, response)def _encode_pair(self, prompt, response):prompt_ids = self.tokenizer.encode(prompt, add_special_tokens=False)response_ids = self.tokenizer.encode(response, add_special_tokens=False)total_len = len(prompt_ids) + len(response_ids) + 3if total_len > self.max_seq_len:max_prompt_len = self.max_seq_len - len(response_ids) - 3prompt_ids = prompt_ids[:max_prompt_len]input_ids = ([self.tokenizer.bos_token_id] +prompt_ids +[self.tokenizer.sep_token_id] +response_ids +[self.tokenizer.eos_token_id])labels = ([-100] * (len(prompt_ids) + 2) +response_ids +[self.tokenizer.eos_token_id])return {"input_ids": input_ids, "labels": labels}
7.4 微调模型合并、权重导出脚本开发
7.4.1 LoRA 权重合并
def merge_lora_weights(base_model, lora_model, save_path):# 获取基座模型和 LoRA 模型的状态字典base_state_dict = base_model.state_dict()lora_state_dict = lora_model.state_dict()# 创建合并后的状态字典merged_state_dict = {}# 先复制基座模型权重for name, param in base_state_dict.items():merged_state_dict[name] = param.clone()# 遍历 LoRA 权重，计算增量并加到基座模型上for name, param in lora_state_dict.items():# 找到 LoRA A 矩阵if "lora_A" in name:# 获取对应的基座模型权重名称base_name = name.replace("_lora_A.weight", ".weight")if base_name in merged_state_dict:# 找到对应的 LoRA B 矩阵lora_B_name = name.replace("_lora_A.weight", "_lora_B.weight")if lora_B_name in lora_state_dict:# 获取 A 和 B 矩阵A = paramB = lora_state_dict[lora_B_name]# 计算 ΔW = BA * (alpha / rank)delta_W = torch.matmul(B, A)merged_state_dict[base_name] += delta_W * (lora_model.config.lora_alpha / lora_model.config.lora_rank)# 保存合并后的权重torch.save(merged_state_dict, save_path)def export_model_for_inference(model, save_path):# 设置模型为评估模式model.eval()# 冻结所有参数for name, param in model.named_parameters():param.requires_grad = False# 保存模型torch.save(model.state_dict(), save_path)print(f"Model exported to {save_path}")
7.4.2 模型导出配置
def export_to_huggingface_format(model, tokenizer, save_dir):# 创建保存目录os.makedirs(save_dir, exist_ok=True)# 保存配置文件model.config.save_pretrained(save_dir)# 保存分词器tokenizer.save_pretrained(save_dir)# 保存模型权重model.save_pretrained(save_dir)print(f"Model saved in HuggingFace format at {save_dir}")
本章小结：
本章详细介绍了基于原生源码的微调实战技术，包括 LoRA 参数高效微调、全参数分布式训练、行业领域数据预处理和微调模型合并导出。这些技术为企业级场景下的模型定制提供了完整的解决方案。 如需沟通：lxb20110121
