第 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 torch
import torch.nn as nn
from torch.nn import functional as F

class LoRAModule(nn.Module):
    def __init__(self, in_dim, out_dim, rank=8, alpha=32, dropout=0.0):
        super().__init__()
        # 低秩矩阵的秩，通常设置为 8-64
        self.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 = 0
        nn.init.zeros_(self.B.weight)

    def forward(self, x):
        # 对输入应用 dropout（如果启用）
        x = self.dropout(x)
        # 计算 BAx：先经过 A 降维，再经过 B 升维，最后乘以缩放因子
        return self.B(self.A(x)) * self.scaling

class 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 层添加 LoRA
        for name, module in self.base_model.named_modules():
            # 只在 query projection 层添加 LoRA
            if 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()]:
                pass
        
        return outputs

7.1.3 LoRA 训练脚本

class LoRATrainer:
    def __init__(self, model, tokenizer, config):
        # 保存模型和分词器
        self.model = model
        self.tokenizer = tokenizer
        self.config = config
        
        # 只优化 LoRA 参数，使用 AdamW 优化器
        self.optimizer = torch.optim.AdamW(
            self.model.lora_layers.parameters(),  # 仅 LoRA 参数
            lr=config.lr,                        # 学习率，通常 1e-4 ~ 1e-3
            betas=(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()
        
        # 将数据移到 GPU
        input_ids = batch["input_ids"].cuda()
        labels = batch["labels"].cuda()
        
        # 前向传播获取 logits
        outputs = self.model(input_ids)
        logits = outputs.logits
        
        # 位移处理：logits 去掉最后一个 token，labels 去掉第一个 token
        shift_logits = logits[..., :-1, :].contiguous()
        shift_labels = labels[..., 1:].contiguous()
        
        # 计算交叉熵损失，忽略 padding token
        loss = F.cross_entropy(
            shift_logits.view(-1, shift_logits.size(-1)),  # 展平 logits
            shift_labels.view(-1),                          # 展平 labels
            ignore_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 超参调优

参数	推荐值	说明
rank	8-64	低秩矩阵秩，越大越接近全参数微调
alpha	32-128	缩放因子，通常设为 rank 的 4 倍
lr	1e-4-1e-3	LoRA 学习率，高于全参数微调
weight_decay	0-0.01	权重衰减，防止过拟合
dropout	0.0-0.1	LoRA 层 dropout
epochs	3-10	训练轮数

7.1.5 超参调优策略

1. Rank 选择：

   • 小任务（分类）：rank=4-8
   • 中任务（问答）：rank=8-16
   • 大任务（生成）：rank=16-64

2. 学习率调优：

   • 初始 lr=5e-4
   • 验证集不提升时降低 lr

7.2 全参数微调分布式训练代码

7.2.1 分布式训练配置

import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.utils.data.distributed import DistributedSampler

def setup_distributed():
    # 初始化分布式进程组，使用 NCCL 后端（GPU 间通信最快）
    dist.init_process_group(backend="nccl")
    # 获取当前进程的 rank（全局唯一编号）
    rank = dist.get_rank()
    # 获取总进程数（world_size）
    world_size = dist.get_world_size()
    
    # 设置当前进程使用的 GPU
    torch.cuda.set_device(rank)
    
    return rank, world_size

class FullParameterTrainer:
    def __init__(self, model, tokenizer, config):
        self.model = model
        self.tokenizer = tokenizer
        self.config = config
        
        # 初始化分布式环境
        self.rank, self.world_size = setup_distributed()
        
        # 将模型移到当前 GPU
        self.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-5
            betas=(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()
                
                # 将数据移到当前 GPU
                input_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 json
import re

class FinancialDataProcessor:
    def __init__(self, tokenizer, max_seq_len=2048):
        # 保存分词器和最大序列长度
        self.tokenizer = tokenizer
        self.max_seq_len = max_seq_len
    
    def 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 id
        prompt_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) - 3
            prompt_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 = tokenizer
        self.max_seq_len = max_seq_len
    
    def 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) + 3
        
        if total_len > self.max_seq_len:
            max_prompt_len = self.max_seq_len - len(response_ids) - 3
            prompt_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 = param
                    B = 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