当前位置：首页 > news >正文

学习记录：初次学习使用transformers进行大模型微调

news 来源：原创 2025/5/3 5:03:25

初次使用transformers进行大模型微调

环境：

电脑配置：
笔记本电脑：I5（6核12线程） + 16G + RTX3070（8G显存）
需要自行解决科学上网

Python环境：
python版本:3.8.8
大模型：microsoft/DialoGPT-medium（微软的对话大模型,模型小，笔记本也能学习微调）
数据集：daily_dialog （日常对话数据集）

其他：
模型及数据集：使用来源于抱抱脸

微调大模型

准备工作：

下载模型：

找到自己想要的模型：

打开抱抱脸官网——点击Model：
输入要搜索的模型(这里以DialoGPT-medium为例)：
复制名称到代码中替换要下载的模型名称：

在这里插入图片描述
模型下载：

import os
from transformers import AutoModel, AutoTokenizer

# 因为使用了科学上网，需要进行处理
os.environ["HTTP_PROXY"] = "http://127.0.0.1:xxxx"
os.environ["HTTPS_PROXY"] = "http://127.0.0.1:xxxx"

if __name__ == '__main__':
    # model_name = 'google-t5/t5-small'  # 要下载的模型名称
    model_name = 'microsoft/DialoGPT-medium'  # 要下载的模型名称 需要到抱抱脸进行复制
    cache_dir = r'xxxx'  # 模型保存位置

    # 加载模型时指定下载路径
    model = AutoModel.from_pretrained(model_name, cache_dir=cache_dir)

下载数据集：

找到自己想要的模型：

打开抱抱脸官网——点击Datasets：
输入要搜索的内容，点击对应数据集进入：
找到适合用的模型后，点击复制

开始微调训练

代码示例：

# 系统模块
import os

# 第三方库
from transformers import AutoTokenizer, AutoModelForCausalLM, TrainingArguments, Trainer
from datasets import load_dataset

# 设置代理（注意：可能需要根据实际网络环境调整或移除）
os.environ["HTTP_PROXY"] = "http://127.0.0.1:xxxx"  # HTTP代理设置
os.environ["HTTPS_PROXY"] = "http://127.0.0.1:xxxx"  # HTTPS代理设置

if __name__ == '__main__':
    # 数据准备阶段 --------------------------------------------------------------
    # 加载完整数据集（daily_dialog包含日常对话数据集）
    full_dataset = load_dataset("daily_dialog", trust_remote_code=True)

    # 创建子数据集（仅使用训练集前500条样本，用于快速实验）
    dataset = {
        "train": full_dataset["train"].select(range(500))  # select保持数据集结构
    }

    # 模型加载阶段 --------------------------------------------------------------
    # 模型配置参数
    model_name = "microsoft/DialoGPT-medium"  # 使用微软的对话生成预训练模型
    cache_dir = r'xxx'  # 本地模型缓存路径

    # 加载分词器（重要：设置填充token与EOS token一致）
    tokenizer = AutoTokenizer.from_pretrained(model_name, cache_dir=cache_dir)
    tokenizer.pad_token = tokenizer.eos_token  # 将填充token设置为与EOS相同

    # 加载预训练模型（使用因果语言模型结构）
    model = AutoModelForCausalLM.from_pretrained(model_name, cache_dir=cache_dir)


    # 数据预处理阶段 ------------------------------------------------------------
    def tokenize_function(examples):
        """将对话数据转换为模型输入格式的预处理函数"""
        # 将多轮对话用EOS token连接，并在结尾添加EOS
        dialogues = [
            tokenizer.eos_token.join(dialog) + tokenizer.eos_token
            for dialog in examples["dialog"]
        ]

        # 对文本进行分词处理
        tokenized = tokenizer(
            dialogues,
            truncation=True,  # 启用截断
            max_length=512,  # 最大序列长度
            padding="max_length"  # 填充到最大长度（静态填充）
        )

        # 创建标签（对于因果语言模型，标签与输入相同）
        tokenized["labels"] = tokenized["input_ids"].copy()
        return tokenized


    # 应用预处理（保留数据集结构）
    tokenized_dataset = {
        "train": dataset["train"].map(
            tokenize_function,
            batched=True,  # 批量处理提升效率
            batch_size=50,  # 每批处理50个样本
            remove_columns=["dialog", "act", "emotion"]  # 移除原始文本列
        )
    }

    # 数据验证（检查预处理结果）
    print("Sample keys:", tokenized_dataset["train"][0].keys())  # 应包含input_ids, attention_mask, labels
    print("Input IDs:", tokenized_dataset["train"][0]["input_ids"][:5])  # 检查前5个token

    # 训练配置阶段 --------------------------------------------------------------
    training_args = TrainingArguments(
        output_dir="./dialo_finetuned",  # 输出目录
        per_device_train_batch_size=2,  # 每个设备的批次大小（根据显存调整）
        gradient_accumulation_steps=8,  # 梯度累积步数（模拟更大batch size）
        learning_rate=1e-5,  # 初始学习率（可调超参数）
        num_train_epochs=3,  # 训练轮次（根据需求调整）
        fp16=True,  # 启用混合精度训练（需要GPU支持）
        logging_steps=10,  # 每10步记录日志
        # 可添加的优化参数：
        # evaluation_strategy="steps",    # 添加验证策略
        # save_strategy="epoch",          # 保存策略
        # warmup_steps=100,               # 学习率预热步数
    )

    # 创建训练器
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=tokenized_dataset["train"],  # 训练数据集
        # 可扩展功能：
        # eval_dataset=tokenized_dataset["validation"],  # 添加验证集
        # data_collator=...,             # 自定义数据整理器
        # compute_metrics=...,           # 添加评估指标
    )

    # 训练执行阶段 --------------------------------------------------------------
    trainer.train()  # 启动训练

    # 模型保存阶段 --------------------------------------------------------------
    model.save_pretrained("./dialo_finetuned")  # 保存模型权重
    tokenizer.save_pretrained("./dialo_finetuned")  # 保存分词器
    # 推荐使用以下方式统一保存：
    trainer.save_model("./dialo_finetuned")       # 官方推荐保存方式

微调后使用

代码：

from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
from transformers import TextStreamer
from collections import deque
import torch


def optimized_generation(text, tokenizer, model):
    inputs = tokenizer(text, return_tensors="pt").to(model.device)

    outputs = model.generate(
        **inputs,
        max_new_tokens=150,
        temperature=0.9,  # 越高越有创意 (0-1)
        top_k=50,  # 限制候选词数量
        top_p=0.95,  # 核采样阈值
        repetition_penalty=1.2,  # 抑制重复
        num_beams=3,  # 束搜索宽度
        early_stopping=True,
        do_sample=True
    )

    return tokenizer.decode(outputs[0], skip_special_tokens=True)


# 单轮对话
def simple_chat(model_path, text, max_length=100):
    """
    单轮对话
    :param text:
    :param max_length:
    :return:
    """

    # 加载模型和分词器
    tokenizer = AutoTokenizer.from_pretrained(model_path)
    model = AutoModelForCausalLM.from_pretrained(model_path)

    # 确保pad_token设置正确
    tokenizer.pad_token = tokenizer.eos_token

    # inputs = tokenizer(text + tokenizer.eos_token, return_tensors="pt")
    # outputs = model.generate(
    #     inputs.input_ids,
    #     max_length=max_length,
    #     pad_token_id=tokenizer.eos_token_id,
    #     temperature=0.7,
    #     do_sample=True
    # )
    # response = tokenizer.decode(outputs[0], skip_special_tokens=True)
    response = optimized_generation(text + tokenizer.eos_token, tokenizer, model)
    return response[len(text):]  # 去除输入文本


# 多轮对话
class DialogueBot:
    def __init__(self, model_path, max_history=3):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        self.model = AutoModelForCausalLM.from_pretrained(model_path).to("cuda")
        self.max_history = max_history
        self.history = deque(maxlen=max_history * 2)  # 每轮包含用户和机器人各一条

        # 确保pad_token设置
        if self.tokenizer.pad_token is None:
            self.tokenizer.pad_token = self.tokenizer.eos_token

    def generate_response(self, user_input):
        # 添加用户输入（带EOS）
        self.history.append(f"User: {user_input}{self.tokenizer.eos_token}")

        # 构建prompt并编码
        prompt = self._build_prompt()
        inputs = self.tokenizer(
            prompt,
            return_tensors="pt",
            max_length=512,
            truncation=True
        ).to(self.model.device)

        # 流式输出
        # streamer = TextStreamer(self.tokenizer)

        # 生成回复
        outputs = self.model.generate(
            inputs.input_ids,
            attention_mask=inputs.attention_mask,
            max_new_tokens=150,
            temperature=0.85,
            top_p=0.95,
            eos_token_id=self.tokenizer.eos_token_id,
            pad_token_id=self.tokenizer.eos_token_id,
            do_sample=True,
            # streamer=streamer,
            early_stopping=True
        )

        # 解码并处理回复
        full_response = self.tokenizer.decode(
            outputs[0][inputs.input_ids.shape[-1]:],
            skip_special_tokens=True
        )
        # 清理无效内容（按第一个EOS截断）
        clean_response = full_response.split(self.tokenizer.eos_token)[0].strip()

        # 添加机器人回复到历史（带EOS）
        self.history.append(f"Bot: {clean_response}{self.tokenizer.eos_token}")

        return clean_response

    def _build_prompt(self):
        return "".join(self.history)


if __name__ == '__main__':
    # 指定模型路径
    model_path = "./dialo_finetuned"

    # 测试单轮对话
    print(simple_chat(model_path, "Hello, how are you?"))

 	# 使用示例 多轮对话
    # bot = DialogueBot(model_path)
    # while True:
    #     user_input = input("You: ")
    #     if user_input.lower() == "exit":
    #         break
    #     print("Bot:", bot.generate_response(user_input))