LLM训练

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年的微调数据准备已经形成了一套系统化的最佳实践。 数据采集 多源数据融合 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论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 高质量的训练数据是LLM能力的上限。但高质量数据的获取成本高昂、数量有限。数据增强——通过变换、生成、筛选等方式扩充数据——是突破数据瓶颈的关键手段。 一、数据增强方法 1.1 文本变换 class TextAugmentor: def synonym_replace(self, text, replace_rate=0.1): """同义词替换""" words = text.split() n_replace = int(len(words) * replace_rate) for _ in range(n_replace): idx = random.randint(0, len(words)-1) synonyms = self.get_synonyms(words[idx]) if synonyms: words[idx] = random.choice(synonyms) return " ".join(words) def back_translation(self, text): """回译增强""" # 中文→英文→中文 en = await self.translate(text, "zh", "en") zh = await self.translate(en, "en", "zh") return zh def random_deletion(self, text, delete_rate=0.1): """随机删除""" words = text.split() kept = [w for w in words if random.random() > delete_rate] return " ".join(kept) if kept else words[0] 1.2 AI生成增强 class AIGenerationAugmentor: async def generate_variations(self, instruction, n=5): """生成指令变体""" prompt = f""" 为以下指令生成{n}个不同的表述方式: {instruction} 要求: 1. 保持语义相同 2. 改变表述方式（正式/口语/简洁/详细） 3. 适合不同教育水平的用户 """ return await self.llm.call(prompt) async def generate_edge_cases(self, instruction): """生成边界情况""" prompt = f""" 对于以下指令，生成边界情况的变体: {instruction} 考虑: 1. 极端简短的输入 2. 包含错误的输入 3. 多语言混合输入 4. 模糊/有歧义的输入 """ return await self.llm.call(prompt) 1.3 Self-Instruct生成 class SelfInstructGenerator: async def generate_dataset(self, seed_tasks, target_size=10000): """Self-Instruct生成大规模数据""" dataset = list(seed_tasks) while len(dataset) < target_size: # 1. 采样种子 seeds = random.sample(dataset, min(3, len(dataset))) # 2. 生成新指令 new_instruction = await self.llm.generate( f"基于以下示例生成一个新的不同的指令:\n{seeds}" ) # 3. 质量过滤 if self.passes_quality_check(new_instruction): # 4. 生成回答 response = await self.llm.generate(new_instruction) # 5. 质量验证 if self.verify_quality(new_instruction, response): dataset.append({ "instruction": new_instruction, "response": response }) return dataset 二、合成数据质量 2.1 质量过滤 class SyntheticDataFilter: def filter(self, dataset): filtered = [] for sample in dataset: # 1. 多样性检查 if self.too_similar(sample, filtered): continue # 2. 复杂度检查 if self.too_simple(sample): continue # 3. 事实准确性 if not self.factually_correct(sample): continue # 4. 格式规范 if not self.well_formatted(sample): continue filtered.append(sample) return filtered 2.2 去偏 class SyntheticDataDebiaser: async def debias(self, dataset): """去除合成数据中的偏见""" # 1. 分析分布 distributions = self.analyze_distributions(dataset) # 2. 识别偏差 biases = self.identify_biases(distributions) # 3. 补充不足 for bias in biases: additional = await self.generate_compensating_data(bias) dataset.extend(additional) # 4. 重新平衡 dataset = self.rebalance(dataset) return dataset 三、特定任务增强 3.1 代码数据增强 class CodeAugmentor: async def augment_code(self, code_snippet): """代码数据增强""" augmentations = [] # 1. 变量重命名 augmentations.append(await self.rename_variables(code_snippet)) # 2. 注释添加/修改 augmentations.append(await self.add_comments(code_snippet)) # 3. 等价重构 augmentations.append(await self.refactor(code_snippet)) # 4. 语言转换 augmentations.append(await self.translate_language(code_snippet, "Python", "JavaScript")) return augmentations 3.2 推理数据增强 class ReasoningAugmentor: async def generate_reasoning_chains(self, question, answer): """生成多种推理路径""" prompt = f""" 问题: {question} 答案: {answer} 请生成3种不同的推理路径来到达这个答案: 1. 直接推理路径 2. 反证法路径 3. 类比推理路径 """ return await self.llm.call(prompt) 四、数据配比 class AugmentedDataMixer: def mix(self, real_data, synthetic_data, ratio=0.3): """混合真实和合成数据""" # 合成数据占比不应过高 n_synthetic = int(len(real_data) * ratio / (1 - ratio)) synthetic_sample = random.sample(synthetic_data, min(n_synthetic, len(synthetic_data))) mixed = real_data + synthetic_sample random.shuffle(mixed) return mixed 五、评估合成数据质量 class SyntheticDataEvaluator: async def evaluate(self, real_data, synthetic_data): metrics = { "diversity": self.compute_diversity(synthetic_data), "fidelity": await self.compute_fidelity(real_data, synthetic_data), "novelty": self.compute_novelty(real_data, synthetic_data), "utility": await self.compute_utility(real_data, synthetic_data) } # 效用测试：用合成数据训练，在真实数据上测试 model = train(synthetic_data) metrics["downstream_performance"] = evaluate(model, real_data) return metrics 结语 数据增强是突破数据瓶颈的有效手段。2026年的趋势是"AI生成数据训练AI"——用强模型生成高质量数据来训练弱模型，实现知识蒸馏。 ...

引言 世界在变，知识在更新。一个训练于2025年的模型不知道2026年的新闻。如何让模型"持续学习"新知识，同时不忘记旧知识？ 这就是持续学习（Continual Learning）要解决的核心问题。 一、挑战：灾难性遗忘 # 灾难性遗忘示例 model = train_on_task_A(model, data_A) # 学会任务A model = train_on_task_B(model, data_B) # 学会任务B，但忘了任务A 缓解策略 class ContinualLearning: # 策略1: 经验回放 def replay_based(self, new_data, old_data_sample): """混入旧数据""" mixed = new_data + old_data_sample return self.train(mixed) # 策略2: 弹性权重巩固（EWC） def ewc(self, model, new_data, old_params, fisher_matrix): """EWC正则化""" for name, param in model.named_parameters(): loss = task_loss + lambda_ * (fisher_matrix[name] * (param - old_params[name])**2).sum() # 策略3: LoRA适配器 def lora_per_task(self, base_model, task_data): """每个任务一个LoRA适配器""" lora = LoRA(r=8) lora.train(task_data) return lora # base_model不变 二、知识更新方法 2.1 RAG优先 # 对于事实性知识更新，RAG通常是更好的选择 # 不需要修改模型参数，只需更新知识库 2.2 增量微调 class IncrementalFineTuner: async def incremental_update(self, model, new_knowledge): """增量知识更新""" # 1. 构建增量数据 incremental_data = self.format_knowledge(new_knowledge) # 2. 混入旧数据（防遗忘） replay_data = self.sample_old_data(ratio=0.3) train_data = incremental_data + replay_data # 3. 小学习率微调 config = SFTConfig( learning_rate=1e-6, # 比初始SFT小10倍 num_train_epochs=1, ) return self.train(model, train_data, config) 2.3 多LoRA管理 class MultiLoRAManager: def __init__(self, base_model): self.base_model = base_model self.lora_adapters = {} # {domain: lora_adapter} async def update_domain(self, domain, new_data): """更新特定领域的LoRA""" if domain in self.lora_adapters: # 在现有LoRA基础上继续训练 lora = self.lora_adapters[domain] else: # 创建新LoRA lora = LoRA(r=8) lora.train(new_data) self.lora_adapters[domain] = lora async def generate(self, prompt, domain=None): """生成时选择合适的LoRA""" if domain and domain in self.lora_adapters: self.base_model.load_adapter(self.lora_adapters[domain]) return await self.base_model.generate(prompt) 三、评估 class ContinualLearningEvaluator: async def evaluate(self, model, old_benchmarks, new_benchmarks): """评估持续学习效果""" results = { "old_performance": {}, # 旧任务性能（遗忘程度） "new_performance": {}, # 新任务性能（学习效果） "transfer": {} # 知识迁移效果 } for bench in old_benchmarks: results["old_performance"][bench] = await run_benchmark(model, bench) for bench in new_benchmarks: results["new_performance"][bench] = await run_benchmark(model, bench) # 遗忘率 forgetting = 1 - (results["old_performance"]["avg"] / baseline_old_performance) results["forgetting_rate"] = forgetting return results 四、生产实践 4.1 更新策略 事实性知识更新 → RAG（不修改模型） 领域适配 → LoRA微调 能力提升 → SFT + DPO 紧急修正 → 小数据快速微调 4.2 版本管理 class ModelVersionManager: def __init__(self): self.versions = {} def save_version(self, model, version_id, metadata): """保存模型版本""" self.versions[version_id] = { "model": model, "metadata": metadata, "timestamp": time.time(), "performance": metadata.get("performance", {}) } def rollback(self, version_id): """回滚到之前的版本""" return self.versions[version_id]["model"] 4.3 监控 # 持续学习监控指标 metrics = { "new_task_accuracy": "新任务的准确率", "old_task_accuracy": "旧任务的准确率（遗忘指标）", "general_capability": "通用能力（不应下降）", "safety_score": "安全分数（不应下降）", "latency": "推理延迟（不应增加）" } 结语 持续学习是LLM在动态世界中保持有用的关键能力。2026年的最佳实践是"混合策略"——RAG处理事实更新，LoRA处理领域适配，SFT/DPO处理能力提升。 ...