Python 大模型微调

作者：追风剑情发布于：2026-6-10 16:13 分类：AI

一、什么情况下需要做模型微调？

当你有一个预训练模型（如 BERT、ResNet、Whisper 等），并且希望将其应用到一个与预训练任务不完全相同的下游任务时，就需要微调。

二、什么情况下需要做全参数微调？

当你数据充足且需要最高性能，或者任务与预训练任务差异很大时，选择全参数微调。通常需要数千到数万条标注数据，以防止过拟合。数据越少，过拟合风险越高。

三、什么情况下只需要做全连接层微调？

如果预训练模型的特征提取器已经能够很好地表征你的数据。或者数据非常少（几十到几百条）、希望节省资源、或快速验证特征质量时，优先选择仅微调全连接层（线性探测）。

示例：

import os
# 国内镜像
os.environ['HF_ENDPOINT'] = 'https://hf-mirror.com'
import torch
from torch.utils.data import Dataset, DataLoader
from transformers import AutoTokenizer, AutoModelForSequenceClassification, AdamW
from sklearn.metrics import accuracy_score
import numpy as np

# ---------- 1. 准备小数据集（中文评论情感） ----------
# 训练数据集
train_texts = [
    # 负面 10 条
    "太难吃了，服务态度也很差。",
    "等了40分钟还没上菜，太失望了。",
    "价格贵得离谱，完全不值。",
    "菜品不新鲜，吃完肚子不舒服。",
    "环境嘈杂，根本无法安静吃饭。",
    "点错了菜还不给换，差评。",
    "分量太少，两个人不够吃。",
    "餐具不干净，有油渍。",
    "外卖包装破损，汤洒了一袋子。",
    "口味一般，不会再来第二次。",
    # 正面 10 条
    "味道很棒，服务也很周到。",
    "环境优雅，适合约会。",
    "价格实惠，分量足。",
    "上菜速度快，味道正宗。",
    "服务员很热情，主动倒水。",
    "菜品精致，拍照很好看。",
    "交通方便，位置好找。",
    "团购超值，性价比高。",
    "孩子很喜欢，说下次还要来。",
    "整体体验非常好，推荐给大家。",
]
train_labels = [0,0,0,0,0,0,0,0,0,0, 1,1,1,1,1,1,1,1,1,1]

# 验证数据集
val_texts = [
    "味道还不错，就是价格稍贵。",   # 偏正面
    "上菜太慢了，等了半小时。",     # 负面
    "服务员很贴心，送了小礼物。",   # 正面
    "菜品太咸了，没法吃。",         # 负面
]
val_labels = [1, 0, 1, 0]

# ---------- 2. 加载预训练模型和分词器 ----------
model_name = "distilbert-base-multilingual-cased"  # 支持中文的小模型
# 加载一个与预训练模型配套的分词器
tokenizer = AutoTokenizer.from_pretrained(model_name)
# 给模型添加一个分类头。
# num_labels: 标签类别数
model = AutoModelForSequenceClassification.from_pretrained(model_name, num_labels=2)

# ---------- 3. 自定义Dataset类 ----------
class TextDataset(Dataset):
    def __init__(self, texts, labels, tokenizer, max_len=128):
        self.texts = texts
        self.labels = labels
        self.tokenizer = tokenizer
        self.max_len = max_len

    def __len__(self):
        return len(self.texts)

    def __getitem__(self, idx):
        text = self.texts[idx]
        label = self.labels[idx]
        encoding = self.tokenizer(
            text,                       # 要编码的原始文本字符串
            truncation=True,            # 是否截断超长序列
            padding='max_length',       # 填充方式：填充到固定长度
            max_length=self.max_len,    # 最大序列长度（截断和填充的目标长度）
            return_tensors='pt'         # 返回 PyTorch 张量格式
        )
        return {
            # 每个 token 在词汇表中的数字 ID
            'input_ids': encoding['input_ids'].flatten(),
            # 掩码列表 1:真实token；0:填充token
            'attention_mask': encoding['attention_mask'].flatten(),
            # 转成张量
            'label': torch.tensor(label, dtype=torch.long)
        }

# 训练数据集
train_dataset = TextDataset(train_texts, train_labels, tokenizer)
# 验证数据集
val_dataset = TextDataset(val_texts, val_labels, tokenizer)

train_loader = DataLoader(train_dataset, batch_size=2, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=2)

# ---------- 4. 设置优化器和训练参数 ----------
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# 将模型加载进CPU或GPU
model.to(device)

# 全参数训练
# 学习率通常比从头训练小10~100倍
optimizer = AdamW(model.parameters(), lr=2e-5)

# ---仅训练分类层(全链接层)---
# 冻结除分类头外的所有参数
#for name, param in model.named_parameters():
#    if 'classifier' in name or 'pre_classifier' in name:
#        param.requires_grad = True
#    else:
#        param.requires_grad = False
# 只将需要更新的参数传给优化器
#optimizer = AdamW(filter(lambda p: p.requires_grad, model.parameters()), lr=2e-5)
# --- end ---

# 训练次数
epochs = 10

# ---------- 5. 训练循环（微调核心） ----------
for epoch in range(epochs):
    # 启动训练模式
    model.train()
    total_loss = 0
    for batch in train_loader:
        # 数据与模型必须在同一设备中进行计算(例如：GPU)
        input_ids = batch['input_ids'].to(device)
        attention_mask = batch['attention_mask'].to(device)
        labels = batch['label'].to(device)

        # 将模型所有可训练参数的梯度清零
        optimizer.zero_grad()
        # 执行前向传播
        outputs = model(input_ids, attention_mask=attention_mask, labels=labels)
        # 获取损失值
        loss = outputs.loss
        # 执行反向传播
        loss.backward()
        # 根据计算出的梯度更新模型参数
        optimizer.step()
        # 保存累计损失
        total_loss += loss.item()

    # 模型验证/评估
    # 启动评估模式
    model.eval()
    # preds: 预测标签列表
    # true_labels: 真实标签列表
    preds, true_labels = [], []
    # 临时禁用梯度计算
    with torch.no_grad():
        for batch in val_loader:
            # 数据与模型必须在同一设备中进行计算(例如：GPU)
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['label'].to(device)
            
            # 前向传播
            outputs = model(input_ids, attention_mask=attention_mask)
            # 获取模型计算出的原始分数
            logits = outputs.logits
            # 返回分数最高的预测类别
            pred = torch.argmax(logits, dim=1)
            # pred.cpu().numpy() 将预测张量从 GPU 移到 CPU，并转换为 NumPy 数组
            # preds.extend() 追加到总列表末尾
            preds.extend(pred.cpu().numpy())
            true_labels.extend(labels.cpu().numpy())

    acc = accuracy_score(true_labels, preds)
    print(f"Epoch {epoch+1}/{epochs} | Loss: {total_loss/len(train_loader):.4f} | Val Acc: {acc:.4f}")

# ---------- 6. 测试一条新评论 ----------
test_text = "这个产品太垃圾了，千万别买。"
inputs = tokenizer(test_text, return_tensors='pt', truncation=True, padding=True).to(device)
model.eval()
with torch.no_grad():
    logits = model(**inputs).logits
    pred = torch.argmax(logits, dim=1).item()
print(f"\n测试评论: {test_text}")
print(f"预测结果: {'正面' if pred == 1 else '负面'}")

全参数训练