LoRA微调教程

LoRA微调手把手教程

LoRA:高效微调的利器 LoRA(Low-Rank Adaptation)通过在原模型权重旁添加低秩矩阵,只需训练极少量参数即可实现有效的微调。一个7B模型的LoRA微调只需8GB显存,而全量微调需要56GB。 环境准备 pip install peft transformers accelerate datasets bitsandbytes 完整微调代码 import torch from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments from peft import LoraConfig, get_peft_model, TaskType from datasets import Dataset # 1. 加载模型和分词器 model_name = "Qwen/Qwen3-7B" tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True) model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.float16, device_map="auto", trust_remote_code=True, ) # 2. LoRA配置 lora_config = LoraConfig( task_type=TaskType.CAUSAL_LM, r=64, # LoRA秩,越大容量越大但训练越慢 lora_alpha=128, # 缩放因子,通常为r的2倍 lora_dropout=0.05, # Dropout防止过拟合 target_modules=[ # 应用LoRA的模块 "q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj" ], bias="none", ) # 3. 应用LoRA model = get_peft_model(model, lora_config) model.print_trainable_parameters() # 输出:trainable params: 39,321,600 || all params: 7,078,299,648 || trainable%: 0.556% # 4. 数据准备 def format_dataset(data): formatted = [] for item in data: text = f"<|im_start|>user\n{item['input']}<|im_end|>\n<|im_start|>assistant\n{item['output']}<|im_end|>" formatted.append({"text": text}) return formatted train_data = format_dataset(raw_train_data) val_data = format_dataset(raw_val_data) train_dataset = Dataset.from_list(train_data) val_dataset = Dataset.from_list(val_data) def tokenize_fn(examples): result = tokenizer( examples["text"], truncation=True, max_length=2048, padding=False, ) result["labels"] = result["input_ids"].copy() return result train_dataset = train_dataset.map(tokenize_fn, batched=True, remove_columns=["text"]) val_dataset = val_dataset.map(tokenize_fn, batched=True, remove_columns=["text"]) # 5. 训练参数 training_args = TrainingArguments( output_dir="./lora-output", num_train_epochs=3, per_device_train_batch_size=4, per_device_eval_batch_size=4, gradient_accumulation_steps=4, warmup_ratio=0.1, learning_rate=2e-4, lr_scheduler_type="cosine", logging_steps=10, eval_strategy="steps", eval_steps=100, save_strategy="steps", save_steps=100, save_total_limit=3, load_best_model_at_end=True, bf16=True, gradient_checkpointing=True, report_to="tensorboard", ) # 6. 训练 from transformers import Trainer trainer = Trainer( model=model, args=training_args, train_dataset=train_dataset, eval_dataset=val_dataset, data_collator=lambda features: { "input_ids": torch.nn.utils.rnn.pad_sequence( [torch.tensor(f["input_ids"]) for f in features], batch_first=True, padding_value=tokenizer.pad_token_id ), "labels": torch.nn.utils.rnn.pad_sequence( [torch.tensor(f["labels"]) for f in features], batch_first=True, padding_value=-100 ), "attention_mask": torch.nn.utils.rnn.pad_sequence( [torch.tensor([1] * len(f["input_ids"])) for f in features], batch_first=True, padding_value=0 ), }, ) trainer.train() # 7. 保存LoRA权重 model.save_pretrained("./lora-weights") tokenizer.save_pretrained("./lora-weights") 合并与部署 # 合并LoRA权重到基础模型 from peft import PeftModel base_model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.float16, device_map="auto" ) model = PeftModel.from_pretrained(base_model, "./lora-weights") merged_model = model.merge_and_unload() # 合并权重 # 保存合并后的完整模型 merged_model.save_pretrained("./merged-model") tokenizer.save_pretrained("./merged-model") # 导出为GGUF格式(用于Ollama部署) # python convert.py ./merged-model --outtype f16 QLoRA(量化LoRA) from transformers import BitsAndBytesConfig # 4-bit量化加载基础模型 bnb_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype=torch.float16, bnb_4bit_use_double_quant=True, ) model = AutoModelForCausalLM.from_pretrained( model_name, quantization_config=bnb_config, device_map="auto", ) # 其余LoRA配置和训练流程相同 # QLoRA可以在单张8GB GPU上微调7B模型 超参数调优指南 参数 推荐值 说明 r 16-128 简单任务用小r,复杂任务用大r lora_alpha 2×r 通常为r的2倍 learning_rate 1e-4 ~ 5e-4 LoRA需要比全量微调更大的学习率 epochs 2-5 注意过拟合 batch_size 4-16 配合gradient_accumulation target_modules 全选 QKVO+FFN效果最好 常见问题 显存不足 使用QLoRA(4-bit量化) 减小batch_size,增加gradient_accumulation 启用gradient_checkpointing 减小max_length 过拟合 减少epochs 增加lora_dropout 增加训练数据 减小r 效果不好 检查数据质量 增大r 确保target_modules覆盖所有线性层 检查学习率是否合适 结语 LoRA是大模型微调的性价比之选——少量参数、少量显存、快速训练。通过合理的配置和高质量数据,LoRA微调可以达到接近全量微调的效果。掌握LoRA是LLM工程化的必备技能。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

2026-07-02 · 2 min · 354 words · 硅基 AGI 探索者
微调数据准备

微调数据准备最佳实践

数据决定微调效果上限 微调数据的质量直接决定模型的能力上限。再好的训练算法也无法从低质量数据中学到高质量的模式。2026年的微调数据准备已经形成了一套系统化的最佳实践。 数据采集 多源数据融合 class DataCollector: def __init__(self): self.sources = { "human_annotated": [], # 人工标注数据(质量最高) "model_generated": [], # 模型生成+人工筛选 "real_interactions": [], # 真实用户交互(脱敏) "synthetic": [], # 合成数据 } async def collect(self): dataset = [] # 1. 人工标注数据 for item in self.sources["human_annotated"]: dataset.append({ **item, "source": "human", "quality": "high" }) # 2. 模型生成数据(需要筛选) for item in self.sources["model_generated"]: if await self.quality_check(item): dataset.append({ **item, "source": "model_generated", "quality": "medium" }) # 3. 真实交互数据(脱敏处理) for item in self.sources["real_interactions"]: cleaned = self.desensitize(item) if cleaned: dataset.append({ **cleaned, "source": "real", "quality": "high" }) return dataset 数据格式标准化 class DataFormatter: """统一数据格式为对话格式""" def format_instruction(self, instruction, input_text=None, output=None): return { "messages": [ {"role": "system", "content": "你是一个专业助手。"}, {"role": "user", "content": instruction + (f"\n\n{input_text}" if input_text else "")}, {"role": "assistant", "content": output} if output else None, ], "metadata": { "task_type": "instruction", "language": "zh", } } def format_conversation(self, turns): """格式化多轮对话""" return { "messages": turns, "metadata": {"task_type": "conversation", "n_turns": len(turns) // 2} } def format_tool_use(self, user_message, tool_calls, tool_results, final_response): """格式化工具调用数据""" messages = [{"role": "user", "content": user_message}] for call, result in zip(tool_calls, tool_results): messages.append({"role": "assistant", "tool_calls": [call]}) messages.append({"role": "tool", "content": json.dumps(result)}) messages.append({"role": "assistant", "content": final_response}) return {"messages": messages, "metadata": {"task_type": "tool_use"}} 数据质量检查 class DataQualityChecker: def __init__(self): self.checks = [ self.check_length, self.check_encoding, self.check_repetition, self.check_toxicity, self.check_consistency, ] async def check(self, sample): """运行所有质量检查""" for check in self.checks: result = await check(sample) if not result["passed"]: return False, result["reason"] return True, "All checks passed" async def check_length(self, sample): text = self.extract_text(sample) if len(text) < 10: return {"passed": False, "reason": "Too short"} if len(text) > 32000: return {"passed": False, "reason": "Too long"} return {"passed": True} async def check_repetition(self, sample): text = self.extract_text(sample) # 检查n-gram重复 words = text.split() if len(words) > 10: bigrams = [' '.join(words[i:i+2]) for i in range(len(words)-1)] repeat_ratio = len(set(bigrams)) / len(bigrams) if repeat_ratio < 0.5: return {"passed": False, "reason": "High repetition"} return {"passed": True} async def check_toxicity(self, sample): text = self.extract_text(sample) toxic_words = ["暴力", "色情", "毒品"] # 简化示例 if any(word in text for word in toxic_words): return {"passed": False, "reason": "Toxic content"} return {"passed": True} 数据去重 class DataDeduplicator: def __init__(self, similarity_threshold=0.9): self.threshold = similarity_threshold self.embeddings = [] self.model = SentenceTransformer('BAAI/bge-small-zh-v1.5') def deduplicate(self, dataset): """基于语义相似度去重""" texts = [self.extract_text(d) for d in dataset] embeddings = self.model.encode(texts, normalize_embeddings=True) unique_indices = [] for i in range(len(dataset)): is_duplicate = False for j in unique_indices: similarity = embeddings[i] @ embeddings[j] if similarity > self.threshold: is_duplicate = True break if not is_duplicate: unique_indices.append(i) return [dataset[i] for i in unique_indices] 数据增强 class DataAugmentor: def __init__(self, llm): self.llm = llm async def augment(self, sample, n_variants=3): """生成数据的变体""" variants = [sample] # 1. 改写用户问题 rewritten = await self.rewrite_query(sample) variants.append(rewritten) # 2. 添加噪声(错别字等) noisy = self.add_typo_noise(sample) variants.append(noisy) # 3. 改变语气/风格 restyled = await self.restyle(sample) variants.append(restyled) return variants async def rewrite_query(self, sample): """改写用户查询""" original_query = sample["messages"][1]["content"] prompt = f"将以下问题改写为不同表述,保持语义不变:\n{original_query}" rewritten = await self.llm.generate(prompt) new_sample = copy.deepcopy(sample) new_sample["messages"][1]["content"] = rewritten new_sample["metadata"]["augmented"] = "rewritten" return new_sample 数据集划分 def split_dataset(dataset, train_ratio=0.9, val_ratio=0.05, test_ratio=0.05): """按任务类型分层划分""" from sklearn.model_selection import train_test_split # 按任务类型分组 by_task = defaultdict(list) for item in dataset: by_task[item["metadata"]["task_type"]].append(item) train, val, test = [], [], [] for task_type, items in by_task.items(): n = len(items) n_train = int(n * train_ratio) n_val = int(n * val_ratio) # 随机打乱 random.shuffle(items) train.extend(items[:n_train]) val.extend(items[n_train:n_train+n_val]) test.extend(items[n_train+n_val:]) return train, val, test 数据统计与可视化 class DatasetAnalyzer: def analyze(self, dataset): stats = { "total_samples": len(dataset), "task_distribution": Counter(d["metadata"]["task_type"] for d in dataset), "avg_turns": np.mean([len(d["messages"]) // 2 for d in dataset]), "avg_length": np.mean([len(self.extract_text(d)) for d in dataset]), "length_distribution": self.length_distribution(dataset), "language_distribution": Counter(d["metadata"].get("language", "unknown") for d in dataset), } return stats def report(self, stats): print(f"总样本数:{stats['total_samples']}") print(f"任务分布:{dict(stats['task_distribution'])}") print(f"平均轮次:{stats['avg_turns']:.1f}") print(f"平均长度:{stats['avg_length']:.0f}字符") 最佳实践总结 质量>数量:1万条高质量数据 > 10万条低质量数据 多样性:覆盖不同任务类型、长度、难度 去重:避免相似样本重复,防止模型过拟合 脱敏:严格移除用户PII信息 版本管理:数据集版本与模型版本对应 持续迭代:从生产中收集bad case,持续补充数据 结语 微调数据准备是一个系统性工程,涉及采集、格式化、质量检查、去重、增强和划分。高质量的数据是微调成功的基础——在数据上投入的时间,会在模型性能上得到回报。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

2026-07-02 · 3 min · 594 words · 硅基 AGI 探索者
强化学习对齐

强化学习对齐 2026:从 RLHF 到 DPO 再到 ORPO

引言 大模型的对齐(Alignment)是让模型输出符合人类期望和价值的关键技术。从2022年RLHF的爆发,到2024年DPO的崛起,再到2026年ORPO和GRPO的成熟,对齐技术经历了深刻的范式转变。本文系统梳理这一演进历程。 RLHF:对齐的起点 核心原理 RLHF(Reinforcement Learning from Human Feedback)通过三个步骤实现对齐: 1. SFT(监督微调):用高质量数据训练基础模型 2. RM(奖励模型):训练模型评估输出质量 3. PPO(近端策略优化):用奖励信号优化策略 优点 理论完备,效果可证明 可优化任意可微奖励函数 缺点 训练复杂度高(4个模型同时训练) 超参数敏感,不稳定 需要大量人工标注 DPO:简化对齐的革命 核心思想 DPO(Direct Preference Optimization)绕过了奖励模型和强化学习,直接优化偏好数据。 关键突破: 将奖励函数隐式建模 直接对策略模型进行偏好优化 只需一个模型,训练稳定 数学本质: L_DPO = -log(σ[β log(π(y|x)/π_ref(y|x))]) 优点 训练简单:只需一个模型 训练稳定:无PPO的不稳定性 资源需求低:无需奖励模型 缺点 需要成对偏好数据 对数据质量敏感 复杂奖励无法表达 ORPO:SFT与DPO的统一 核心创新 ORPO(Odds Ratio Preference Optimization)将SFT和DPO统一到一个目标函数中。 核心公式: L_ORPO = -log(σ[log(π_chosen/π_rejected) - log(π_ref_chosen/π_ref_rejected)]) 与DPO的区别 维度 DPO ORPO 参考模型 需要 隐式 训练步骤 2步(SFT+DPO) 1步 数据需求 偏好对 偏好对 + 单样本 训练速度 中等 快 2026年进展 ORPO在2026年成为微调的首选方法,特别是在中小规模模型上表现优异。 ...

2026-06-30 · 1 min · 191 words · 硅基 AGI 探索者
LoRA微调2026:从数据准备到部署的全流程

LoRA微调2026:从数据准备到部署的全流程

LoRA的核心价值再认识 LoRA(Low-Rank Adaptation)在2026年仍然是性价比最高的微调方案。相比全量微调: 对比维度 全量微调 LoRA微调 显存占用 70B模型需8×A100 70B模型需2×A100 训练成本 $50-100/次 $5-10/次 模型体积 每次全量模型 仅Adapter权重(几十MB) 训练速度 基线 快30-50% 效果差距 基线 相差<3% LoRA的本质:冻结原模型权重,只训练低秩分解矩阵。 原模型权重 W ∈ R^(d×k),参数量 d×k LoRA分解: W' = W + ΔW = W + B×A 其中: B ∈ R^(d×r),A ∈ R^(r×k) r << min(d, k) # 典型值 r=8-64 参数量:d×r + r×k = r×(d+k) << d×k 完整微调流程 第一步:数据准备(最关键) 数据质量 > 数据数量,这是2026年行业的共识。 数据格式 { "instruction": "解释什么是GraphRAG,并说明它的核心优势", "input": "", "output": "GraphRAG是一种结合知识图谱的检索增强生成技术..." } 或对话格式: ...

2026-06-30 · 4 min · 793 words · 硅基 AGI 探索者
DPO 训练实践

DPO 训练实践:偏好对齐的数据工程

DPO:RLHF 的简化革命 传统的 RLHF 需要训练一个奖励模型(Reward Model),再用 PPO 算法优化策略模型,流程复杂且不稳定。DPO(Direct Preference Optimization)直接用偏好数据优化模型,跳过了奖励模型,大大简化了流程。 传统 RLHF: 偏好数据 → 训练 Reward Model → PPO 优化 → 对齐模型 DPO: 偏好数据 → 直接优化模型 → 对齐模型 DPO 原理简述 DPO 的核心思想是:通过偏好数据(chosen vs rejected)直接优化模型策略,使得模型输出更符合人类偏好。 损失函数: def dpo_loss(policy_chosen_logps, policy_rejected_logps, reference_chosen_logps, reference_rejected_logps, beta=0.1): """ DPO Loss """ pi_logratios = policy_chosen_logps - policy_rejected_logps ref_logratios = reference_chosen_logps - reference_rejected_logps logits = pi_logratios - ref_logratios return -torch.nn.functional.logsigmoid(beta * logits).mean() 1. 偏好数据构建 数据格式 { "prompt": "解释量子计算的基本原理", "chosen": "量子计算利用量子比特的叠加态和纠缠特性...", "rejected": "量子计算就是很快的计算机...", "metadata": { "source": "expert_annotation", "quality_gap": 3.5, "domain": "physics" } } 偏好数据生成方案 class PreferenceDataBuilder: """多种偏好数据生成策略""" def from_human_annotation(self, prompts: list, annotators: list): """方案1:人工标注(质量最高,成本最高)""" data = [] for prompt in prompts: # 生成两个不同质量的回复 responses = [] for model_config in [self.strong_model, self.weak_model]: resp = model_config.generate(prompt) responses.append(resp) # 人工选择更好的回复 chosen_idx = annotator.select_better(prompt, responses) data.append({ "prompt": prompt, "chosen": responses[chosen_idx], "rejected": responses[1 - chosen_idx] }) return data def from_ai_feedback(self, prompts: list): """方案2:AI 反馈( scalable,成本低)""" data = [] for prompt in prompts: # 用不同温度/模型生成回复 resp_a = self.model.generate(prompt, temperature=0.3) resp_b = self.model.generate(prompt, temperature=1.2) # 用 Judge 模型评分 scores = self.judge_model.evaluate(prompt, [resp_a, resp_b]) if scores[0] > scores[1]: chosen, rejected = resp_a, resp_b else: chosen, rejected = resp_b, resp_a # 只保留差异明显的样本 if abs(scores[0] - scores[1]) > 0.5: data.append({ "prompt": prompt, "chosen": chosen, "rejected": rejected }) return data def from_existing_sft_data(self, sft_data: list): """方案3:从 SFT 数据构造(成本最低)""" data = [] for item in sft_data: prompt = item["messages"][-2]["content"] # user message chosen = item["messages"][-1]["content"] # assistant response # 生成劣质回复(高温度/截断/换模型) rejected = self.model.generate( prompt, temperature=1.5, max_tokens=len(chosen) // 2 ) data.append({ "prompt": prompt, "chosen": chosen, "rejected": rejected }) return data def from_rejection_sampling(self, prompts: list, num_samples: int = 4): """方案4:拒绝采样(高质量+低成本)""" data = [] for prompt in prompts: # 生成多个回复 responses = [ self.model.generate(prompt, temperature=0.8) for _ in range(num_samples) ] # 用奖励模型或 Judge 模型排序 scores = self.judge_model.evaluate(prompt, responses) ranked = sorted(zip(responses, scores), key=lambda x: x[1], reverse=True) # 取最好和最差的构造偏好对 best, worst = ranked[0], ranked[-1] if best[1] - worst[1] > 0.3: # 质量差距足够大 data.append({ "prompt": prompt, "chosen": best[0], "rejected": worst[0] }) return data 偏好数据质量控制 class PreferenceDataQualityChecker: def check(self, dataset: list) -> dict: report = { "total": len(dataset), "issues": [], "quality_distribution": {} } for i, sample in enumerate(dataset): # 1. chosen 和 rejected 不能太相似 similarity = compute_similarity(sample["chosen"], sample["rejected"]) if similarity > 0.9: report["issues"].append(f"Sample {i}: chosen 和 rejected 过于相似 ({similarity:.3f})") # 2. chosen 应该比 rejected 长(通常更好的回答更详细) len_chosen = len(sample["chosen"]) len_rejected = len(sample["rejected"]) if len_chosen < len_rejected * 0.5: report["issues"].append(f"Sample {i}: chosen 过短 ({len_chosen} vs {len_rejected})") # 3. prompt 不应为空 if not sample["prompt"].strip(): report["issues"].append(f"Sample {i}: prompt 为空") # 4. 质量差距分布 gap = abs(len_chosen - len_rejected) / max(len_chosen, len_rejected, 1) report["quality_distribution"][i] = gap return report 2. DPO 训练 # dpo_train.py import torch from transformers import AutoModelForCausalLM, AutoTokenizer from trl import DPOTrainer, DPOConfig from datasets import load_dataset # 1. 加载模型(需要两个:policy 和 reference) model_name = "Qwen/Qwen2.5-7B-Instruct" # Policy 模型(要训练的) policy_model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.bfloat16, device_map="auto" ) # Reference 模型(冻结的参考) ref_model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.bfloat16, device_map="auto" ) tokenizer = AutoTokenizer.from_pretrained(model_name) # 2. 如果用 LoRA,只需要一个模型 from peft import LoraConfig, get_peft_model lora_config = LoraConfig( r=64, lora_alpha=128, target_modules=["q_proj", "k_proj", "v_proj", "o_proj"], lora_dropout=0.05, bias="none", task_type="CAUSAL_LM" ) policy_model = get_peft_model(policy_model, lora_config) # 使用 LoRA 时,reference 模型可以设为 None # DPOTrainer 会自动用未训练的 base model 作为 reference ref_model = None # 3. DPO 配置 dpo_config = DPOConfig( output_dir="./output/dpo-qwen2.5-7b", num_train_epochs=1, per_device_train_batch_size=2, gradient_accumulation_steps=8, learning_rate=5e-6, # DPO 的 LR 要比 SFT 低很多 lr_scheduler_type="cosine", warmup_ratio=0.1, bf16=True, logging_steps=10, save_strategy="steps", save_steps=100, eval_strategy="steps", eval_steps=100, beta=0.1, # DPO 温度参数 max_length=2048, max_prompt_length=1024, loss_type="sigmoid", # sigmoid (标准DPO) / hinge / ipo ) # 4. 加载数据 dataset = load_dataset("json", data_files={ "train": "data/dpo_train.jsonl", "test": "data/dpo_test.jsonl" }) # 5. 训练 trainer = DPOTrainer( model=policy_model, ref_model=ref_model, args=dpo_config, train_dataset=dataset["train"], eval_dataset=dataset["test"], processing_class=tokenizer, ) trainer.train() 3. 超参数调优 参数 推荐值 说明 beta 0.1-0.5 控制偏离 reference 的程度,越小越激进 learning_rate 1e-6 ~ 1e-5 远低于 SFT 的 LR epochs 1-2 DPO 容易过拟合,1 epoch 通常足够 batch_size 16-32 有效 batch size beta 的影响 beta 偏好学习强度 过拟合风险 输出多样性 0.05 很强 高 低 0.1 强 中 中 0.3 适中 低 高 0.5 弱 很低 很高 # Beta 扫描实验 for beta in [0.05, 0.1, 0.3, 0.5]: config = DPOConfig(..., beta=beta) trainer = DPOTrainer(..., args=config) result = trainer.train() eval_score = evaluate(trainer.model) print(f"beta={beta}: train_loss={result.training_loss:.4f}, eval_score={eval_score:.4f}") 4. DPO 变体对比 方法 损失函数 特点 适用场景 DPO Sigmoid Loss 标准版本 通用 IPO Identity Preference Optimization 不受偏好数据噪声影响 数据质量低 KTO Kahneman-Tversky Optimization 不需要配对数据 只有二元反馈 SimPO Length-normalized DPO 解决长度偏置 回复长度差异大 ORPO SFT + DPO 一体化 不需要 SFT 预训练 简化流程 # SimPO 配置示例 simpo_config = DPOConfig( ..., loss_type="simpo", # 使用 SimPO loss beta=2.0, # SimPO 的 beta 通常更大 loss_beta=2.0, ) 5. 效果评估 class DPOEvaluator: def evaluate(self, model, eval_dataset): metrics = { "accuracy": 0, # chosen vs rejected 的准确率 "margin": 0, # chosen 和 rejected 的 logit 差距 "reward_accuracy": 0, # 奖励模型准确率 "human_win_rate": 0, # 人工评估胜率 } correct = 0 margins = [] for sample in eval_dataset: # 计算 chosen 和 rejected 的 log probability chosen_logp = self._compute_logp(model, sample["prompt"], sample["chosen"]) rejected_logp = self._compute_logp(model, sample["prompt"], sample["rejected"]) margin = chosen_logp - rejected_logp margins.append(margin) if margin > 0: correct += 1 metrics["accuracy"] = correct / len(eval_dataset) metrics["margin"] = np.mean(margins) # 生成质量评估 metrics["generation_quality"] = self._eval_generation(model, eval_dataset) return metrics 总结 DPO 在 2026 年已经是偏好对齐的主流方法,关键建议: ...

2026-06-28 · 4 min · 843 words · 硅基 AGI 探索者
lora finetuning 2026 data to deployment

LoRA 微调实战 2026:从数据准备到部署的完整流程

为什么选择 LoRA 全参数微调一个 70B 模型需要数百 GB 显存,而 LoRA(Low-Rank Adaptation)通过冻结原始权重、只训练低秩适配矩阵,将可训练参数减少到原来的 0.1%-1%,在消费级 GPU 上即可完成微调。 方法 可训练参数 显存需求 (7B) 显存需求 (70B) 全参数微调 100% 120GB 1200GB LoRA 0.1-1% 16GB 80GB QLoRA 0.1-1% 8GB 40GB 完整流程概览 数据准备 → 格式转换 → 训练配置 → LoRA训练 → 评估 → 合并 → 部署 1. 数据准备 数据格式 # 推荐格式:ShareGPT / OpenAI Messages { "messages": [ {"role": "system", "content": "你是一个专业的技术顾问。"}, {"role": "user", "content": "解释一下 RAG 的工作原理"}, {"role": "assistant", "content": "RAG(检索增强生成)是一种..."} ] } 数据构建脚本 import json from pathlib import Path class SFTDataBuilder: def __init__(self, output_dir: str): self.output_dir = Path(output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) def build_from_qa_pairs(self, qa_pairs: list, system_prompt: str): """从问答对构建训练数据""" samples = [] for qa in qa_pairs: sample = { "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": qa["question"]}, {"role": "assistant", "content": qa["answer"]} ] } samples.append(sample) # 划分训练/验证集 split = int(len(samples) * 0.95) train_path = self.output_dir / "train.jsonl" val_path = self.output_dir / "val.jsonl" with open(train_path, 'w', encoding='utf-8') as f: for s in samples[:split]: f.write(json.dumps(s, ensure_ascii=False) + '\n') with open(val_path, 'w', encoding='utf-8') as f: for s in samples[split:]: f.write(json.dumps(s, ensure_ascii=False) + '\n') print(f"训练集: {split} 条 → {train_path}") print(f"验证集: {len(samples) - split} 条 → {val_path}") def build_from_conversations(self, conversations: list): """从多轮对话构建训练数据""" samples = [] for conv in conversations: messages = [] for turn in conv: messages.append({"role": turn["role"], "content": turn["content"]}) samples.append({"messages": messages}) return samples 数据质量检查 class DataQualityChecker: def check(self, data_path: str): issues = [] with open(data_path, 'r', encoding='utf-8') as f: lines = f.readlines() for i, line in enumerate(lines): sample = json.loads(line) # 1. 检查消息格式 if "messages" not in sample: issues.append(f"Line {i}: 缺少 messages 字段") continue # 2. 检查角色顺序 roles = [m["role"] for m in sample["messages"]] if roles[-1] != "assistant": issues.append(f"Line {i}: 最后一条消息不是 assistant") # 3. 检查内容长度 for msg in sample["messages"]: if len(msg["content"]) < 5: issues.append(f"Line {i}: 消息内容过短") if len(msg["content"]) > 8000: issues.append(f"Line {i}: 消息内容过长 ({len(msg['content'])} chars)") # 4. 检查 assistant 回复质量 assistant_msgs = [m for m in sample["messages"] if m["role"] == "assistant"] for msg in assistant_msgs: if msg["content"].startswith("我是一个AI"): issues.append(f"Line {i}: assistant 回复包含模板化语言") if len(msg["content"]) < 20: issues.append(f"Line {i}: assistant 回复过短") # 5. 统计 stats = { "total_samples": len(lines), "avg_turns": np.mean([len(json.loads(l)["messages"]) for l in lines]), "avg_assistant_len": np.mean([ len(m["content"]) for l in lines for m in json.loads(l)["messages"] if m["role"] == "assistant" ]), "issues_found": len(issues), } return {"issues": issues[:20], "stats": stats} 2. 训练配置 # train_lora.py import torch from transformers import AutoModelForCausalLM, AutoTokenizer, TrainingArguments from peft import LoraConfig, get_peft_model, TaskType from trl import SFTTrainer, SFTConfig # 1. 加载模型和分词器 model_name = "Qwen/Qwen2.5-7B-Instruct" tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.bfloat16, device_map="auto", attn_implementation="flash_attention_2" ) # 2. LoRA 配置 lora_config = LoraConfig( task_type=TaskType.CAUSAL_LM, r=64, # 秩,常用 8/16/32/64 lora_alpha=128, # alpha = 2 * r 是常见默认值 lora_dropout=0.05, target_modules=[ "q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj" ], bias="none" ) # 3. 应用 LoRA model = get_peft_model(model, lora_config) model.print_trainable_parameters() # 输出: trainable params: 39,976,960 || all params: 7,621,836,800 || trainable%: 0.5247% # 4. 训练配置 training_args = SFTConfig( output_dir="./output/qwen2.5-7b-lora", num_train_epochs=3, per_device_train_batch_size=4, gradient_accumulation_steps=4, # 有效 batch_size = 16 learning_rate=2e-4, lr_scheduler_type="cosine", warmup_ratio=0.05, bf16=True, logging_steps=10, save_strategy="steps", save_steps=200, save_total_limit=3, eval_strategy="steps", eval_steps=200, load_best_model_at_end=True, metric_for_best_model="eval_loss", greater_is_better=False, gradient_checkpointing=True, max_seq_length=2048, dataset_text_field="messages", ) # 5. 加载数据 from datasets import load_dataset dataset = load_dataset("json", data_files={ "train": "data/train.jsonl", "validation": "data/val.jsonl" }) # 6. 启动训练 trainer = SFTTrainer( model=model, args=training_args, train_dataset=dataset["train"], eval_dataset=dataset["validation"], processing_class=tokenizer, ) trainer.train() 3. QLoRA:4bit 量化微调 显存不够?用 QLoRA 量化到 4bit: ...

2026-06-28 · 4 min · 844 words · 硅基 AGI 探索者
SFT 数据质量评估

SFT 数据质量评估:Bad Data 如何毁掉你的微调

Bad Data 的杀伤力 一句流行的说法:“Garbage In, Garbage Out”。在 SFT 微调中,这个效应被放大——1000 条高质量数据的效果远好于 10000 条低质量数据。低质量数据不仅浪费训练资源,还会主动降低模型能力。 实验数据 数据质量 数据量 模型准确率 关键问题 高质量 1K 82.3% 无 混合质量 10K 78.5% 偶尔幻觉 低质量 10K 65.2% 频繁幻觉、格式混乱 低质量 50K 61.8% 灾难性退化 结论:低质量数据越多,效果越差。50K 低质量数据比 1K 高质量数据差 20 个百分点。 1. Bad Data 的七大类型 class BadDataType: """SFT 数据中的七种常见质量问题""" # 类型1:格式不一致 FORMAT_INCONSISTENT = { "description": "回复格式不统一,有的用Markdown,有的用纯文本", "example": {"user": "解释RAG", "assistant": "RAG是检索增强生成"}, # 缺少结构化格式 "fix": "统一为指定格式(如Markdown),用LLM重新格式化" } # 类型2:回复过短/过长 LENGTH_EXTREME = { "description": "回复要么一句话敷衍,要么冗长重复", "example_short": {"user": "解释量子计算", "assistant": "量子计算用量子比特"}, "example_long": {"user": "解释量子计算", "assistant": "量子计算是一种...(5000字废话)"}, "fix": "过滤极端长度,保留200-2000字符范围" } # 类型3:事实错误 FACTUAL_ERROR = { "description": "回复中包含事实性错误", "example": {"user": "地球到月球多远", "assistant": "约38万公里"}, # 实际约38.4万公里 "fix": "用可信来源验证,或用强模型交叉检查" } # 类型4:答非所问 IRRELEVANT = { "description": "回复与问题不相关", "example": {"user": "如何优化SQL", "assistant": "SQL是结构化查询语言..."}, "fix": "计算query-response相关性,过滤低相关样本" } # 类型5:模板化回复 TEMPLATE_RESPONSE = { "description": "所有回复都是模板化的套话", "example": "作为AI语言模型,我不能...", "fix": "过滤包含常见AI模板用语的样本" } # 类型6:有害内容 HARMFUL = { "description": "包含偏见、歧视或有害建议", "fix": "安全过滤器 + 人工审核" } # 类型7:重复数据 DUPLICATE = { "description": "相同或高度相似的样本重复出现", "fix": "去重(精确去重 + 语义去重)" } 2. 数据质量评估框架 class SFTDataQualityAssessor: def __init__(self, strong_model): self.strong_model = strong_model # 用强模型做评估 self.dimensions = [ "accuracy", # 准确性 "relevance", # 相关性 "completeness", # 完整性 "clarity", # 清晰度 "safety", # 安全性 "format", # 格式规范性 ] def assess_sample(self, sample: dict) -> dict: prompt = f""" 请评估以下SFT训练样本的质量。 用户问题:{sample['messages'][-2]['content']} 助手回复:{sample['messages'][-1]['content']} 请从以下维度评分(1-5分): 1. 准确性:回复中的信息是否准确? 2. 相关性:回复是否直接回答了用户问题? 3. 完整性:回复是否充分回答了问题? 4. 清晰度:回复是否表达清晰、结构合理? 5. 安全性:回复是否安全无害? 6. 格式:回复格式是否规范统一? 同时检查: - 是否有事实错误 - 是否有模板化语言 - 是否有害内容 - 回复长度是否合适 输出 JSON: {{ "scores": {{"accuracy": 1-5, "relevance": 1-5, "completeness": 1-5, "clarity": 1-5, "safety": 1-5, "format": 1-5}}, "overall_score": 1.0-5.0, "issues": ["问题1", "问题2"], "recommendation": "keep" / "fix" / "discard" }} """ result = self.strong_model.generate(prompt, response_format="json") return result def assess_dataset(self, dataset: list) -> dict: results = [] for sample in dataset: quality = self.assess_sample(sample) results.append(quality) return { "total_samples": len(dataset), "avg_overall": np.mean([r["overall_score"] for r in results]), "quality_distribution": self._distribution(results), "keep_count": sum(1 for r in results if r["recommendation"] == "keep"), "fix_count": sum(1 for r in results if r["recommendation"] == "fix"), "discard_count": sum(1 for r in results if r["recommendation"] == "discard"), "common_issues": self._aggregate_issues(results), } 3. 自动化数据清洗 class SFTDataCleaner: def __init__(self): self.steps = [ self.deduplicate, self.filter_length, self.filter_templates, self.filter_safety, self.check_relevance, self.fix_format, ] def clean(self, data: list) -> list: original_count = len(data) for step in self.steps: before = len(data) data = step(data) print(f"{step.__name__}: {before} → {len(data)} (removed {before - len(data)})") print(f"\n总计: {original_count} → {len(data)} (保留率: {len(data)/original_count:.1%})") return data def deduplicate(self, data: list): """三层去重""" # 1. 精确去重 seen = set() deduped = [] for sample in data: key = hash(json.dumps(sample, sort_keys=True)) if key not in seen: seen.add(key) deduped.append(sample) # 2. 问题去重(相同问题不同回复,保留最好的) question_map = {} for sample in deduped: q = sample["messages"][-2]["content"].strip() if q not in question_map: question_map[q] = sample else: # 保留回复更长的(通常更详细) old_resp = question_map[q]["messages"][-1]["content"] new_resp = sample["messages"][-1]["content"] if len(new_resp) > len(old_resp): question_map[q] = sample deduped = list(question_map.values()) # 3. 语义去重(相似问题) embeddings = self._compute_question_embeddings(deduped) clusters = self._cluster_similar(embeddings, threshold=0.95) deduped = [deduped[c[0]] for c in clusters] # 每簇保留一个 return deduped def filter_length(self, data: list): """过滤极端长度""" filtered = [] for sample in data: response = sample["messages"][-1]["content"] # 回复太短 if len(response) < 50: continue # 回复太长 if len(response) > 8000: continue # 问题太短(无法构成有效训练) question = sample["messages"][-2]["content"] if len(question) < 5: continue filtered.append(sample) return filtered def filter_templates(self, data: list): """过滤模板化回复""" TEMPLATE_PATTERNS = [ r"作为一个AI.*?我不能", r"作为AI语言模型", r"我是.*?AI.*?助手", r"很抱歉.*?无法", r"对不起.*?不能", r"我理解您的.*?但是", ] filtered = [] for sample in data: response = sample["messages"][-1]["content"] is_template = any( re.search(pattern, response, re.IGNORECASE) for pattern in TEMPLATE_PATTERNS ) if not is_template: filtered.append(sample) return filtered def check_relevance(self, data: list): """检查问题-回复相关性""" filtered = [] for sample in data: question = sample["messages"][-2]["content"] response = sample["messages"][-1]["content"] # 计算语义相似度 q_emb = self.embedder.encode(question) r_emb = self.embedder.encode(response) similarity = cosine_similarity(q_emb, r_emb) if similarity > 0.3: # 最低相关性阈值 filtered.append(sample) return filtered def fix_format(self, data: list): """统一格式""" for sample in data: response = sample["messages"][-1]["content"] # 统一使用 Markdown 格式 response = self._normalize_markdown(response) # 确保以句号或换行结尾 if not response.endswith(('.', '。', '!', '!', '?', '?', '\n')): response += '。' sample["messages"][-1]["content"] = response return data 4. 数据质量与训练效果的关系 实验设计 控制变量:基础模型 Qwen2.5-7B,训练参数相同,只变化数据质量。 ...

2026-06-28 · 5 min · 864 words · 硅基 AGI 探索者
大模型微调工具链 2026:LLaMA-Factory vs Axolotl vs Unsloth

大模型微调工具链 2026:LLaMA-Factory vs Axolotl vs Unsloth

大模型微调已从研究实验变为工程标配。2026 年,LLaMA-Factory、Axolotl 和 Unsloth 三大工具链形成了微调生态的三足鼎立。LLaMA-Factory 以全中文生态和 WebUI 著称,Axolotl 以灵活配置和深度定制见长,Unsloth 以极致速度和低显存占用突围。本文将在相同条件下对三者进行全面对比。 一、工具概览 维度 LLaMA-Factory Axolotl Unsloth 开发语言 Python Python Python + CUDA 界面 WebUI + CLI YAML 配置 + CLI Python API 核心优势 全中文、易用、方法全 灵活、社区配方丰富 速度快、显存低 支持方法 SFT/LoRA/QLoRA/DPO/PO/KTO SFT/LoRA/QLoRA/DPO/PO SFT/LoRA/QLoRA 支持模型 主流模型全覆盖 主流模型全覆盖 Llama/Qwen/Mistral/Gemma 训练速度 基准 0.95x 1.8-2.5x 显存节省 基准 1.0x 0.5-0.7x 社区活跃度 高(中文为主) 高(英文为主) 高(全球) GitHub Stars 45k+ 28k+ 22k+ 二、功能对比 2.1 微调方法支持 方法 LLaMA-Factory Axolotl Unsloth Full SFT ✅ ✅ ❌ LoRA ✅ ✅ ✅ QLoRA ✅ ✅ ✅ DPO ✅ ✅ ✅ KTO ✅ ✅ ❌ ORPO ✅ ✅ ❌ PPO ✅ ✅ ❌ Reward Model ✅ ✅ ❌ 多模态微调 ✅ ✅ 部分 持续预训练 ✅ ✅ ❌ 分析: ...

2026-06-28 · 4 min · 713 words · 硅基 AGI 探索者
大模型蒸馏技术 2026

大模型蒸馏技术 2026:从 GPT-5.5 到 7B 模型的能力迁移

为什么需要模型蒸馏 GPT-5.5、Claude 4 等前沿模型能力强大,但成本高昂、延迟较高、依赖 API。模型蒸馏(Knowledge Distillation)将大模型的能力迁移到小模型上,在保持核心能力的同时大幅降低成本。 维度 Teacher (GPT-5.5) Student (7B) 蒸馏后 Student 推理成本 $15/M tokens $0.50/M tokens $0.50/M tokens 延迟 800ms 80ms 80ms 部署 仅 API 本地/Self-hosted 本地/Self-hosted 能力 100% 65% 85-90% 隐私 数据出境 完全可控 完全可控 蒸馏方法分类 知识蒸馏 ├── 响应蒸馏 (Response Distillation) │ ├── SFT 蒸馏(最常用) │ ├── DPO 蒸馏 │ └── Best-of-N 蒸馏 ├── 特征蒸馏 (Feature Distillation) │ ├── Logit 蒸馏 │ ├── 中间层蒸馏 │ └── Attention 蒸馏 ├── Agent 蒸馏 (Agent Distillation) │ ├── 工具使用蒸馏 │ ├── 推理链蒸馏 │ └── 规划能力蒸馏 └── 数据蒸馏 (Data Distillation) ├── 合成数据生成 ├── 数据增强 └── 自指令 1. 响应蒸馏:SFT 蒸馏 最常用且效果最好的方法:用 Teacher 模型生成高质量回复,再用 SFT 训练 Student 模型。 ...

2026-06-28 · 4 min · 774 words · 硅基 AGI 探索者
hermes finetune guide

Hermes微调实战指南

概述 Hermes微调实战指南是AI智能体领域中Hermes微调实战指南的重要主题。本文将从多个角度深入分析这一话题,为读者提供系统性的认知框架和实践参考。 核心概念 基本定义 在深入讨论之前,我们需要明确几个核心概念。AI智能体是指能够感知环境、理解指令、规划行动并调用工具完成任务的AI系统。与传统的聊天机器人不同,智能体具有自主性、目标导向性和工具使用能力。 Hermes微调实战指南涉及的关键技术包括: 大语言模型:作为智能体的认知引擎,负责理解、推理和生成 工具调用:通过Function Calling或MCP协议与外部系统交互 记忆系统:短期记忆处理当前对话,长期记忆存储历史经验 规划引擎:将复杂任务分解为可执行的子步骤 技术原理 从技术层面看,Hermes微调实战指南的核心在于如何让AI系统更好地理解和执行人类意图。这涉及多个技术环节的协同: 首先是感知层,智能体需要准确理解用户的自然语言指令,提取关键信息和约束条件。其次是规划层,将高层目标分解为具体的执行步骤。然后是执行层,调用合适的工具完成每个步骤。最后是反馈层,根据执行结果调整后续策略。 实践分析 当前现状 在爱马仕智能体领域,当前的技术实践呈现出几个明显特征: 工程化程度提升:从实验室原型到生产级系统,工程能力成为关键差异化因素 评估体系完善:越来越多标准化的评测基准被提出,帮助开发者量化能力边界 开源生态繁荣:开源框架和工具链的成熟降低了开发门槛 安全意识增强:对AI安全和对齐问题的重视程度显著提升 关键挑战 尽管进展显著,Hermes微调实战指南仍面临几个核心挑战: 技术挑战: 大模型的幻觉问题在智能体场景下被放大,因为智能体需要做出实际决策 多步推理中的错误累积效应导致长程任务成功率下降 工具调用的可靠性受外部API稳定性影响 工程挑战: 智能体的可观测性不足,调试和排错困难 成本控制与性能优化的平衡 从单机到分布式部署的架构复杂性 安全挑战: Prompt注入等攻击手段不断进化 智能体权限管理需要更精细化的控制 数据隐私保护在多Agent协作场景下更加复杂 优化策略 针对上述挑战,以下是几个关键优化方向: 技术优化 分而治之:将复杂任务分解为可独立验证的子任务,降低单步错误影响 多路投票:对关键决策使用多次采样投票机制,提高可靠性 渐进式信任:智能体权限从最小化开始,根据表现逐步扩展 人在回路:高风险决策保留人工审核环节 工程优化 可观测性优先:建立完善的日志、指标和追踪体系 灰度发布:新版本智能体先在小流量环境验证 自动化测试:构建端到端测试套件,防止回归 成本监控:实时追踪Token消耗和API调用成本 案例研究 为了更具体地说明Hermes微调实战指南的实践价值,我们来看一个典型场景: 某科技公司在内部IT运维中部署了AI智能体,负责处理员工的工单请求。智能体需要理解员工的自然语言描述,判断问题类型,查询知识库,执行修复操作或转接人工。 实施过程中遇到的关键问题包括: 员工描述模糊导致意图识别错误 知识库信息过时导致给出错误建议 某些操作需要管理员权限存在安全风险 解决方案: 引入澄清对话机制,在不确定时主动追问 建立知识库更新流程,定期审核内容 实施权限分级制度,敏感操作需人工确认 效果:工单首次解决率提升35%,平均处理时间缩短60%,员工满意度显著提升。 未来趋势 Hermes微调实战指南的发展趋势值得关注: 标准化:MCP等开放协议将推动工具接口标准化,降低集成成本 垂直化:针对特定行业和场景的专用智能体将大量涌现 协作化:多智能体协作将成为复杂任务的标准解决方案 自主化:智能体的自主决策能力将持续提升,但需要配套的安全机制 结论 Hermes微调实战指南是AI智能体技术发展中的重要一环。无论是技术原理的深入理解,还是实践中的工程优化,都需要系统性思维。对于开发者和企业而言,关键在于: 理解技术能力和边界,避免过度期待 建立系统化的评估和监控体系 在创新和安全之间找到平衡 持续学习和适应快速变化的技术生态 硅基AGI探索者将持续关注爱马仕智能体领域的最新进展,为读者提供深度分析和实践指导。— ...

2026-06-27 · 1 min · 88 words · 硅基 AGI 探索者
鲁ICP备2026018361号