开始学习
课程 › 0基础Agent开发课 / RAG与向量数据库 / HyDE与查询改写-从Query端提升RAG检索质量
— 21 min read

HyDE与查询改写-从Query端提升RAG检索质量

> **[进阶选读]** 本篇适合已掌握基础 RAG 和混合检索、希望从"用户问题"端进一步提升检索质量的读者。HyDE（假设文档嵌入）和查询改写是两种不修改索引、只优化查询的技术，适合知识库不便频繁更新的场景。

HyDE 与查询改写：从 Query 端提升 RAG 检索质量

[进阶选读] 本篇适合已掌握基础 RAG 和混合检索、希望从"用户问题"端进一步提升检索质量的读者。HyDE（假设文档嵌入）和查询改写是两种不修改索引、只优化查询的技术，适合知识库不便频繁更新的场景。

1.1 检索质量差的根源：语义鸿沟

用户查询

LLM 生成
假设文档

假设文档
(Hypothetical Doc)

Embedding
向量化

向量检索

返回相关真实文档

原始查询 + 真实文档
送入 LLM

生成最终回答

HyDE 原理——用 LLM 生成假设文档，让查询向量更接近真实文档的向量空间

RAG 系统的检索质量问题，很大一部分不是向量数据库的问题，而是 Query 和文档之间的语义鸿沟。

用户的表达方式和文档的表达方式往往存在显著差异：

  • 用户问："怎么让我的 Python 代码跑快一点？"
  • 文档标题："Python 性能优化最佳实践" / "使用 Cython 加速数值计算" / "asyncio 并发编程模式"

用户的问题是口语化的动作描述，文档是技术术语的概念聚合。直接向量化后做相似性搜索，两者在高维空间中的距离可能并不近，导致检索遗漏关键文档。

本章介绍四种从 Query 端入手的改善策略，它们不需要修改向量数据库，也不需要重新索引文档，只在查询阶段做额外处理。

1.2 HyDE：假设性文档嵌入

1.2.1 原理

HyDE（Hypothetical Document Embeddings，Gao et al. 2022）的思路简单而有效：

  1. 用 LLM 根据用户的问题生成一段假设性答案（不需要正确，只需要和真实答案风格相似）
  2. 对这段假设性答案做向量化
  3. 用假设性答案的向量去检索，而不是用原始问题的向量

为什么有效： 假设性答案的文本风格与知识库文档相近（都是陈述性文字、专业术语、段落结构），两者在向量空间中的距离自然更近；而用户的口语化问题与文档风格差异大，向量距离较远。

HyDE 检索

向量化

相似度更高

用户 Query
'怎么让代码跑快'

LLM 生成
假设性答案

假设答案：
'可以使用 Cython、
numba JIT 编译...'

假设答案向量

文档：Python 性能优化

传统检索

直接向量化

相似度较低

用户 Query
'怎么让代码跑快'

Query 向量

文档：Python 性能优化

1.2.2 实现代码

python 
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.documents import Document

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(
    collection_name="knowledge_base",
    embedding_function=embeddings,
)

def hyde_retrieve(query: str, k: int = 4) -> list[Document]:
    """用 HyDE 方法检索文档"""

    # Step 1：生成假设性答案
    response = llm.invoke([
        SystemMessage(content="""你正在模拟一个专业文档。
根据给定的问题，生成一段可能出现在专业技术文档中的答案段落。
注意：使用正式、技术性的语言风格，包含相关术语，就像真实文档内容一样。
不需要保证答案准确，只需要风格与技术文档相符。"""),
        HumanMessage(content=f"问题：{query}")
    ])

    hypothetical_answer = response.content

    # Step 2：用假设性答案的向量检索
    results = vectorstore.similarity_search_by_vector(
        embeddings.embed_query(hypothetical_answer),
        k=k,
    )

    return results

def compare_retrieval_methods(query: str, k: int = 3):
    """对比标准检索和 HyDE 检索的结果差异"""
    print(f"查询：{query}\n")

    # 标准检索
    standard_results = vectorstore.similarity_search(query, k=k)
    print("=== 标准检索结果 ===")
    for i, doc in enumerate(standard_results):
        print(f"{i+1}. {doc.page_content[:100]}...")

    print("\n=== HyDE 检索结果 ===")
    hyde_results = hyde_retrieve(query, k=k)
    for i, doc in enumerate(hyde_results):
        print(f"{i+1}. {doc.page_content[:100]}...")

# 使用示例
compare_retrieval_methods("怎么让我的 Python 代码跑快一点？")

1.3 Step-back Prompting：退一步提问

Step-back Prompting（退一步提问，Google DeepMind 提出的技术）：将具体的、细节性的问题"退一步"抽象为更通用的原则性问题，然后检索通用问题的答案，再结合原始问题给出答案。

适用场景： 用户问的是具体实例（"这段代码为什么报错"），但知识库存储的是通用原理（"Python 类型错误的常见原因"）。

python 
def step_back_retrieve(query: str, vectorstore, k: int = 4) -> dict:
    """Step-back Prompting 检索"""

    # Step 1：生成更通用的退一步问题
    response = llm.invoke([
        SystemMessage(content="""给定一个具体的问题，生成一个更通用、更抽象的相关问题，
这个通用问题的答案可以帮助回答原始具体问题。

示例：
具体问题：为什么 pandas DataFrame 的 .loc 和 .iloc 有时候行为不一样？
通用问题：pandas 中 .loc 和 .iloc 的区别是什么？

只输出通用问题，不加任何解释。"""),
        HumanMessage(content=f"具体问题：{query}")
    ])

    abstract_query = response.content.strip()

    # Step 2：分别检索原始问题和通用问题
    original_results = vectorstore.similarity_search(query, k=k // 2)
    abstract_results = vectorstore.similarity_search(abstract_query, k=k // 2)

    # Step 3：合并去重
    seen_contents = set()
    combined = []
    for doc in original_results + abstract_results:
        if doc.page_content[:100] not in seen_contents:
            seen_contents.add(doc.page_content[:100])
            combined.append(doc)

    return {
        "abstract_query": abstract_query,
        "documents": combined,
    }

# 使用示例
result = step_back_retrieve(
    "为什么我的 LangGraph 节点在并行执行时状态会丢失？",
    vectorstore,
)
print(f"退一步问题：{result['abstract_query']}")

1.4 Multi-hop 子问题分解

Multi-hop（多跳推理）是指通过多个关联步骤才能得出答案的推理方式，每一"跳"利用上一步的结果进行下一步检索。复杂问题往往需要多步推理才能回答，单次检索无法获取所有必要信息。子问题分解将一个复杂问题拆解为多个可以独立检索和回答的子问题，再将结果汇总。

python 
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, Send
import operator
from typing import Annotated

class DecompositionState(TypedDict):
    original_question: str
    sub_questions: list[str]
    sub_answers: Annotated[list[dict], operator.add]  # Reducer：并行收集子答案
    final_answer: str

class SubQuestionState(TypedDict):
    sub_question: str
    sub_answers: Annotated[list[dict], operator.add]

def decompose_question(state: DecompositionState) -> Command:
    """将复杂问题分解为子问题"""
    response = llm.invoke([
        SystemMessage(content="""将给定的复杂问题分解为2-4个更简单的子问题，
每个子问题应该可以独立检索和回答，子问题的答案组合起来可以回答原始问题。
每行输出一个子问题，不加序号或其他符号。
如果问题本身不复杂，只输出原始问题本身。"""),
        HumanMessage(content=state["original_question"])
    ])

    sub_questions = [q.strip() for q in response.content.strip().split("\n") if q.strip()]
    if not sub_questions:
        sub_questions = [state["original_question"]]

    return Command(
        goto=[Send("answer_sub_question", {"sub_question": q, "sub_answers": []}) for q in sub_questions],
        update={"sub_questions": sub_questions}
    )

def answer_sub_question(state: SubQuestionState) -> dict:
    """检索并回答单个子问题"""
    docs = vectorstore.similarity_search(state["sub_question"], k=3)
    context = "\n\n".join([doc.page_content for doc in docs])

    response = llm.invoke([
        SystemMessage(content="基于提供的文档简洁回答问题。如果文档不足以回答，说明文档未涵盖此内容。"),
        HumanMessage(content=f"文档：\n{context}\n\n问题：{state['sub_question']}")
    ])

    return {
        "sub_answers": [{
            "question": state["sub_question"],
            "answer": response.content,
        }]
    }

def synthesize_answer(state: DecompositionState) -> dict:
    """汇总所有子答案，生成最终答案"""
    qa_pairs = "\n\n".join([
        f"子问题：{sa['question']}\n子答案：{sa['answer']}"
        for sa in state["sub_answers"]
    ])

    response = llm.invoke([
        SystemMessage(content="基于以下子问题的答案，综合回答原始问题。保持答案简洁、连贯。"),
        HumanMessage(content=f"原始问题：{state['original_question']}\n\n{qa_pairs}")
    ])

    return {"final_answer": response.content}

# 构建图
builder = StateGraph(DecompositionState)
builder.add_node("decompose_question", decompose_question)
builder.add_node("answer_sub_question", answer_sub_question)
builder.add_node("synthesize_answer", synthesize_answer)

builder.add_edge(START, "decompose_question")
builder.add_edge("answer_sub_question", "synthesize_answer")
builder.add_edge("synthesize_answer", END)

multihop_graph = builder.compile()

result = multihop_graph.invoke({
    "original_question": "LangGraph 和 LangChain 的关系是什么？分别适合什么场景？",
    "sub_questions": [],
    "sub_answers": [],
    "final_answer": "",
})
print(result["final_answer"])

1.5 查询扩展：Query Expansion

查询扩展（Query Expansion）通过生成原始查询的多个同义表达变体，同时检索，合并结果，提升召回率。

python 
def query_expansion_retrieve(query: str, k: int = 4) -> list[Document]:
    """查询扩展：生成多个变体，合并检索结果"""

    # 生成查询变体
    response = llm.invoke([
        SystemMessage(content="""为给定的检索查询生成3个不同的表述变体：
1. 使用同义词或近义词改写
2. 从不同角度描述同一需求
3. 使用更技术性或更口语化的表达

每行输出一个变体，不加序号。"""),
        HumanMessage(content=f"原始查询：{query}")
    ])

    variants = [query] + [
        q.strip() for q in response.content.strip().split("\n") if q.strip()
    ]

    # 对每个变体检索
    all_docs = []
    seen_contents = set()

    for variant in variants:
        results = vectorstore.similarity_search(variant, k=k)
        for doc in results:
            content_key = doc.page_content[:150]
            if content_key not in seen_contents:
                seen_contents.add(content_key)
                all_docs.append(doc)

    # 返回去重后的前 k*2 个结果（实际使用时可以加入重排序）
    return all_docs[:k * 2]

1.6 各技术的成本与收益对比

技术 额外 LLM 调用 检索延迟增加 召回率提升 精确率影响 最适合场景
HyDE 1 次（生成假设答案） 低 中高 基本无影响 Query 风格与文档差异大
Step-back Prompting 1 次（生成通用问题） 低 中 略有提升 具体实例 vs 通用原理
子问题分解 1 次分解 + N 次子问答 高（并行可优化） 高 高 复杂多跳问题
查询扩展 1 次（生成变体） 中（多次检索） 高 略降（更多噪音） 提升召回率优先
组合使用 2-3 次 中高 最高 取决于后处理 高质量要求场景

成本收益原则：

  • 响应时间敏感的场景（如实时对话）：优先选 HyDE，单次额外调用，改善最显著。
  • 高质量知识库问答（可接受较高延迟）：子问题分解 + 重排序的组合效果最好。
  • 通用增量优化：查询扩展可作为其他策略的补充，与 HyDE 或 Step-back 叠加使用。

1.7 查询改写完整流程图

简单问题
单跳回答

具体实例问题
需要通用原理

复杂多跳问题
需要多个信息源

低召回率场景
需要提升覆盖

用户原始 Query

问题分析

HyDE
生成假设性答案
用答案向量检索

Step-back
退一步抽象
双路检索

子问题分解
并行检索各子问题
Send API 实现

查询扩展
生成同义变体
合并检索结果

结果合并
去重 + 重排序

Re-ranking
Cross-encoder 精排

Top-K 上下文

LLM 生成答案

最终答案

1.8 组合使用策略

在实际系统中，多种技术可以按需叠加。以下是一个实用的组合检索器：

python 
from enum import Enum
from typing import Literal

class QueryStrategy(Enum):
    STANDARD = "standard"
    HYDE = "hyde"
    STEPBACK = "step_back"
    MULTIHOP = "multihop"
    EXPANDED = "expanded"

def intelligent_retrieve(
    query: str,
    strategy: Literal["auto", "hyde", "step_back", "multihop", "expanded"] = "auto",
    k: int = 4,
) -> list[Document]:
    """
    智能检索：根据策略选择最适合的查询改写方式。
    strategy="auto" 时，由 LLM 自动判断最优策略。
    """

    if strategy == "auto":
        # 让 LLM 判断最合适的策略
        response = llm.invoke([
            SystemMessage(content="""根据问题特征选择最合适的检索策略：
- hyde：问题是问句形式，与文档表述风格差异较大
- step_back：问题过于具体，需要从通用原理角度检索
- multihop：问题复杂，需要组合多个信息点才能回答
- expanded：问题较简单，但需要提高召回覆盖率
- standard：直接检索即可

只输出策略名称，不加其他内容。"""),
            HumanMessage(content=f"问题：{query}")
        ])
        strategy = response.content.strip().lower()
        if strategy not in ["hyde", "step_back", "multihop", "expanded"]:
            strategy = "standard"

    if strategy == "hyde":
        return hyde_retrieve(query, k=k)
    elif strategy == "step_back":
        result = step_back_retrieve(query, vectorstore, k=k)
        return result["documents"]
    elif strategy == "multihop":
        # 对简单场景的轻量版多跳检索（不使用完整 LangGraph 图）
        sub_docs = []
        sub_response = llm.invoke([
            HumanMessage(content=f"将以下问题拆解为2-3个子问题，每行一个：\n{query}")
        ])
        for sub_q in sub_response.content.strip().split("\n")[:3]:
            if sub_q.strip():
                sub_docs.extend(vectorstore.similarity_search(sub_q.strip(), k=2))
        return sub_docs[:k]
    elif strategy == "expanded":
        return query_expansion_retrieve(query, k=k)
    else:
        return vectorstore.similarity_search(query, k=k)


# 集成到 RAG 链
from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "基于以下文档内容准确回答问题：\n\n{context}"),
    ("human", "{question}"),
])

def format_docs(docs: list[Document]) -> str:
    return "\n\n---\n\n".join([doc.page_content for doc in docs])

rag_chain = (
    {
        "context": RunnableLambda(lambda x: format_docs(
            intelligent_retrieve(x["question"], strategy="auto")
        )),
        "question": lambda x: x["question"],
    }
    | prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke({"question": "什么是 RAG 系统中的 Chunking 策略？"})
print(answer)

查询改写最大的优点是不需要重建索引。知识库不用动，只改查询这一端，代价低，适合在已有系统上快速验证效果。

←上一节Self-RAG与CRAG-让RAG系统自我纠错下一节文档预处理全攻略-OCR-PDF解析与表格提取→
本页目录