advanced rag patterns

高级 RAG 模式:HyDE、CRAG、Self-RAG 与 FLARE

高级 RAG 模式:HyDE、CRAG、Self-RAG 与 FLARE 引言 基础 RAG(检索→增强→生成)在真实场景中常遇到三大瓶颈:语义鸿沟(用户问题与文档用词不一致)、检索噪声(相关度低的片段污染上下文)、一次性检索(无法根据生成过程动态调整)。高级 RAG 模式针对这些问题提出了系统性解法。 本文深入解析四种最具代表性的高级 RAG 模式,给出可运行的 Python 实现,并在最后提供综合对比。 1. HyDE(Hypothetical Document Embeddings) 核心思想 HyDE 的核心洞察:用 LLM 先生成一个「假设答案」,再用这个假设的嵌入去检索,而非直接用用户问题检索。 原因是:假设答案在语义空间中比原始问题更接近真实文档,尤其当用户问题很短或措辞与知识库不一致时。 用户问题 → LLM 生成假设文档 → 嵌入假设文档 → 向量检索 → 真实文档 → 最终生成 算法流程 接收用户查询 q 令 LLM 生成 k 个假设文档 {d₁, d₂, ..., dₖ}(不要求事实正确,只要求语义相关) 对每个假设文档计算嵌入向量 hᵢ = embed(dᵢ) 取平均(或最大池化)得到查询嵌入 h_q = mean(h₁, ..., hₖ) 在向量库中检索 Top-K 最相似的真实文档 将真实文档作为上下文,令 LLM 生成最终答案 代码实现 import numpy as np from typing import List, Dict from openai import OpenAI client = OpenAI() def hyde_retrieve( query: str, index, # faiss/chroma 等向量索引 embed_model, # 嵌入模型 k_hypothetical: int = 3, # 假设文档数量 top_k: int = 5, # 最终检索数量 temperature: float = 0.7, ) -> List[Dict]: """ HyDE 检索:生成假设文档 → 嵌入 → 检索真实文档 """ # Step 1: 生成假设文档 hypothetical_docs = [] for _ in range(k_hypothetical): resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "你是一个知识丰富的助手。请根据用户问题,撰写一段可能回答该问题的文档片段。不需要事实完全准确,重点是覆盖相关概念和关键词。"}, {"role": "user", "content": f"问题:{query}"} ], temperature=temperature, ) hypothetical_docs.append(resp.choices[0].message.content) # Step 2: 嵌入假设文档 hyp_embeddings = embed_model.encode(hypothetical_docs) # shape: (k, dim) query_embedding = hyp_embeddings.mean(axis=0, keepdims=True) # 平均池化 # Step 3: 检索真实文档 distances, indices = index.search(query_embedding.astype(np.float32), top_k) results = [{"index": idx, "score": dist} for idx, dist in zip(indices[0], distances[0])] return results # 单文档 HyDE(k=1 的简化版) def hyde_single(query: str, index, embed_model, top_k: int = 5): """单假设文档版本,速度更快""" resp = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "写一个可能回答问题的段落,覆盖相关关键词。"}, {"role": "user", "content": query} ], temperature=0.7, ) hypo_doc = resp.choices[0].message.content q_emb = embed_model.encode([hypo_doc]) distances, indices = index.search(q_emb.astype(np.float32), top_k) return indices[0].tolist(), distances[0].tolist() 适用场景与局限 场景 适合度 说明 短查询 / 口语化问题 ⭐⭐⭐⭐⭐ 假设文档补充了语义上下文 专业术语与用户措辞差异大 ⭐⭐⭐⭐⭐ 假设文档桥接了术语鸿沟 多跳推理问题 ⭐⭐⭐⭐ 假设文档可覆盖多个推理步骤 低延迟要求 ⭐⭐ 需要额外一次 LLM 调用 事实敏感场景(医疗/法律) ⭐⭐⭐ 假设文档可能引入错误事实,需后处理 2. CRAG(Corrective RAG) 核心思想 CRAG 引入检索质量评估器,在检索后对文档相关性进行打分,动态决定: ...

2026-06-25 · 5 min · 984 words · 硅基 AGI 探索者
prompt chaining advanced

高级 Prompt 链式调用:构建复杂推理流水线

链式调用原理 单次 LLM 调用的能力是有限的。链式调用 (Prompt Chaining) 将复杂任务分解为多个步骤,每步的输出作为下一步的输入,形成流水线: [用户输入] → [Step 1: 理解意图] → [Step 2: 检索知识] → [Step 3: 生成答案] → [最终输出] 核心假设:分解后的子任务更简单、更可靠,错误隔离更容易。 链式架构模式 1. 顺序链 (Sequential Chain) 最简单的模式——线性传递: from typing import Dict, Any import openai def llm_call(prompt: str, model: str = "gpt-4") -> str: resp = openai.chat.completions.create( model=model, messages=[{"role": "user", "content": prompt}], temperature=0.3, ) return resp.choices[0].message.content class SequentialChain: def __init__(self): self.steps = [] def add_step(self, name: str, prompt_template: str): """prompt_template 中用 {prev_output} 引用上一步输出""" self.steps.append({"name": name, "template": prompt_template}) return self def run(self, initial_input: str) -> Dict[str, str]: results = {} current_output = initial_input for step in self.steps: prompt = step["template"].format(prev_output=current_output) current_output = llm_call(prompt) results[step["name"]] = current_output print(f"[{step['name']}] 完成") results["final"] = current_output return results # 示例:技术文档翻译流水线 chain = SequentialChain() chain.add_step("extract_keypoints", "分析以下技术文档,提取关键技术概念和术语(JSON数组格式):\n{prev_output}") chain.add_step("translate", "将以下内容翻译为中文,保持技术术语准确:\n{prev_output}") chain.add_step("review", "审校以下翻译,修正不准确之处,输出最终版本:\n{prev_output}") 2. 并行链 (Parallel Chain) 多个独立子任务并行执行后汇总: ...

2026-06-24 · 4 min · 799 words · 硅基 AGI 探索者
鲁ICP备2026018361号