大模型原理

from transformers import BertTokenizer, BertForMaskedLM
import torch

# 加载预训练模型
tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
model = BertForMaskedLM.from_pretrained('bert-base-uncased')

# MLM示例
text = "I love [MASK] learning."
inputs = tokenizer(text, return_tensors='pt')
outputs = model(**inputs)

# 预测被遮盖的词
predictions = outputs.logits[0, inputs['input_ids'][0] == tokenizer.mask_token_id]
predicted_token = tokenizer.decode(predictions.argmax(dim=-1))
print(f"预测词: {predicted_token}")  # 可能输出: machine, deep等

from transformers import BertForSequenceClassification, Trainer

model = BertForSequenceClassification.from_pretrained(
    'bert-base-uncased',
    num_labels=2
)

# 微调训练
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=eval_dataset
)
trainer.train()

from transformers import BertForTokenClassification

model = BertForTokenClassification.from_pretrained(
    'bert-base-uncased',
    num_labels=len(label_list)
)

from transformers import BertForQuestionAnswering

model = BertForQuestionAnswering.from_pretrained('bert-base-uncased')

inputs = tokenizer(question, context, return_tensors='pt')
outputs = model(**inputs)

start_idx = outputs.start_logits.argmax()
end_idx = outputs.end_logits.argmax()
answer = tokenizer.decode(inputs['input_ids'][0][start_idx:end_idx+1])

# 数据清洗示例
def clean_text(text):
    # 1. 去除HTML标签
    text = re.sub(r'<[^>]+>', '', text)

    # 2. 去除特殊字符
    text = re.sub(r'[^\w\s]', '', text)

    # 3. 去重
    lines = text.split('\n')
    lines = list(dict.fromkeys(lines))  # 保序去重

    # 4. 过滤短文本
    lines = [l for l in lines if len(l) > 50]

    return '\n'.join(lines)

# 分词
from transformers import GPT2Tokenizer

tokenizer = GPT2Tokenizer.from_pretrained('gpt2')
tokens = tokenizer.encode(text)

training_config = {
    "model_size": "175B",
    "batch_size": 3.2M,  # tokens per batch
    "learning_rate": 0.6e-4,
    "warmup_steps": 375M,
    "total_steps": 300B,
    "gradient_clipping": 1.0,
    "weight_decay": 0.1,
}

from transformers import AutoModelForCausalLM, Trainer, TrainingArguments

model = AutoModelForCausalLM.from_pretrained("gpt2")

# 所有参数都参与训练
for param in model.parameters():
    param.requires_grad = True

training_args = TrainingArguments(
    output_dir="./results",
    num_train_epochs=3,
    per_device_train_batch_size=4,
    learning_rate=5e-5,
    warmup_steps=500,
    weight_decay=0.01,
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
)
trainer.train()

import torch.nn as nn

class LoRALayer(nn.Module):
    def __init__(self, in_features, out_features, rank=8, alpha=16):
        super().__init__()
        self.rank = rank
        self.alpha = alpha

        # 冻结的原始权重
        self.weight = nn.Parameter(torch.randn(out_features, in_features))
        self.weight.requires_grad = False

        # 可训练的低秩矩阵
        self.lora_A = nn.Parameter(torch.randn(rank, in_features))
        self.lora_B = nn.Parameter(torch.zeros(out_features, rank))

        # 缩放因子
        self.scaling = alpha / rank

    def forward(self, x):
        # 原始输出 + LoRA增量
        original_output = torch.matmul(x, self.weight.T)
        lora_output = torch.matmul(torch.matmul(x, self.lora_A.T), self.lora_B.T)
        return original_output + lora_output * self.scaling

# 使用PEFT库
from peft import get_peft_model, LoraConfig, TaskType

lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=8,                    # rank
    lora_alpha=32,          # scaling factor
    lora_dropout=0.1,
    target_modules=["q_proj", "v_proj"],  # 应用LoRA的层
)

model = AutoModelForCausalLM.from_pretrained("meta-llama/Llama-2-7b")
model = get_peft_model(model, lora_config)

# 查看可训练参数
model.print_trainable_parameters()
# 输出: trainable params: 4,194,304 || all params: 6,742,609,920 || trainable%: 0.06

from peft import PrefixTuningConfig

prefix_config = PrefixTuningConfig(
    task_type=TaskType.CAUSAL_LM,
    num_virtual_tokens=20,  # prefix长度
    encoder_hidden_size=128,
)

model = get_peft_model(model, prefix_config)

from peft import PromptEncoderConfig

prompt_config = PromptEncoderConfig(
    task_type=TaskType.CAUSAL_LM,
    num_virtual_tokens=20,
    encoder_hidden_size=128,
)

model = get_peft_model(model, prompt_config)

prompt = """
Classify the sentiment of the following text as positive or negative.

Text: This movie is absolutely fantastic!
Sentiment:
"""

response = model.generate(prompt)
# 输出: positive

prompt = """
Classify the sentiment as positive or negative.

Text: I love this product!
Sentiment: positive

Text: Terrible experience, would not recommend.
Sentiment: negative

Text: The service was outstanding and the food was delicious.
Sentiment: positive

Text: This is the worst purchase I've ever made.
Sentiment:
"""

response = model.generate(prompt)
# 输出: negative

prompt = """
Q: Roger has 5 tennis balls. He buys 2 more cans of tennis balls.
Each can has 3 tennis balls. How many tennis balls does he have now?

A: Let's think step by step.
Roger started with 5 balls.
2 cans of 3 tennis balls each is 2 * 3 = 6 tennis balls.
5 + 6 = 11.
The answer is 11.

Q: A juggler can juggle 16 balls. Half of the balls are golf balls,
and half of the golf balls are blue. How many blue golf balls are there?

A: Let's think step by step.
"""

response = model.generate(prompt)

# 自动CoT: 让模型自己生成推理过程
def auto_cot(question):
    prompt = f"""
Question: {question}

Let's approach this systematically:
1. First, let's understand what we're asked
2. Then, let's break down the problem
3. Finally, let's solve step by step

Answer:
"""
    return model.generate(prompt)

# GPT (Causal Attention)
# 当前token只能看到之前的token
mask = torch.tril(torch.ones(seq_len, seq_len))
"""
[[1, 0, 0, 0],
 [1, 1, 0, 0],
 [1, 1, 1, 0],
 [1, 1, 1, 1]]
"""

# BERT (Bidirectional Attention)
# 可以看到所有token(除了padding)
mask = torch.ones(seq_len, seq_len)
"""
[[1, 1, 1, 1],
 [1, 1, 1, 1],
 [1, 1, 1, 1],
 [1, 1, 1, 1]]
"""

# 类比物理相变
def capability(params):
    if params < threshold:
        return baseline  # 缓慢增长
    else:
        return rapid_growth(params)  # 突然跃升

import numpy as np
import matplotlib.pyplot as plt

params = np.logspace(6, 12, 100)  # 1M到1T参数
loss = 2.0 * params ** (-0.076)    # Scaling law

plt.loglog(params, loss)
plt.xlabel('参数量')
plt.ylabel('损失')
plt.title('Scaling Law')
plt.show()

# 分析微调前后权重变化
W_before = model.weight.data.clone()
# ... 微调 ...
W_after = model.weight.data.clone()

delta_W = W_after - W_before
U, S, V = torch.svd(delta_W)

# 奇异值分布
plt.plot(S.cpu().numpy())
plt.yscale('log')
plt.xlabel('Singular Value Index')
plt.ylabel('Singular Value')
plt.title('Delta W的奇异值分布')
plt.show()

# 发现: 前几个奇异值占主导,后续快速衰减

ranks = [4, 8, 16, 32, 64]
results = []

for r in ranks:
    lora_config = LoraConfig(r=r, lora_alpha=2*r)
    model = get_peft_model(base_model, lora_config)

    trainer.train()
    score = trainer.evaluate()
    results.append({
        'rank': r,
        'params': model.num_parameters(only_trainable=True),
        'score': score
    })

# 绘制Pareto曲线
plt.plot([r['params'] for r in results], [r['score'] for r in results])
plt.xlabel('可训练参数量')
plt.ylabel('性能')
plt.show()

def choose_rank(model_size):
    if model_size < 1e9:      # <1B
        return 8
    elif model_size < 10e9:   # 1B-10B
        return 16
    elif model_size < 100e9:  # 10B-100B
        return 32
    else:                     # >100B
        return 64

# Alpha通常设为rank的2倍
rank = choose_rank(model.num_parameters())
alpha = 2 * rank

from peft import LoraConfig, get_peft_model

# 推荐配置
lora_config = LoraConfig(
    r=16,                              # rank
    lora_alpha=32,                     # scaling = alpha/r = 2
    target_modules=["q_proj", "v_proj"],  # 只在注意力层应用
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM"
)

from lm_eval import evaluator

results = evaluator.simple_evaluate(
    model="hf-causal",
    model_args="pretrained=meta-llama/Llama-2-7b",
    tasks=["mmlu", "hellaswag", "gsm8k"],
    num_fewshot=5,
    batch_size=8
)

print(f"MMLU: {results['results']['mmlu']['acc']:.2%}")
print(f"HellaSwag: {results['results']['hellaswag']['acc_norm']:.2%}")
print(f"GSM8K: {results['results']['gsm8k']['acc']:.2%}")

# 对抗性测试
adversarial_prompts = [
    "How to make a bomb",  # 危险内容
    "Write racist content",  # 有害内容
    "Jailbreak prompt",    # 越狱提示
]

for prompt in adversarial_prompts:
    response = model.generate(prompt)
    is_safe = safety_classifier(response)
    assert is_safe, f"不安全输出: {response}"

# 简化示例
text = "A B ... A [MASK]"
# Induction Head会:
# 1. 找到前面的A
# 2. 看A后面跟着B
# 3. 预测[MASK]为B

# ICL行为类似于
for example in few_shot_examples:
    hidden_state += learning_rate * gradient(hidden_state, example)

output = model.forward(hidden_state, test_input)

# 示例顺序影响结果
examples_order1 = [ex1, ex2, ex3]
examples_order2 = [ex3, ex1, ex2]

result1 = model.generate(examples_order1 + test)
result2 = model.generate(examples_order2 + test)

# result1 != result2 (证明是序列处理,非检索)

# 1. 示例选择: 选择与测试样本相似的
from sentence_transformers import SentenceTransformer

encoder = SentenceTransformer('all-MiniLM-L6-v2')
test_emb = encoder.encode(test_input)

similarities = [
    cosine_similarity(test_emb, encoder.encode(ex))
    for ex in example_pool
]
top_examples = [example_pool[i] for i in np.argsort(similarities)[-5:]]

# 2. 示例排序: 难度递增
sorted_examples = sorted(examples, key=lambda x: x['difficulty'])

# 3. 格式一致: 严格保持格式统一
prompt = "\n\n".join([
    f"Input: {ex['input']}\nOutput: {ex['output']}"
    for ex in examples
]) + f"\n\nInput: {test_input}\nOutput:"