RAG与微调

引言 参数高效微调（PEFT）技术让普通开发者也能微调大模型。其中LoRA和QLoRA是最流行的两种方案。2026年，随着模型规模进一步扩大，这两种技术的差异和选择变得更加重要。 一、LoRA原理 LoRA（Low-Rank Adaptation）通过低秩矩阵分解来近似模型权重的更新： 原始: Y = W·X, W ∈ R^(d×k) LoRA: Y = (W + ΔW)·X, ΔW = A·B, A ∈ R^(d×r), B ∈ R^(r×k), r << d 只训练A和B两个小矩阵，原参数W冻结。 from peft import LoraConfig, get_peft_model config = LoraConfig( r=8, # 秩 lora_alpha=16, target_modules=["q_proj", "v_proj", "k_proj", "o_proj"], lora_dropout=0.05, bias="none", task_type="CAUSAL_LM" ) model = get_peft_model(base_model, config) 优势 训练参数减少99%+ 不增加推理延迟（训练后可合并权重） 效果接近全参数微调 劣势 仍需要模型以全精度加载到GPU 大模型仍需要大量GPU内存 二、QLoRA原理 QLoRA在LoRA基础上增加了4位量化： 基础模型: 4-bit量化存储 → 大幅减少内存 LoRA适配器: 全精度训练 → 保持训练精度 梯度: 通过4-bit基础模型反传 → 计算LoRA梯度 from transformers import BitsAndBytesConfig import torch bnb_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_use_double_quant=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype=torch.bfloat16 ) model = AutoModelForCausalLM.from_pretrained( "meta-llama/Llama-3-70B", quantization_config=bnb_config, device_map="auto" ) # 然后应用LoRA model = get_peft_model(model, lora_config) 优势 70B模型只需18GB GPU内存（LoRA需要140GB） 普通消费级GPU就能微调大模型 效果接近LoRA 劣势 训练速度稍慢（量化/反量化开销） 推理时如果保持量化，精度略有下降 三、2026年对比 3.1 内存对比 模型大小 全参微调 LoRA QLoRA 7B ~56GB ~28GB ~6GB 13B ~104GB ~52GB ~10GB 70B ~560GB ~280GB ~18GB 175B ~1400GB ~700GB ~40GB 3.2 性能对比 指标 LoRA QLoRA 差距 训练速度 1.0x 0.85x QLoRA慢15% 最终精度 100% 98-99% QLoRA略低 推理速度 1.0x 1.0x* 合并后相同 训练稳定性 高 中高 QLoRA偶有梯度异常 *合并权重后推理速度相同；如果不合并，QLoRA推理更慢 ...

引言 传统RAG是"一刀切"的——每次查询都检索，每次检索都使用相同的策略。但不是所有问题都需要检索（“1+1等于几”），不同问题需要不同的检索策略。 Self-RAG的核心思想是：让模型自己决定是否需要检索、检索什么、检索结果是否相关、是否需要再次检索。这种"自我反思"的能力让RAG系统更加智能和高效。 一、Self-RAG原理 1.1 反思令牌 Self-RAG引入了"反思令牌"（Reflection Tokens），让模型在生成过程中输出特定的控制信号： [Retrieve] - 是否需要检索？ [IsRel] - 检索结果是否相关？ [IsSup] - 生成是否被检索结果支持？ [IsUse] - 生成是否有用？ 1.2 工作流程 用户问题 ↓ 模型判断 [Retrieve]? → No → 直接生成 → Yes → 执行检索 ↓ 评估每个文档 [IsRel]? → 过滤不相关 ↓ 基于相关文档生成 ↓ 评估生成 [IsSup]? → No → 重新生成 → Yes → 评估 [IsUse]? ↓ 输出最终答案 二、实现 2.1 基本框架 class SelfRAG: def __init__(self, llm, retriever): self.llm = llm self.retriever = retriever async def generate(self, question): # 1. 判断是否需要检索 need_retrieve = await self.predict_retrieve_token(question) if not need_retrieve: # 不需要检索，直接生成 answer = await self.llm.generate(question) return {"answer": answer, "retrieved": False} # 2. 执行检索 docs = await self.retriever.search(question, top_k=5) # 3. 评估每个文档的相关性 relevant_docs = [] for doc in docs: is_rel = await self.predict_isrel_token(question, doc) if is_rel: relevant_docs.append(doc) if not relevant_docs: # 没有相关文档，直接生成 answer = await self.llm.generate(question) return {"answer": answer, "retrieved": True, "relevant": 0} # 4. 基于相关文档生成多个候选 candidates = [] for doc in relevant_docs: candidate = await self.llm.generate( f"基于以下信息回答问题:\n{doc.text}\n\n问题: {question}" ) # 5. 评估每个候选是否被文档支持 is_sup = await self.predict_issup_token(candidate, doc) # 6. 评估是否有用 is_use = await self.predict_isuse_token(question, candidate) candidates.append({ "answer": candidate, "support_score": is_sup, "usefulness_score": is_use, "source_doc": doc }) # 7. 选择最佳候选 best = max(candidates, key=lambda x: x["support_score"] * x["usefulness_score"]) return { "answer": best["answer"], "retrieved": True, "relevant_docs": len(relevant_docs), "source": best["source_doc"] } 2.2 反思令牌预测 class ReflectionTokenPredictor: def __init__(self, llm): self.llm = llm async def predict_retrieve(self, question): """预测是否需要检索""" prompt = f""" 问题: {question} 这个问题是否需要查阅外部资料才能准确回答？ 考虑: - 问题是否涉及事实性知识？ - 你的训练数据中是否有足够的知识？ - 问题是否涉及最新信息？ 输出: "RETRIEVE" 或 "NO_RETRIEVE" """ result = await self.llm.call(prompt) return "RETRIEVE" in result async def predict_isrel(self, question, document): """预测文档是否与问题相关""" prompt = f""" 问题: {question} 文档: {document.text[:500]} 这个文档是否包含回答问题所需的信息？ 输出: "RELEVANT" 或 "IRRELEVANT" """ result = await self.llm.call(prompt) return "RELEVANT" in result async def predict_issup(self, answer, document): """预测答案是否被文档支持""" prompt = f""" 文档: {document.text[:500]} 答案: {answer} 答案中的所有陈述是否都能从文档中找到支持？ 输出: "FULLY_SUPPORTED" / "PARTIALLY_SUPPORTED" / "NOT_SUPPORTED" """ result = await self.llm.call(prompt) if "FULLY" in result: return 1.0 elif "PARTIALLY" in result: return 0.5 else: return 0.0 async def predict_isuse(self, question, answer): """预测答案是否有用""" prompt = f""" 问题: {question} 答案: {answer} 这个答案对提问者有多大帮助？（1-5分） """ result = await self.llm.call(prompt) return int(result.strip()) / 5.0 三、训练Self-RAG 3.1 训练数据构建 class SelfRAGTrainingDataBuilder: async def build_training_data(self, base_dataset): """构建Self-RAG训练数据""" training_samples = [] for sample in base_dataset: question = sample["question"] answer = sample["answer"] relevant_docs = sample.get("relevant_docs", []) # 1. 生成Retrieve标签 retrieve_label = "RETRIEVE" if relevant_docs else "NO_RETRIEVE" # 2. 生成IsRel标签 for doc in relevant_docs: isrel_label = "RELEVANT" training_samples.append({ "input": f"问题: {question}\n文档: {doc}", "label": isrel_label, "token_type": "IsRel" }) # 3. 生成IsSup标签 for doc in relevant_docs: issup_label = "FULLY_SUPPORTED" # 假设标准答案被文档支持 training_samples.append({ "input": f"文档: {doc}\n答案: {answer}", "label": issup_label, "token_type": "IsSup" }) # 4. 生成IsUse标签 isuse_label = "5" # 标准答案评为5分 training_samples.append({ "input": f"问题: {question}\n答案: {answer}", "label": isuse_label, "token_type": "IsUse" }) return training_samples 3.2 模型微调 class SelfRAGTrainer: def __init__(self, base_model): self.base_model = base_model async def train(self, training_data): """微调模型学习反思令牌预测""" # 1. 格式化训练数据 formatted = self.format_training_data(training_data) # 2. 微调 # 方式1: 在基础模型上微调（需要大量数据） # 方式2: 使用few-shot prompting（更实用） # 使用方式2 self.model = self.create_few_shot_model(formatted) def create_few_shot_model(self, training_data): """使用few-shot examples创建Self-RAG模型""" # 为每种反思令牌选择最好的示例 retrieve_examples = self.select_examples(training_data, "Retrieve", n=3) isrel_examples = self.select_examples(training_data, "IsRel", n=3) issup_examples = self.select_examples(training_data, "IsSup", n=3) isuse_examples = self.select_examples(training_data, "IsUse", n=3) system_prompt = f""" 你是一个Self-RAG助手。在回答问题时，你需要: 1. 首先判断是否需要检索外部资料: {self.format_examples(retrieve_examples)} 2. 如果检索，评估每个文档的相关性: {self.format_examples(isrel_examples)} 3. 评估答案是否被文档支持: {self.format_examples(issup_examples)} 4. 评估答案的有用程度: {self.format_examples(isuse_examples)} """ return FewShotModel(self.base_model, system_prompt) 四、生产实践 4.1 混合模式 class HybridSelfRAG: """Self-RAG与规则结合的混合模式""" async def generate(self, question): # 规则前置：某些类型的问题不需要检索 if self.is_simple_factual(question): return await self.llm.generate(question) if self.is_current_event(question): need_retrieve = True else: need_retrieve = await self.predict_retrieve(question) if need_retrieve: return await self.rag_generate(question) else: return await self.llm.generate(question) def is_simple_factual(self, question): """简单事实型问题""" simple_patterns = [ r"^\d+[\+\-\*/]\d+", # 数学计算 r"什么是.*定义", # 定义类 ] return any(re.match(p, question) for p in simple_patterns) def is_current_event(self, question): """时事类问题""" event_keywords = ["最新", "今天", "最近", "现在", "current", "latest"] return any(kw in question.lower() for kw in event_keywords) 4.2 成本优化 class CostAwareSelfRAG: """成本感知的Self-RAG""" def __init__(self): self.token_budget = 100000 # 每次会话的Token预算 self.token_used = 0 async def generate(self, question): # 估算检索+生成的Token消耗 retrieve_cost = 2000 # 检索+反思令牌的估算成本 direct_cost = 500 # 直接生成的估算成本 remaining_budget = self.token_budget - self.token_used if remaining_budget < direct_cost: # 预算不足，使用最省钱的模式 return await self.llm.generate(question) # 正常Self-RAG流程 need_retrieve = await self.predict_retrieve(question) if need_retrieve and remaining_budget > retrieve_cost: result = await self.rag_generate(question) self.token_used += retrieve_cost else: result = await self.llm.generate(question) self.token_used += direct_cost return result 五、评估 class SelfRAGEvaluator: async def evaluate(self, self_rag, test_cases): metrics = { "retrieve_accuracy": [], # 检索决策准确率 "relevance_accuracy": [], # 相关性判断准确率 "support_accuracy": [], # 支持度评估准确率 "answer_quality": [], # 答案质量 "efficiency": [] # 效率（Token消耗） } for case in test_cases: result = await self_rag.generate(case.question) # 检索决策是否正确 should_retrieve = bool(case.relevant_docs) actual_retrieve = result["retrieved"] metrics["retrieve_accuracy"].append(should_retrieve == actual_retrieve) # 答案质量 quality = await self.evaluate_answer(result["answer"], case.expected_answer) metrics["answer_quality"].append(quality) # 效率 metrics["efficiency"].append(result.get("tokens_used", 0)) return {k: np.mean(v) for k, v in metrics.items()} 结语 Self-RAG代表了RAG系统从"机械式"走向"反思式"的方向。不是每次都检索，不是每次都使用相同策略——而是根据问题特征动态调整。 ...

引言 在RAG系统中，向量检索负责从海量文档中快速召回候选结果，但召回结果的排序不一定准确。重排序（Reranking）作为第二步，对候选结果进行更精确的排序，将最相关的文档排在最前面。 2026年，重排序已经成为RAG系统的标配组件。一个好的重排序模型可以将RAG的准确率提升20-40%。本文将深入探讨重排序技术。 一、为什么需要重排序 1.1 向量检索的局限 向量检索使用双塔模型（Bi-encoder）：查询和文档分别编码，通过向量相似度排序。这种方式速度快，但精度有限——因为查询和文档的交互发生在最后的相似度计算阶段。 1.2 重排序的优势 重排序使用交叉编码器（Cross-encoder）：将查询和文档拼接在一起输入模型，模型可以充分理解查询和文档的关系。精度更高，但速度更慢。 向量检索: encode(query) · encode(doc) → similarity score (快，精度中等) 重排序: encode([query, doc]) → relevance score (慢，精度高) 1.3 两阶段架构 查询 → 向量检索（召回Top-100）→ 重排序（精排Top-10）→ 生成 向量检索负责"快而全"，重排序负责"准而精"。 二、重排序模型 2.1 交叉编码器 class CrossEncoderReranker: def __init__(self, model_name="cross-encoder/ms-marco-MiniLM-L-12-v2"): self.model = load_model(model_name) async def rerank(self, query, documents, top_k=10): # 1. 构建query-document对 pairs = [(query, doc.text) for doc in documents] # 2. 模型打分 scores = self.model.predict(pairs) # 3. 排序 ranked = sorted(zip(documents, scores), key=lambda x: -x[1]) return [doc for doc, score in ranked[:top_k]] 2.2 LLM重排序 class LLMReranker: def __init__(self, llm): self.llm = llm async def rerank(self, query, documents, top_k=10): # 方法1: 逐个打分 scores = [] for doc in documents: score = await self.score_document(query, doc) scores.append(score) # 方法2: 两两比较（更准确但更慢） # ranked = await self.pairwise_rank(query, documents) # 排序 ranked = sorted(zip(documents, scores), key=lambda x: -x[1]) return [doc for doc, score in ranked[:top_k]] async def score_document(self, query, document): """使用LLM对query-document对打分""" prompt = f""" 查询: {query} 文档: {document.text[:500]} 请评估文档与查询的相关性: 0 - 完全不相关 1 - 弱相关 2 - 部分相关 3 - 相关 4 - 高度相关 只输出数字。 """ result = await self.llm.call(prompt) return int(result.strip()) async def pairwise_rank(self, query, documents): """两两比较排序""" # 使用类似冒泡排序的方法 for i in range(len(documents)): for j in range(i+1, len(documents)): comparison = await self.compare(query, documents[i], documents[j]) if comparison == "B is better": documents[i], documents[j] = documents[j], documents[i] return documents 2.3 ColBERT ColBERT是一种延迟交互模型，介于双塔和交叉编码器之间： ...

引言 RAG系统的第一步是将文档分块（Chunking）。这个看似简单的步骤，实际上对RAG质量有巨大影响——分块太大，检索不精确；分块太小，丢失上下文。 2026年，分块策略已经从简单的"按字数切分"发展出多种智能策略。本文将系统对比这些策略。 一、为什么分块很重要 1.1 检索精度 文档: "Python是一种解释型语言。它由Guido van Rossum于1991年创建。Python强调代码可读性..." 分块A（太大大）: 整个文档作为一个块 → 检索时可能返回很多不相关的内容 分块B（太小）: "Python是一种解释型语言。" → 缺少上下文，不知道在说什么 分块C（合适）: "Python是一种解释型语言。它由Guido van Rossum于1991年创建。" → 信息完整且精确 1.2 上下文完整性 好的分块应该保持语义完整性——一个块应该表达一个完整的意思。 1.3 检索效率 块的大小直接影响向量索引的大小和检索速度。 二、分块策略分类 2.1 固定大小分块 class FixedSizeChunker: def __init__(self, chunk_size=500, overlap=50): self.chunk_size = chunk_size # 字符数或token数 self.overlap = overlap def chunk(self, text): chunks = [] start = 0 while start < len(text): end = start + self.chunk_size chunk = text[start:end] chunks.append(chunk) start = end - self.overlap # 重叠部分 return chunks 优势：简单、快速、可预测。 劣势：可能在句子中间截断，破坏语义。 最佳实践：设置50-100 token的重叠，减少截断影响。 2.2 句子分块 class SentenceChunker: def __init__(self, max_sentences=5, max_tokens=500): self.max_sentences = max_sentences self.max_tokens = max_tokens def chunk(self, text): sentences = self.split_sentences(text) chunks = [] current_chunk = [] current_tokens = 0 for sentence in sentences: sent_tokens = self.count_tokens(sentence) if current_tokens + sent_tokens > self.max_tokens or \ len(current_chunk) >= self.max_sentences: if current_chunk: chunks.append(" ".join(current_chunk)) current_chunk = [sentence] current_tokens = sent_tokens else: current_chunk.append(sentence) current_tokens += sent_tokens if current_chunk: chunks.append(" ".join(current_chunk)) return chunks 优势：保持句子完整，语义更好。 劣势：块大小不均匀。 ...

引言 “我的RAG系统效果怎么样？“这是每个RAG开发者都会问的问题。但回答这个问题并不简单——RAG系统包含检索和生成两个环节，每个环节都有多个质量维度。 2026年，RAG评估已经形成了一套比较完整的指标体系。从检索的准确率和召回率，到生成的忠实度和相关性，再到端到端的用户满意度。本文将系统介绍这些指标。 一、RAG评估框架 1.1 评估维度 RAG系统 ├── 检索质量 │ ├── 准确率（Precision） │ ├── 召回率（Recall） │ ├── 排序质量（NDCG/MRR） │ └── 多样性（Diversity） ├── 生成质量 │ ├── 忠实度（Faithfulness） │ ├── 答案相关性（Answer Relevance） │ ├── 完整性（Completeness） │ └── 正确性（Correctness） └── 端到端质量 ├── 用户满意度 ├── 响应延迟 └── 成本效率 1.2 评估方法 人工评估：最准确但成本高 自动评估：使用LLM或规则自动评估 基准测试：在标准数据集上测试 A/B测试：线上对比不同版本 二、检索质量指标 2.1 准确率与召回率 class RetrievalMetrics: def precision_at_k(self, retrieved, relevant, k): """Top-K准确率""" retrieved_k = retrieved[:k] relevant_set = set(relevant) retrieved_set = set(retrieved_k) return len(relevant_set & relevant_set) / k def recall_at_k(self, retrieved, relevant, k): """Top-K召回率""" retrieved_k = retrieved[:k] relevant_set = set(relevant) retrieved_set = set(retrieved_k) return len(retrieved_set & relevant_set) / len(relevant_set) def average_precision(self, retrieved, relevant): """平均准确率""" relevant_set = set(relevant) precisions = [] num_relevant = 0 for i, doc in enumerate(retrieved): if doc in relevant_set: num_relevant += 1 precisions.append(num_relevant / (i + 1)) if not precisions: return 0.0 return np.mean(precisions) def mean_average_precision(self, queries_results): """MAP""" return np.mean([self.average_precision(r["retrieved"], r["relevant"]) for r in queries_results]) 2.2 排序质量 def ndcg_at_k(self, retrieved, relevance_scores, k): """NDCG@K""" # DCG dcg = sum( (2 ** relevance_scores.get(doc, 0) - 1) / np.log2(i + 2) for i, doc in enumerate(retrieved[:k]) ) # IDCG（理想排序的DCG） ideal_order = sorted(relevance_scores.values(), reverse=True)[:k] idcg = sum( (2 ** score - 1) / np.log2(i + 2) for i, score in enumerate(ideal_order) ) return dcg / idcg if idcg > 0 else 0 def mrr(self, queries_results): """平均倒数排名""" reciprocal_ranks = [] for result in queries_results: for i, doc in enumerate(result["retrieved"]): if doc in result["relevant"]: reciprocal_ranks.append(1 / (i + 1)) break else: reciprocal_ranks.append(0) return np.mean(reciprocal_ranks) 2.3 上下文相关性 class ContextRelevanceMetric: async def measure(self, question, retrieved_context): """评估检索到的上下文与问题的相关性""" prompt = f""" 问题: {question} 检索到的上下文: {self.format_context(retrieved_context)} 请评估上下文与问题的相关性: 1. 有多少段落直接相关？ 2. 有多少段落间接相关？ 3. 有多少段落完全不相关？ 4. 相关性评分（0-1） """ result = await self.llm.call(prompt) return result["relevance_score"] 三、生成质量指标 3.1 忠实度（Faithfulness） 答案是否忠实于检索到的上下文，不产生幻觉： ...

引言 真实世界的信息不只存在于文本中。图片、视频、音频、表格——这些非文本模态包含着文本无法替代的信息。传统的文本RAG无法处理"找一张包含红色汽车的产品图片"这类查询。 2026年，多模态RAG已经成为AI系统的标配能力。它能够跨模态检索信息、融合多模态上下文、生成多模态输出。本文将深入探讨多模态RAG的架构与实践。 一、多模态RAG的核心挑战 1.1 表示鸿沟 不同模态的数据需要不同的编码方式： 文本：文本embedding模型 图像：视觉embedding模型 音频：音频embedding模型 如何让不同模态的表示在同一个空间中可比？ 1.2 检索鸿沟 用户可能用文本查询图像，也可能用图像查询文本： 文本→图像: "找一张展示AI架构的图" 图像→文本: "这张图对应的技术文档在哪？" 图像→图像: "找类似的图片" 1.3 融合鸿沟 检索到的多模态信息如何融合？文本和图像的信息如何整合在一起喂给LLM？ 二、架构设计 2.1 整体架构 ┌─────────────────────────────────────────────┐ │ 多模态查询处理 │ │ (文本/图像/音频 → 统一表示) │ ├─────────────────────────────────────────────┤ │ 跨模态检索引擎 │ │ ┌─────────┬─────────┬─────────┐ │ │ │文本检索 │图像检索 │音频检索 │ │ │ └─────────┴─────────┴─────────┘ │ ├─────────────────────────────────────────────┤ │ 多模态融合层 │ │ (跨模态排序、上下文组装) │ ├─────────────────────────────────────────────┤ │ 多模态生成 │ │ (文本+图像+表格 → 统一回答) │ └─────────────────────────────────────────────┘ 2.2 多模态编码 class MultiModalEncoder: def __init__(self): self.text_encoder = load_model("text-embedding-3-large") self.image_encoder = load_model("clip-vit-large") self.audio_encoder = load_model("audio-embedding-model") async def encode(self, content, modality): """编码不同模态的内容""" if modality == "text": return await self.text_encoder.embed(content) elif modality == "image": return await self.image_encoder.encode(content) elif modality == "audio": return await self.audio_encoder.encode(content) elif modality == "mixed": # 混合内容：分别编码后融合 return await self.encode_mixed(content) async def encode_mixed(self, content): """编码混合模态内容""" embeddings = [] for item in content: emb = await self.encode(item["data"], item["modality"]) embeddings.append({"embedding": emb, "modality": item["modality"]}) # 使用跨模态对齐模型将不同模态的embedding映射到统一空间 unified = await self.align_embeddings(embeddings) return unified 2.3 跨模态对齐 使用CLIP等模型实现跨模态对齐： ...

引言 向量检索擅长语义匹配——“这段文字和我的问题有多相似”。但它不擅长关系推理——“A的上级的上级是谁”。知识图谱检索擅长关系推理，但不擅长模糊语义匹配。 2026年，混合RAG——将图检索和向量检索结合——已经成为处理复杂知识问答的最佳方案。本文将深入探讨这种混合架构。 一、为什么需要混合检索 1.1 向量检索的局限 问题: "爱因斯坦的博士导师是谁？" 向量检索: 搜索"爱因斯坦 博士 导师" → 可能找到: "爱因斯坦在苏黎世联邦理工学院学习" (语义相关但缺关键信息) 可能找不到: "Alfred Kleiner是爱因斯坦的博士论文导师" (语义距离较远) 1.2 图检索的局限 问题: "量子力学的哲学意义是什么？" 图检索: 需要遍历"量子力学"→"哲学意义"的边 → 但"哲学意义"是一个抽象概念，知识图谱中可能没有对应的节点 1.3 混合的优势 问题: "爱因斯坦的博士导师的研究领域是什么？" 混合检索: 1. 向量检索: "爱因斯坦 博士 导师" → 找到相关文档 2. 实体识别: 识别出"Alfred Kleiner" 3. 图检索: 查询"Alfred Kleiner" → "研究领域" → "实验物理学" 4. 向量检索: "Alfred Kleiner 实验物理学" → 找到详细描述 5. 综合答案: "爱因斯坦的博士导师Alfred Kleiner的研究领域是实验物理学..." 二、混合RAG架构 2.1 整体架构 用户问题 ↓ ┌─────────────────┐ │ 查询分析器 │ → 识别实体、关系、意图 └────────┬────────┘ ↓ ┌─────────────────────────────────────┐ │ 混合检索引擎 │ ├──────────┬──────────┬───────────────┤ │ 向量检索 │ 图谱检索 │ 关键词检索 │ └──────────┴──────────┴───────────────┘ ↓ ┌─────────────────┐ │ 结果融合器 │ → 排序、去重、互补 └────────┬────────┘ ↓ ┌─────────────────┐ │ 推理生成器 │ → 基于融合结果生成答案 └─────────────────┘ 2.2 查询分析器 class QueryAnalyzer: async def analyze(self, question): """分析查询，提取检索线索""" # 1. 实体识别 entities = await self.ner.extract(question) # 2. 关系识别 relations = await self.relation_extractor.extract(question, entities) # 3. 意图识别 intent = await self.intent_classifier.classify(question) # 4. 生成不同检索策略的查询 return { "entities": entities, "relations": relations, "intent": intent, "vector_query": await self.generate_vector_query(question, entities), "graph_query": self.generate_graph_query(entities, relations), "keyword_query": await self.generate_keyword_query(question) } 2.3 图谱构建 class KnowledgeGraphBuilder: async def build_from_documents(self, documents): """从文档构建知识图谱""" for doc in documents: # 1. 实体抽取 entities = await self.entity_extractor.extract(doc.text) # 2. 关系抽取 relations = await self.relation_extractor.extract(doc.text, entities) # 3. 添加到图谱 for entity in entities: await self.graph.add_node( id=entity.id, label=entity.text, type=entity.type, properties={"source": doc.id, "embedding": entity.embedding} ) for relation in relations: await self.graph.add_edge( source=relation.subject, target=relation.object, label=relation.predicate, properties={"confidence": relation.confidence} ) # 4. 同时添加到向量库 await self.vector_store.add( id=doc.id, text=doc.text, embedding=doc.embedding, metadata={"entities": [e.id for e in entities]} ) 2.4 混合检索引擎 class HybridRetrievalEngine: def __init__(self, vector_store, graph_store, keyword_index): self.vector_store = vector_store self.graph_store = graph_store self.keyword_index = keyword_index async def retrieve(self, query_analysis, top_k=10): # 1. 并行执行三种检索 vector_task = self.vector_search(query_analysis["vector_query"], top_k*2) graph_task = self.graph_search(query_analysis) keyword_task = self.keyword_search(query_analysis["keyword_query"], top_k*2) vector_results, graph_results, keyword_results = await asyncio.gather( vector_task, graph_task, keyword_task ) # 2. 融合结果 fused = self.fuse_results(vector_results, graph_results, keyword_results) # 3. 重排 reranked = await self.rerank(fused, query_analysis) return reranked[:top_k] async def graph_search(self, query_analysis): """图谱检索""" results = [] # 1. 实体匹配：在图谱中找到查询实体的对应节点 for entity in query_analysis["entities"]: matched_nodes = await self.graph_store.find_nodes( label=entity.text, fuzzy=True ) # 2. 关系遍历：沿着关系边扩展 for node in matched_nodes: for relation in query_analysis["relations"]: neighbors = await self.graph_store.traverse( start=node, edge_label=relation.predicate, max_depth=2 ) results.extend(neighbors) # 3. 获取关联文档 for result in results: if result.has_property("source"): doc = await self.document_store.get(result.source) result.content = doc.text return results def fuse_results(self, vector_results, graph_results, keyword_results): """融合三种检索结果""" fused = {} # 为每种检索结果分配权重 weights = {"vector": 0.4, "graph": 0.4, "keyword": 0.2} for result_type, results, weight in [ ("vector", vector_results, weights["vector"]), ("graph", graph_results, weights["graph"]), ("keyword", keyword_results, weights["keyword"]) ]: for rank, result in enumerate(results): doc_id = result.id if doc_id not in fused: fused[doc_id] = { "result": result, "score": 0, "sources": [] } # 倒数排名融合，加上权重 fused[doc_id]["score"] += weight * (1 / (60 + rank)) fused[doc_id]["sources"].append(result_type) # 按融合分数排序 sorted_results = sorted(fused.values(), key=lambda x: -x["score"]) # 标记同时被多种检索命中的文档（置信度更高） for item in sorted_results: item["multi_source"] = len(item["sources"]) > 1 return [item["result"] for item in sorted_results] 三、GraphRAG模式 3.1 社区检测 将知识图谱划分为社区，每个社区是一组紧密相关的实体： ...

引言 基础RAG（检索-增强-生成）已经成为了AI应用的标配：用户提问→检索相关文档→LLM基于文档生成回答。但这种简单模式在面对复杂问题时力不从心。 2026年，RAG架构已经演进出多种高级模式：多跳检索、自适应检索、推理增强检索、迭代检索等。这些模式让RAG系统能够处理更复杂的问题、提供更准确的答案。本文将系统介绍这些高级模式。 一、基础RAG的局限 1.1 单次检索不够 复杂问题需要多步推理和多轮检索： 问题: "对比GPT-4和Claude在代码生成任务上的表现" 基础RAG: 一次检索"GPT-4 Claude 代码生成对比" → 可能找不到直接的对比文档 高级RAG: 检索1: "GPT-4 代码生成性能" 检索2: "Claude 代码生成性能" 检索3: "GPT-4 vs Claude benchmark" → 综合多个来源生成对比 1.2 检索质量不稳定 简单向量检索可能返回语义相关但信息不足的文档。 1.3 缺乏推理能力 基础RAG是"检索+拼接"，缺乏对检索结果的推理和整合。 二、多跳检索 2.1 原理 多跳检索（Multi-hop Retrieval）通过多轮检索逐步收集信息，每轮检索基于前一轮的结果。 问题: "那位发明了Transformer的作者后来创立了哪家公司？" 跳1: 检索"Transformer论文作者" → 得到作者列表（包括Ashish Vaswani等） 跳2: 检索"Ashish Vaswani 创立公司" → 得到Adept AI 答案: Adept AI 2.2 实现 class MultiHopRetriever: def __init__(self, retriever, llm, max_hops=3): self.retriever = retriever self.llm = llm self.max_hops = max_hops async def retrieve(self, question): accumulated_context = [] for hop in range(self.max_hops): # 1. 根据当前上下文生成下一步检索查询 if hop == 0: query = question else: query = await self.generate_next_query( question, accumulated_context ) if query is None: # 不需要更多检索 break # 2. 执行检索 results = await self.retriever.search(query, top_k=5) # 3. 过滤和评估结果 relevant = await self.filter_relevant(results, question, accumulated_context) accumulated_context.extend(relevant) # 4. 检查是否已有足够信息回答 if await self.has_enough_info(question, accumulated_context): break return accumulated_context async def generate_next_query(self, question, context): """基于当前上下文生成下一步查询""" prompt = f""" 原始问题: {question} 已知信息: {context} 还需要什么信息才能回答问题？请生成一个搜索查询。 如果已经有足够信息，返回"DONE"。 """ response = await self.llm.call(prompt) if "DONE" in response: return None return response.strip() 三、自适应检索 3.1 原理 不是所有问题都需要检索，也不是所有问题都需要相同类型的检索。自适应检索根据问题特征选择最优的检索策略。 ...

引言 大模型的对齐（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年成为微调的首选方法，特别是在中小规模模型上表现优异。 ...

引言 检索增强生成（RAG）是2026年最实用的大模型应用架构之一。从基础的"检索-生成"流水线，到Agentic RAG的智能检索代理，RAG技术正在经历深刻的架构演进。本文系统梳理RAG的发展脉络和技术选型。 RAG 的演进阶段 第一阶段：基础 RAG（Naive RAG） 最原始的RAG架构，包含三个步骤： 将文档切分为片段 对片段进行向量化并存储 检索最相关的片段，拼接后输入模型 优点： 简单、快速、易于实现 缺点： 检索质量依赖单一向量相似度，无法处理复杂查询 第二阶段：增强 RAG（Enhanced RAG） 在基础RAG之上增加优化： 查询优化： 查询重写：将用户问题转化为更适合检索的形式 查询分解：将复杂问题拆分为多个子查询 多路检索：同时使用向量检索和关键词检索 后处理： 重排序（Reranking）：对检索结果进行精排 上下文压缩：去除冗余片段，保留关键信息 自适应窗口：根据问题类型调整上下文长度 第三阶段：Agentic RAG Agentic RAG的核心思想：让Agent自主决定如何检索、何时检索、检索什么。 ┌─────────────────────────────────────────────┐ │ Agentic RAG 架构 │ │ │ │ ┌──────────┐ ┌──────────┐ │ │ │ Agent │───▶│ 检索 │ │ │ │ 规划器 │ │ 策略 │ │ │ └──────────┘ └──────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ │ │ │ 反思 │◀───│ 多路 │ │ │ │ 与调整 │ │ 检索 │ │ │ └──────────┘ └──────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────────────────────┐ │ │ │ 最终生成与验证 │ │ │ └──────────────────────────┘ │ └─────────────────────────────────────────────┘ 主流 RAG 框架对比 LangChain / LangGraph 特点： 生态最丰富，组件最全 适合： 快速原型开发、复杂RAG管线 ...