大语言模型预训练：数据准备与训练策略完全指南

# 预训练数据收集 Pipeline 示例
import requests
from typing import List, Dict
import hashlib

class DataCollector:
    """预训练数据收集器"""
    
    def __init__(self, output_dir: str, target_size_gb: float = 100):
        self.output_dir = output_dir
        self.target_bytes = int(target_size_gb * 1e9)
        self.collected_bytes = 0
        self.seen_hashes = set()
    
    def fetch_common_crawl(self, snapshot: str) -> List[str]:
        """从 Common Crawl 获取指定快照的 URL 列表"""
        index_url = f"https://index.commoncrawl.org/{snapshot}-index"
        # 实际实现需要解析 CDX 索引
        return []  # 简化示例
    
    def download_and_dedup(self, url: str) -> bool:
        """下载网页并去重"""
        try:
            resp = requests.get(url, timeout=10)
            content = resp.text
            
            # 计算内容哈希（用于去重）
            content_hash = hashlib.sha256(
                content.encode()
            ).hexdigest()
            
            if content_hash in self.seen_hashes:
                return False  # 重复内容，跳过
            
            self.seen_hashes.add(content_hash)
            
            # 写入文件
            filename = f"{content_hash[:12]}.txt"
            with open(f"{self.output_dir}/{filename}", "w") as f:
                f.write(content)
            
            self.collected_bytes += len(content.encode())
            return True
        except Exception:
            return False
    
    def collect_stats(self) -> Dict:
        """收集统计信息"""
        return {
            "total_bytes": self.collected_bytes,
            "total_gb": self.collected_bytes / 1e9,
            "progress": self.collected_bytes / self.target_bytes,
            "unique_pages": len(self.seen_hashes),
        }

# 使用示例
# collector = DataCollector("./raw_data", target_size_gb=500)
# urls = collector.fetch_common_crawl("CC-MAIN-2024-10")
# for url in urls[:100000]:
#     collector.download_and_dedup(url)
# print(collector.collect_stats())

# 预训练数据质量过滤 Pipeline
import re
from typing import Optional

class TextFilter:
    """文本质量过滤器"""
    
    def __init__(self, min_length: int = 200,
                 max_special_char_ratio: float = 0.1,
                 max_repetition_ratio: float = 0.3,
                 perplexity_threshold: float = 1000):
        self.min_length = min_length
        self.max_special_char_ratio = max_special_char_ratio
        self.max_repetition_ratio = max_repetition_ratio
        self.perplexity_threshold = perplexity_threshold
    
    def is_valid_length(self, text: str) -> bool:
        """检查文本长度是否合理"""
        return len(text.strip()) >= self.min_length
    
    def has_html_tags(self, text: str) -> bool:
        """检查是否残留 HTML 标签"""
        html_pattern = r'<[^>]+>'
        return bool(re.search(html_pattern, text))
    
    def special_char_ratio(self, text: str) -> float:
        """计算特殊字符占比"""
        if not text:
            return 0
        special_chars = sum(1 for c in text 
                          if not c.isalnum() and not c.isspace()
                          and c not in '，。！？、；：""''（）【】《》')
        return special_chars / len(text)
    
    def repetition_ratio(self, text: str) -> float:
        """计算重复内容占比（基于 n-gram）"""
        words = text.split()
        if len(words) < 10:
            return 0
        # 检查连续重复的 trigram
        trigrams = [tuple(words[i:i+3]) for i in range(len(words)-2)]
        if not trigrams:
            return 0
        unique_trigrams = set(trigrams)
        return 1 - len(unique_trigrams) / len(trigrams)
    
    def filter(self, text: str) -> Optional[str]:
        """综合过滤，返回清洗后的文本或 None"""
        if not self.is_valid_length(text):
            return None
        if self.has_html_tags(text):
            return None
        if self.special_char_ratio(text) > self.max_special_char_ratio:
            return None
        if self.repetition_ratio(text) > self.max_repetition_ratio:
            return None
        # perplexity 检查需要加载语言模型，此处省略
        return text.strip()

# 使用示例
# filter = TextFilter(min_length=200)
# with open("raw_data.txt") as f:
#     for line in f:
#         cleaned = filter.filter(line)
#         if cleaned:
#             # 写入清洗后的数据
#             pass

# 训练 BPE 分词器示例
from tokenizers import Tokenizer
from tokenizers.models import BPE
from tokenizers.trainers import BpeTrainer
from tokenizers.pre_tokenizers import Whitespace
from tokenizers.processors import TemplateProcessing

# 创建 BPE 分词器
tokenizer = Tokenizer(BPE(unk_token="[UNK]"))

# 设置预分词器
tokenizer.pre_tokenizer = Whitespace()

# 配置训练器
trainer = BpeTrainer(
    vocab_size=50000,           # 词汇表大小
    min_frequency=2,            # 最小出现频率
    special_tokens=[
        "[PAD]", "[UNK]", "[CLS]", 
        "[SEP]", "[MASK]"
    ],
    show_progress=True
)

# 训练分词器
files = ["corpus_part_1.txt", "corpus_part_2.txt"]
tokenizer.train(files, trainer)

# 保存分词器
tokenizer.save("tokenizer.json")
tokenizer.model.save("tokenizer", "bpe")

# 使用分词器
from tokenizers import Tokenizer
tokenizer = Tokenizer.from_file("tokenizer.json")

text = "预训练是大语言模型的核心环节"
encoded = tokenizer.encode(text)
print(f"文本: {text}")
print(f"Token 数: {len(encoded.ids)}")
print(f"Token IDs: {encoded.ids}")
print(f"Tokens: {encoded.tokens}")
# 输出: Tokens: ['预', '训', '练', '是', '大', '语', '言', '模', '型', '的', '核', '心', '环', '节']

# 预训练学习率调度与优化器配置
import torch
from torch.optim import AdamW
from transformers import get_cosine_schedule_with_warmup

# 模型参数
model = ...  # 你的模型
total_steps = 100000           # 总训练步数
warmup_steps = 2000            # Warmup 步数 (2%)
peak_lr = 3e-4                 # 峰值学习率
weight_decay = 0.1             # 权重衰减

# 优化器：AdamW
optimizer = AdamW(
    model.parameters(),
    lr=peak_lr,
    betas=(0.9, 0.999),
    eps=1e-8,
    weight_decay=weight_decay
)

# 学习率调度：Warmup + Cosine Decay
scheduler = get_cosine_schedule_with_warmup(
    optimizer,
    num_warmup_steps=warmup_steps,
    num_training_steps=total_steps
)

# 训练循环
for step in range(total_steps):
    # 前向传播
    outputs = model(batch)
    loss = outputs.loss
    
    # 反向传播
    loss.backward()
    
    # 梯度裁剪（防止梯度爆炸）
    torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
    
    # 参数更新
    optimizer.step()
    scheduler.step()  # 更新学习率
    optimizer.zero_grad()
    
    if step % 100 == 0:
        current_lr = scheduler.get_last_lr()[0]
        print(f"Step {step}: loss={loss:.4f}, lr={current_lr:.2e}")

# 预训练监控与评估框架
import json
import time
from typing import Dict, Optional
from dataclasses import dataclass

@dataclass
class TrainingMetrics:
    step: int
    train_loss: float
    val_loss: Optional[float]
    learning_rate: float
    gradient_norm: float
    throughput_tok_s: float
    gpu_memory_gb: float

class TrainingMonitor:
    """训练监控器"""
    
    def __init__(self, log_file: str = "training.log"):
        self.log_file = log_file
        self.history = []
    
    def record(self, metrics: TrainingMetrics):
        """记录当前步的指标"""
        self.history.append(metrics)
        
        # 写入日志
        log_entry = {
            "step": metrics.step,
            "train_loss": metrics.train_loss,
            "val_loss": metrics.val_loss,
            "lr": metrics.learning_rate,
            "grad_norm": metrics.gradient_norm,
            "throughput": metrics.throughput_tok_s,
        }
        with open(self.log_file, "a") as f:
            f.write(json.dumps(log_entry) + "\n")
        
        # 检查异常
        self._check_anomalies(metrics)
    
    def _check_anomalies(self, metrics: TrainingMetrics):
        """检查训练异常"""
        if metrics.train_loss != metrics.train_loss:  # NaN 检查
            raise ValueError(f"Step {metrics.step}: Loss is NaN!")
        if metrics.gradient_norm > 100:
            print(f"WARNING: Step {metrics.step}: "
                  f"Gradient norm {metrics.gradient_norm:.1f} is very high!")
        if metrics.gradient_norm < 1e-6:
            print(f"WARNING: Step {metrics.step}: "
                  f"Gradient norm {metrics.gradient_norm:.2e} is near zero!")
    
    def get_summary(self) -> Dict:
        """获取训练摘要"""
        if not self.history:
            return {}
        latest = self.history[-1]
        return {
            "latest_step": latest.step,
            "latest_loss": latest.train_loss,
            "avg_throughput": sum(
                m.throughput_tok_s for m in self.history[-100:]
            ) / min(100, len(self.history)),
            "total_steps_trained": len(self.history),
        }

# 使用示例
# monitor = TrainingMonitor()
# for step, batch in enumerate(dataloader):
#     loss = model(batch).loss
#     loss.backward()
#     optimizer.step()
#     
#     monitor.record(TrainingMetrics(
#         step=step,
#         train_loss=loss.item(),
#         val_loss=evaluate_every_1000_steps(),
#         learning_rate=scheduler.get_last_lr()[0],
#         gradient_norm=get_grad_norm(model),
#         throughput_tok_s=compute_throughput(),
#         gpu_memory_gb=get_gpu_memory(),
#     ))