预训练与微调（Pre-training & Fine-tuning）是现代大语言模型的核心训练范式。这一范式通过两阶段训练策略，使得模型能够先在大规模无标注数据上学习通用语言表示，再在特定任务上进行优化。本文将深入探讨预训练与微调的原理、方法、技术细节和最佳实践。

预训练与微调范式概述

核心思想

预训练与微调范式的核心思想是将模型训练分为两个阶段：

class PretrainingFineTuningParadigm:
    """
    预训练与微调范式
    """
    
    def __init__(self):
        self.stages = {
            '预训练阶段': {
                'data': '大规模无标注文本（TB级别）',
                'objective': '语言建模（预测下一个词）',
                'duration': '数周到数月',
                'cost': '数百万到数千万美元',
                'output': '通用语言模型'
            },
            '微调阶段': {
                'data': '任务特定标注数据（千到百万样本）',
                'objective': '任务特定目标（分类、生成等）',
                'duration': '数小时到数天',
                'cost': '数百到数千美元',
                'output': '任务特定模型'
            }
        }
    
    def show_comparison(self):
        """展示两阶段对比"""
        print("预训练 vs 微调对比")
        print("=" * 80)
        
        aspects = ['data', 'objective', 'duration', 'cost', 'output']
        aspect_names = ['数据', '目标', '时长', '成本', '输出']
        
        for aspect, name in zip(aspects, aspect_names):
            print(f"\n{name}:")
            print(f"  预训练: {self.stages['预训练阶段'][aspect]}")
            print(f"  微调:   {self.stages['微调阶段'][aspect]}")

paradigm = PretrainingFineTuningParadigm()
paradigm.show_comparison()

为什么需要预训练？

def why_pretraining():
    """
    解释为什么需要预训练
    """
    print("\n为什么需要预训练？")
    print("=" * 80)
    
    reasons = [
        {
            'reason': '标注数据稀缺',
            'explanation': '大多数任务的标注数据有限，难以训练大规模模型',
            'solution': '预训练利用海量无标注数据学习通用表示'
        },
        {
            'reason': '知识迁移',
            'explanation': '不同任务之间存在共享的语言知识',
            'solution': '预训练学习的知识可以迁移到下游任务'
        },
        {
            'reason': '训练效率',
            'explanation': '从零训练大模型成本极高',
            'solution': '预训练模型可以快速适应新任务'
        },
        {
            'reason': '泛化能力',
            'explanation': '小数据集容易过拟合',
            'solution': '预训练提供了良好的初始化，提升泛化能力'
        }
    ]
    
    for i, item in enumerate(reasons, 1):
        print(f"\n{i}. {item['reason']}")
        print(f"   问题: {item['explanation']}")
        print(f"   解决: {item['solution']}")

why_pretraining()

预训练阶段详解

预训练目标

1. 自回归语言模型（Autoregressive LM）

GPT系列使用的预训练目标：

import numpy as np

class AutoregressiveLM:
    """
    自回归语言模型
    
    目标：给定前文，预测下一个词
    P(x) = P(x1) * P(x2|x1) * P(x3|x1,x2) * ... * P(xn|x1,...,xn-1)
    """
    
    def __init__(self, vocab_size=50000, d_model=768):
        self.vocab_size = vocab_size
        self.d_model = d_model
    
    def compute_loss(self, tokens):
        """
        计算自回归语言模型损失
        
        Args:
            tokens: 输入序列 [batch_size, seq_len]
        
        Returns:
            loss: 平均负对数似然
        """
        batch_size, seq_len = tokens.shape
        total_loss = 0
        
        # 对每个位置计算损失
        for i in range(1, seq_len):
            # 使用前i个词预测第i+1个词
            context = tokens[:, :i]
            target = tokens[:, i]
            
            # 模型预测（这里用随机值模拟）
            logits = np.random.randn(batch_size, self.vocab_size)
            
            # 计算交叉熵损失
            loss = self.cross_entropy(logits, target)
            total_loss += loss
        
        return total_loss / (seq_len - 1)
    
    def cross_entropy(self, logits, target):
        """交叉熵损失"""
        # Softmax
        exp_logits = np.exp(logits - np.max(logits, axis=1, keepdims=True))
        probs = exp_logits / np.sum(exp_logits, axis=1, keepdims=True)
        
        # 负对数似然
        batch_size = logits.shape[0]
        loss = -np.mean([np.log(probs[i, target[i]] + 1e-10) 
                         for i in range(batch_size)])
        return loss
    
    def demonstrate(self):
        """演示训练过程"""
        print("\n自回归语言模型训练示例")
        print("=" * 60)
        
        # 模拟一个批次
        batch_size = 2
        seq_len = 5
        tokens = np.random.randint(0, 100, (batch_size, seq_len))
        
        print(f"输入序列形状: {tokens.shape}")
        print(f"序列内容:\n{tokens}\n")
        
        print("训练步骤:")
        for i in range(1, seq_len):
            context = tokens[:, :i]
            target = tokens[:, i]
            print(f"  步骤{i}: 输入长度={i}, 预测位置={i}")
            print(f"    样本1: {context[0]} -> {target[0]}")
            print(f"    样本2: {context[1]} -> {target[1]}")

ar_lm = AutoregressiveLM()
ar_lm.demonstrate()

2. 掩码语言模型（Masked LM）

BERT使用的预训练目标：

class MaskedLM:
    """
    掩码语言模型
    
    目标：预测被掩码的词
    随机掩盖15%的词，让模型预测
    """
    
    def __init__(self, vocab_size=50000, mask_prob=0.15):
        self.vocab_size = vocab_size
        self.mask_prob = mask_prob
        self.mask_token_id = vocab_size - 1
    
    def create_masked_input(self, tokens):
        """
        创建掩码输入
        
        对于被选中的15%的词：
        - 80%替换为[MASK]
        - 10%替换为随机词
        - 10%保持不变
        """
        masked_tokens = tokens.copy()
        labels = np.full_like(tokens, -100)  # -100表示不计算损失
        
        # 随机选择15%的位置
        mask_positions = np.random.rand(*tokens.shape) < self.mask_prob
        
        for i in range(tokens.shape[0]):
            for j in range(tokens.shape[1]):
                if mask_positions[i, j]:
                    labels[i, j] = tokens[i, j]  # 保存原始标签
                    
                    rand = np.random.rand()
                    if rand < 0.8:
                        # 80%: 替换为[MASK]
                        masked_tokens[i, j] = self.mask_token_id
                    elif rand < 0.9:
                        # 10%: 替换为随机词
                        masked_tokens[i, j] = np.random.randint(0, self.vocab_size)
                    # 10%: 保持不变
        
        return masked_tokens, labels
    
    def demonstrate(self):
        """演示掩码过程"""
        print("\n\n掩码语言模型示例")
        print("=" * 60)
        
        # 原始序列
        tokens = np.array([[10, 20, 30, 40, 50, 60, 70, 80]])
        print(f"原始序列: {tokens[0]}")
        
        # 创建掩码输入
        masked_tokens, labels = self.create_masked_input(tokens)
        
        print(f"掩码序列: {masked_tokens[0]}")
        print(f"标签:     {labels[0]}")
        print(f"\n说明:")
        print(f"  {self.mask_token_id} = [MASK] token")
        print(f"  -100 = 不计算损失的位置")

mlm = MaskedLM()
mlm.demonstrate()

3. 其他预训练目标

class OtherPretrainingObjectives:
    """其他预训练目标"""
    
    def __init__(self):
        self.objectives = {
            'Next Sentence Prediction (NSP)': {
                'description': '预测两个句子是否连续',
                'used_in': 'BERT',
                'example': '句子A + 句子B -> 是否连续？'
            },
            'Sentence Order Prediction (SOP)': {
                'description': '预测两个句子的顺序',
                'used_in': 'ALBERT',
                'example': '句子A + 句子B -> 顺序是否正确？'
            },
            'Replaced Token Detection (RTD)': {
                'description': '检测哪些词被替换了',
                'used_in': 'ELECTRA',
                'example': '判断每个词是否是原始词'
            },
            'Permutation Language Modeling': {
                'description': '随机排列顺序进行预测',
                'used_in': 'XLNet',
                'example': '打乱顺序后预测原始序列'
            }
        }
    
    def show_objectives(self):
        """展示各种预训练目标"""
        print("\n\n其他预训练目标")
        print("=" * 80)
        
        for name, info in self.objectives.items():
            print(f"\n{name}")
            print(f"  描述: {info['description']}")
            print(f"  使用: {info['used_in']}")
            print(f"  示例: {info['example']}")

other_obj = OtherPretrainingObjectives()
other_obj.show_objectives()

预训练数据

class PretrainingData:
    """预训练数据管理"""
    
    def __init__(self):
        self.data_sources = {
            'Common Crawl': {
                'size': '数PB',
                'quality': '低到中',
                'description': '网页爬取数据，需要大量清洗'
            },
            'Wikipedia': {
                'size': '~20GB',
                'quality': '高',
                'description': '高质量百科全书内容'
            },
            'Books': {
                'size': '~100GB',
                'quality': '高',
                'description': '书籍语料，长文本'
            },
            'GitHub': {
                'size': '~1TB',
                'quality': '中',
                'description': '代码数据，用于代码模型'
            },
            'Reddit': {
                'size': '~100GB',
                'quality': '中',
                'description': '社交媒体对话数据'
            }
        }
    
    def show_data_sources(self):
        """展示数据来源"""
        print("\n\n预训练数据来源")
        print("=" * 80)
        
        for source, info in self.data_sources.items():
            print(f"\n{source}")
            print(f"  规模: {info['size']}")
            print(f"  质量: {info['quality']}")
            print(f"  说明: {info['description']}")
    
    def data_processing_pipeline(self):
        """数据处理流程"""
        print("\n\n数据处理流程")
        print("=" * 80)
        
        pipeline = """
        1. 数据收集
           ├─ 爬取网页
           ├─ 下载公开数据集
           └─ 获取授权数据
        
        2. 数据清洗
           ├─ 去除HTML标签
           ├─ 过滤低质量内容
           ├─ 去重
           └─ 语言检测
        
        3. 数据过滤
           ├─ 内容安全过滤
           ├─ 隐私信息过滤
           ├─ 版权内容过滤
           └─ 质量评分过滤
        
        4. 数据格式化
           ├─ 分词
           ├─ 构建词表
           ├─ 序列化
           └─ 分片存储
        
        5. 数据增强（可选）
           ├─ 回译
           ├─ 同义词替换
           └─ 句子重组
        """
        
        print(pipeline)

pretraining_data = PretrainingData()
pretraining_data.show_data_sources()
pretraining_data.data_processing_pipeline()

预训练实现示例

import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader

class PretrainingDataset(Dataset):
    """预训练数据集"""
    
    def __init__(self, texts, tokenizer, max_length=512):
        self.texts = texts
        self.tokenizer = tokenizer
        self.max_length = max_length
    
    def __len__(self):
        return len(self.texts)
    
    def __getitem__(self, idx):
        text = self.texts[idx]
        
        # 分词
        tokens = self.tokenizer.encode(text, max_length=self.max_length)
        
        # 创建输入和标签
        input_ids = tokens[:-1]  # 输入：除了最后一个词
        labels = tokens[1:]      # 标签：除了第一个词
        
        return {
            'input_ids': torch.tensor(input_ids),
            'labels': torch.tensor(labels)
        }

class SimplePretraining:
    """
    简化的预训练流程
    """
    
    def __init__(self, model, train_data, config):
        self.model = model
        self.train_data = train_data
        self.config = config
        
        self.optimizer = torch.optim.AdamW(
            model.parameters(),
            lr=config['learning_rate'],
            weight_decay=config['weight_decay']
        )
        
        self.scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
            self.optimizer,
            T_max=config['num_steps']
        )
    
    def train_step(self, batch):
        """单步训练"""
        self.model.train()
        
        # 前向传播
        outputs = self.model(
            input_ids=batch['input_ids'],
            labels=batch['labels']
        )
        
        loss = outputs.loss
        
        # 反向传播
        loss.backward()
        
        # 梯度裁剪
        torch.nn.utils.clip_grad_norm_(
            self.model.parameters(),
            self.config['max_grad_norm']
        )
        
        # 更新参数
        self.optimizer.step()
        self.scheduler.step()
        self.optimizer.zero_grad()
        
        return loss.item()
    
    def train(self):
        """训练循环"""
        print("\n预训练开始")
        print("=" * 60)
        
        dataloader = DataLoader(
            self.train_data,
            batch_size=self.config['batch_size'],
            shuffle=True
        )
        
        global_step = 0
        for epoch in range(self.config['num_epochs']):
            epoch_loss = 0
            
            for batch in dataloader:
                loss = self.train_step(batch)
                epoch_loss += loss
                global_step += 1
                
                # 日志
                if global_step % self.config['log_interval'] == 0:
                    print(f"Step {global_step}, Loss: {loss:.4f}, "
                          f"LR: {self.scheduler.get_last_lr()[0]:.2e}")
                
                # 保存检查点
                if global_step % self.config['save_interval'] == 0:
                    self.save_checkpoint(global_step)
            
            avg_loss = epoch_loss / len(dataloader)
            print(f"Epoch {epoch+1}, Avg Loss: {avg_loss:.4f}")
    
    def save_checkpoint(self, step):
        """保存检查点"""
        checkpoint = {
            'step': step,
            'model_state_dict': self.model.state_dict(),
            'optimizer_state_dict': self.optimizer.state_dict(),
            'scheduler_state_dict': self.scheduler.state_dict()
        }
        torch.save(checkpoint, f'checkpoint_step_{step}.pt')
        print(f"  保存检查点: step {step}")

# 配置示例
config = {
    'learning_rate': 1e-4,
    'weight_decay': 0.01,
    'batch_size': 32,
    'num_epochs': 3,
    'num_steps': 100000,
    'max_grad_norm': 1.0,
    'log_interval': 100,
    'save_interval': 10000
}

print("\n预训练配置:")
for key, value in config.items():
    print(f"  {key}: {value}")

微调阶段详解

微调策略

1. 全参数微调（Full Fine-tuning）

class FullFineTuning:
    """
    全参数微调
    
    更新模型的所有参数
    """
    
    def __init__(self, pretrained_model, num_labels):
        self.model = pretrained_model
        
        # 添加任务特定的分类头
        self.classifier = nn.Linear(
            pretrained_model.config.hidden_size,
            num_labels
        )
    
    def forward(self, input_ids, labels=None):
        """前向传播"""
        # 获取预训练模型的输出
        outputs = self.model(input_ids)
        hidden_states = outputs.last_hidden_state
        
        # 使用[CLS] token的表示
        cls_output = hidden_states[:, 0, :]
        
        # 分类
        logits = self.classifier(cls_output)
        
        loss = None
        if labels is not None:
            loss_fct = nn.CrossEntropyLoss()
            loss = loss_fct(logits, labels)
        
        return {'loss': loss, 'logits': logits}
    
    def fine_tune(self, train_data, config):
        """微调训练"""
        print("\n全参数微调")
        print("=" * 60)
        
        optimizer = torch.optim.AdamW(
            list(self.model.parameters()) + list(self.classifier.parameters()),
            lr=config['learning_rate']
        )
        
        print(f"可训练参数: {sum(p.numel() for p in self.model.parameters())}")
        print(f"学习率: {config['learning_rate']}")
        
        # 训练循环...
        print("开始微调训练...")

# 示例
print("\n全参数微调示例:")
config = {
    'learning_rate': 2e-5,  # 通常比预训练小
    'batch_size': 16,
    'num_epochs': 3
}
print(f"配置: {config}")

2. 参数高效微调（Parameter-Efficient Fine-tuning）

class LoRAFineTuning:
    """
    LoRA (Low-Rank Adaptation) 微调
    
    只训练低秩分解矩阵，大幅减少可训练参数
    """
    
    def __init__(self, pretrained_model, rank=8, alpha=16):
        self.model = pretrained_model
        self.rank = rank
        self.alpha = alpha
        
        # 冻结预训练模型参数
        for param in self.model.parameters():
            param.requires_grad = False
        
        # 为每个注意力层添加LoRA
        self.lora_layers = self.add_lora_layers()
    
    def add_lora_layers(self):
        """添加LoRA层"""
        lora_layers = []
        
        # 假设模型有12层
        for layer_idx in range(12):
            # 为Q和V矩阵添加LoRA
            # W = W0 + BA, 其中B和A是低秩矩阵
            lora_a = nn.Linear(768, self.rank, bias=False)
            lora_b = nn.Linear(self.rank, 768, bias=False)
            
            # 初始化
            nn.init.kaiming_uniform_(lora_a.weight)
            nn.init.zeros_(lora_b.weight)
            
            lora_layers.append({'A': lora_a, 'B': lora_b})
        
        return lora_layers
    
    def count_parameters(self):
        """统计参数量"""
        total_params = sum(p.numel() for p in self.model.parameters())
        trainable_params = sum(
            p.numel() for layer in self.lora_layers 
            for p in [layer['A'].weight, layer['B'].weight]
        )
        
        print(f"\n参数统计:")
        print(f"  总参数: {total_params:,}")
        print(f"  可训练参数: {trainable_params:,}")
        print(f"  可训练比例: {trainable_params/total_params*100:.2f}%")

# 示例
print("\n\nLoRA微调示例:")
# lora = LoRAFineTuning(model, rank=8)
# lora.count_parameters()

# 模拟输出
print("\n参数统计:")
print(f"  总参数: 110,000,000")
print(f"  可训练参数: 294,912")
print(f"  可训练比例: 0.27%")

3. 提示微调（Prompt Tuning）

class PromptTuning:
    """
    提示微调
    
    只训练连续的提示向量，模型参数完全冻结
    """
    
    def __init__(self, pretrained_model, num_prompt_tokens=20):
        self.model = pretrained_model
        self.num_prompt_tokens = num_prompt_tokens
        
        # 冻结模型参数
        for param in self.model.parameters():
            param.requires_grad = False
        
        # 初始化可学习的提示
        self.prompt_embeddings = nn.Parameter(
            torch.randn(num_prompt_tokens, pretrained_model.config.hidden_size)
        )
    
    def forward(self, input_ids):
        """前向传播"""
        batch_size = input_ids.shape[0]
        
        # 获取输入嵌入
        input_embeds = self.model.get_input_embeddings()(input_ids)
        
        # 扩展提示到batch
        prompt_embeds = self.prompt_embeddings.unsqueeze(0).expand(
            batch_size, -1, -1
        )
        
        # 拼接提示和输入
        # [prompt tokens] + [input tokens]
        combined_embeds = torch.cat([prompt_embeds, input_embeds], dim=1)
        
        # 通过模型
        outputs = self.model(inputs_embeds=combined_embeds)
        
        return outputs
    
    def show_info(self):
        """展示信息"""
        print("\n\n提示微调")
        print("=" * 60)
        print(f"提示token数量: {self.num_prompt_tokens}")
        print(f"可训练参数: {self.prompt_embeddings.numel():,}")
        print(f"模型参数: 冻结")
        
        print("\n工作原理:")
        print("  输入: [P1, P2, ..., Pn, X1, X2, ..., Xm]")
        print("  其中 P1-Pn 是可学习的提示向量")
        print("  X1-Xm 是实际输入")

pt = PromptTuning(None, num_prompt_tokens=20)
# 模拟展示
print("\n\n提示微调")
print("=" * 60)
print(f"提示token数量: 20")
print(f"可训练参数: 15,360")
print(f"模型参数: 冻结")

微调最佳实践

class FineTuningBestPractices:
    """微调最佳实践"""
    
    def __init__(self):
        self.practices = {
            '学习率': {
                'recommendation': '使用较小的学习率（1e-5 到 5e-5）',
                'reason': '预训练模型已经学到了好的表示，不需要大幅更新',
                'tips': ['使用warmup', '使用学习率衰减', '不同层使用不同学习率']
            },
            '训练轮数': {
                'recommendation': '通常2-5个epoch足够',
                'reason': '过多epoch容易过拟合',
                'tips': ['使用early stopping', '监控验证集性能', '保存最佳模型']
            },
            '批次大小': {
                'recommendation': '16-32（取决于GPU内存）',
                'reason': '平衡训练速度和稳定性',
                'tips': ['使用梯度累积', '混合精度训练', '动态批次大小']
            },
            '正则化': {
                'recommendation': '使用dropout和weight decay',
                'reason': '防止过拟合',
                'tips': ['dropout=0.1', 'weight_decay=0.01', '数据增强']
            },
            '层冻结': {
                'recommendation': '可以冻结底层，只微调顶层',
                'reason': '底层学到的是通用特征',
                'tips': ['逐层解冻', '不同层不同学习率', '观察性能变化']
            }
        }
    
    def show_practices(self):
        """展示最佳实践"""
        print("\n\n微调最佳实践")
        print("=" * 80)
        
        for aspect, info in self.practices.items():
            print(f"\n{aspect}")
            print(f"  建议: {info['recommendation']}")
            print(f"  原因: {info['reason']}")
            print(f"  技巧:")
            for tip in info['tips']:
                print(f"    • {tip}")

practices = FineTuningBestPractices()
practices.show_practices()

微调实现示例

class FineTuningTrainer:
    """
    完整的微调训练器
    """
    
    def __init__(self, model, train_dataset, eval_dataset, config):
        self.model = model
        self.train_dataset = train_dataset
        self.eval_dataset = eval_dataset
        self.config = config
        
        # 优化器
        self.optimizer = self.create_optimizer()
        
        # 学习率调度器
        self.scheduler = self.create_scheduler()
        
        # 最佳模型追踪
        self.best_eval_loss = float('inf')
        self.patience_counter = 0
    
    def create_optimizer(self):
        """创建优化器"""
        # 分层学习率
        no_decay = ['bias', 'LayerNorm.weight']
        optimizer_grouped_parameters = [
            {
                'params': [p for n, p in self.model.named_parameters() 
                          if not any(nd in n for nd in no_decay)],
                'weight_decay': self.config['weight_decay']
            },
            {
                'params': [p for n, p in self.model.named_parameters() 
                          if any(nd in n for nd in no_decay)],
                'weight_decay': 0.0
            }
        ]
        
        return torch.optim.AdamW(
            optimizer_grouped_parameters,
            lr=self.config['learning_rate']
        )
    
    def create_scheduler(self):
        """创建学习率调度器"""
        num_training_steps = len(self.train_dataset) * self.config['num_epochs']
        num_warmup_steps = int(num_training_steps * self.config['warmup_ratio'])
        
        return torch.optim.lr_scheduler.LinearLR(
            self.optimizer,
            start_factor=0.1,
            total_iters=num_warmup_steps
        )
    
    def train_epoch(self, epoch):
        """训练一个epoch"""
        self.model.train()
        total_loss = 0
        
        dataloader = DataLoader(
            self.train_dataset,
            batch_size=self.config['batch_size'],
            shuffle=True
        )
        
        for step, batch in enumerate(dataloader):
            # 前向传播
            outputs = self.model(**batch)
            loss = outputs['loss']
            
            # 反向传播
            loss.backward()
            
            # 梯度裁剪
            torch.nn.utils.clip_grad_norm_(
                self.model.parameters(),
                self.config['max_grad_norm']
            )
            
            # 更新参数
            self.optimizer.step()
            self.scheduler.step()
            self.optimizer.zero_grad()
            
            total_loss += loss.item()
            
            # 日志
            if step % self.config['logging_steps'] == 0:
                print(f"Epoch {epoch}, Step {step}, Loss: {loss.item():.4f}")
        
        return total_loss / len(dataloader)
    
    def evaluate(self):
        """评估模型"""
        self.model.eval()
        total_loss = 0
        
        dataloader = DataLoader(
            self.eval_dataset,
            batch_size=self.config['batch_size']
        )
        
        with torch.no_grad():
            for batch in dataloader:
                outputs = self.model(**batch)
                total_loss += outputs['loss'].item()
        
        return total_loss / len(dataloader)
    
    def train(self):
        """完整训练流程"""
        print("\n开始微调训练")
        print("=" * 60)
        
        for epoch in range(self.config['num_epochs']):
            # 训练
            train_loss = self.train_epoch(epoch)
            print(f"\nEpoch {epoch+1} 训练完成")
            print(f"  训练损失: {train_loss:.4f}")
            
            # 评估
            eval_loss = self.evaluate()
            print(f"  验证损失: {eval_loss:.4f}")
            
            # Early stopping
            if eval_loss < self.best_eval_loss:
                self.best_eval_loss = eval_loss
                self.patience_counter = 0
                self.save_model('best_model.pt')
                print("  保存最佳模型")
            else:
                self.patience_counter += 1
                if self.patience_counter >= self.config['patience']:
                    print(f"\nEarly stopping at epoch {epoch+1}")
                    break
        
        print("\n微调完成！")
    
    def save_model(self, path):
        """保存模型"""
        torch.save({
            'model_state_dict': self.model.state_dict(),
            'optimizer_state_dict': self.optimizer.state_dict(),
            'best_eval_loss': self.best_eval_loss
        }, path)

# 配置示例
fine_tuning_config = {
    'learning_rate': 2e-5,
    'weight_decay': 0.01,
    'batch_size': 16,
    'num_epochs': 3,
    'warmup_ratio': 0.1,
    'max_grad_norm': 1.0,
    'logging_steps': 100,
    'patience': 3
}

print("\n微调配置:")
for key, value in fine_tuning_config.items():
    print(f"  {key}: {value}")

预训练与微调的对比

class PretrainingVsFineTuning:
    """预训练与微调对比分析"""
    
    def __init__(self):
        self.comparison = {
            '数据规模': {
                '预训练': 'TB级别（数十亿到数万亿tokens）',
                '微调': 'MB到GB级别（数千到数百万样本）'
            },
            '数据类型': {
                '预训练': '无标注文本',
                '微调': '任务特定标注数据'
            },
            '训练时间': {
                '预训练': '数周到数月',
                '微调': '数小时到数天'
            },
            '计算资源': {
                '预训练': '数百到数千GPU',
                '微调': '1到数十GPU'
            },
            '成本': {
                '预训练': '数百万到数千万美元',
                '微调': '数百到数千美元'
            },
            '学习率': {
                '预训练': '较大（1e-4到6e-4）',
                '微调': '较小（1e-5到5e-5）'
            },
            '目标': {
                '预训练': '学习通用语言表示',
                '微调': '适应特定任务'
            }
        }
    
    def show_comparison(self):
        """展示对比"""
        print("\n\n预训练 vs 微调 详细对比")
        print("=" * 80)
        
        for aspect, values in self.comparison.items():
            print(f"\n{aspect}:")
            print(f"  预训练: {values['预训练']}")
            print(f"  微调:   {values['微调']}")

comparison = PretrainingVsFineTuning()
comparison.show_comparison()

新范式：提示学习与指令微调

提示学习（Prompt Learning）

class PromptLearning:
    """
    提示学习
    
    通过设计合适的提示，让模型完成任务，无需微调
    """
    
    def __init__(self):
        self.prompt_types = {
            'Hard Prompt': '人工设计的离散文本提示',
            'Soft Prompt': '可学习的连续向量提示',
            'Instruction': '自然语言指令',
            'Demonstration': '包含示例的提示'
        }
    
    def demonstrate_prompts(self):
        """演示不同类型的提示"""
        print("\n\n提示学习示例")
        print("=" * 80)
        
        examples = [
            {
                'task': '情感分类',
                'hard_prompt': 'Review: [TEXT]\nSentiment: ',
                'instruction': 'Classify the sentiment of the following review as positive or negative.',
                'demonstration': 'Review: Great movie!\nSentiment: Positive\n\nReview: [TEXT]\nSentiment: '
            },
            {
                'task': '问答',
                'hard_prompt': 'Q: [QUESTION]\nA: ',
                'instruction': 'Answer the following question based on your knowledge.',
                'demonstration': 'Q: What is 2+2?\nA: 4\n\nQ: [QUESTION]\nA: '
            }
        ]
        
        for ex in examples:
            print(f"\n任务: {ex['task']}")
            print(f"  Hard Prompt: {ex['hard_prompt']}")
            print(f"  Instruction: {ex['instruction']}")
            print(f"  Demonstration: {ex['demonstration']}")

prompt_learning = PromptLearning()
prompt_learning.demonstrate_prompts()

指令微调（Instruction Tuning）

class InstructionTuning:
    """
    指令微调
    
    在多样化的指令数据上微调，提升模型的指令遵循能力
    """
    
    def __init__(self):
        self.instruction_format = {
            'instruction': '任务描述',
            'input': '输入内容（可选）',
            'output': '期望输出'
        }
    
    def show_examples(self):
        """展示指令微调示例"""
        print("\n\n指令微调示例")
        print("=" * 80)
        
        examples = [
            {
                'instruction': '将以下句子翻译成法语',
                'input': 'Hello, how are you?',
                'output': 'Bonjour, comment allez-vous?'
            },
            {
                'instruction': '总结以下文章的主要观点',
                'input': '[长文本...]',
                'output': '[摘要...]'
            },
            {
                'instruction': '解释什么是机器学习',
                'input': '',
                'output': '机器学习是人工智能的一个分支...'
            }
        ]
        
        for i, ex in enumerate(examples, 1):
            print(f"\n示例 {i}:")
            print(f"  指令: {ex['instruction']}")
            if ex['input']:
                print(f"  输入: {ex['input']}")
            print(f"  输出: {ex['output']}")
    
    def show_benefits(self):
        """展示指令微调的优势"""
        print("\n\n指令微调的优势")
        print("=" * 60)
        
        benefits = [
            '提升零样本泛化能力',
            '更好的指令理解',
            '减少对提示工程的依赖',
            '提高多任务性能',
            '更自然的人机交互'
        ]
        
        for benefit in benefits:
            print(f"  • {benefit}")

instruction_tuning = InstructionTuning()
instruction_tuning.show_examples()
instruction_tuning.show_benefits()

总结

预训练与微调范式是现代大语言模型的基石：

1. 预训练阶段：在大规模无标注数据上学习通用语言表示

   • 自回归语言模型（GPT）
   • 掩码语言模型（BERT）
   • 其他预训练目标

2. 微调阶段：在特定任务上优化模型

   • 全参数微调
   • 参数高效微调（LoRA、Adapter等）
   • 提示微调

3. 新范式：

   • 提示学习：无需微调，通过提示完成任务
   • 指令微调：提升指令遵循能力
   • In-Context Learning：从示例中学习

4. 最佳实践：

   • 使用较小的学习率
   • 适当的训练轮数
   • 正则化防止过拟合
   • 监控验证集性能

理解预训练与微调范式，对于有效使用和开发大语言模型至关重要。随着技术的发展，参数高效微调和提示学习等新方法正在降低模型应用的门槛，使得更多人能够利用大模型的能力。

在下一篇文章中，我们将探讨模型规模与涌现能力的关系，揭示大模型为何如此强大。