什么是HyDE？如何利用假设性文档来提升RAG的质量？

检索增强生成技术，通常被称为RAG，已成为利用大型语言模型构建应用程序时最常用的方法之一。

与让大型语言模型完全依据其训练数据来回答问题不同，RAG系统会从外部知识库中检索相关信息，并将这些信息作为上下文提供给模型。

其基本原理非常简单：

• 将用户的问题转化为嵌入向量。

• 在向量数据库中搜索语义上相似的文档片段。

• 将检索到的这些片段传递给大型语言模型。

• 基于这些片段生成答案。

然而，这个看似简单的过程存在一个重大缺陷：用户提出的问题与包含答案的文档在表达方式上可能存在很大差异。

例如，用户可能会提出这样的问题：

为什么我的AWS Glue作业在处理了几百万条记录后速度会显著下降？

而知识库中相关的文档则可能会这样描述：

当Spark执行器遇到过多的数据洗牌操作、分布不均的数据分区、内存压力过大，或者频繁需要将数据写入磁盘时，性能就会下降。

虽然这两个问题讨论的是同一个现象，但它们使用的词汇、表达结构以及细节层次都不同。因此，直接使用查询生成的嵌入向量可能无法使它们在嵌入空间中处于相近的位置。

正是为了解决这个问题，人们才设计了“假设性文档嵌入技术”或HyDE。

目录

• 先决条件

• 什么是HyDE？

• HyDE的原理机制

• 最小化实现方案

• 为什么“幻觉”机制不会自动破坏HyDE的功能？

• 生产环境中的使用注意事项

• 总结

先决条件

要想充分理解本文的内容，你需要了解并掌握以下知识。

你需要了解的内容包括：

• 对RAG技术及其应用场景有基本的了解。

• 从概念层面理解向量嵌入技术的原理。

• 具备Python编程语言的相关知识。

你还需要具备以下条件：

• 一个安装了numpy、sentence-transformers以及Anthropic的本地Python环境。

• 如果想要运行HyDE代码示例，还需要一个Anthropic API密钥（该密钥可在console.anthropic.com获取）。

什么是HyDE？

HyDE的全称是“假设性文档嵌入技术”。这种技术的原理非常简单：在查询时，会让大型语言模型生成一份能够回答用户问题的假设性文档，然后使用这份文档而不是原始查询内容来索引数据库进行搜索。整个流程就是这样的，其余的部分都属于工程实现的细节。

图2：HyDE的工作流程

这份假设性文档并不是最终答案，它仅仅起到了连接用户查询与知识库中存储的实际文档的桥梁作用。

这一区别非常重要。

生成的文档可能会包含不正确的信息，但这并不一定意味着系统失败了，因为系统并不会直接将这份文档呈现给用户。它的目的仅仅是生成一个更丰富、更具语义意义的信息表示形式而已。

最初的HyDE方案使用语言模型来生成假设性文档，并通过无监督的密集检索算法将这些文档映射到嵌入空间中。这种嵌入向量实际上起到了搜索指令的作用，用于从语料库中检索出实际对应的文档。

为什么HyDE能够发挥作用

这一机制背后的直觉是几何学上的。密集检索算法会将文本映射到语义空间中，而两段文本之间的相似度实际上就是它们对应向量之间夹角的余弦值。

当我们将一个查询语句进行嵌入处理后，再将其与一篇文档进行比较时，其实是在测量两种本质上并不应该相近的文本形式之间的“角度”。我们的嵌入模型虽然被训练用来将语义上相似的文本放在一起，但它并没有被训练来将查询语句与其对应的答案放在一起——因为这两者属于完全不同的类型。

HyDE通过让比较的两方具有相同的结构，从而弥补了这一缺陷。假设性文档与实际文档在向量空间中处于相同的区域，因为它们使用的是相同的语言风格、词汇表以及描述细节的深度。这样一来，向量搜索实际上就是在将答案与答案进行比较，因此得到的相似度信号也会更加准确。

整个机制就是这样的。其余的所有环节——比如提示语的设计、模型选择以及缓存策略等——其实都是建立在这个几何学原理基础之上的。

HyDE的运作机制

首先，假设用户提出了这样的问题：为什么我的Lambda函数在很长时间没有被调用之后，响应速度会变慢呢？

然后你可以用这样一个简短的指令来向大型语言模型提出这个问题：“写一篇来自技术文档的文字，用来回答这个问题。”

大语言模型会给出如下类似的回答：

“AWS Lambda会回收那些已经闲置了一段时间的执行环境。当函数被再次调用时，就会发生‘冷启动’现象，这时系统需要重新设置运行环境并加载依赖项，因此在这次调用之前会存在额外的延迟。”

现在你需要将这段生成的内容嵌入到你的系统中。注意，这里指的是生成出来的文本，而不是原始的问题本身。

你可以利用这种嵌入技术来搜索你的向量存储系统。由于这个假设性的文本是按照真实文档的格式编写的，因此在实际的AWS文档中，与这些假设性文本相关的信息在向量空间中会位于彼此附近的位置。

接下来，你需要取出排名前k条的搜索结果，并将它们连同原始的用户问题一起传递给生成器。生成器会利用这些真实的文档来生成答案，而那些假设性的内容则会被忽略掉。

大语言模型被使用了两次，但用途不同：第一次是用来将用户的问题转换成文档格式；第二次则是利用这些生成的文档来回答问题。其中，第一次调用成本较低，风险也较小；而第二次调用才真正具有实际意义。

图3：Naive RAG与HyDE处理流程的对比。

最小化实现方案

Naive RAG的处理流程可能如下所示：

import numpy as np
from sentence_transformers import SentenceTransformer

collection = [
    "AWS Lambda会在函数闲置一段时间后回收这些执行环境，因此下次调用时会发生‘冷启动’现象，需要重新设置运行环境和加载依赖项。",
    "Apache Airflow使用有向无环图来安排任务，其中每个节点代表一项工作任务。",
    "AWS Glue会从源数据中推断出数据结构，并自动将这些信息添加到Glue数据目录中。",
    "Amazon Bedrock通过一个统一的API提供基础模型，并负责透明的资源管理任务。",
    "DynamoDB会根据分区键将数据分配到不同的节点上，从而决定数据的物理存储位置。"
]

embedder = SentenceTransformer("all-MiniLM-L6-v2")
collection_embeddings = embedder.encode(collection, normalize_embeddings=True)

def retrieve(query: str, k: int = 2) -> list[str]:
    query_embedding = embedder.encode(query, normalize_embeddings=True)
    scores = collection_embeddings @ query_embedding
    top_k = np.argsort(scores)[::-1][:k]
    return [collection[i] for i in top_k]

query = "为什么我的Lambda函数在一段时间没有被调用后，响应时间会变长呢？"
for passage in retrieve(query):
    print(passage)

在这个示例集合中，系统很可能会返回排名第一的正确答案。但如果将文档数量增加到五万份，并且查询语句的多样性也增加，那么正确答案在排名中的位置就很可能下降。

接下来需要注意的那一行代码，是位于`retrieve`函数内部、执行`embedder.encode(query, ...)这一操作的那一行。正是在这一行中，原始问题被转换成了向量形式，而HyDE也正是修改了这一行代码。

在HyDE版本中，用于生成假设性回答的函数只有一个：

import numpy as np
from anthropic import Anthropic
from sentence_transformers import SentenceTransformer

# collection。在实际应用中，这个变量代表你的向量存储库。

collection = [
    "AWS Lambda会在函数长时间未被调用后回收空闲的执行环境，因此下次调用该函数时会导致冷启动，从而需要重新加载运行时所需资源及依赖项。",
    "Apache Airflow使用有向无环图来安排任务，其中每个节点代表一项工作任务。",
    "AWS Glue会从源数据中推断出数据结构，并自动将这些信息添加到Glue数据目录中。",
    "Amazon Bedrock通过一个统一的API提供基础模型，并能够透明地处理这些模型的配置与管理。",
    "DynamoDB会根据分区键将数据分配到不同的节点上，从而决定数据的物理存储位置。",
]

embedder = SentenceTransformer("all-MiniLM-L6-v2")
corpus_embeddings = embedder.encode(collection, normalize_embeddings=True)

client = Anthropic()

# HyDE：首先生成一个假设性的回答文本，将其转换成向量形式，然后再使用这个向量在数据集合中进行搜索。

HYDE_PROMPT = (
    "请编写一段来自技术文档的简短文字，用来回答以下问题。写作风格应符合官方文档的要求：表述清晰、准确无误，不要使用模糊或回避性的语言。只需要提供答案段落，长度为两到四句话。\n\n
    问题：{query}"
)

def generate_hypothetical(query: str) -> str:
    """让大语言模型生成一段用于回答该问题的虚假文档文字。"""
    message = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=200,
        messages=[
            {"role": "user", "content": HYDE_PROMPT.format(query=query)}
        ],
    )
    return message.content[0].text

def retrieve_hyde(query: str, k: int = 2) -> list[str]:
    """生成一个假设性的回答文本，将其转换成向量形式，然后使用这个向量在数据集合中进行搜索。」
    hypothetical = generate_hypothetical(query)
    hyde_embedding = embedder.encode(hypothetical, normalize_embeddings=True)
    scores = corpus embeddings @ hyde_embedding
    top_k_indices = np.argsort(scores)[::-1][:k]
    return [collection[i] for i in top_k_indices]

if __name__ == "__main__":
    query = (
        "为什么我的Lambda函数在很长时间没有被调用后，响应时间会变长呢？"
    )
    for passage in retrieve_hyde(query):
        print(passage)

整个技术流程就是这样的。其中多了一个调用大语言模型的步骤，多了一个额外的函数，但其他部分都与基线版本完全相同。生成出来的假设性回答文本在被转换成向量形式后就会被丢弃，永远不会被传回到生成器中。

而基线版本的实现方式是直接将问题本身转换为向量形式，然后使用余弦相似度算法在数据集合中搜索匹配的答案。正是这一行代码——`embedder.encode(query, ...)`——使得问题被转换成了与答案格式不同的向量形式，而这正是本文所讨论的、导致检索效果不佳的原因所在。

HyDE方法的区别仅在于一点：在进行嵌入处理之前，会要求大型语言模型生成一段用于技术文档的文本，这段文本旨在回答某个问题，因此计算得到的向量实际上是针对这段文本生成的，而不是针对原始问题本身。其他所有环节都保持不变——仍然使用相同的嵌入模型、余弦相似度搜索算法以及top-k选择机制。

这段假设性的文本仅被用来生成用于搜索的向量，并不会被用于其他任何目的。这种差异并非源于检索方法本身的不同，而仅仅是因为用于比较的文本形式发生了变化而已。

为什么“幻觉”机制并不会自动破坏HyDE的效果

乍一看，HyDE似乎有些矛盾之处：为什么让语言模型在检索事实之前先生成相关文本，反而能提高事实检索的效率呢？

答案在于，HyDE将生成的文本作为检索结果的表现形式，而非视为可靠的知识来源。

假设用户询问：“7月18日数据库中断的原因是什么？”大型语言模型无法从内部事故报告中得知真正的原因，因此它必须自己编造一个解释。

于是它可能会这样回答：

“7月18日的数据库中断是由于主副本上的故障转移设置出现错误，从而导致依赖服务出现了连锁性的连接超时问题。工程师们通过将流量重新路由到备用区域并重建连接池，才恢复了服务。”

这段描述完全是编造出来的。实际上，导致中断的原因可能是硬盘故障、部署错误、证书过期等等。但请注意，这段文字中出现的词汇——如“中断”、“故障转移”、“备份副本”、“连锁超时”等——无论真正原因是什么，这些术语都会出现在任何关于数据库事故的分析报告中。

无论具体原因是什么，数据库事故的分析报告在词汇选择、表达方式和结构上都是相似的。

大型语言模型生成的这段描述可能还会涉及到连接饱和、锁竞争、存储延迟、部署失败或资源耗尽等问题。其中一些细节可能是错误的，但这一点并不重要。因为这些术语本身就会使生成的结果落在与真实事故分析报告相同的“词空间”中。

当将这段编造的文本进行嵌入处理后，得到的向量就会位于与真实事故分析报告相同的区域范围内。通过向量搜索，系统就能找到正确的分析结果；只有到了这个时候，生成器才会读取真实的文档并给出准确的答案。

虽然这个假设在事实描述上是不正确的，但它所关注的“文本结构”却是正确的。嵌入过程看到的是文本的结构，而检索到的文档所提供的则是具体的事实信息。

这里真正的风险并不在于这种“幻觉”机制本身，而在于人们会如何利用这种机制。如果系统错误地将这段编造的文本当作真实的信息传递给最终答案生成器，那么用户就会得到错误的结论。

这种缓解措施属于架构层面的，而非统计方法：必须将假设内容严格限制在检索阶段之内，绝不允许其影响到生成过程。下一节将会详细阐述这一点。

生产过程中的安全防护机制

HyDE在检索路径中加入了大语言模型，这就带来了一些新的技术挑战。以下是一些可以采取的安全防护措施，这些措施能够提升系统的安全性与可靠性：

设置超时机制与备用方案

如果假设生成过程耗时过长或出现故障，应立即切换回传统的检索方式，而不要让用户因此受到影响。

def retrieve_with_fallback(query: str, k: int = 2) -> list[str]:
    try:
        hypothetical = generate_hypothetical(query)
        search_vector = embedder.encode(hypothetical, normalize_embeddings=True)
    except Exception:
        logger.exception(
            "HyDE生成失败；将切换回原始查询方式。"
        )
        # 切换回对原始查询内容的编码处理
        search_vector = embedder.encode(query, normalize.embeddings=True)

    scores = corpus embeddings @ search_vector
    top_k = np.argsort(scores)[::-1][:k]
    return [collection[i] for i in top_k]

可以在客户端本身设置明确的超时时间，例如 [Anthropic(timeout=3.0)]。

限制生成内容的长度

过长的假设生成内容往往会引入无关的概念，从而导致嵌入结果的质量下降。因此应限制大语言模型生成的输出长度。

message = client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=200,   # 限制生成内容的长度
    messages=[{"role": "user", "content": HYDE_PROMPT.format(query=query)}],
)

对于技术文档领域而言，200个标记通常就已经足够用来生成一条目标文本了；超过这个数量的话，检索难度往往会增加。

在将数据发送给外部模型提供商之前，必须保护其中包含的敏感信息

在执行假设生成操作之前，必须从输入数据中去除所有个人身份识别信息，并且这一措施必须在接口层面得到严格执行，而不能依赖后续的处理流程。

PII_PATTERNS = {
    "email": r'\b[\w.-]+@[\w.-]+\.\w+\b',
    "ssn":   r'\b\d{3}-\d{2}-\d{4}\b',
    "card":  r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',
}

def scrub_pii(text: str) -> str:
    for label, pattern in PII_PATTERNS.items():
        text = re.sub(pattern, f"[REDACTED_{label.upper()}]", text)
    return text

def safe_generate_hypothetical(query: str) -> str:
    return generate_hypothetical(scrub_pii(query))

对于受监管的数据而言，这些措施就是最低要求；在实际应用中，还可以增加更多的安全控制措施。

追踪每一个处理环节

如果无法了解每一个处理阶段的详细情况，就根本无法排查检索过程中出现的问题。因此，必须收集所有查询请求的相关信息，包括查询内容、提示语、生成的假设结果、处理耗时、检索到的数据以及相似度评分等。import time import logging logger = logging.getLogger(__name__) def traced_retrieve_hyde(query: str, k: int = 2) -> HyDEContext: t0 = time.time() hypothetical = generate_hypothetical(query) gen_ms = int((time.time() - t0) * 1000) t1 = time.time() search_vector = embedder.encode(hypothetical, normalize_embeddings=True) embed_ms = int((time.time() - t1) * 1000) scores = corpus embeddings @ search_vector top_k = np.argsort(scores)[::-1][:k] logger.info( "hyde_retrieval", extra={ "query": query, "prompt_version": "v1", "hypothetical": hypothetical, "gen_latency_ms": gen_ms, "embed_latency_ms": embed_ms, "retrieved_ids": top_k.tolist(), "similarity_scores": [float(scores[i]) for i in top_k], }, ) return HyDEContext( original_query=query, hypothetical=hypothetical, retrieved_documents=[collection[i] for i in top_k], )

这种结构化的日志记录是构建延迟监控面板、检测系统异常以及进行离线检索评估的基础。

何时使用HyDE，何时不应使用它

在以下情况下可以使用HyDE：

• 当你的嵌入模型无法充分理解你所处理的领域时。

• 当你没有标记好的查询-文档对来微调检索系统时。

• 当用户提出的是非结构化问题，而你的文档内容较为正式或技术性较强时。

• 当你愿意在检索之前多调用一次大型语言模型时。

在以下情况下应避免使用HyDE：

• 当你的应用对延迟有严格要求时。

• 当通用的大型语言模型可能会生成与领域内容不符的术语时。

• 当你的查询中已经包含了明确的关键词、标识符或错误代码时。

• 当使用BM25或混合搜索方法已经能够获得相关结果时。

• 当你拥有足够的标记数据可以直接微调检索系统时。

总结

HyDE虽然只是一个简单的改进措施，但它却能产生显著的效果。你并不需要更改索引、嵌入模型或生成器，只需要修改其中一行代码——即当查询到来时究竟应该嵌入哪些信息。这一小小的改变就能彻底改变搜索的机制，从“问题与答案”的匹配方式转变为“答案与答案”之间的对比，进而提升检索效果。

这种技术并非神奇之物。它通过牺牲一定的延迟和计算成本来提高召回率，而只有当查询与文档之间的信息不对称性确实是整个系统中的瓶颈时，这种方法才会真正发挥作用。在这种情况下，HyDE无疑是RAG工具箱中性价比最高的解决方案之一。