十一：手撕代码与项目深挖
LoRA 线性层¶
import torch
import torch.nn as nn
import math
class LoRALinear(nn.Module):
def __init__(self, in_features: int, out_features: int, r: int = 8,
lora_alpha: int = 16, bias: bool = True, dropout: float = 0.0):
super().__init__()
self.r = r
self.lora_alpha = lora_alpha
self.scaling = lora_alpha / r
# 原始线性层，权重冻结
self.linear = nn.Linear(in_features, out_features, bias=bias)
self.linear.weight.requires_grad = False
if bias:
self.linear.bias.requires_grad = False
# LoRA 低秩矩阵 A (r, in_features) 和 B (out_features, r)
self.lora_A = nn.Parameter(torch.zeros(r, in_features))
self.lora_B = nn.Parameter(torch.zeros(out_features, r))
self.lora_dropout = nn.Dropout(dropout) if dropout > 0 else nn.Identity()
self.reset_lora_parameters()
def reset_lora_parameters(self):
# A 使用 Kaiming 初始化，B 初始化为零，使初始时 ΔW = 0
nn.init.kaiming_uniform_(self.lora_A, a=math.sqrt(5))
nn.init.zeros_(self.lora_B)
def forward(self, x: torch.Tensor) -> torch.Tensor:
# 原始输出 (冻结)
result = self.linear(x)
# LoRA 旁路：x -> A^T -> B^T，然后乘以缩放因子
lora_out = self.lora_dropout(x) @ self.lora_A.T # (..., r)
lora_out = lora_out @ self.lora_B.T # (..., out_features)
result = result + self.scaling * lora_out
return result
合并 LoRA 权重¶
def merge_lora_to_linear(lora_layer: LoRALinear) -> nn.Linear:
"""将 LoRA 权重合并到原始线性层，返回一个新的标准线性层"""
# 原始权重 (out_features, in_features)
W = lora_layer.linear.weight.data
A = lora_layer.lora_A.data # (r, in_features)
B = lora_layer.lora_B.data # (out_features, r)
# 计算增量 ΔW = (alpha / r) * B @ A
delta_W = lora_layer.scaling * torch.matmul(B, A) # (out_features, in_features)
merged_weight = W + delta_W
# 构建新的线性层
new_linear = nn.Linear(lora_layer.in_features, lora_layer.out_features,
bias=lora_layer.linear.bias is not None)
new_linear.weight.data = merged_weight
if lora_layer.linear.bias is not None:
new_linear.bias.data = lora_layer.linear.bias.data
return new_linear
SFT Data Collator（带 loss masking）¶
from typing import List, Dict
class SFTDataCollator:
"""
将对话样本转换为训练批次，并对 prompt 部分进行 loss masking。
要求每个样本包含 'input_ids' 和 'labels'，其中 prompt 部分 labels 已设为 -100。
在 batch 内进行右侧填充，填充 token 的 label 也设为 -100。
"""
def __init__(self, pad_token_id: int = 0):
self.pad_token_id = pad_token_id
def __call__(self, features: List[Dict[str, torch.Tensor]]) -> Dict[str, torch.Tensor]:
max_len = max(f['input_ids'].size(0) for f in features)
padded_input_ids, padded_labels, attention_masks = [], [], []
for f in features:
input_ids = f['input_ids']
labels = f['labels']
seq_len = input_ids.size(0)
pad_len = max_len - seq_len
# 右侧填充
padded_input = torch.cat([input_ids, torch.full((pad_len,), self.pad_token_id, dtype=input_ids.dtype)])
padded_label = torch.cat([labels, torch.full((pad_len,), -100, dtype=labels.dtype)])
attn_mask = torch.cat([torch.ones(seq_len, dtype=torch.long), torch.zeros(pad_len, dtype=torch.long)])
padded_input_ids.append(padded_input)
padded_labels.append(padded_label)
attention_masks.append(attn_mask)
return {
'input_ids': torch.stack(padded_input_ids),
'labels': torch.stack(padded_labels),
'attention_mask': torch.stack(attention_masks)
}
序列打包（packing）及注意力掩码生成¶
def pack_sequences(
input_ids_list: List[torch.Tensor],
labels_list: List[torch.Tensor],
eos_token_id: int = 2,
pad_token_id: int = 0
) -> Dict[str, torch.Tensor]:
"""
将多个样本拼接为一个长序列，生成分块对角因果注意力掩码 (4D)。
返回 input_ids (1, total_len), labels (1, total_len), attention_mask (1, 1, total_len, total_len)
"""
# 存储每个片段的起止位置
segments = []
all_ids = []
all_labels = []
offset = 0
for inp, lab in zip(input_ids_list, labels_list):
L = inp.size(0)
all_ids.append(inp)
all_labels.append(lab)
segments.append((offset, offset + L - 1))
offset += L
# 插入 EOS 分隔符，其 label 设为 -100
all_ids.append(torch.tensor([eos_token_id], dtype=inp.dtype))
all_labels.append(torch.tensor([-100], dtype=lab.dtype))
offset += 1
# 移除末尾多余的分隔符
if all_ids:
all_ids.pop()
all_labels.pop()
offset -= 1
input_ids = torch.cat(all_ids, dim=0)
labels = torch.cat(all_labels, dim=0)
total_len = input_ids.size(0)
# 构建分块对角因果掩码
# 先构造下三角因果掩码
causal_mask = torch.tril(torch.ones(total_len, total_len, dtype=torch.bool))
# 构造块对角掩码：允许同一样本内互相看到
block_mask = torch.zeros(total_len, total_len, dtype=torch.bool)
for start, end in segments:
block_mask[start:end+1, start:end+1] = True
# 最终掩码 = 因果掩码 AND 块对角掩码
final_mask = causal_mask & block_mask # (total_len, total_len)
# 扩展到4D: (1, 1, total_len, total_len)
final_mask = final_mask.unsqueeze(0).unsqueeze(0)
return {
'input_ids': input_ids.unsqueeze(0), # (1, total_len)
'labels': labels.unsqueeze(0),
'attention_mask': final_mask
}
DPO 损失函数¶
def dpo_loss(
policy_logits: torch.Tensor, # (batch, seq_len, vocab_size) 当前策略的 logits
ref_logits: torch.Tensor, # 参考模型的 logits
labels: torch.Tensor, # (batch, seq_len) 用于构建回答的 token ids
chosen_mask: torch.Tensor, # (batch,) bool 表示该样本是否为 chosen (True) 或 rejected (False)
beta: float = 0.1,
ignore_index: int = -100
) -> torch.Tensor:
"""
计算 DPO 损失。假设数据已组织为：前半 batch 是 chosen，后半是 rejected 与其配对。
chosen_mask 标记每个样本的归属。
"""
# 计算每个 token 的对数似然，忽略 ignore_index 的位置
loss_fct = torch.nn.CrossEntropyLoss(reduction='none', ignore_index=ignore_index)
# 形状 (batch, seq_len)
log_probs_policy = -loss_fct(policy_logits.permute(0, 2, 1), labels)
log_probs_ref = -loss_fct(ref_logits.permute(0, 2, 1), labels)
# 对序列求和（或平均）得到整个回答的对数似然
# 注意：应只考虑回答部分，即 labels != ignore_index 的位置
mask = (labels != ignore_index).float()
sum_log_policy = (log_probs_policy * mask).sum(dim=-1) # (batch,)
sum_log_ref = (log_probs_ref * mask).sum(dim=-1)
# 拆分 chosen 和 rejected
chosen_mask = chosen_mask.bool()
chosen_log_policy = sum_log_policy[chosen_mask]
rejected_log_policy = sum_log_policy[~chosen_mask]
chosen_log_ref = sum_log_ref[chosen_mask]
rejected_log_ref = sum_log_ref[~chosen_mask]
# 计算隐式奖励差
chosen_reward = beta * (chosen_log_policy - chosen_log_ref)
rejected_reward = beta * (rejected_log_policy - rejected_log_ref)
# DPO 损失 = -log σ(chosen_reward - rejected_reward)
loss = -torch.log(torch.sigmoid(chosen_reward - rejected_reward)).mean()
return loss
多头自注意力（含因果掩码）¶
class MultiHeadSelfAttention(nn.Module):
def __init__(self, d_model: int, n_heads: int, dropout: float = 0.1):
super().__init__()
assert d_model % n_heads == 0
self.d_model = d_model
self.n_heads = n_heads
self.d_k = d_model // n_heads
self.W_q = nn.Linear(d_model, d_model)
self.W_k = nn.Linear(d_model, d_model)
self.W_v = nn.Linear(d_model, d_model)
self.W_o = nn.Linear(d_model, d_model)
self.dropout = nn.Dropout(dropout)
def forward(self, x: torch.Tensor, mask: torch.Tensor = None) -> torch.Tensor:
B, T, C = x.shape
# 线性变换并切分为多头
q = self.W_q(x).view(B, T, self.n_heads, self.d_k).transpose(1, 2) # (B, n_heads, T, d_k)
k = self.W_k(x).view(B, T, self.n_heads, self.d_k).transpose(1, 2)
v = self.W_v(x).view(B, T, self.n_heads, self.d_k).transpose(1, 2)
# 计算注意力分数
attn_scores = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(self.d_k) # (B, n_heads, T, T)
# 构造因果掩码（下三角），若外部未提供
if mask is None:
causal_mask = torch.tril(torch.ones(T, T, device=x.device)).view(1, 1, T, T)
attn_scores = attn_scores.masked_fill(causal_mask == 0, float('-inf'))
else:
# mask 预期为 (B, 1, 1, T) 或 (B, 1, T, T)，根据实际情况适配
attn_scores = attn_scores.masked_fill(mask == 0, float('-inf'))
attn_weights = F.softmax(attn_scores, dim=-1)
attn_weights = self.dropout(attn_weights)
# 加权求和
out = torch.matmul(attn_weights, v) # (B, n_heads, T, d_k)
out = out.transpose(1, 2).contiguous().view(B, T, C)
return self.W_o(out)
RoPE 位置编码及其应用到 Q, K¶
def apply_rotary_emb(x: torch.Tensor, cos: torch.Tensor, sin: torch.Tensor) -> torch.Tensor:
"""
对输入 x 应用旋转位置编码。x 形状为 (..., seq_len, dim)
cos, sin 形状为 (seq_len, dim//2) 或 (..., seq_len, dim//2)
"""
# 将 x 按最后一维分成两半，分别进行旋转
dim = x.shape[-1]
x1, x2 = x[..., :dim//2], x[..., dim//2:]
cos = cos.unsqueeze(-2) # 便于广播
sin = sin.unsqueeze(-2)
rotated = torch.cat([x1 * cos - x2 * sin, x1 * sin + x2 * cos], dim=-1)
return rotated
def precompute_rotary_embeddings(seq_len: int, dim: int, theta: float = 10000.0):
"""预计算 cos 和 sin 表"""
freqs = 1.0 / (theta ** (torch.arange(0, dim, 2, dtype=torch.float32) / dim)) # (dim/2,)
t = torch.arange(seq_len, dtype=torch.float32)
freqs = torch.outer(t, freqs) # (seq_len, dim/2)
cos = torch.cos(freqs)
sin = torch.sin(freqs)
return cos, sin
Top-P (nucleus) 采样¶
def top_p_sampling(logits: torch.Tensor, p: float = 0.9, temperature: float = 1.0) -> int:
"""从 logits 中根据 Top-P 采样返回一个 token id"""
if temperature > 0:
logits = logits / temperature
probs = torch.softmax(logits, dim=-1)
# 按概率降序排列
sorted_probs, sorted_indices = torch.sort(probs, descending=True)
cumsum_probs = torch.cumsum(sorted_probs, dim=-1)
# 去除累积概率超过 p 的尾部，但至少保留第一个 token
mask = (cumsum_probs - sorted_probs) < p # 等效于 cumsum_probs - sorted_probs < p 确保第一个保留？
# 更直接的方法：保留累积概率 <= p 的部分，然后至少保留一个
mask = cumsum_probs <= p
mask[0] = True
# 重新归一化
filtered_probs = sorted_probs * mask.float()
filtered_probs = filtered_probs / filtered_probs.sum()
# 采样
sampled_idx = torch.multinomial(filtered_probs, 1).item()
return sorted_indices[sampled_idx].item()
Top-K 采样¶
def top_k_sampling(logits: torch.Tensor, k: int = 50, temperature: float = 1.0) -> int:
"""从 logits 中根据 Top-K 采样返回一个 token id"""
if temperature > 0:
logits = logits / temperature
probs = torch.softmax(logits, dim=-1)
# 选出概率最高的 k 个 token
topk_probs, topk_indices = torch.topk(probs, k, dim=-1)
# 重新归一化
topk_probs = topk_probs / topk_probs.sum()
sampled_idx = torch.multinomial(topk_probs, 1).item()
return topk_indices[sampled_idx].item()
KV Cache 更新逻辑¶
def update_kv_cache(
new_k: torch.Tensor, # (batch, num_heads, 1, head_dim)
new_v: torch.Tensor, # 同上
cache_k: torch.Tensor, # (batch, num_heads, past_len, head_dim) 或 None
cache_v: torch.Tensor,
max_cache_len: int = None
):
"""
将新 token 的 K,V 拼接到历史缓存中。若 cache 为空，则直接返回 new_k, new_v。
若设置了 max_cache_len，则仅保留最近的 max_cache_len 个 token。
"""
if cache_k is None:
return new_k, new_v
# 沿 seq_len 维度拼接 (dim=2)
updated_k = torch.cat([cache_k, new_k], dim=2)
updated_v = torch.cat([cache_v, new_v], dim=2)
if max_cache_len is not None and updated_k.size(2) > max_cache_len:
updated_k = updated_k[:, :, -max_cache_len:, :]
updated_v = updated_v[:, :, -max_cache_len:, :]
return updated_k, updated_v
加载QLoRA模型并对话的脚本¶
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
from peft import PeftModel
def load_qlora_model(base_model_name: str, adapter_path: str):
# 4-bit量化配置
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_compute_dtype=torch.bfloat16,
bnb_4bit_use_double_quant=True,
)
# 加载基座模型
model = AutoModelForCausalLM.from_pretrained(
base_model_name,
quantization_config=bnb_config,
device_map="auto",
trust_remote_code=True,
)
tokenizer = AutoTokenizer.from_pretrained(base_model_name, trust_remote_code=True)
tokenizer.pad_token = tokenizer.eos_token
# 加载LoRA适配器
model = PeftModel.from_pretrained(model, adapter_path)
model.eval()
return model, tokenizer
def chat_loop(model, tokenizer, max_new_tokens=256):
print("模型加载完毕，开始对话。输入 'quit' 退出。")
while True:
user_input = input("User: ")
if user_input.lower() == 'quit':
break
# 构建对话模板（以ChatML为例）
prompt = f"<|im_start|>user\n{user_input}<|im_end|>\n<|im_start|>assistant\n"
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
outputs = model.generate(
**inputs,
max_new_tokens=max_new_tokens,
do_sample=True,
temperature=0.7,
top_p=0.9,
pad_token_id=tokenizer.eos_token_id,
)
response = tokenizer.decode(outputs[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
print(f"Assistant: {response}")
if __name__ == "__main__":
model, tokenizer = load_qlora_model("meta-llama/Llama-2-7b-hf", "./my_lora_adapter")
chat_loop(model, tokenizer)
带梯度累积、裁剪和调度器的微调训练循环¶
import torch
from torch.optim import AdamW
from torch.optim.lr_scheduler import CosineAnnealingLR
from tqdm import tqdm
def train_one_epoch(model, dataloader, optimizer, scheduler, accumulation_steps, max_grad_norm):
model.train()
total_loss = 0
optimizer.zero_grad()
for step, batch in enumerate(tqdm(dataloader)):
outputs = model(**batch)
loss = outputs.loss / accumulation_steps
loss.backward()
total_loss += loss.item() * accumulation_steps
if (step + 1) % accumulation_steps == 0:
# 梯度裁剪
torch.nn.utils.clip_grad_norm_(model.parameters(), max_grad_norm)
optimizer.step()
scheduler.step()
optimizer.zero_grad()
return total_loss / len(dataloader)
使用示例：
optimizer = AdamW(model.parameters(), lr=2e-5)
scheduler = CosineAnnealingLR(optimizer, T_max=num_training_steps)
train_one_epoch(model, train_loader, optimizer, scheduler, accumulation_steps=4, max_grad_norm=1.0)
使用HuggingFace Trainer自定义加权多任务损失¶
from transformers import Trainer
import torch.nn.functional as F
class WeightedMultiTaskTrainer(Trainer):
def __init__(self, task_weight_dict, *args, **kwargs):
super().__init__(*args, **kwargs)
self.task_weight_dict = task_weight_dict # e.g. {'code': 2.0, 'chat': 1.0}
def compute_loss(self, model, inputs, return_outputs=False):
labels = inputs.pop("labels")
# 取出任务类型权重并拓展到与labels相同形状
task_types = inputs.pop("task_type", None) # 假设数据中有 task_type 字段
outputs = model(**inputs)
logits = outputs.logits
# 计算每个token的交叉熵
loss_fct = torch.nn.CrossEntropyLoss(reduction='none', ignore_index=-100)
per_token_loss = loss_fct(logits.view(-1, logits.size(-1)), labels.view(-1))
per_token_loss = per_token_loss.view(labels.size())
if task_types is not None:
# 根据任务类型赋予权重
weights = torch.ones_like(labels, dtype=torch.float)
for task, weight in self.task_weight_dict.items():
mask = (task_types == task)
weights[mask] = weight
# 忽略 -100 的位置
valid_mask = (labels != -100).float()
weighted_loss = (per_token_loss * weights * valid_mask).sum() / (weights * valid_mask).sum()
else:
# 默认平均
weighted_loss = per_token_loss.sum() / (labels != -100).sum()
return (weighted_loss, outputs) if return_outputs else weighted_loss
使用：
trainer = WeightedMultiTaskTrainer(
task_weight_dict={'code': 2.0, 'math': 1.5, 'chat': 1.0},
model=model,
args=training_args,
train_dataset=train_dataset,
# ...
)
trainer.train()
统计SFT数据集指令与回复长度分布¶
import json
import matplotlib.pyplot as plt
from transformers import AutoTokenizer
def analyze_length_distribution(jsonl_file: str, tokenizer_name: str = 'gpt2'):
tokenizer = AutoTokenizer.from_pretrained(tokenizer_name)
instr_lens, resp_lens = [], []
with open(jsonl_file, 'r', encoding='utf-8') as f:
for line in f:
obj = json.loads(line)
instr = obj.get('instruction', '') + obj.get('input', '')
resp = obj.get('output', '')
instr_lens.append(len(tokenizer.encode(instr)))
resp_lens.append(len(tokenizer.encode(resp)))
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 5))
ax1.hist(instr_lens, bins=50, alpha=0.7)
ax1.set_title('Instruction Length Distribution')
ax2.hist(resp_lens, bins=50, alpha=0.7, color='orange')
ax2.set_title('Response Length Distribution')
plt.show()
return instr_lens, resp_lens
基于MinHash的文本去重¶
import re
import hashlib
from collections import defaultdict
import random
class MinHash:
def __init__(self, num_perm: int = 128, seed: int = 42):
self.num_perm = num_perm
self.prime = 2**61 - 1
random.seed(seed)
self.a = [random.randint(1, self.prime - 1) for _ in range(num_perm)]
self.b = [random.randint(1, self.prime - 1) for _ in range(num_perm)]
def _hash(self, x, i):
return (self.a[i] * x + self.b[i]) % self.prime
def signature(self, tokens: set) -> list:
token_hashes = [hash(t) & 0xFFFFFFFF for t in tokens]
if not token_hashes:
return [float('inf')] * self.num_perm
sig = []
for i in range(self.num_perm):
sig.append(min(self._hash(h, i) for h in token_hashes))
return sig
@staticmethod
def similarity(sig1, sig2):
return sum(x == y for x, y in zip(sig1, sig2)) / len(sig1)
def deduplicate_texts(texts, threshold=0.8, ngram_size=3):
mh = MinHash()
signatures = []
for text in texts:
words = text.split()
ngrams = set()
for i in range(len(words) - ngram_size + 1):
ngrams.add(' '.join(words[i:i+ngram_size]))
signatures.append(mh.signature(ngrams))
keep_indices = []
for i, text in enumerate(texts):
is_dup = False
for j in keep_indices:
if mh.similarity(signatures[i], signatures[j]) >= threshold:
is_dup = True
break
if not is_dup:
keep_indices.append(i)
return [texts[i] for i in keep_indices]
数据去污染检测（n-gram重叠）¶
from typing import List, Tuple
def get_ngrams(text: str, n: int = 13) -> set:
"""返回文本的n-gram集合（字符级）"""
text = text.lower()
return set(text[i:i+n] for i in range(len(text) - n + 1))
def detect_contamination(train_samples: List[str], eval_samples: List[str], n: int = 13, threshold: float = 0.6) -> List[Tuple[int, int, float]]:
"""
检查每个训练样本与评测样本的最大n-gram重叠率。
返回可疑污染列表，包含 (train_idx, eval_idx, overlap_ratio)。
"""
eval_ngrams = [get_ngrams(s, n) for s in eval_samples]
suspicious = []
for i, train_text in enumerate(train_samples):
train_ngrams = get_ngrams(train_text, n)
if not train_ngrams:
continue
for j, e_ng in enumerate(eval_ngrams):
overlap = train_ngrams & e_ng
ratio = len(overlap) / len(train_ngrams)
if ratio >= threshold:
suspicious.append((i, j, ratio))
return suspicious
Self-Instruct指令生成pipeline¶
import openai
import random
import json
openai.api_key = "your-api-key"
def generate_self_instruct(seed_tasks: list, num_to_generate: int = 10, model="gpt-4"):
generated = []
for _ in range(num_to_generate):
few_shot = random.sample(seed_tasks, min(6, len(seed_tasks)))
prompt = "你是一个智能助手，请生成一个全新的、不同于以下示例的指令，并为该指令提供一个高质量的回答。\n\n"
for task in few_shot:
prompt += f"指令: {task['instruction']}\n回答: {task['output']}\n\n"
prompt += "现在，请生成一个全新的指令和回答。\n指令:"
response = openai.ChatCompletion.create(
model=model,
messages=[{"role": "user", "content": prompt}],
temperature=0.8,
max_tokens=256
)
content = response.choices[0].message.content.strip()
# 简单解析（实际需更鲁棒）
parts = content.split('\n回答:', 1)
if len(parts) == 2:
instr = parts[0].replace('指令:', '').strip()
ans = parts[1].strip()
# 去重
if not any(instr == t['instruction'] for t in generated):
generated.append({'instruction': instr, 'output': ans})
return generated
Evol-Instruct深度进化¶
def evolve_instruction(seed_instruction: str, model="gpt-4") -> str:
prompt = f"""请将以下指令改写得更加复杂，增加更多的限制条件或需要多步推理的要求。确保改写后的指令仍然是可以被合理回答的。
原指令: {seed_instruction}
请只输出改写后的指令，不要添加额外文本。"""
response = openai.ChatCompletion.create(
model=model,
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
max_tokens=200
)
return response.choices[0].message.content.strip()
多轮对话转换为ChatML格式并生成input_ids和labels¶
from transformers import AutoTokenizer
def convert_multi_turn_to_chatml_and_tokenize(conversations: list, tokenizer, max_length=2048):
"""
conversations: [{"role": "system"/"user"/"assistant", "content": "..."}, ...]
返回 input_ids 和 labels (list of ints)。
"""
chatml = ""
for turn in conversations:
role = turn["role"]
content = turn["content"]
chatml += f"<|im_start|>{role}\n{content}<|im_end|>\n"
# 若最后不是 assistant 结尾，补充一个空的 assistant 标记以便生成
if conversations[-1]["role"] != "assistant":
chatml += "<|im_start|>assistant\n"
# 此时标签不需要计算该部分
tokenized = tokenizer(chatml, truncation=True, max_length=max_length, return_tensors=None)
input_ids = tokenized['input_ids']
# 构造labels：找到所有 assistant 部分保留token，其余设-100
labels = [-100] * len(input_ids)
# 简单处理：通过查找特殊token来确定assistant段，这里提供一个更鲁棒的方法：
# 用tokenizer编码 "<|im_start|>assistant\n" 找到起始token id，但序列可能变化，此处简化：
# 更好的做法是标记数据时保留原始边界。
# 这里仅演示：假设我们可以通过分词后的token序列寻找 assistant 标记的位置。
# 以下代码为伪代码，实际使用时建议在构造chatml时记录位置。
# （省略详细边界查找，实际项目中使用现成库如tokenizers处理）
return input_ids, labels
更实用的方式是使用HuggingFace的apply_chat_template方法（如果tokenizer已配置chat_template）。
IFEval约束遵循准确率评估¶
import json
import re
from typing import List, Dict, Callable
# 预定义约束检查函数注册
CONSTRAINT_CHECKERS = {}
def register_checker(name: str):
def decorator(func):
CONSTRAINT_CHECKERS[name] = func
return func
return decorator
@register_checker('word_count_between')
def check_word_count(text: str, params: dict) -> bool:
words = text.split()
return params.get('min', 0) <= len(words) <= params.get('max', float('inf'))
@register_checker('must_contain')
def check_must_contain(text: str, params: dict) -> bool:
return params['substring'] in text
@register_checker('forbidden_words')
def check_forbidden_words(text: str, params: dict) -> bool:
return all(word not in text for word in params.get('words', []))
@register_checker('json_format')
def check_json(text: str, params=None) -> bool:
try:
json.loads(text.strip())
return True
except:
return False
def ifeval_evaluate(responses: List[str], constraints_list: List[List[Dict]]) -> float:
"""
responses: 模型生成的回答列表
constraints_list: 每条指令对应的约束列表，每个约束是一个字典 {'type': ..., 'params': {...}}
返回严格约束遵循率
"""
total = len(responses)
strict_correct = 0
for resp, constraints in zip(responses, constraints_list):
if all(CONSTRAINT_CHECKERS[c['type']](resp, c.get('params', {})) for c in constraints):
strict_correct += 1
return strict_correct / total if total > 0 else 0.0
GPT-as-judge比较两个模型回复¶
import openai
import json
import random
openai.api_key = "your-api-key"
def gpt_judge(prompt: str, response_a: str, response_b: str, model="gpt-4") -> str:
"""返回 'A' / 'B' / 'tie'"""
# 随机交换位置以消除位置偏差
if random.random() < 0.5:
first, second = "A", "B"
resp1, resp2 = response_a, response_b
else:
first, second = "B", "A"
resp1, resp2 = response_b, response_a
judge_prompt = f"""请作为一个公正的评判者，比较以下两个AI助手对用户问题的回复。
用户问题: {prompt}
助手{first}的回复: {resp1}
助手{second}的回复: {resp2}
请从准确性、有用性、流畅度等方面综合评判，哪个回复更好？如果难分高下，请选择 'tie'。
请只输出 'A', 'B', 或 'tie'，并简要说明理由。
格式：{{"winner": "A/B/tie", "reason": "..."}}"""
response = openai.ChatCompletion.create(
model=model,
messages=[{"role": "user", "content": judge_prompt}],
temperature=0.0,
max_tokens=256
)
content = response.choices[0].message.content.strip()
try:
result = json.loads(content)
winner = result['winner']
# 映射回原始A/B
if winner == first:
return 'A'
elif winner == second:
return 'B'
else:
return 'tie'
except:
return 'tie'
vLLM部署及Python API调用示例¶
部署命令（服务器端）：
python -m vllm.entrypoints.openai.api_server \
--model /path/to/sft-model \
--gpu-memory-utilization 0.9 \
--max-model-len 4096 \
--served-model-name my-sft-model
Python客户端调用：
import requests
def chat_vllm(prompt: str, history=None):
url = "http://localhost:8000/v1/chat/completions"
messages = [{"role": "system", "content": "你是一个有帮助的助手。"}]
if history:
messages.extend(history)
messages.append({"role": "user", "content": prompt})
payload = {
"model": "my-sft-model",
"messages": messages,
"max_tokens": 256,
"temperature": 0.7,
"top_p": 0.9,
}
response = requests.post(url, json=payload)
return response.json()['choices'][0]['message']['content']
# 使用示例
reply = chat_vllm("你好，请介绍一下人工智能。")
print(reply)
编写脚本实现A/B测试，批量运行两个模型对比，并统计胜率。¶
import torch
from tqdm import tqdm
from transformers import AutoModelForCausalLM, AutoTokenizer
from typing import List, Dict
def ab_test(
prompts: List[str],
model_a_path: str,
model_b_path: str,
tokenizer_path: str,
judge_model=None, # 可选的裁判模型，若没有则用简单规则
max_new_tokens: int = 128,
device: str = "cuda"
) -> Dict[str, int]:
"""
批量运行两个模型，对每个prompt分别生成回答，然后利用内置或外部裁判统计胜率。
返回字典 {'A_wins': int, 'B_wins': int, 'ties': int}
"""
tokenizer = AutoTokenizer.from_pretrained(tokenizer_path)
model_a = AutoModelForCausalLM.from_pretrained(model_a_path).to(device)
model_b = AutoModelForCausalLM.from_pretrained(model_b_path).to(device)
model_a.eval()
model_b.eval()
results = {'A_wins': 0, 'B_wins': 0, 'ties': 0}
for prompt in tqdm(prompts):
inputs = tokenizer(prompt, return_tensors="pt").to(device)
with torch.no_grad():
gen_a = model_a.generate(**inputs, max_new_tokens=max_new_tokens, do_sample=False)
gen_b = model_b.generate(**inputs, max_new_tokens=max_new_tokens, do_sample=False)
resp_a = tokenizer.decode(gen_a[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
resp_b = tokenizer.decode(gen_b[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
if judge_model is not None:
# 使用GPT-as-judge或其他裁判模型
winner = judge_model.compare(prompt, resp_a, resp_b)
else:
# 简单规则：谁更长谁赢（仅示例，实际无意义）
winner = 'A' if len(resp_a) > len(resp_b) else ('B' if len(resp_b) > len(resp_a) else 'tie')
if winner == 'A':
results['A_wins'] += 1
elif winner == 'B':
results['B_wins'] += 1
else:
results['ties'] += 1
return results
使用示例：
prompts = ["解释什么是机器学习", "写一首关于秋天的诗"]
stats = ab_test(prompts, "path/to/modelA", "path/to/modelB", "path/to/tokenizer")
print(stats) # {'A_wins': 2, 'B_wins': 0, 'ties': 0}
实现一个基于困惑度的数据过滤函数，过滤高PPL的回复。¶
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from typing import List
def filter_by_ppl(
texts: List[str],
model_name: str = "gpt2",
threshold: float = 100.0,
device: str = "cuda"
) -> List[str]:
"""
计算每个文本的困惑度，返回 PPL 低于阈值的文本。
"""
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(model_name).to(device)
model.eval()
kept = []
for text in texts:
inputs = tokenizer(text, return_tensors="pt").to(device)
with torch.no_grad():
outputs = model(**inputs, labels=inputs.input_ids)
loss = outputs.loss
ppl = torch.exp(loss).item()
if ppl < threshold:
kept.append(text)
return kept
注意：实际应用中可使用更大的参考模型，并可对每个样本计算长度归一化的PPL。
使用DeepSpeed配置文件启动SFT训练，写出命令行和关键配置。¶
配置文件 ds_config.json：
{
"train_batch_size": 16,
"gradient_accumulation_steps": 4,
"optimizer": {
"type": "AdamW",
"params": {
"lr": 2e-5,
"weight_decay": 0.1
}
},
"fp16": {
"enabled": false
},
"bf16": {
"enabled": true
},
"zero_optimization": {
"stage": 2,
"offload_optimizer": {
"device": "cpu",
"pin_memory": true
},
"overlap_comm": true,
"contiguous_gradients": true
},
"gradient_clipping": 1.0,
"steps_per_print": 10
}
启动命令：
deepspeed --num_gpus=4 train_sft.py \
--deepspeed ds_config.json \
--model_name_or_path meta-llama/Llama-2-7b-hf \
--data_path ./sft_data.jsonl \
--output_dir ./output \
--per_device_train_batch_size 4 \
--gradient_accumulation_steps 4 \
--learning_rate 2e-5 \
--num_train_epochs 2 \
--bf16 \
--logging_steps 10
自定义回调函数，在每个epoch结束时用几个prompt测试生成质量。¶
from transformers import TrainerCallback, TrainerControl, TrainerState, TrainingArguments
import torch
class GenerationTestCallback(TrainerCallback):
def __init__(self, tokenizer, test_prompts: list, max_new_tokens=50):
self.tokenizer = tokenizer
self.test_prompts = test_prompts
self.max_new_tokens = max_new_tokens
def on_epoch_end(self, args: TrainingArguments, state: TrainerState, control: TrainerControl, **kwargs):
model = kwargs['model']
model.eval()
print(f"\n=== Epoch {state.epoch:.2f} 生成测试 ===")
for prompt in self.test_prompts:
inputs = self.tokenizer(prompt, return_tensors='pt').to(model.device)
with torch.no_grad():
outputs = model.generate(**inputs, max_new_tokens=self.max_new_tokens, do_sample=False,
pad_token_id=self.tokenizer.eos_token_id)
generated = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
print(f"Prompt: {prompt}\nGenerated: {generated}\n")
model.train()
使用：
实现一个简单的“拒绝采样”数据筛选逻辑。¶
从生成的大量候选中，通过评判模型（或规则）选择最佳回答作为训练数据。
def rejection_sampling(
prompt: str,
candidates: List[str],
judge_model, # 一个能打分的模型或函数
score_threshold: float = 0.8
) -> str:
"""
对多个候选回答打分，返回得分最高且超过阈值的回答。若没有，返回空字符串或进行重试。
"""
best_score = -1.0
best_response = ""
for resp in candidates:
score = judge_model.evaluate(prompt, resp) # 假设返回0-1之间的分数
if score > best_score:
best_score = score
best_response = resp
if best_score >= score_threshold:
return best_response
else:
return "" # 所有候选均不合格，可触发重新生成
具体应用时，judge_model 可以是GPT-4、一个专门的打分模型，甚至基于规则（如包含关键词等）。
实现早期停止（Early Stopping）类，可监控验证损失。¶
class EarlyStopping:
def __init__(self, patience: int = 3, min_delta: float = 0.0, mode: str = 'min'):
self.patience = patience
self.min_delta = min_delta
self.mode = mode
self.counter = 0
self.best_score = None
self.early_stop = False
def __call__(self, current_score: float) -> bool:
if self.best_score is None:
self.best_score = current_score
elif (self.mode == 'min' and current_score > self.best_score - self.min_delta) or \
(self.mode == 'max' and current_score < self.best_score + self.min_delta):
self.counter += 1
if self.counter >= self.patience:
self.early_stop = True
else:
self.best_score = current_score
self.counter = 0
return self.early_stop
训练循环中使用：
stopper = EarlyStopping(patience=3)
for epoch in range(epochs):
val_loss = validate()
if stopper(val_loss):
print("Early stopping triggered")
break
编写代码计算两个模型输出之间的KL散度。¶
import torch
import torch.nn.functional as F
def compute_kl_divergence(model1, model2, tokenizer, prompt: str, max_len: int = 50):
inputs = tokenizer(prompt, return_tensors='pt').to(model1.device)
gen_out = model1.generate(**inputs, max_new_tokens=max_len, do_sample=False,
output_scores=True, return_dict_in_generate=True)
generated_ids = gen_out.sequences[0]
input_len = inputs.input_ids.shape[1]
logits1 = []
logits2 = []
with torch.no_grad():
past_kv1, past_kv2 = None, None
cur_input = inputs.input_ids
for step in range(generated_ids.shape[1] - input_len):
out1 = model1(cur_input, past_key_values=past_kv1, use_cache=True)
out2 = model2(cur_input, past_key_values=past_kv2, use_cache=True)
logits1.append(out1.logits[:, -1, :])
logits2.append(out2.logits[:, -1, :])
past_kv1 = out1.past_key_values
past_kv2 = out2.past_key_values
# 使用实际生成的token作为下一步输入
cur_input = generated_ids[:, input_len + step].unsqueeze(1)
logits1 = torch.cat(logits1, dim=0)
logits2 = torch.cat(logits2, dim=0)
p = F.softmax(logits1, dim=-1)
log_q = F.log_softmax(logits2, dim=-1)
kl = F.kl_div(log_q, p, reduction='batchmean') # KL(p || q)
return kl.item()
实现一个“温度采样”数据采样器，用于平衡不同任务的数据量。¶
import numpy as np
from torch.utils.data import Sampler
from collections import Counter
class TemperatureSampler(Sampler):
def __init__(self, dataset, task_labels: list, temperature: float = 0.5):
self.dataset = dataset
self.task_labels = task_labels
self.task_counts = Counter(task_labels)
self.temperature = temperature
self.task_probs = self._compute_probs()
def _compute_probs(self):
counts = np.array([self.task_counts[t] for t in sorted(self.task_counts.keys())])
probs = counts ** self.temperature
probs = probs / probs.sum()
return probs
def __iter__(self):
# 根据每个任务的概率，随机采样任务，再从该任务中均匀采样一个样本
tasks = sorted(self.task_counts.keys())
task_ids = np.arange(len(tasks))
while True:
chosen_task = np.random.choice(task_ids, p=self.task_probs)
# 获取该任务的所有样本索引
indices = [i for i, t in enumerate(self.task_labels) if t == tasks[chosen_task]]
yield np.random.choice(indices)
def __len__(self):
# 无法确定精确长度，返回估计值
return len(self.dataset)
使用时将 task_labels 传入，可自动平衡高频和低频任务。
如何用HuggingFace datasets流式加载海量JSONL文件并预处理？¶
from datasets import load_dataset
from transformers import AutoTokenizer
tokenizer = AutoTokenizer.from_pretrained("gpt2")
def preprocess(example):
# 假设 example 包含 'instruction' 和 'output'
prompt = f"<|im_start|>user\n{example['instruction']}<|im_end|>\n<|im_start|>assistant\n"
answer = example['output']
full_text = prompt + answer + "<|im_end|>"
tokenized = tokenizer(full_text, truncation=True, max_length=2048)
# 构造labels（prompt部分设为-100）
prompt_len = len(tokenizer(prompt)['input_ids'])
labels = tokenized['input_ids'].copy()
labels[:prompt_len] = [-100] * prompt_len
tokenized['labels'] = labels
return tokenized
# 流式加载：不将全量数据加载到内存
dataset = load_dataset("json", data_files="sft_data.jsonl", split="train", streaming=True)
dataset = dataset.map(preprocess, remove_columns=dataset.column_names)
# 此时dataset是一个可迭代对象，可配合torch的DataLoader使用
编写一个简单的PPO微调循环框架（伪代码或简化版）。¶
# 假设已有一个SFT模型 actor_model，参考模型 ref_model，奖励模型 reward_model，价值模型 critic_model
# 使用一个简单的PPO训练循环（省略优势计算细节）
import torch
from torch.optim import AdamW
actor_optimizer = AdamW(actor_model.parameters(), lr=1e-5)
critic_optimizer = AdamW(critic_model.parameters(), lr=5e-6)
for epoch in range(ppo_epochs):
for batch in dataloader:
prompts = batch['prompt']
# 生成回答
responses = actor_model.generate(prompts)
# 计算奖励
rewards = reward_model.score(prompts, responses)
# 计算价值估计
values = critic_model(prompts, responses)
# 计算优势（简化：实际应用GAE）
advantages = rewards - values.detach()
# PPO损失
old_log_probs = actor_model.log_prob(prompts, responses).detach()
for _ in range(update_epochs):
new_log_probs = actor_model.log_prob(prompts, responses)
ratio = torch.exp(new_log_probs - old_log_probs)
surr1 = ratio * advantages
surr2 = torch.clamp(ratio, 1-epsilon, 1+epsilon) * advantages
actor_loss = -torch.min(surr1, surr2).mean()
# 添加KL惩罚
kl = torch.distributions.kl_divergence(
actor_model.distribution(prompts), ref_model.distribution(prompts)
).mean()
loss = actor_loss + beta * kl
actor_optimizer.zero_grad()
loss.backward()
actor_optimizer.step()
# 更新价值模型
critic_loss = torch.nn.functional.mse_loss(values, rewards)
critic_optimizer.zero_grad()
critic_loss.backward()
critic_optimizer.step()
实际完整PPO实现非常复杂，建议直接使用库如 trl 或 DeepSpeed-Chat。
实现将SFT数据转为标准OpenAI messages格式的函数。¶
def sft_to_openai_messages(instruction: str, input_text: str = "", output: str = "") -> list:
"""
将Alpaca风格数据转为OpenAI API的messages格式。
"""
messages = [{"role": "system", "content": "你是一个有帮助的AI助手。"}]
if input_text:
user_content = f"{instruction}\n{input_text}"
else:
user_content = instruction
messages.append({"role": "user", "content": user_content})
messages.append({"role": "assistant", "content": output})
return messages
# 使用：
example = {"instruction": "解释什么是黑洞", "input": "", "output": "黑洞是时空的一个区域，引力极强..."}
openai_format = sft_to_openai_messages(**example)
编写一个对话模拟器，让两个模型互相对话生成微调数据。¶
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
def simulate_dialogue(
model_a, tokenizer_a,
model_b, tokenizer_b,
initial_prompt: str,
num_turns: int = 4,
max_new_tokens: int = 100
) -> list:
"""
两个模型交替对话，A发起，B回复，A再回复，如此循环。返回对话历史列表。
"""
dialogue_history = [{"role": "user", "content": initial_prompt}]
# 模型A生成第一个回复
prompt_a = tokenizer_a.apply_chat_template(dialogue_history, tokenize=False, add_generation_prompt=True)
inputs = tokenizer_a(prompt_a, return_tensors="pt").to(model_a.device)
gen_a = model_a.generate(**inputs, max_new_tokens=max_new_tokens, do_sample=True, temperature=0.8)
reply_a = tokenizer_a.decode(gen_a[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
dialogue_history.append({"role": "assistant", "content": reply_a})
for turn in range(1, num_turns):
# B模型回复
prompt_b = tokenizer_b.apply_chat_template(dialogue_history, tokenize=False, add_generation_prompt=True)
inputs = tokenizer_b(prompt_b, return_tensors="pt").to(model_b.device)
gen_b = model_b.generate(**inputs, max_new_tokens=max_new_tokens, do_sample=True, temperature=0.8)
reply_b = tokenizer_b.decode(gen_b[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
dialogue_history.append({"role": "user", "content": reply_b}) # B的回复作为下一轮的“用户”消息
# A模型回复
prompt_a = tokenizer_a.apply_chat_template(dialogue_history, tokenize=False, add_generation_prompt=True)
inputs = tokenizer_a(prompt_a, return_tensors="pt").to(model_a.device)
gen_a = model_a.generate(**inputs, max_new_tokens=max_new_tokens, do_sample=True, temperature=0.8)
reply_a = tokenizer_a.decode(gen_a[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
dialogue_history.append({"role": "assistant", "content": reply_a})
return dialogue_history
实现一个“数据配方”配置解析器，能按比例采样多个数据集。¶
import random
from torch.utils.data import Dataset, ConcatDataset, Subset
class RecipeSampler:
def __init__(self, datasets: list, ratios: list):
"""
datasets: List of Dataset objects
ratios: List of float, 每个数据集的采样比例（概率）
"""
self.datasets = datasets
self.ratios = ratios
self.total_samples = sum(len(ds) for ds in datasets)
# 构建每个数据集的索引区间
self._dataset_offsets = self._compute_offsets()
def _compute_offsets(self):
indices = []
for ds, ratio in zip(self.datasets, self.ratios):
# 根据比例计算应采样的数量（按比例从总样本数中分配）
n_samples = int(self.total_samples * ratio)
# 如果比例分配导致超过数据集大小，则截断
n_samples = min(n_samples, len(ds))
indices.append(random.sample(range(len(ds)), n_samples))
return indices
def sample(self, num_samples: int):
combined = []
for ds_idx, ds in enumerate(self.datasets):
combined.extend([ (ds_idx, idx) for idx in self._dataset_offsets[ds_idx] ])
random.shuffle(combined)
return combined[:num_samples]
更简单的方式是使用 torch.utils.data.WeightedRandomSampler，按每个数据集的权重分配采样概率，然后对合并后的数据集应用该采样器。
使用bitsandbytes加载4-bit量化模型，并用LoRA微调，给出代码框架。¶
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, TaskType
# 量化配置
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_compute_dtype=torch.bfloat16,
bnb_4bit_use_double_quant=True,
)
model = AutoModelForCausalLM.from_pretrained(
"meta-llama/Llama-2-7b-hf",
quantization_config=bnb_config,
device_map="auto",
trust_remote_code=True,
)
tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-2-7b-hf")
tokenizer.pad_token = tokenizer.eos_token
# LoRA配置
lora_config = LoraConfig(
r=8,
lora_alpha=16,
target_modules=["q_proj", "v_proj"],
lora_dropout=0.05,
bias="none",
task_type=TaskType.CAUSAL_LM,
)
model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 后续可使用HuggingFace Trainer或其他训练循环进行微调
编写一个微调模型回归测试脚本，自动跑100个固定prompt并计算指标。¶
import json
from transformers import AutoModelForCausalLM, AutoTokenizer
def regression_test(model, tokenizer, prompt_file: str = "test_prompts.json", output_file: str = "results.json"):
with open(prompt_file, 'r', encoding='utf-8') as f:
prompts = [line.strip() for line in f.readlines() if line.strip()]
results = []
for i, prompt in enumerate(prompts):
inputs = tokenizer(prompt, return_tensors="pt").to(model.device)
gen = model.generate(**inputs, max_new_tokens=100, do_sample=False,
pad_token_id=tokenizer.eos_token_id)
response = tokenizer.decode(gen[0][inputs.input_ids.shape[1]:], skip_special_tokens=True)
results.append({"id": i, "prompt": prompt, "response": response})
# 可添加自动评分（例如检查是否包含特定关键词、长度、格式等）
# 这里仅记录输出
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(results, f, ensure_ascii=False, indent=2)
print(f"Results saved to {output_file}")
# 调用
# regression_test(my_model, tokenizer)
更进一步可以加入自动化指标计算（如BLEU、ROUGE、或基于关键字的规则），并与基线模型结果对比，若有显著退化则告警。
以上代码覆盖了从数据工程、训练、评估到部署的多个关键环节，可根据实际需求进行调整和扩展。