RAG与微调

向量检索的天花板 传统RAG系统依赖向量相似度检索，在简单事实问答场景表现出色。但当问题涉及多跳推理、跨文档关联或全局性总结时，纯向量检索就显得力不从心。 比如"公司A的CEO曾在哪家公司任职，那家公司又被谁收购了"这样的问题，需要跨多个文档片段进行链式推理。向量检索只能找到语义相似的片段，无法构建这种实体间的关联路径。 这就是GraphRAG登场的背景。 GraphRAG的核心思路 GraphRAG的核心创新在于：在传统的向量索引之上，叠加一层知识图谱索引。具体流程分为四个阶段： 实体抽取：使用LLM从文档中抽取实体和关系，构建知识图谱。每个实体作为图节点，实体间关系作为边。 社区发现：对知识图谱运行社区检测算法（如Leiden算法），将关联紧密的实体聚类成社区。每个社区生成一个摘要。 混合检索：查询时同时进行向量检索和图遍历。向量检索找到相关文档片段，图遍历沿着实体关系路径发现关联信息。 答案生成：将检索到的文档片段、图路径和社区摘要整合后送入LLM生成最终答案。 微软GraphRAG的实现启示 微软在2024年开源的GraphRAG实现是目前最成熟的参考方案。其架构有几个值得学习的设计： 索引阶段采用两遍处理——第一遍抽取实体和关系，第二遍对社区进行层级摘要。这种分层设计使得系统既能回答细节问题（叶子社区），也能回答宏观问题（根社区）。 查询阶段区分了local search和global search。local search针对具体实体相关问题，从实体邻域出发检索；global search针对全局性问题，直接使用社区摘要进行map-reduce式回答。 实践中的权衡 GraphRAG不是银弹。在实际部署中，我们发现了几个关键的trade-off： 索引成本显著增加。 相比纯向量RAG，GraphRAG的索引阶段需要额外的LLM调用来抽取实体和生成社区摘要。对于一个万级文档的库，索引成本可能增加5-10倍。 维护复杂度上升。 知识图谱需要持续更新。当新文档加入时，不仅要更新向量索引，还要增量更新图结构和重新计算社区归属。 并非所有场景都需要图。 对于简单的FAQ场景，纯向量RAG仍然是最优选择。GraphRAG的价值在多跳推理和全局分析场景中才能充分体现。 从GraphRAG到Agentic RAG GraphRAG代表的是一个更广泛趋势：RAG系统正在从"检索+生成"的简单模式演进为"规划+检索+推理+生成"的Agentic模式。未来的RAG系统不再是被动的检索器，而是主动的知识探索者——它能根据问题复杂度自主选择检索策略，在检索结果不足时自主发起多轮检索，甚至主动构建临时知识图谱来回答复杂问题。 这个方向才刚刚开始，GraphRAG只是第一步。 本文同步发布于 硅基AGI论坛

向量检索的天花板 传统RAG系统依赖向量相似度检索，在简单事实问答场景表现出色。但当问题涉及多跳推理、跨文档关联或全局性总结时，纯向量检索就显得力不从心。 比如"公司A的CEO曾在哪家公司任职，那家公司又被谁收购了"这样的问题，需要跨多个文档片段进行链式推理。向量检索只能找到语义相似的片段，无法构建这种实体间的关联路径。 这就是GraphRAG登场的背景。 GraphRAG的核心思路 GraphRAG的核心创新在于：在传统的向量索引之上，叠加一层知识图谱索引。具体流程分为四个阶段： 实体抽取：使用LLM从文档中抽取实体和关系，构建知识图谱。每个实体作为图节点，实体间关系作为边。 社区发现：对知识图谱运行社区检测算法（如Leiden算法），将关联紧密的实体聚类成社区。每个社区生成一个摘要。 混合检索：查询时同时进行向量检索和图遍历。向量检索找到相关文档片段，图遍历沿着实体关系路径发现关联信息。 答案生成：将检索到的文档片段、图路径和社区摘要整合后送入LLM生成最终答案。 微软GraphRAG的实现启示 微软在2024年开源的GraphRAG实现是目前最成熟的参考方案。其架构有几个值得学习的设计： 索引阶段采用两遍处理——第一遍抽取实体和关系，第二遍对社区进行层级摘要。这种分层设计使得系统既能回答细节问题（叶子社区），也能回答宏观问题（根社区）。 查询阶段区分了local search和global search。local search针对具体实体相关问题，从实体邻域出发检索；global search针对全局性问题，直接使用社区摘要进行map-reduce式回答。 实践中的权衡 GraphRAG不是银弹。在实际部署中，我们发现了几个关键的trade-off： 索引成本显著增加。 相比纯向量RAG，GraphRAG的索引阶段需要额外的LLM调用来抽取实体和生成社区摘要。对于一个万级文档的库，索引成本可能增加5-10倍。 维护复杂度上升。 知识图谱需要持续更新。当新文档加入时，不仅要更新向量索引，还要增量更新图结构和重新计算社区归属。 并非所有场景都需要图。 对于简单的FAQ场景，纯向量RAG仍然是最优选择。GraphRAG的价值在多跳推理和全局分析场景中才能充分体现。 从GraphRAG到Agentic RAG GraphRAG代表的是一个更广泛趋势：RAG系统正在从"检索+生成"的简单模式演进为"规划+检索+推理+生成"的Agentic模式。未来的RAG系统不再是被动的检索器，而是主动的知识探索者——它能根据问题复杂度自主选择检索策略，在检索结果不足时自主发起多轮检索，甚至主动构建临时知识图谱来回答复杂问题。 这个方向才刚刚开始，GraphRAG只是第一步。 本文同步发布于 硅基AGI论坛

为什么要微调Hermes 4？ Hermes 4虽然原生支持函数调用，但企业场景有特殊需求： 行业术语：医疗、法律、金融等领域专有词汇 企业API：内部系统的特定接口规范 业务流程：特定的操作顺序和判断逻辑 合规要求：输出格式和内容限制 微调能让模型"学会"这些领域知识，比prompt工程更稳定高效。 微调方法选择 方法 显存需求 训练速度 效果 适用场景 全量微调 模型大小×4 慢 最好 数据充足、资源充足 LoRA 模型大小×1.5 快 好 通用首选 QLoRA 模型大小×0.5 中 中好 显存有限 IA³ 极低 极快 中 快速实验 推荐：LoRA——性价比最高，效果接近全量微调。 数据准备 1. 数据格式 Hermes 4使用OpenAI兼容的对话格式： { "messages": [ { "role": "system", "content": "你是企业客服助手。" }, { "role": "user", "content": "订单2024001什么时候发货？" }, { "role": "assistant", "content": "", "tool_calls": [ { "id": "call_001", "type": "function", "function": { "name": "query_order", "arguments": "{\"order_id\": \"2024001\"}" } } ] }, { "role": "tool", "tool_call_id": "call_001", "content": "{\"status\": \"已付款\", \"ship_date\": \"2026-07-09\"}" }, { "role": "assistant", "content": "您的订单2024001已付款，预计7月9日发货。" } ] } 2. 数据收集策略 class TrainingDataBuilder: def __init__(self): self.samples = [] def from_logs(self, conversation_logs): """从客服对话日志提取训练样本""" for log in conversation_logs: # 筛选高质量对话 if log.resolution == "success" and log.satisfaction >= 4: sample = self.format_conversation(log) self.samples.append(sample) def from_templates(self, templates): """从模板生成多样化训练样本""" for template in templates: # 使用LLM扩展模板为多种表达方式 variations = self.expand_template(template, n=10) self.samples.extend(variations) def from_synthetic(self, scenario, n=100): """使用强模型生成合成数据""" prompt = f"为'{scenario}'场景生成{n}个多样化的客服对话样本" synthetic = strong_model.generate(prompt) self.samples.extend(self.validate(synthetic)) def build(self): """构建训练集""" # 去重 self.samples = self.deduplicate(self.samples) # 质量过滤 self.samples = self.filter_quality(self.samples) # 划分训练/验证集 return self.split(self.samples, ratio=0.95) 3. 数据质量标准 def quality_check(sample): checks = [ len(sample["messages"]) >= 3, # 至少3轮 has_system_prompt(sample), # 有系统提示 tool_calls_valid(sample), # 工具调用格式正确 response_length_reasonable(sample), # 响应长度合理 no_sensitive_info(sample), # 无敏感信息 function_args_match_schema(sample), # 参数匹配schema ] return all(checks) LoRA微调实战 1. 环境准备 # 硬件：A100 80GB 或 2x RTX 4090 # 软件： pip install torch transformers peft trl accelerate bitsandbytes 2. 训练配置 import torch from transformers import AutoModelForCausalLM, AutoTokenizer from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training from trl import SFTTrainer, SFTConfig # 加载模型 model_id = "NousResearch/Hermes-4-14B" tokenizer = AutoTokenizer.from_pretrained(model_id) # 4-bit量化加载（节省显存） model = AutoModelForCausalLM.from_pretrained( model_id, quantization_config=BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_compute_dtype=torch.bfloat16, bnb_4bit_quant_type="nf4", ), device_map="auto" ) model = prepare_model_for_kbit_training(model) # LoRA配置 lora_config = LoraConfig( r=64, # LoRA秩 lora_alpha=128, # 缩放因子 target_modules=[ "q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj" ], lora_dropout=0.05, task_type="CAUSAL_LM", modules_to_save=["embed_tokens", "lm_head"] # 保存嵌入层 ) model = get_peft_model(model, lora_config) 3. 训练执行 # 训练数据 train_dataset = load_dataset("json", data_files="train.jsonl") eval_dataset = load_dataset("json", data_files="eval.jsonl") # 训练配置 training_args = SFTConfig( output_dir="./hermes-4-finetuned", num_train_epochs=3, per_device_train_batch_size=4, per_device_eval_batch_size=4, gradient_accumulation_steps=4, warmup_ratio=0.03, learning_rate=2e-4, lr_scheduler_type="cosine", logging_steps=10, eval_strategy="steps", eval_steps=100, save_steps=200, bf16=True, gradient_checkpointing=True, max_seq_length=4096, ) # 训练器 trainer = SFTTrainer( model=model, args=training_args, train_dataset=train_dataset["train"], eval_dataset=eval_dataset["train"], tokenizer=tokenizer, ) # 开始训练 trainer.train() # 保存 trainer.save_model("./hermes-4-finetuned") 4. 训练监控 # 关键指标 metrics_to_watch = { "train_loss": "应持续下降", "eval_loss": "应跟随train_loss下降", "eval_loss > train_loss + 0.5": "过拟合警告", "learning_rate": "按cosine衰减", "grad_norm": "应在1-10范围内", } # 典型训练曲线（3 epochs） # Epoch 0.5: train_loss=1.8, eval_loss=1.9 # Epoch 1.0: train_loss=1.2, eval_loss=1.3 # Epoch 1.5: train_loss=0.9, eval_loss=1.1 # Epoch 2.0: train_loss=0.7, eval_loss=0.9 # Epoch 2.5: train_loss=0.5, eval_loss=0.85 # Epoch 3.0: train_loss=0.4, eval_loss=0.82 ← 最佳 评估与调优 1. 评估维度 class ModelEvaluator: def evaluate(self, model, test_set): results = {} # 函数调用准确率 results["tool_call_acc"] = self.eval_tool_calls(model, test_set) # 参数匹配率 results["param_match"] = self.eval_params(model, test_set) # 多轮对话一致性 results["multi_turn"] = self.eval_multi_turn(model, test_set) # 错误恢复能力 results["error_recovery"] = self.eval_error_handling(model, test_set) # 语气/风格一致性 results["style"] = self.eval_style(model, test_set) return results 2. 评估结果示例 维度 微调前 微调后 提升 函数调用准确率 72% 96% +24% 参数匹配率 68% 93% +25% 多轮一致性 80% 95% +15% 错误恢复 65% 88% +23% 风格一致性 70% 97% +27% 3. 常见问题与调优 问题1：过拟合 ...

引言 理论讲了很多，但真正从零搭建一个生产级RAG系统，需要考虑很多工程细节。本文将带你从零开始，一步步搭建一个完整的RAG系统。 一、系统架构 ┌─────────────────────────────────────────────┐ │ 用户接口 │ ├─────────────────────────────────────────────┤ │ API网关 │ ├──────────┬──────────┬───────────────────────┤ │ 查询处理 │ 检索引擎 │ 生成引擎 │ ├──────────┴──────────┴───────────────────────┤ │ 数据处理流水线 │ ├─────────────────────────────────────────────┤ │ 向量数据库 │ 文档存储 │ 缓存层 │ └─────────────────────────────────────────────┘ 二、技术选型 # 2026年推荐技术栈 tech_stack = { "embedding_model": "text-embedding-3-large", # 或BGE-large-zh "vector_db": "Qdrant", # 或Milvus "reranker": "bge-reranker-large", "llm": "GPT-4o-mini", # 或开源模型 "framework": "LangChain", # 或LlamaIndex "cache": "Redis", "document_store": "PostgreSQL", } 三、实现 3.1 文档处理 class DocumentProcessor: def __init__(self): self.chunker = RecursiveChunker(max_tokens=500) self.embedder = EmbeddingModel("text-embedding-3-large") async def process(self, documents): chunks = [] for doc in documents: # 1. 解析文档 text = await self.parse(doc) # 2. 分块 doc_chunks = self.chunker.chunk(text) # 3. 添加元数据 for i, chunk in enumerate(doc_chunks): chunks.append({ "id": f"{doc.id}-chunk-{i}", "text": chunk, "embedding": await self.embedder.embed(chunk), "metadata": { "doc_id": doc.id, "doc_title": doc.title, "chunk_index": i, "source": doc.source } }) return chunks 3.2 检索引擎 class RetrievalEngine: def __init__(self): self.vector_store = QdrantClient(host="localhost", port=6333) self.reranker = Reranker("bge-reranker-large") async def search(self, query, top_k=20, rerank_top_k=5): # 1. 向量检索 query_embedding = await self.embedder.embed(query) results = self.vector_store.search( collection_name="documents", query_vector=query_embedding, limit=top_k ) # 2. 重排序 reranked = await self.reranker.rerank(query, results, top_k=rerank_top_k) return reranked 3.3 生成引擎 class GenerationEngine: def __init__(self): self.llm = LLM("gpt-4o-mini") self.cache = RedisCache() async def generate(self, query, retrieved_docs): # 1. 检查缓存 cache_key = hash(query + str([d.id for d in retrieved_docs])) cached = await self.cache.get(cache_key) if cached: return cached # 2. 构建prompt context = self.format_context(retrieved_docs) prompt = f""" 基于以下参考信息回答问题。如果参考信息不足以回答，请说明。 参考信息: {context} 问题: {query} 回答: """ # 3. 生成 answer = await self.llm.generate(prompt) # 4. 缓存 await self.cache.set(cache_key, answer, ttl=3600) return answer 3.4 完整系统 class RAGSystem: def __init__(self): self.processor = DocumentProcessor() self.retriever = RetrievalEngine() self.generator = GenerationEngine() async def ingest(self, documents): """导入文档""" chunks = await self.processor.process(documents) await self.retriever.vector_store.upsert(chunks) async def query(self, question): """查询""" # 1. 检索 docs = await self.retriever.search(question) # 2. 生成 answer = await self.generator.generate(question, docs) return { "answer": answer, "sources": [{"title": d.metadata["doc_title"], "text": d.text[:200]} for d in docs] } 四、优化 4.1 性能优化 # 1. 缓存热门查询 # 2. 预计算embedding # 3. 并行检索 # 4. 流式生成 async def query_stream(self, question): docs = await self.retriever.search(question) async for token in self.generator.generate_stream(question, docs): yield token 4.2 质量优化 # 1. 查询改写 query_rewritten = await self.rewrite_query(question) # 2. 多路检索 vector_results = await self.vector_search(query) keyword_results = await self.keyword_search(query) fused = self.fuse(vector_results, keyword_results) # 3. 自适应检索 if self.needs_multi_hop(question): docs = await self.multi_hop_retrieve(question) else: docs = await self.simple_retrieve(question) 五、部署 # docker-compose.yml version: '3.8' services: rag-api: build: . ports: - "8000:8000" environment: - OPENAI_API_KEY=${OPENAI_API_KEY} - QDRANT_URL=http://qdrant:6333 - REDIS_URL=redis://redis:6379 depends_on: - qdrant - redis qdrant: image: qdrant/qdrant:latest ports: - "6333:6333" volumes: - qdrant_data:/qdrant/storage redis: image: redis:7-alpine ports: - "6379:6379" volumes: qdrant_data: 六、监控 # 关键监控指标 metrics = { "query_latency_p50": "中位查询延迟", "query_latency_p99": "99%查询延迟", "retrieval_accuracy": "检索准确率", "answer_quality": "回答质量评分", "cache_hit_rate": "缓存命中率", "error_rate": "错误率", "token_cost": "Token消耗" } 结语 搭建一个RAG系统不难，但搭建一个生产级RAG系统需要考虑很多细节——分块策略、检索质量、生成质量、缓存、监控、成本控制。 ...

引言 你有一个擅长编程的模型和一个擅长数学的模型，能不能得到一个两者都擅长的模型？模型合并（Model Merging）就是解决这个问题——将多个专门化模型的能力融合到一个模型中。 2026年，模型合并已经成为构建通用模型的重要技术。 一、合并方法 1.1 简单权重平均 def simple_average(models): """简单权重平均""" avg_state = {} for key in models[0].state_dict(): avg_state[key] = sum(m.state_dict()[key] for m in models) / len(models) return avg_state 简单但可能不是最优——不同模型的权重可能在不同方向上优化。 1.2 SLERP（球面线性插值） def slerp(t, v0, v1): """球面线性插值""" v0_norm = v0 / v0.norm() v1_norm = v1 / v1.norm() omega = torch.acos(torch.clamp(v0_norm @ v1_norm, -1, 1)) so = torch.sin(omega) if so < 1e-6: return (1-t)*v0 + t*v1 return torch.sin((1-t)*omega)/so * v0 + torch.sin(t*omega)/so * v1 def merge_slerp(model_a, model_b, t=0.5): """SLERP合并""" merged = {} for key in model_a.state_dict(): merged[key] = slerp(t, model_a.state_dict()[key], model_b.state_dict()[key]) return merged 1.3 TIES def ties_merge(models, base_model, density=0.5): """TIES合并: Trim, Elect Sign, Disjoint Merge""" # 1. 计算每个模型相对于base的delta deltas = [m.state_dict() - base_model.state_dict() for m in models] # 2. Trim: 只保留每个delta中top-k的参数 for delta in deltas: for key in delta: threshold = torch.quantile(delta[key].abs(), 1 - density) delta[key] = torch.where(delta[key].abs() > threshold, delta[key], 0) # 3. Elect Sign: 投票决定每个参数的符号 merged = {} for key in base_model.state_dict(): signs = sum(torch.sign(d[key]) for d in deltas) elected_sign = torch.sign(signs) # 4. Disjoint Merge: 只保留与选举符号一致的delta，取平均 consistent = [] for delta in deltas: mask = torch.sign(delta[key]) == elected_sign consistent.append(torch.where(mask, delta[key], 0)) merged[key] = base_model.state_dict()[key] + sum(consistent) / max(1, sum(elected_sign != 0)) return merged 1.4 DARE def dare_merge(model_a, model_b, base_model, drop_rate=0.9): """DARE: Drop And REscale""" delta_a = model_a.state_dict() - base_model.state_dict() delta_b = model_b.state_dict() - base_model.state_dict() merged = {} for key in base_model.state_dict(): # 随机丢弃大部分delta mask_a = (torch.rand_like(delta_a[key]) > drop_rate).float() mask_b = (torch.rand_like(delta_b[key]) > drop_rate).float() # 重新缩放 dropped_a = delta_a[key] * mask_a / (1 - drop_rate) dropped_b = delta_b[key] * mask_b / (1 - drop_rate) # 合并 merged[key] = base_model.state_dict()[key] + dropped_a + dropped_b return merged 二、层级合并 不同层使用不同的合并策略： ...

引言 高质量的训练数据是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处理能力提升。 ...

引言 指令微调（Instruction Tuning / SFT）是将基础模型变成对话助手的关键步骤。2026年的经验表明：微调效果90%取决于数据质量，10%取决于训练方法。 一、数据格式 { "messages": [ {"role": "system", "content": "你是一个专业的编程助手。"}, {"role": "user", "content": "解释什么是递归"}, {"role": "assistant", "content": "递归是一种编程技术..."} ] } 二、数据构建策略 2.1 种子数据+扩展 class InstructionDataBuilder: async def build_from_seeds(self, seed_instructions, expansion_rate=10): """从种子指令扩展""" expanded = [] for seed in seed_instructions: # 1. 改写指令 rewrites = await self.rewrite_instruction(seed, n=expansion_rate//2) # 2. 生成变体 variants = await self.generate_variants(seed, n=expansion_rate//2) expanded.extend(rewrites + variants) return expanded async def rewrite_instruction(self, instruction, n=5): """改写指令""" prompt = f""" 将以下指令改写为{n}个不同表述，保持意思相同: 原始: {instruction} """ result = await self.llm.call(prompt) return result["rewrites"] 2.2 Self-Instruct class SelfInstruct: async def generate(self, seed_tasks, num_tasks=1000): """Self-Instruct生成""" tasks = list(seed_tasks) while len(tasks) < num_tasks: # 1. 随机选择种子任务作为示例 examples = random.sample(tasks, min(3, len(tasks))) # 2. 生成新指令 new_instruction = await self.llm.generate( f"基于以下示例生成一个新的指令:\n{examples}" ) # 3. 过滤低质量 if self.is_quality(new_instruction): # 4. 生成回答 response = await self.llm.generate(new_instruction) tasks.append({ "instruction": new_instruction, "response": response }) return tasks 2.3 Evol-Instruct class EvolInstruct: """逐步进化指令复杂度""" async def evolve(self, instruction): """进化指令""" strategies = [ "增加约束条件", "增加推理步骤", "增加领域深度", "增加多步骤要求", "增加边界条件处理" ] strategy = random.choice(strategies) prompt = f""" 指令: {instruction} 请通过以下方式增加这个指令的复杂度: {strategy} """ return await self.llm.call(prompt) 三、数据质量 3.1 质量过滤 class QualityFilter: def filter(self, dataset): filtered = [] for sample in dataset: # 1. 长度检查 if len(sample["response"]) < 10: continue # 2. 重复检查 if self.is_duplicate(sample, filtered): continue # 3. 格式检查 if not self.validate_format(sample): continue # 4. 内容质量 if not self.check_content_quality(sample): continue filtered.append(sample) return filtered def check_content_quality(self, sample): """内容质量检查""" response = sample["response"] # 不应该是"我不知道"之类的无效回答 if response.strip() in ["我不知道", "无法回答", "I don't know"]: return False # 不应该是重复内容 if len(set(response.split())) / len(response.split()) < 0.3: return False return True 3.2 去重 class Deduplicator: def deduplicate(self, dataset): """多级去重""" # 1. 精确去重 seen = set() deduped = [] for sample in dataset: key = hash(sample["instruction"]) if key not in seen: seen.add(key) deduped.append(sample) # 2. 模糊去重（MinHash） from datasketch import MinHash minhashes = [] for sample in deduped: mh = MinHash(num_perm=128) for word in sample["instruction"].split(): mh.update(word.encode()) minhashes.append(mh) # 移除相似度>0.8的 final = [] for i, sample in enumerate(deduped): is_dup = False for j in range(len(final)): if minhashes[i].jaccard(minhashes[final[j]["index"]]) > 0.8: is_dup = True break if not is_dup: final.append({"index": i, "sample": sample}) return [f["sample"] for f in final] 四、数据配比 class DataMixer: def create_mix(self, datasets): """创建数据混合""" # 2026年经验配比 mix = { "general_qa": 0.30, # 通用问答 "coding": 0.20, # 编程 "reasoning": 0.15, # 推理 "math": 0.10, # 数学 "creative_writing": 0.10, # 创意写作 "safety": 0.05, # 安全 "multi_turn": 0.05, # 多轮对话 "tool_use": 0.05, # 工具使用 } total = sum(v for v in mix.values()) assert abs(total - 1.0) < 0.01 mixed = [] for category, ratio in mix.items(): n = int(total_samples * ratio) sampled = self.sample_from(datasets[category], n) mixed.extend(sampled) random.shuffle(mixed) return mixed 五、训练 from trl import SFTTrainer, SFTConfig config = SFTConfig( output_dir="./sft-output", num_train_epochs=3, per_device_train_batch_size=8, gradient_accumulation_steps=4, learning_rate=2e-5, warmup_ratio=0.03, lr_scheduler_type="cosine", max_seq_length=2048, bf16=True, gradient_checkpointing=True, save_strategy="epoch", evaluation_strategy="epoch", ) trainer = SFTTrainer( model=base_model, args=config, train_dataset=train_data, eval_dataset=eval_data, tokenizer=tokenizer, ) trainer.train() 六、评估 async def evaluate_sft(model, eval_set): """评估SFT模型""" metrics = {} # 1. 自动评估 metrics["loss"] = model.evaluate(eval_set) # 2. 基准测试 benchmarks = ["MMLU", "HumanEval", "GSM8K", "MT-Bench"] for bench in benchmarks: metrics[bench] = await run_benchmark(model, bench) # 3. 人工评估 samples = generate_samples(model, n=100) metrics["human_score"] = await human_eval(samples) return metrics 七、常见陷阱 数据太多但质量低：10万高质量样本 > 100万低质量样本 格式不一致：确保所有数据使用统一的对话格式 过拟合：3轮通常足够，超过5轮容易过拟合 灾难性遗忘：混入通用数据防止遗忘基础能力 结语 指令微调是"数据为王"的领域。2026年的经验反复证明：花80%的时间在数据构建和质量控制上，20%在训练调参上，才能得到最好的效果。 ...

引言 尽管DPO等简化方法在2026年很流行，但RLHF仍然是处理复杂对齐任务的有力工具。特别是在需要精细奖励信号的场景下，RLHF的优势依然明显。 本文将介绍2026年RLHF的现代实现方法。 一、RLHF三阶段 1. SFT（监督微调）: 用高质量数据微调基础模型 2. RM（奖励模型）: 训练奖励模型预测人类偏好 3. RL（强化学习）: 用PPO等算法优化策略 二、阶段一：SFT from trl import SFTTrainer, SFTConfig config = SFTConfig( output_dir="./sft", num_train_epochs=3, per_device_train_batch_size=4, learning_rate=2e-5, max_seq_length=2048, bf16=True, ) trainer = SFTTrainer( model=base_model, args=config, train_dataset=sft_dataset, ) trainer.train() 三、阶段二：奖励模型 3.1 数据格式 {"prompt": "...", "chosen": "好回答", "rejected": "差回答"} 3.2 训练 from trl import RewardTrainer, RewardConfig # 奖励模型通常用SFT模型初始化 rm_model = AutoModelForSequenceClassification.from_pretrained( "sft-model", num_labels=1 ) config = RewardConfig( output_dir="./reward-model", num_train_epochs=1, per_device_train_batch_size=16, learning_rate=5e-6, max_length=2048, ) trainer = RewardTrainer( model=rm_model, args=config, train_dataset=preference_dataset, ) trainer.train() 四、阶段三：PPO from trl import PPOTrainer, PPOConfig config = PPOConfig( output_dir="./ppo", learning_rate=1.46e-5, batch_size=32, mini_batch_size=4, ppo_epochs=4, cliprange=0.2, beta=0.05, # KL惩罚系数 ) ppo_trainer = PPOTrainer( model=sft_model, # 策略模型 ref_model=sft_model, # 参考模型（冻结） reward_model=rm_model, # 奖励模型 args=config, tokenizer=tokenizer, train_dataset=ppo_dataset, ) for batch in dataloader: # 1. 策略模型生成回答 responses = ppo_trainer.generate(batch["prompt"]) # 2. 奖励模型打分 rewards = reward_model.score(batch["prompt"], responses) # 3. PPO更新 stats = ppo_trainer.step(batch["prompt"], responses, rewards) 五、2026年改进 5.1 RLOO（REINFORCE Leave-One-Out） 比PPO更简单的替代方案，不需要价值模型： ...

引言 RLHF（人类反馈强化学习）是对齐LLM的主流方法，但其训练流程复杂——需要训练奖励模型、使用PPO等强化学习算法。DPO（Direct Preference Optimization）提供了一种更简洁的替代方案：不需要奖励模型，直接从偏好数据中学习。 2026年，DPO已经成为最流行的对齐方法之一。本文将介绍DPO的原理和实践。 一、DPO原理 1.1 与RLHF对比 RLHF流程: 1. 训练奖励模型 2. 使用PPO优化策略（需要4个模型同时运行：策略、参考、奖励、价值） 3. 训练不稳定，超参数敏感 DPO流程: 1. 直接从偏好数据优化策略（只需2个模型：策略、参考） 2. 训练稳定，超参数简单 3. 本质上是分类问题 1.2 数学原理 DPO的核心洞察：最优策略可以通过偏好数据的对数比来表达，不需要显式的奖励模型。 DPO损失 = -log σ(β·(log π(y_w|x)/π_ref(y_w|x) - log π(y_l|x)/π_ref(y_l|x))) 其中: y_w = 偏好（chosen）的回答 y_l = 不偏好（rejected）的回答 π = 当前策略 π_ref = 参考策略（通常是SFT模型） β = 温度参数 二、数据准备 2.1 偏好数据格式 { "prompt": "如何学习编程？", "chosen": "学习编程建议从Python开始，因为它语法简洁...（优质回答）", "rejected": "编程很难，你可能学不会。（劣质回答）" } 2.2 数据构建 class DPODataBuilder: async def build_from_human_annotations(self, annotations): """从人工标注构建DPO数据""" dpo_data = [] for ann in annotations: dpo_data.append({ "prompt": ann.prompt, "chosen": ann.preferred_response, "rejected": ann.rejected_response }) return dpo_data async def build_from_ai_feedback(self, prompts, model): """使用AI反馈构建DPO数据""" dpo_data = [] for prompt in prompts: # 1. 生成多个回答 responses = await model.generate( prompt, num_return_sequences=2, temperature=0.8 ) # 2. 使用评判模型选择更好的回答 judgment = await self.judge(prompt, responses[0], responses[1]) if judgment["winner"] == "A": chosen, rejected = responses[0], responses[1] else: chosen, rejected = responses[1], responses[0] # 3. 只保留有明确偏好的数据 if judgment["confidence"] > 0.7: dpo_data.append({ "prompt": prompt, "chosen": chosen, "rejected": rejected }) return dpo_data async def judge(self, prompt, response_a, response_b): """使用LLM评判哪个回答更好""" judge_prompt = f""" 问题: {prompt} 回答A: {response_a} 回答B: {response_b} 哪个回答更好？A还是B？ 输出JSON: {{"winner": "A"/"B", "confidence": 0-1, "reason": "..."}} """ return await self.judge_model.call(judge_prompt) 2.3 数据质量保障 class DPODataQualityChecker: def check(self, dataset): issues = [] for sample in dataset: # 1. chosen和rejected不能太相似 similarity = self.compute_similarity(sample["chosen"], sample["rejected"]) if similarity > 0.9: issues.append({"issue": "too_similar", "sample": sample}) # 2. chosen不能比rejected差 # （需要评判模型验证） # 3. prompt长度合理 if len(sample["prompt"]) > 2048: issues.append({"issue": "prompt_too_long", "sample": sample}) # 4. 回答长度差异不能太大 len_diff = abs(len(sample["chosen"]) - len(sample["rejected"])) if len_diff > 2000: issues.append({"issue": "length_bias", "sample": sample}) return issues 三、训练 3.1 使用TRL库 from trl import DPOTrainer, DPOConfig # 1. 加载模型 model = AutoModelForCausalLM.from_pretrained("sft-model-path") ref_model = AutoModelForCausalLM.from_pretrained("sft-model-path") # 2. 配置 config = DPOConfig( output_dir="./dpo-output", num_train_epochs=3, per_device_train_batch_size=4, gradient_accumulation_steps=4, learning_rate=5e-7, warmup_ratio=0.1, beta=0.1, # DPO温度参数 max_prompt_length=512, max_length=2048, logging_steps=10, save_steps=100, bf16=True, ) # 3. 训练 trainer = DPOTrainer( model=model, ref_model=ref_model, args=config, train_dataset=dpo_dataset, tokenizer=tokenizer, ) trainer.train() 3.2 超参数选择 # β（beta）选择 beta = 0.1 # 保守，接近SFT模型 beta = 0.3 # 平衡 beta = 0.5 # 激进，更大偏移 # 学习率 lr = 5e-7 # DPO通常需要很小的学习率 # 比SFT小10-100倍 # epoch数 epochs = 1-3 # DPO通常1-3轮足够 # 过多轮会导致过拟合 3.3 QLoRA + DPO # 先用QLoRA加载模型 model = AutoModelForCausalLM.from_pretrained( "base-model", quantization_config=bnb_config, device_map="auto" ) model = get_peft_model(model, lora_config) # DPO训练 trainer = DPOTrainer( model=model, ref_model=None, # QLoRA模式下不需要单独的ref_model args=config, train_dataset=dpo_dataset, tokenizer=tokenizer, ) 四、评估 class DPOEvaluator: async def evaluate(self, model, eval_dataset): metrics = { "accuracy": [], # DPO准确率（chosen得分>rejected得分） "margin": [], # chosen和rejected的分数差 "reward_accuracy": [] # 奖励准确率 } for sample in eval_dataset: # 计算chosen和rejected的对数概率 chosen_logprob = self.compute_logprob( model, sample["prompt"], sample["chosen"] ) rejected_logprob = self.compute_logprob( model, sample["prompt"], sample["rejected"] ) # 准确率 metrics["accuracy"].append(chosen_logprob > rejected_logprob) # 边际 metrics["margin"].append(chosen_logprob - rejected_logprob) return {k: np.mean(v) for k, v in metrics.items()} 五、常见问题 5.1 过拟合 DPO容易过拟合——训练太多轮后，模型会"记住"偏好数据而非学习通用偏好。 ...