HyDE (2022) - 零样本密集检索技术详解

约 3314 字大约 11 分钟

论文HyDE密集检索零样本学习

论文基本信息

标题: Precise Zero-Shot Dense Retrieval without Relevance Labels
作者: Luyu Gao, Xueguang Ma, Jimmy Lin, Jamie Callan
机构: University of Waterloo, Carnegie Mellon University
年份: 2022
发表: arXiv:2212.10496
论文链接: arXiv论文
官方代码: GitHub仓库
引用次数: 500+ (截至2025年)

核心贡献

HyDE（Hypothetical Document Embeddings）是一种创新的零样本密集检索方法，通过巧妙地结合大型语言模型和无监督编码器，实现了无需相关性标签的高效检索。主要贡献包括：

无需标注数据：完全不需要相关性标签，解决了密集检索的数据依赖问题
假设文档生成：利用语言模型生成捕获相关性模式的假设文档
两步检索流程：生成 → 编码 → 检索，简单而高效
显著性能提升：大幅优于现有无监督检索器Contriever，接近有监督方法

架构概述

HyDE工作流程

核心思想

HyDE的核心洞察是：直接对查询和文档进行语义匹配很难，但可以通过一个"中介"来桥接查询和文档的语义鸿沟。

查询通常很短，信息有限
假设文档是由LLM生成的、看似合理的答案
编码器将假设文档和真实文档都映射到统一的向量空间
密集瓶颈会过滤掉假设文档中的错误细节，只保留语义相关性

技术细节分析

步骤1：假设文档生成

使用指令遵循语言模型（如InstructGPT）生成假设文档：

from openai import OpenAI

client = OpenAI(api_key="your-api-key")

def generate_hypothetical_document(query: str) -> str:
    """
    为给定查询生成假设文档
    """
    prompt = f"""请根据以下查询生成一个详细的答案。

查询：{query}

答案："""

    response = client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=256,
        temperature=0.7
    )

    return response.choices[0].message.content

# 示例
query = "什么是量子计算？"
hyp_doc = generate_hypothetical_document(query)
print(f"假设文档：{hyp_doc}")

关键点：

假设文档不需要完全准确，只需要捕捉相关性的"模式"
LLM会生成看似合理的结构和内容
这些不准确的细节会在后续步骤中被过滤

步骤2：编码与检索

使用无监督对比学习编码器生成嵌入：

from sentence_transformers import SentenceTransformer
import faiss
import numpy as np

class HyDERetriever:
    def __init__(self, encoder_name="facebook/contriever"):
        """
        初始化HyDE检索器
        """
        self.encoder = SentenceTransformer(encoder_name)
        self.index = None
        self.documents = None

    def build_index(self, documents: list[str]):
        """
        为文档集合构建FAISS索引
        """
        self.documents = documents
        doc_embeddings = self.encoder.encode(documents)

        # 创建FAISS索引（内积搜索）
        self.index = faiss.IndexFlatIP(doc_embeddings.shape[1])
        self.index = faiss.IndexIDMap(self.index)
        self.index.add_with_ids(
            doc_embeddings,
            np.arange(len(documents))
        )

    def retrieve(self, query: str, top_k: int = 10) -> list[tuple[int, float]]:
        """
        使用HyDE进行检索
        """
        # 1. 生成假设文档（实际应用中可调用LLM）
        # 这里简化为使用查询本身
        hypothetical_doc = self._generate_hypothetical(query)

        # 2. 编码假设文档
        query_embedding = self.encoder.encode([hypothetical_doc])

        # 3. 向量检索
        scores, indices = self.index.search(query_embedding, top_k)

        results = []
        for idx, score in zip(indices[0], scores[0]):
            results.append((int(idx), float(score)))

        return results

    def _generate_hypothetical(self, query: str) -> str:
        """
        生成假设文档（简化版本）
        实际应用中应调用LLM
        """
        # 这里可以使用查询扩展或其他技术
        return f"{query} 相关的详细信息和解释"

# 使用示例
documents = [
    "量子计算是一种利用量子力学原理进行计算的新型计算模式",
    "经典计算机使用比特作为基本信息单位，而量子计算机使用量子比特",
    "量子纠缠和量子叠加是量子计算的两大核心特性",
    # ... 更多文档
]

retriever = HyDERetriever()
retriever.build_index(documents)

query = "量子计算的基本原理是什么？"
results = retriever.retrieve(query, top_k=3)

for idx, score in results:
    print(f"文档 {idx}: {documents[idx]} (相关度: {score:.4f})")

密集瓶颈的过滤机制

工作原理：

编码器的输出是一个固定长度的密集向量（例如768维）
这个"瓶颈"迫使模型压缩信息，只能保留最重要的语义特征
假设文档中的具体细节（可能是错误的）被过滤掉
只保留与查询相关的语义模式

性能评估

主要实验结果

HyDE在多个标准检索数据集上进行了评估：

数据集	任务类型	HyDE	Contriever	BM25	有监督检索器
MS MARCO	Web搜索	32.1	24.3	21.5	35.8
Natural Questions	问答	45.2	38.7	32.1	48.3
BEIR	混合任务	42.8	35.4	28.9	46.7
FEVER	事实验证	67.5	58.2	52.3	71.2

多语言性能

语言	任务	HyDE	Contriever
英语	MS MARCO	32.1	24.3
斯瓦希里语	Swahili-Squad	28.7	21.5
韩语	Ko-StrategyQA	35.2	27.8
日语	Ja-QuAC	31.9	25.4
德语	German-QuAC	33.6	26.7

关键发现：

HyDE在所有语言上都显著优于Contriever
零样本跨语言迁移能力强
性能接近有监督的微调检索器

不同查询长度的影响

查询类型	平均长度	HyDE	Contriever
短查询	3-5词	29.3	24.7
中查询	6-10词	33.8	26.5
长查询	11+词	36.2	28.1

实践应用

完整HyDE实现示例

from typing import List, Tuple
import numpy as np
from sentence_transformers import SentenceTransformer
import faiss

class CompleteHyDERetriever:
    """完整的HyDE检索器实现"""

    def __init__(
        self,
        encoder_name: str = "facebook/contriever",
        llm_generator = None  # 可以是任何LLM接口
    ):
        self.encoder = SentenceTransformer(encoder_name)
        self.llm_generator = llm_generator
        self.index = None
        self.documents = None

    def build_index(self, documents: List[str]):
        """构建文档索引"""
        self.documents = documents
        doc_embeddings = self.encoder.encode(
            documents,
            show_progress_bar=True
        )

        # 归一化（使用内积作为相似度）
        faiss.normalize_L2(doc_embeddings)

        # 创建FAISS索引
        self.index = faiss.IndexFlatIP(doc_embeddings.shape[1])
        self.index.add(doc_embeddings)

    def generate_hypothetical(self, query: str) -> str:
        """
        使用LLM生成假设文档

        实际应用中，这里应该调用真实的LLM API
        """
        if self.llm_generator:
            return self.llm_generator(query)

        # 简化版本：使用查询扩展
        prompt = f"""请详细回答以下问题，提供全面的解释和相关信息。

问题：{query}

详细回答："""

        # 这里应该调用实际的LLM
        # 示例使用伪代码：
        # return call_llm_api(prompt)

        # 临时返回查询本身
        return query

    def retrieve(
        self,
        query: str,
        top_k: int = 10
    ) -> List[Tuple[int, float, str]]:
        """
        执行HyDE检索

        返回: [(doc_id, score, document_text), ...]
        """
        # 1. 生成假设文档
        hypothetical_doc = self.generate_hypothetical(query)

        # 2. 编码
        query_embedding = self.encoder.encode([hypothetical_doc])
        faiss.normalize_L2(query_embedding)

        # 3. 检索
        scores, indices = self.index.search(query_embedding, top_k)

        # 4. 格式化结果
        results = []
        for score, idx in zip(scores[0], indices[0]):
            results.append((
                int(idx),
                float(score),
                self.documents[int(idx)]
            ))

        return results

    def batch_retrieve(
        self,
        queries: List[str],
        top_k: int = 10
    ) -> List[List[Tuple[int, float, str]]]:
        """批量检索"""
        # 生成所有假设文档
        hypothetical_docs = [
            self.generate_hypothetical(q) for q in queries
        ]

        # 批量编码
        query_embeddings = self.encoder.encode(hypothetical_docs)
        faiss.normalize_L2(query_embeddings)

        # 批量检索
        scores, indices = self.index.search(query_embeddings, top_k)

        # 格式化结果
        all_results = []
        for query_scores, query_indices in zip(scores, indices):
            results = []
            for score, idx in zip(query_scores, query_indices):
                results.append((
                    int(idx),
                    float(score),
                    self.documents[int(idx)]
                ))
            all_results.append(results)

        return all_results

# 使用示例
if __name__ == "__main__":
    # 准备文档集合
    documents = [
        "机器学习是人工智能的一个分支，它使计算机能够在没有明确编程的情况下学习。",
        "深度学习是机器学习的一个子领域，使用多层神经网络。",
        "自然语言处理（NLP）是AI的一个重要应用领域，涉及计算机与人类语言之间的交互。",
        "Transformer是一种基于自注意力机制的神经网络架构，彻底改变了NLP领域。",
        "BERT和GPT是两种流行的预训练语言模型，基于Transformer架构。",
        # ... 更多文档
    ]

    # 创建检索器
    retriever = CompleteHyDERetriever()
    retriever.build_index(documents)

    # 单个查询
    query = "什么是深度学习？它与机器学习有什么关系？"
    results = retriever.retrieve(query, top_k=3)

    print(f"查询: {query}\n")
    print("检索结果:")
    for rank, (idx, score, doc) in enumerate(results, 1):
        print(f"{rank}. [{score:.4f}] {doc}")

    # 批量查询
    queries = [
        "解释Transformer的工作原理",
        "NLP有哪些应用场景？",
        "BERT和GPT的区别是什么？"
    ]

    batch_results = retriever.batch_retrieve(queries, top_k=2)
    print("\n批量检索结果:")
    for query, results in zip(queries, batch_results):
        print(f"\n查询: {query}")
        for rank, (idx, score, doc) in enumerate(results, 1):
            print(f"  {rank}. [{score:.4f}] {doc}")

与RAG系统集成

HyDE在检索增强生成（RAG）系统中特别有用：

class HyDE_RAG_System:
    """结合HyDE的RAG系统"""

    def __init__(self, retriever, llm_client):
        self.retriever = retriever
        self.llm_client = llm_client

    def query(self, user_query: str, top_k: int = 5) -> str:
        """
        使用RAG回答用户查询
        """
        # 1. 使用HyDE检索相关文档
        retrieved_docs = self.retriever.retrieve(user_query, top_k)

        # 2. 构建上下文
        context = "\n\n".join([
            f"文档 {i+1}: {doc}"
            for i, (_, _, doc) in enumerate(retrieved_docs)
        ])

        # 3. 使用LLM生成答案
        prompt = f"""基于以下参考文档回答用户问题。

参考文档：
{context}

用户问题：{user_query}

请提供详细的答案："""

        response = self.llm_client.generate(prompt)
        return response

# 使用示例
# rag_system = HyDE_RAG_System(hyde_retriever, gpt4_client)
# answer = rag_system.query("什么是量子纠缠？")
# print(answer)

与其他方法的比较

vs. 稀疏检索（BM25）

特性	HyDE	BM25
匹配方式	语义向量匹配	关键词匹配
需要训练	无（使用预训练模型）	无
词汇鸿沟	✅ 解决	❌ 存在
性能	高	中等
效率	需要GPU	CPU即可
可解释性	低	高

优势：HyDE通过语义理解解决了词汇鸿沟问题，能找到语义相关但关键词不匹配的文档。

vs. 有监督密集检索

特性	HyDE	有监督方法
需要标注数据	❌ 不需要	✅ 需要
性能	接近有监督	最高
部署成本	低	高（数据标注）
跨域迁移	✅ 强	⚠️ 需要微调
训练时间	无需训练	长