LLM训练技术详解

1
import re
2
from bs4 import BeautifulSoup
3
from langdetect import detect
4

5
class DataCleaner:
6
    def __init__(self):
7
        self.min_length = 100  # 最小字符数
8
        self.max_length = 100000  # 最大字符数
9

10
    def clean(self, text):
11
        """数据清洗管道"""
12
        # 1. 移除 HTML 标签
13
        text = self._remove_html(text)
14

15
        # 2. 语言检测（保留目标语言）
16
        if not self._is_valid_language(text):
17
            return None
18

19
        # 3. 去重（URL级别 + 内容级别）
20
        if self._is_duplicate(text):
21
            return None
22

23
        # 4. 质量过滤
24
        if not self._quality_filter(text):
25
            return None
26

27
        # 5. 移除个人信息（PII）
28
        text = self._remove_pii(text)
29

30
        # 6. 标准化
31
        text = self._normalize(text)
32

33
        return text
34

35
    def _remove_html(self, text):
36
        """移除 HTML 标签"""
37
        soup = BeautifulSoup(text, 'html.parser')
38
        return soup.get_text()
39

40
    def _is_valid_language(self, text, target_lang='en'):
41
        """语言检测"""
42
        try:
43
            lang = detect(text[:1000])  # 只检测前 1000 字符
44
            return lang == target_lang
45
        except:
46
            return False
47

48
    def _quality_filter(self, text):
49
        """质量过滤"""
50
        # 长度检查
51
        if len(text) < self.min_length or len(text) > self.max_length:
52
            return False
53

54
        # 字母/数字比例（过滤乱码）
55
        alnum_ratio = sum(c.isalnum() for c in text) / len(text)
56
        if alnum_ratio < 0.5:
57
            return False
58

59
        # 重复行检查（过滤模板内容）
60
        lines = text.split('\n')
61
        unique_lines = len(set(lines))
62
        if unique_lines / len(lines) < 0.3:
63
            return False
64

65
        # 停用词密度（过滤低质量内容）
66
        stopwords = {'the', 'a', 'an', 'is', 'are', 'was', 'were'}
67
        words = text.lower().split()
68
        stopword_ratio = sum(w in stopwords for w in words) / len(words)
69
        if stopword_ratio < 0.05:  # 正常文本应有一定停用词
70
            return False
71

72
        return True
73

74
    def _remove_pii(self, text):
75
        """移除个人信息"""
76
        # 邮箱
77
        text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '[EMAIL]', text)
78

79
        # 电话号码
80
        text = re.sub(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', '[PHONE]', text)
81

82
        # IP 地址
83
        text = re.sub(r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b', '[IP]', text)
84

85
        return text
86

87
    def _normalize(self, text):
88
        """文本标准化"""
89
        # 统一空白字符
90
        text = re.sub(r'\s+', ' ', text)
91

92
        # 移除控制字符
93
        text = ''.join(char for char in text if ord(char) >= 32 or char == '\n')
94

95
        return text.strip()
96

97
# 使用
98
cleaner = DataCleaner()
99
cleaned_text = cleaner.clean(raw_text)

1
from datasketch import MinHash, MinHashLSH
2

3
class Deduplicator:
4
    def __init__(self, threshold=0.8):
5
        self.lsh = MinHashLSH(threshold=threshold, num_perm=128)
6
        self.seen = set()
7

8
    def is_duplicate(self, text, doc_id):
9
        """使用 MinHash LSH 检测近似重复"""
10
        # 创建 MinHash 签名
11
        minhash = MinHash(num_perm=128)
12
        for word in text.split():
13
            minhash.update(word.encode('utf8'))
14

15
        # 查询相似文档
16
        result = self.lsh.query(minhash)
17

18
        if result:
19
            return True  # 发现重复
20

21
        # 添加到索引
22
        self.lsh.insert(doc_id, minhash)
23
        return False
24

25
# 使用
26
dedup = Deduplicator(threshold=0.8)
27
for doc_id, text in enumerate(documents):
28
    if not dedup.is_duplicate(text, doc_id):
29
        # 保留非重复文档
30
        cleaned_docs.append(text)

1
from tokenizers import Tokenizer
2
from tokenizers.models import BPE
3
from tokenizers.trainers import BpeTrainer
4
from tokenizers.pre_tokenizers import Whitespace
5

6
# 训练 BPE tokenizer
7
tokenizer = Tokenizer(BPE(unk_token="<unk>"))
8
tokenizer.pre_tokenizer = Whitespace()
9

10
trainer = BpeTrainer(
11
    vocab_size=50000,
12
    special_tokens=["<pad>", "<s>", "</s>", "<unk>", "<mask>"]
13
)
14

15
# 在数据上训练
16
tokenizer.train(files=["train.txt"], trainer=trainer)
17

18
# 保存
19
tokenizer.save("tokenizer.json")
20

21
# 使用
22
encoded = tokenizer.encode("Hello, world!")
23
print(f"Tokens: {encoded.tokens}")
24
print(f"IDs: {encoded.ids}")

1
import sentencepiece as spm
2

3
# 训练
4
spm.SentencePieceTrainer.train(
5
    input='train.txt',
6
    model_prefix='spm_model',
7
    vocab_size=32000,
8
    character_coverage=0.9995,  # 覆盖 99.95% 字符
9
    model_type='bpe',
10
    pad_id=0,
11
    unk_id=1,
12
    bos_id=2,
13
    eos_id=3
14
)
15

16
# 加载
17
sp = spm.SentencePieceProcessor()
18
sp.load('spm_model.model')
19

20
# 编码
21
tokens = sp.encode_as_pieces("你好，世界！")
22
ids = sp.encode_as_ids("你好，世界！")

1
import torch
2
import torch.nn as nn
3
from transformers import GPT2Config, GPT2LMHeadModel
4
from torch.utils.data import DataLoader
5
from datasets import load_dataset
6

7
# 1. 配置模型
8
config = GPT2Config(
9
    vocab_size=50257,
10
    n_positions=2048,
11
    n_embd=768,
12
    n_layer=12,
13
    n_head=12
14
)
15

16
model = GPT2LMHeadModel(config)
17
model = model.cuda()
18

19
# 2. 准备数据
20
dataset = load_dataset("openwebtext", split="train")
21

22
def tokenize_function(examples):
23
    return tokenizer(
24
        examples["text"],
25
        truncation=True,
26
        max_length=2048,
27
        padding="max_length"
28
    )
29

30
tokenized_dataset = dataset.map(tokenize_function, batched=True)
31
dataloader = DataLoader(tokenized_dataset, batch_size=8, shuffle=True)
32

33
# 3. 优化器（使用 AdamW + 学习率调度）
34
optimizer = torch.optim.AdamW(
35
    model.parameters(),
36
    lr=6e-4,
37
    betas=(0.9, 0.95),
38
    weight_decay=0.1
39
)
40

41
from transformers import get_cosine_schedule_with_warmup
42
scheduler = get_cosine_schedule_with_warmup(
43
    optimizer,
44
    num_warmup_steps=2000,
45
    num_training_steps=100000
46
)
47

48
# 4. 训练循环
49
model.train()
50
for epoch in range(num_epochs):
51
    for batch in dataloader:
52
        input_ids = batch['input_ids'].cuda()
53
        attention_mask = batch['attention_mask'].cuda()
54

55
        # 前向传播
56
        outputs = model(
57
            input_ids=input_ids,
58
            attention_mask=attention_mask,
59
            labels=input_ids  # 自回归：输入即标签
60
        )
61

62
        loss = outputs.loss
63

64
        # 反向传播
65
        loss.backward()
66

67
        # 梯度裁剪（防止梯度爆炸）
68
        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
69

70
        optimizer.step()
71
        scheduler.step()
72
        optimizer.zero_grad()
73

74
        if step % 100 == 0:
75
            print(f"Step {step}, Loss: {loss.item():.4f}")

1
import deepspeed
2

3
# DeepSpeed 配置
4
ds_config = {
5
    "train_batch_size": 64,
6
    "gradient_accumulation_steps": 8,
7
    "fp16": {
8
        "enabled": True
9
    },
10
    "zero_optimization": {
11
        "stage": 2,  # ZeRO-2: 分片优化器状态
12
        "offload_optimizer": {
13
            "device": "cpu"
14
        }
15
    }
16
}
17

18
# 初始化 DeepSpeed
19
model_engine, optimizer, _, _ = deepspeed.initialize(
20
    model=model,
21
    model_parameters=model.parameters(),
22
    config=ds_config
23
)
24

25
# 训练
26
for batch in dataloader:
27
    loss = model_engine(batch)
28
    model_engine.backward(loss)
29
    model_engine.step()

1
def format_instruction(example):
2
    """将数据格式化为训练格式"""
3
    if example.get("input"):
4
        prompt = f"""Below is an instruction that describes a task, paired with an input that provides further context. Write a response that appropriately completes the request.
5

6
### Instruction:
7
{example['instruction']}
8

9
### Input:
10
{example['input']}
11

12
### Response:
13
{example['output']}"""
14
    else:
15
        prompt = f"""Below is an instruction that describes a task. Write a response that appropriately completes the request.
16

17
### Instruction:
18
{example['instruction']}
19

20
### Response:
21
{example['output']}"""
22

23
    return {"text": prompt}
24

25
# 应用到数据集
26
sft_dataset = dataset.map(format_instruction)

1
from transformers import Trainer, TrainingArguments
2

3
# 加载预训练模型
4
model = GPT2LMHeadModel.from_pretrained("gpt2-medium")
5

6
# 训练配置
7
training_args = TrainingArguments(
8
    output_dir="./sft_model",
9
    num_train_epochs=3,
10
    per_device_train_batch_size=4,
11
    gradient_accumulation_steps=4,
12
    learning_rate=2e-5,  # SFT 学习率比预训练小
13
    warmup_steps=100,
14
    logging_steps=10,
15
    save_steps=500,
16
    evaluation_strategy="steps",
17
    eval_steps=500,
18
    fp16=True
19
)
20

21
# 数据整理器（处理 padding）
22
from transformers import DataCollatorForLanguageModeling
23
data_collator = DataCollatorForLanguageModeling(
24
    tokenizer=tokenizer,
25
    mlm=False  # 因果语言模型，不使用 MLM
26
)
27

28
# 训练
29
trainer = Trainer(
30
    model=model,
31
    args=training_args,
32
    train_dataset=train_dataset,
33
    eval_dataset=eval_dataset,
34
    data_collator=data_collator
35
)
36

37
trainer.train()

1
def self_instruct(seed_instructions, model, num_generate=1000):
2
    """使用模型生成新的指令-回答对"""
3
    generated_data = []
4

5
    for _ in range(num_generate):
6
        # 随机选择种子指令作为示例
7
        examples = random.sample(seed_instructions, k=3)
8

9
        prompt = f"""Generate a new instruction-input-output example.
10

11
Examples:
12
{format_examples(examples)}
13

14
New example:
15
Instruction:"""
16

17
        # 生成新指令
18
        new_instruction = model.generate(prompt, max_tokens=100)
19

20
        # 生成对应的输出
21
        output_prompt = f"Instruction: {new_instruction}\nResponse:"
22
        new_output = model.generate(output_prompt, max_tokens=200)
23

24
        generated_data.append({
25
            "instruction": new_instruction,
26
            "output": new_output
27
        })
28

29
    return generated_data

1
evolve_prompts = [
2
    "Add more constraints to make the instruction more specific",
3
    "Deepen the instruction by adding reasoning requirements",
4
    "Complicate the instruction by introducing multiple steps",
5
    "Increase the breadth by asking about related concepts"
6
]
7

8
def evolve_instruction(instruction, model):
9
    """进化指令，增加难度"""
10
    evolve_type = random.choice(evolve_prompts)
11

12
    prompt = f"""Original instruction: {instruction}
13

14
{evolve_type}
15

16
Evolved instruction:"""
17

18
    evolved = model.generate(prompt, max_tokens=150)
19
    return evolved

1
import torch
2
import torch.nn as nn
3

4
class RewardModel(nn.Module):
5
    def __init__(self, base_model):
6
        super().__init__()
7
        self.base_model = base_model
8

9
        # 添加奖励头（输出标量分数）
10
        hidden_size = base_model.config.hidden_size
11
        self.reward_head = nn.Linear(hidden_size, 1)
12

13
    def forward(self, input_ids, attention_mask):
14
        # 获取最后一个 token 的隐藏状态
15
        outputs = self.base_model(
16
            input_ids=input_ids,
17
            attention_mask=attention_mask,
18
            output_hidden_states=True
19
        )
20

21
        last_hidden_state = outputs.hidden_states[-1]
22

23
        # 提取最后一个非 padding token 的状态
24
        sequence_lengths = attention_mask.sum(dim=1) - 1
25
        batch_size = input_ids.shape[0]
26

27
        last_token_hidden = last_hidden_state[
28
            torch.arange(batch_size, device=input_ids.device),
29
            sequence_lengths
30
        ]
31

32
        # 计算奖励分数
33
        reward = self.reward_head(last_token_hidden)
34
        return reward
35

36
# 训练奖励模型
37
def train_reward_model(model, dataloader, epochs=1):
38
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-5)
39

40
    model.train()
41
    for epoch in range(epochs):
42
        for batch in dataloader:
43
            # batch 包含 chosen 和 rejected 两个版本
44
            chosen_ids = batch['chosen_input_ids'].cuda()
45
            chosen_mask = batch['chosen_attention_mask'].cuda()
46
            rejected_ids = batch['rejected_input_ids'].cuda()
47
            rejected_mask = batch['rejected_attention_mask'].cuda()
48

49
            # 计算奖励
50
            reward_chosen = model(chosen_ids, chosen_mask)
51
            reward_rejected = model(rejected_ids, rejected_mask)
52

53
            # Bradley-Terry 模型损失
54
            loss = -torch.log(torch.sigmoid(reward_chosen - reward_rejected)).mean()
55

56
            # 反向传播
57
            optimizer.zero_grad()
58
            loss.backward()
59
            optimizer.step()
60

61
            if step % 100 == 0:
62
                print(f"Loss: {loss.item():.4f}")

1
from trl import PPOTrainer, PPOConfig, AutoModelForCausalLMWithValueHead
2
from transformers import AutoTokenizer
3

4
# 配置
5
ppo_config = PPOConfig(
6
    model_name="sft_model",
7
    learning_rate=1.41e-5,
8
    batch_size=64,
9
    mini_batch_size=8,
10
    gradient_accumulation_steps=8,
11
    ppo_epochs=4,
12
    max_grad_norm=0.5
13
)
14

15
# 加载模型
16
model = AutoModelForCausalLMWithValueHead.from_pretrained("sft_model")
17
ref_model = AutoModelForCausalLMWithValueHead.from_pretrained("sft_model")  # 参考模型
18
tokenizer = AutoTokenizer.from_pretrained("sft_model")
19

20
# 加载奖励模型
21
reward_model = RewardModel.from_pretrained("reward_model").cuda()
22
reward_model.eval()
23

24
# PPO 训练器
25
ppo_trainer = PPOTrainer(
26
    config=ppo_config,
27
    model=model,
28
    ref_model=ref_model,
29
    tokenizer=tokenizer
30
)
31

32
# 训练循环
33
for epoch in range(num_epochs):
34
    for batch in dataloader:
35
        prompts = batch['prompt']
36

37
        # 生成回答
38
        query_tensors = [tokenizer.encode(p, return_tensors="pt")[0] for p in prompts]
39
        response_tensors = ppo_trainer.generate(query_tensors, **generation_kwargs)
40

41
        # 计算奖励
42
        texts = [tokenizer.decode(r) for r in response_tensors]
43
        rewards = []
44
        for text in texts:
45
            input_ids = tokenizer.encode(text, return_tensors="pt").cuda()
46
            with torch.no_grad():
47
                reward = reward_model(input_ids, attention_mask=torch.ones_like(input_ids))
48
            rewards.append(reward.cpu())
49

50
        # PPO 更新
51
        stats = ppo_trainer.step(query_tensors, response_tensors, rewards)
52

53
        if step % 100 == 0:
54
            print(f"Mean reward: {stats['ppo/mean_scores']:.4f}")

1
from trl import DPOTrainer, DPOConfig
2

3
dpo_config = DPOConfig(
4
    beta=0.1,  # KL 惩罚系数
5
    learning_rate=5e-7,
6
    num_train_epochs=1,
7
    per_device_train_batch_size=4,
8
    gradient_accumulation_steps=4
9
)
10

11
dpo_trainer = DPOTrainer(
12
    model=model,
13
    ref_model=ref_model,
14
    args=dpo_config,
15
    train_dataset=preference_dataset,  # 包含 chosen 和 rejected
16
    tokenizer=tokenizer
17
)
18

19
dpo_trainer.train()

1
from torch.cuda.amp import autocast, GradScaler
2

3
scaler = GradScaler()
4

5
for batch in dataloader:
6
    optimizer.zero_grad()
7

8
    # 使用混合精度
9
    with autocast():
10
        outputs = model(batch)
11
        loss = outputs.loss
12

13
    # 缩放损失并反向传播
14
    scaler.scale(loss).backward()
15

16
    # 梯度裁剪
17
    scaler.unscale_(optimizer)
18
    torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
19

20
    # 更新参数
21
    scaler.step(optimizer)
22
    scaler.update()

1
{
2
  "zero_optimization": {
3
    "stage": 3,
4
    "offload_optimizer": {
5
      "device": "cpu",
6
      "pin_memory": true
7
    },
8
    "offload_param": {
9
      "device": "cpu",
10
      "pin_memory": true
11
    },
12
    "overlap_comm": true,
13
    "contiguous_gradients": true,
14
    "sub_group_size": 1e9,
15
    "reduce_bucket_size": 5e8,
16
    "stage3_prefetch_bucket_size": 5e8,
17
    "stage3_param_persistence_threshold": 1e6
18
  }
19
}