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-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.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 层添加 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

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()

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()    # 设置当前进程使用的 GPU    torch.cuda.set_device(rank)    return rank, world_sizeclass 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()

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

import jsonimport reclass 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)

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}

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}")

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}")