RAG 进阶：HyDE、Multi-HyDE 与 Adaptive HyDE 深度解析与实战

# 基础模型配置

BASIC_MODEL_BASE_URL=...
BASIC_MODEL_MODEL=...
BASIC_MODEL_API_KEY=...

EMBEDDING_MODEL_BASE_URL=...
EMBEDDING_MODEL_MODEL=...
EMBEDDING_MODEL_API_KEY=...

[project]
name = "hyde-samples"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "chromadb>=1.4.0",
    "dashscope>=1.25.7",
    "fastembed>=0.4.2",
    "langchain>=1.2.3",
    "langchain-community>=0.4.1",
    "langchain-openai>=1.1.7",
    "onnxruntime<1.20",
    "python-dotenv>=1.2.1"
]

import os
import time
import asyncio
from typing import List
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 加载环境变量
load_dotenv()

# 获取配置
base_url = os.getenv("BASIC_MODEL_BASE_URL")
api_key = os.getenv("BASIC_MODEL_API_KEY")
model_name = os.getenv("BASIC_MODEL_MODEL")

class AsyncHyDE:
    def __init__(self, llm, base_embeddings, vectorstore):
        self.llm = llm
        self.base_embeddings = base_embeddings
        self.vectorstore = vectorstore
        
        # 预定义提示模板 (Multi-Perspective Prompts)
        # 通过不同角色的设定，引导模型生成多样化的假设文档
        self.hyde_prompt_styles = [
            "请为以下查询生成一段详细的假设性回答。这段回答应该包含可能出现在相关文档中的关键词和语义信息。\n查询: {query}\n回答:",
            "请以技术专家的身份，深入解析以下查询涉及的技术原理、底层机制或核心架构。\n查询: {query}\n回答:",
            "请简明扼要地列出解决以下查询所需的关键步骤、命令或核心事实，侧重于实际操作。\n查询: {query}\n回答:",
            "请用通俗易懂的语言解释以下查询的核心概念，适合初学者理解，多用类比。\n查询: {query}\n回答:"
        ]

    async def _generate_single_doc(self, query: str, style_idx: int) -> str:
        """异步生成单个文档"""
        prompt_content = self.hyde_prompt_styles[style_idx % len(self.hyde_prompt_styles)]
        prompt = ChatPromptTemplate.from_template(prompt_content)
        chain = prompt | self.llm | StrOutputParser()
        try:
            return await chain.ainvoke({"query": query})
        except Exception as e:
            print(f"    ! 生成文档 (style={style_idx}) 时出错: {e}")
            return ""

    async def _generate_hypothetical_docs_async(self, query: str, n: int = 1) -> List[str]:
        """并行生成 N 个假设性文档"""
        print(f"  > [Async] 正在并行生成 {n} 个假设文档...")
        
        tasks = []
        for i in range(n):
            tasks.append(self._generate_single_doc(query, i))
            
        # 并发执行所有生成任务，显著降低总延迟
        docs = await asyncio.gather(*tasks)
        # 过滤掉生成失败的空文档
        return [d for d in docs if d]

    def _get_combined_embedding(self, texts: List[str]) -> List[float]:
        """计算多个文本嵌入的平均向量"""
        # 注意：LangChain 的 OpenAIEmbeddings 目前主要提供同步 embed_documents
        # 如果需要异步，可使用 aembed_documents (如果有实现)
        embeddings = self.base_embeddings.embed_documents(texts)
        if not embeddings:
            return []
        
        dim = len(embeddings[0])
        n = len(embeddings)
        # 计算平均向量，融合多视角的语义信息
        avg_embedding = [sum(e[i] for e in embeddings) / n for i in range(dim)]
        return avg_embedding

    async def multi_hyde_retrieve_async(self, query: str, k: int = 4, n: int = 3):
        """策略: Multi-HyDE (并行生成 + 平均嵌入)"""
        print(f"  [Multi-HyDE Async] 执行并行 Multi-HyDE (n={n}): {query}")
        
        # 1. 并行生成假设文档
        start_gen = time.time()
        hypo_docs = await self._generate_hypothetical_docs_async(query, n=n)
        print(f"    > 生成耗时: {time.time()-start_gen:.2f}s")
        
        for i, doc in enumerate(hypo_docs):
            print(f"    - 假设文档 {i+1}: {doc[:50]}...")
            
        if not hypo_docs:
            return []

        # 2. 计算平均嵌入 (同步)
        avg_embedding = self._get_combined_embedding(hypo_docs)
        
        # 3. 向量检索 (同步)
        return await asyncio.to_thread(
            self.vectorstore.similarity_search_by_vector, avg_embedding, k=k
        )

# --- 主程序 ---
async def main():
    if not api_key:
        print("Error: 请先配置 .env 文件中的 BASIC_MODEL_API_KEY")
        return

    # 1. 初始化组件
    print("正在初始化组件...")
    llm = ChatOpenAI(
        model=model_name,
        temperature=0.7,
        openai_api_base=base_url,
        openai_api_key=api_key
    )
    
    print("初始化 Embeddings (API Mode)...")
    base_embeddings = OpenAIEmbeddings(
        openai_api_base=os.getenv("EMBEDDING_MODEL_BASE_URL"),
        openai_api_key=os.getenv("EMBEDDING_MODEL_API_KEY"),
        model=os.getenv("EMBEDDING_MODEL_MODEL"),
        check_embedding_ctx_length=False
    )
    
    # 2. 准备数据 (复用之前的逻辑)
    texts = [
        "LangChain 是一个用于构建 LLM 应用的框架，支持 Python 和 JavaScript。",
        "HyDE (Hypothetical Document Embeddings) 利用 LLM 生成假设性文档来强化检索。",
        "RAG (Retrieval-Augmented Generation) 通过检索外部数据来增强生成模型的准确性。",
        "使用 `pip install langchain` 命令安装 LangChain。",
        "Python 中的 `list.append(x)` 方法用于在列表末尾添加元素。",
        "ChromaDB 的默认持久化目录可以通过 `persist_directory` 参数设置。",
        "大模型在处理特定领域知识时经常产生幻觉，检索增强是解决此问题的关键手段。",
        "相比于微调，RAG 具有成本更低、数据更新更及时的优势。",
        "向量数据库通过计算余弦相似度来衡量文本之间的语义距离。",
    ]
    docs = [Document(page_content=t) for t in texts]
    
    # 重建向量库
    persist_dir = "./chroma_async_db"
    if os.path.exists(persist_dir):
        import shutil
        shutil.rmtree(persist_dir)
            
    vectorstore = Chroma.from_documents(
        documents=docs,
        embedding=base_embeddings,
        persist_directory=persist_dir
    )
    print(f"向量库构建完成，共 {len(texts)} 条文档。")

    # 3. 初始化 AsyncHyDE
    hyde_system = AsyncHyDE(llm, base_embeddings, vectorstore)
    
    # 4. 测试查询
    query = "如何安装 langchain 库？"
    
    print("\n" + "="*50)
    print(f"测试查询 (Async): {query}")
    print("="*50)
    
    print("\n  >>> 方法: Multi-HyDE (Async, n=3) <<<")
    start = time.time()
    
    # 调用异步方法
    res_multi = await hyde_system.multi_hyde_retrieve_async(query, k=2, n=3)
    
    print(f"  > 总耗时: {time.time()-start:.2f}s")
    print("  > 结果:")
    for doc in res_multi:
        print(f"    * {doc.page_content}")
        
    print("-" * 30)

if __name__ == "__main__":
    asyncio.run(main())

正在初始化组件...
初始化 Embeddings (API Mode)...
向量库构建完成，共 9 条文档。

==================================================
测试查询: 如何安装 langchain 库？
==================================================

  >>> 方法 A: Direct Retrieval <<<
  [Direct] 执行直接检索: 如何安装 langchain 库？
  > 耗时: 0.32s
  > 结果:
    * 使用 `pip install langchain` 命令安装 LangChain。
    * LangChain 是一个用于构建 LLM 应用的框架，支持 Python 和 JavaScript。
------------------------------

  >>> 方法 B: Standard HyDE (n=1) <<<
  [Standard HyDE] 执行 Standard HyDE (n=1): 如何安装 langchain 库？
  > 正在生成 1 个假设文档 (使用多样化视角)...
    - 假设文档: 要安装 LangChain 库，您可以使用 Python 的包管理工具 pip。LangChain ...
  > 耗时: 18.07s
  > 结果:
    * 使用 `pip install langchain` 命令安装 LangChain。
    * LangChain 是一个用于构建 LLM 应用的框架，支持 Python 和 JavaScript。
------------------------------

  >>> 方法 C: Multi-HyDE (n=3) <<<
  [Multi-HyDE] 执行 Multi-HyDE (n=3): 如何安装 langchain 库？
  > 正在生成 3 个假设文档 (使用多样化视角)...
    - 假设文档 1: 要安装 LangChain 库，您可以使用 Python 的包管理工具 pip。LangChain ...
    - 假设文档 2: 以技术专家的身份，对“如何安装 langchain 库”这一看似简单的查询进行深入解析，需从多个维度...
    - 假设文档 3: 1. **确保已安装 Python（建议 3.8+）和 pip**  
   - 检查版本：`pyt...
  > 耗时: 81.27s
  > 结果:
    * 使用 `pip install langchain` 命令安装 LangChain。
    * LangChain 是一个用于构建 LLM 应用的框架，支持 Python 和 JavaScript。
------------------------------

  >>> 方法 D: Adaptive HyDE <<<
  [Adaptive] 分析查询意图: 如何安装 langchain 库？
    - 判定策略: DIRECT
  [Direct] 执行直接检索: 如何安装 langchain 库？
  > 耗时: 1.82s
  > 结果:
    * 使用 `pip install langchain` 命令安装 LangChain。
    * LangChain 是一个用于构建 LLM 应用的框架，支持 Python 和 JavaScript。

==================================================
测试查询: 为什么我们需要检索增强生成(RAG)技术？
==================================================

  >>> 方法 A: Direct Retrieval <<<
  [Direct] 执行直接检索: 为什么我们需要检索增强生成(RAG)技术？
  > 耗时: 0.12s
  > 结果:
    * RAG (Retrieval-Augmented Generation) 通过检索外部数据来增强生成模型的准确性。
    * 大模型在处理特定领域知识时经常产生幻觉，检索增强是解决此问题的关键手段。
------------------------------

  >>> 方法 B: Standard HyDE (n=1) <<<
  [Standard HyDE] 执行 Standard HyDE (n=1): 为什么我们需要检索增强生成(RAG)技术？
  > 正在生成 1 个假设文档 (使用多样化视角)...
    - 假设文档: 我们需要检索增强生成（Retrieval-Augmented Generation，简称 RAG）技...
  > 耗时: 13.34s
  > 结果:
    * RAG (Retrieval-Augmented Generation) 通过检索外部数据来增强生成模型的准确性。
    * 大模型在处理特定领域知识时经常产生幻觉，检索增强是解决此问题的关键手段。
------------------------------

  >>> 方法 C: Multi-HyDE (n=3) <<<
  [Multi-HyDE] 执行 Multi-HyDE (n=3): 为什么我们需要检索增强生成(RAG)技术？
  > 正在生成 3 个假设文档 (使用多样化视角)...
    - 假设文档 1: 检索增强生成（Retrieval-Augmented Generation, RAG）技术之所以被广...
    - 假设文档 2: 检索增强生成（Retrieval-Augmented Generation, RAG）技术的出现，源...
    - 假设文档 3: 1. **理解大语言模型（LLM）的局限性**：LLMs 依赖训练数据，无法访问最新或私有信息，且可...
  > 耗时: 62.80s
  > 结果:
    * RAG (Retrieval-Augmented Generation) 通过检索外部数据来增强生成模型的准确性。
    * 大模型在处理特定领域知识时经常产生幻觉，检索增强是解决此问题的关键手段。
------------------------------

  >>> 方法 D: Adaptive HyDE <<<
  [Adaptive] 分析查询意图: 为什么我们需要检索增强生成(RAG)技术？
    - 判定策略: HYDE
  [Multi-HyDE] 执行 Multi-HyDE (n=3): 为什么我们需要检索增强生成(RAG)技术？
  > 正在生成 3 个假设文档 (使用多样化视角)...
    - 假设文档 1: 我们需要检索增强生成（Retrieval-Augmented Generation，简称 RAG）技...
    - 假设文档 2: 检索增强生成（Retrieval-Augmented Generation, RAG）技术的出现，源...
    - 假设文档 3: 1. **理解大语言模型（LLM）的局限性**：LLMs 依赖训练数据，无法访问最新或私有信息，且可...
  > 耗时: 69.15s
  > 结果:
    * RAG (Retrieval-Augmented Generation) 通过检索外部数据来增强生成模型的准确性。
    * 大模型在处理特定领域知识时经常产生幻觉，检索增强是解决此问题的关键手段。