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

引言

基础 RAG（检索→增强→生成）在真实场景中常遇到三大瓶颈：语义鸿沟（用户问题与文档用词不一致）、检索噪声（相关度低的片段污染上下文）、一次性检索（无法根据生成过程动态调整）。高级 RAG 模式针对这些问题提出了系统性解法。

本文深入解析四种最具代表性的高级 RAG 模式，给出可运行的 Python 实现，并在最后提供综合对比。

1. HyDE（Hypothetical Document Embeddings）

核心思想

HyDE 的核心洞察：用 LLM 先生成一个「假设答案」，再用这个假设的嵌入去检索，而非直接用用户问题检索。

原因是：假设答案在语义空间中比原始问题更接近真实文档，尤其当用户问题很短或措辞与知识库不一致时。

用户问题 → LLM 生成假设文档 → 嵌入假设文档 → 向量检索 → 真实文档 → 最终生成

算法流程

  1. 接收用户查询 q
  2. 令 LLM 生成 k 个假设文档 {d₁, d₂, ..., dₖ}（不要求事实正确，只要求语义相关）
  3. 对每个假设文档计算嵌入向量 hᵢ = embed(dᵢ)
  4. 取平均（或最大池化）得到查询嵌入 h_q = mean(h₁, ..., hₖ)
  5. 在向量库中检索 Top-K 最相似的真实文档
  6. 将真实文档作为上下文，令 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 引入检索质量评估器，在检索后对文档相关性进行打分，动态决定：

  • 置信：检索结果质量高 → 直接使用
  • 不置信：检索结果质量低 → 触发网页搜索补充
  • 中等：混合使用检索结果 + 网页搜索结果
检索 → 相关性评估 → [置信/中等/不置信] → 知识精炼 → 生成

检索评估器

CRAG 使用一个轻量级分类模型（如 T5-small 或 DeBERTa）对每篇检索文档计算相关性分数 s ∈ [0, 1]。

from transformers import AutoTokenizer, AutoModelForSequenceClassification
import torch

class RetrievalEvaluator:
    """CRAG 检索质量评估器"""
    def __init__(self, model_name: str = "BAAI/bge-reranker-base"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.model.eval()

    def evaluate(self, query: str, documents: List[str]) -> List[float]:
        """
        对每篇文档计算相关性分数
        返回: List[float], 每个元素 ∈ [0, 1]
        """
        scores = []
        for doc in documents:
            inputs = self.tokenizer(
                query, doc,
                return_tensors="pt",
                truncation=True,
                max_length=512
            )
            with torch.no_grad():
                logits = self.model(**inputs).logits
                score = torch.sigmoid(logits[0][0]).item()
            scores.append(score)
        return scores

知识精炼（Knowledge Refinement）

CRAG 对检索文档进行句子级精炼：只保留高相关性句子，过滤噪声句子。

import re
from typing import List, Tuple

def refine_document(
    query: str,
    document: str,
    evaluator,           # RetrievalEvaluator 实例
    threshold: float = 0.5,
) -> str:
    """
    句子级精炼：只保留相关性高于阈值的句子
    """
    sentences = re.split(r'(?<=[。！？.!?])\s+', document)
    refined = []
    for sent in sentences:
        score = evaluator.evaluate(query, [sent])[0]
        if score >= threshold:
            refined.append(sent)
    return " ".join(refined)

def corrective_retrieve(
    query: str,
    vector_index,
    web_search_fn,       # 网页搜索函数（如 Tavily/SerpAPI）
    evaluator,
    top_k: int = 5,
    confidence_threshold: float = 0.7,
    low_threshold: float = 0.3,
) -> List[str]:
    """
    CRAG 主流程
    """
    # Step 1: 向量检索
    raw_docs = vector_index.search(query, top_k)

    # Step 2: 评估相关性
    scores = evaluator.evaluate(query, raw_docs)

    # Step 3: 决策
    avg_score = np.mean(scores)
    refined_docs = []

    if avg_score >= confidence_threshold:
        # 置信：直接使用，并进行句子级精炼
        for doc, score in zip(raw_docs, scores):
            refined = refine_document(query, doc, evaluator)
            refined_docs.append(refined)

    elif avg_score <= low_threshold:
        # 不置信：触发网页搜索
        web_results = web_search_fn(query)
        refined_docs.extend(web_results[:top_k])

    else:
        # 中等：混合
        for doc, score in zip(raw_docs, scores):
            if score >= threshold:
                refined = refine_document(query, doc, evaluator)
                refined_docs.append(refined)
        web_results = web_search_fn(query)
        refined_docs.extend(web_results[:2])

    return refined_docs

CRAG 核心优势

  1. 自我纠正：不盲目相信检索结果
  2. 知识精炼：句子级过滤大幅降低噪声
  3. 网络互补：检索失败时自动 fallback 到网页搜索

3. Self-RAG：自反思检索增强生成

核心思想

Self-RAG 引入反思令牌（Reflection Tokens），让 LLM 在生成过程中自主决定何时检索、何时停止检索，并对检索结果和最终生成进行自我评价。

四种反思令牌：

  • [Retrieve]：是否需要检索（yes / no / continue）
  • [Relevant]：检索结果是否相关（分数 1-5）
  • [Supported]：生成内容是否被检索结果支持（分数 1-5）
  • [Utility]：最终回答整体质量（分数 1-5）

训练方式

Self-RAG 需要微调 LLM 使其学会输出反思令牌。训练数据通过以下方式构造：

  1. 对原始指令数据集，用 RM（奖励模型）标注反思令牌
  2. 用标注数据微调底座 LLM（通常 7B 级别）
  3. 推理时，LLM 自发输出反思令牌，无需外部控制器
# Self-RAG 推理伪代码（假设模型已微调）
def self_rag_generate(
    query: str,
    retriever,
    llm,                # 已微调的 Self-RAG 模型
    max_new_tokens: int = 512,
) -> str:
    """
    Self-RAG 推理：模型自主决定检索时机
    """
    prompt = f"### Instruction:\n{query}\n\n### Response:\n"
    output = ""
    retrieve_count = 0

    for step in range(max_new_tokens // 50):  # 分段生成
        # 让模型输出下一个片段 + 反思令牌
        gen = llm.generate(
            prompt + output,
            max_new_tokens=50,
            return_dict_in_generate=True,
            output_scores=True,
        )
        segment = gen.sequences[0]
        output += segment

        # 解析反思令牌
        if "[Retrieve] yes" in segment:
            # 触发检索
            docs = retriever.search(query + " " + output[-200:])
            retrieve_count += 1
            prompt += f"\n[Retrieved Context]: {docs}\n"

        if "[Relevant]" in segment:
            # 评估相关性，低分则重新检索
            rel_score = extract_score(segment, "[Relevant]")
            if rel_score <= 2:
                docs = retriever.search(query)  # 重新检索
                prompt += f"\n[Re-retrieved]: {docs}\n"

        if "[Utility]" in segment:
            utility = extract_score(segment, "[Utility]")
            if utility >= 4:
                break  # 质量足够，提前停止

    return output

Self-RAG 与 CRAG 对比

维度Self-RAGCRAG
检索决策LLM 内部自发决定外部评估器 + 规则
是否需要微调✅ 需要❌ 不需要
可控性较低（依赖模型判断）较高（阈值可调）
延迟中等（可能多次检索）低（单次评估）
事实准确性⭐⭐⭐⭐⭐⭐⭐⭐

4. FLARE：主动检索增强生成

核心思想

FLARE（Forward-Looking Active REtrieval）的核心：在生成过程中，当模型对后续内容「不确定」时，主动触发检索。

不确定性度量：LLM 生成 t 个候选续写，计算 token 概率的熵或最小概率，超过阈值则触发检索。

生成 → 检测低概率 token → 触发检索 → 用检索结果继续生成

算法流程

  1. 生成初始回答前缀
  2. 对每个即将生成的 token，预测其概率分布
  3. 若最小概率 p_min < θ（阈值），则： a. 用当前生成前缀作为查询，进行检索 b. 将检索结果加入上下文 c. 重新生成当前位置及后续内容
  4. 重复直到生成结束

代码实现

def flare_generate(
    query: str,
    context_docs: List[str],
    llm,
    retriever,
    threshold: float = 0.2,      # 低概率阈值
    max_steps: int = 10,
) -> str:
    """
    FLARE 主动检索生成
    """
    # 初始上下文
    context = "\n".join(context_docs)
    prompt = f"Context:\n{context}\n\nQuestion: {query}\nAnswer:"

    answer_so_far = ""
    for step in range(max_steps):
        # 生成下一段（带 logits）
        outputs = llm(prompt + answer_so_far, return_dict=True, output_scores=True)
        next_token_probs = torch.softmax(outputs.scores[0], dim=-1)
        min_prob = next_token_probs.max(dim=-1).values.item()  # 简化：用最大概率近似

        if min_prob < threshold:
            # 低置信度：主动检索
            retrieval_query = query + " " + answer_so_far[-100:]
            new_docs = retriever.search(retrieval_query, top_k=3)
            context += "\n" + "\n".join(new_docs)
            prompt = f"Context:\n{context}\n\nQuestion: {query}\nAnswer:"
            # 重新生成（从当前位置）
            continue

        # 高置信度：正常生成
        next_tokens = llm.generate(prompt + answer_so_far, max_new_tokens=50)
        new_text = next_tokens[0, len(prompt):]
        answer_so_far += new_text

        if "[END]" in new_text or step == max_steps - 1:
            break

    return answer_so_far

FLARE 的核心优势

  • 按需检索：只在不确定时才检索，节省 API 调用
  • 动态上下文：随着生成推进，检索查询越来越精准
  • 适合长回答：回答越长，FLARE 优势越明显

综合对比

模式核心机制是否需要额外训练检索次数延迟最适合场景
Naive RAG一次性检索❌1低简单问答
HyDE假设文档嵌入❌1中（多1次LLM）短查询、术语鸿沟大
CRAG检索评估 + 纠正❌1~2中检索质量不稳定
Self-RAG反思令牌 + 自主决策✅ 需微调动态中高高质量要求、有训练资源
FLARE主动检索（不确定性驱动）❌动态中高长回答、多跳推理

工程实践建议

选择指南

  1. 起步阶段：先用 Naive RAG + 重排序（bge-reranker），建立基线
  2. 检索质量差：加入 CRAG 的评估器，过滤低质量文档
  3. 术语鸿沟问题突出：引入 HyDE
  4. 长回答 / 多跳推理：引入 FLARE
  5. 有微调资源：升级到 Self-RAG

组合使用

这些模式可以组合：

HyDE（改善检索） → CRAG（评估+纠正） → FLARE（生成中主动检索）

组合时注意总延迟，建议用异步并发处理独立步骤。

总结

模式一句话总结
HyDE用假设答案的嵌入桥接语义鸿沟
CRAG检索质量不行？评估+纠正+网页补充
Self-RAG让模型学会自己决定要不要检索
FLARE模型不确定时，主动去查

四种模式各有侧重，实际落地时建议从 HyDE + CRAG 入手，逐步引入更复杂的 Self-RAG 和 FLARE。

相关阅读：RAG 分块策略、RAG 重排序指南、GraphRAG 解析

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。