零基础入门：DeepSeek微调教程来了！

需要引入的库：
pip install torch transformers peft datasets matplotlib accelerate safetensors

import torch
import matplotlib.pyplot as plt
from transformers import (
    AutoTokenizer,
    AutoModelForCausalLM,
    TrainingArguments,
    Trainer,
    TrainerCallback
)
from peft import LoraConfig, get_peft_model
from datasets import load_dataset
import os

# 配置路径（根据实际路径修改）
model_path = r"你的模型路径"  # 模型路径
data_path = r"你的数据集路径"  # 数据集路径
output_path = r"你的保存微调后的模型路径"  # 微调后模型保存路径

# 强制使用GPU
assert torch.cuda.is_available(), //"必须使用GPU进行训练！"
device = torch.device("cuda")

# 自定义回调记录Loss
class LossCallback(TrainerCallback):
    def __init__(self):
        self.losses = []

    def on_log(self, args, state, control, logs=None, **kwargs):
        if "loss" in logs:
            self.losses.append(logs["loss"])

# 数据预处理函数
def process_data(tokenizer):
    dataset = load_dataset("json", data_files=data_path, split="train[:1500]")

    def format_example(example):
        instruction = f"诊断问题：{example['Question']}\n详细分析：{example['Complex_CoT']}"
        inputs = tokenizer(
            f"{instruction}\n### 答案：\n{example['Response']}<|endoftext|>",
            padding="max_length",
            truncation=True,
            max_length=512,
            return_tensors="pt"
        )
        return {"input_ids": inputs["input_ids"].squeeze(0), "attention_mask": inputs["attention_mask"].squeeze(0)}

    return dataset.map(format_example, remove_columns=dataset.column_names)

# LoRA配置
peft_config = LoraConfig(
    r=16,
    lora_alpha=32,
    target_modules=["q_proj", "v_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM"
)

# 训练参数配置
training_args = TrainingArguments(
    output_dir=output_path,
    per_device_train_batch_size=2,  # 显存优化设置
    gradient_accumulation_steps=4,  # 累计梯度相当于batch_size=8
    num_train_epochs=3,
    learning_rate=3e-4,
    fp16=True,  # 开启混合精度
    logging_steps=20,
    save_strategy="no",
    report_to="none",
    optim="adamw_torch",
    no_cuda=False,  # 强制使用CUDA
    dataloader_pin_memory=False,  # 加速数据加载
    remove_unused_columns=False  # 防止删除未使用的列
)

def main():
    # 创建输出目录
    os.makedirs(output_path, exist_ok=True)

    # 加载tokenizer
    tokenizer = AutoTokenizer.from_pretrained(model_path)
    tokenizer.pad_token = tokenizer.eos_token

    # 加载模型到GPU
    model = AutoModelForCausalLM.from_pretrained(
        model_path,
        torch_dtype=torch.float16,
        device_map={"": device}  # 强制使用指定GPU
    )
    model = get_peft_model(model, peft_config)
    model.print_trainable_parameters()

    # 准备数据
    dataset = process_data(tokenizer)

    # 训练回调
    loss_callback = LossCallback()

    # 数据加载器
    def data_collator(data):
        batch = {
            "input_ids": torch.stack([torch.tensor(d["input_ids"]) for d in data]).to(device),
            "attention_mask": torch.stack([torch.tensor(d["attention_mask"]) for d in data]).to(device),
            "labels": torch.stack([torch.tensor(d["input_ids"]) for d in data]).to(device)  # 使用input_ids作为labels
        }
        return batch

    # 创建Trainer
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=dataset,
        data_collator=data_collator,
        callbacks=[loss_callback]
    )

    # 开始训练
    print("开始训练...")
    trainer.train()

    # 保存最终模型
    trainer.model.save_pretrained(output_path)
    print(f"模型已保存至：{output_path}")

    # 绘制训练集损失Loss曲线
    plt.figure(figsize=(10, 6))
    plt.plot(loss_callback.losses)
    plt.title("Training Loss Curve")
    plt.xlabel("Steps")
    plt.ylabel("Loss")
    plt.savefig(os.path.join(output_path, "loss_curve.png"))
    print("Loss曲线已保存")

if __name__ == "__main__":
    main()

import torch
import matplotlib.pyplot as plt
from transformers import (  # HuggingFace Transformer模型工具
    AutoTokenizer,
    AutoModelForCausalLM,
    TrainingArguments,
    Trainer,
    TrainerCallback
)
from peft import LoraConfig, get_peft_model  # 参数高效微调库
from datasets import load_dataset  # 数据集加载工具
import os  # 系统路径操作

# 配置路径（根据实际路径修改）
model_path = r"你的模型路径"  # 预训练模型存放路径
data_path = r"你的数据集路径"  # 训练数据路径（JSON格式）
output_path = r"你的保存微调后的模型路径"  # 微调后模型保存位置

# 强制使用GPU（确保CUDA可用）
assert torch.cuda.is_available(), "必须使用GPU进行训练！"
device = torch.device("cuda")  # 指定使用CUDA设备

class LossCallback(TrainerCallback):
    def __init__(self):
        self.losses = []  # 存储损失值的列表

    # 当训练过程中有日志输出时触发
    def on_log(self, args, state, control, logs=None, **kwargs):
        if "loss" in logs:  # 过滤并记录损失值
            self.losses.append(logs["loss"])

def process_data(tokenizer):
    # 从JSON文件加载数据集（仅取前1500条）
    dataset = load_dataset("json", data_files=data_path, split="train[:1500]")

    # 单条数据格式化函数
    def format_example(example):
        # 拼接指令和答案（固定模板）
        instruction = f"诊断问题：{example['Question']}\n详细分析：{example['Complex_CoT']}"
        inputs = tokenizer(
            f"{instruction}\n### 答案：\n{example['Response']}<|endoftext|>",  # 添加结束符
            padding="max_length",  # 填充至最大长度
            truncation=True,       # 超长截断
            max_length=512,        # 最大序列长度
            return_tensors="pt"    # 返回PyTorch张量
        )
        # 返回处理后的输入（移除batch维度）
        return {"input_ids": inputs["input_ids"].squeeze(0), 
                "attention_mask": inputs["attention_mask"].squeeze(0)}

    # 应用格式化函数并移除原始列
    return dataset.map(format_example, remove_columns=dataset.column_names)

instruction = f"诊断问题：{example['Question']}\n详细分析：{example['Complex_CoT']}"

inputs = tokenizer(
    f"{instruction}\n### 答案：\n{example['Response']}<|endoftext|>",  # 添加结束符
    padding="max_length",  # 填充至最大长度
    truncation=True,       # 超长截断
    max_length=512,        # 最大序列长度
    return_tensors="pt"    # 返回PyTorch张量
)

return {"input_ids": inputs["input_ids"].squeeze(0), 
        "attention_mask": inputs["attention_mask"].squeeze(0)}

return dataset.map(format_example, remove_columns=dataset.column_names)

peft_config = LoraConfig(
    r=16,               # LoRA秩（矩阵分解维度）
    lora_alpha=32,      # 缩放系数（控制适配器影响强度）
    target_modules=["q_proj", "v_proj"],  # 要适配的注意力模块（查询/值投影）
    lora_dropout=0.05,  # 防止过拟合的Dropout率
    bias="none",        # 不训练偏置参数
    task_type="CAUSAL_LM"  # 任务类型（因果语言模型）
)

training_args = TrainingArguments(
    output_dir=output_path,        # 输出目录（模型/日志）
    per_device_train_batch_size=2, # 单GPU批次大小（显存优化）
    gradient_accumulation_steps=4, # 梯度累积步数（等效batch_size=8）
    num_train_epochs=3,            # 训练轮次
    learning_rate=3e-4,            # 初始学习率
    fp16=True,                     # 启用混合精度训练（节省显存）
    logging_steps=20,              # 每隔20步记录日志
    save_strategy="no",            # 不保存中间检查点
    report_to="none",              # 禁用第三方报告（如W&B）
    optim="adamw_torch",           # 优化器类型
    no_cuda=False,                 # 强制使用CUDA
    dataloader_pin_memory=False,   # 禁用锁页内存（加速数据加载）
    remove_unused_columns=False    # 保留未使用的列（避免数据错误）
)

def main():
    # 创建输出目录（如果不存在）
    os.makedirs(output_path, exist_ok=True)

    # 加载Tokenizer并设置填充符
    tokenizer = AutoTokenizer.from_pretrained(model_path)
    tokenizer.pad_token = tokenizer.eos_token  # 使用EOS作为填充符

    # 加载预训练模型（半精度+指定GPU）
    model = AutoModelForCausalLM.from_pretrained(
        model_path,
        torch_dtype=torch.float16,       # 半精度加载（节省显存）
        device_map={"": device}          # 指定使用的GPU设备
    )
    # 应用LoRA适配器
    model = get_peft_model(model, peft_config)
    model.print_trainable_parameters()   # 打印可训练参数量

    # 准备训练数据集
    dataset = process_data(tokenizer)

    # 初始化损失记录回调
    loss_callback = LossCallback()

    # 数据整理函数（构造批次）
    def data_collator(data):
        batch = {
            "input_ids": torch.stack([torch.tensor(d["input_ids"]) for d in data]).to(device),
            "attention_mask": torch.stack([torch.tensor(d["attention_mask"]) for d in data]).to(device),
            "labels": torch.stack([torch.tensor(d["input_ids"]) for d in data]).to(device)  # 标签=输入（因果LM任务）
        }
        return batch

    # 初始化Trainer
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=dataset,
        data_collator=data_collator,  # 自定义数据整理
        callbacks=[loss_callback]     # 添加回调
    )

    # 执行训练
    print("开始训练...")
    trainer.train()

    # 保存微调后的模型
    trainer.model.save_pretrained(output_path)
    print(f"模型已保存至：{output_path}")

    # 绘制损失曲线
    plt.figure(figsize=(10, 6))
    plt.plot(loss_callback.losses)
    plt.title("Training Loss Curve")
    plt.xlabel("Steps")
    plt.ylabel("Loss")
    plt.savefig(os.path.join(output_path, "loss_curve.png"))  # 保存为PNG
    print("Loss曲线已保存")

if __name__ == "__main__":
    main()

tokenizer = AutoTokenizer.from_pretrained(model_path)
tokenizer.pad_token = tokenizer.eos_token  # 使用EOS作为填充符

model = AutoModelForCausalLM.from_pretrained(
    model_path,
    torch_dtype=torch.float16,       # 半精度加载（节省显存）
    device_map={"": device}          # 指定使用的GPU设备
)

def data_collator(data):
    batch = {
        "input_ids": torch.stack([torch.tensor(d["input_ids"]) for d in data]).to(device),
        "attention_mask": torch.stack([torch.tensor(d["attention_mask"]) for d in data]).to(device),
        "labels": torch.stack([torch.tensor(d["input_ids"]) for d in data]).to(device)  # 标签=输入（因果LM任务）
    }
    return batch

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=dataset,
    data_collator=data_collator,  # 自定义数据整理
    callbacks=[loss_callback]     # 添加回调
)