解析 ‘Hypothetical Document Refinement’：利用循环节点不断优化‘伪文档’，直到其与向量库的匹配度达到阈值

各位同仁，各位对前沿技术充满热情的开发者们：

欢迎来到今天的技术讲座。今天，我们将深入探讨一个在现代信息检索与生成领域极具潜力的概念——“Hypothetical Document Refinement”，即“伪文档迭代优化”。我们将聚焦于如何利用一个“循环节点”机制，不断生成和优化一个“伪文档”（Hypothetical Document），直到它在语义上与我们庞大的向量库中的真实文档达到预设的匹配度阈值。这不仅仅是一个理论探讨，更是一个结合了大型语言模型（LLM）、向量数据库以及智能控制流的实用工程范式。

第一章：语义搜索的挑战与HyDE的崛起

在信息爆炸的时代，我们面临的核心挑战是如何高效、准确地从海量数据中检索出真正相关的信息。传统的关键词匹配，例如基于TF-IDF或BM25的倒排索引方法，在面对语义模糊、同义词、近义词或概念匹配时显得力不从心。用户可能用一种方式表达他们的意图，而文档可能用另一种方式阐述相同或相似的概念，这时关键词搜索就无法捕捉到深层语义关联。

图1.1：传统关键词搜索的局限性

搜索查询	预期结果示例	关键词搜索表现	语义搜索表现
如何提高工作效率？	关于时间管理、任务优先级、专注力训练的文章	依赖关键词工作效率	关联概念，更广
巴黎的浪漫之都	巴黎旅游、法国文化、埃菲尔铁塔、塞纳河	可能错过法国文化	精准匹配
AI的未来趋势	人工智能发展、机器学习应用、深度学习突破	仅限AI 未来趋势	涵盖技术方向
健康饮食的最佳实践	营养学、食谱、健身建议、生活方式	仅限健康饮食	扩展至相关领域

为了克服这些局限，语义搜索应运而生。它旨在理解查询和文档的含义，而不仅仅是字面上的匹配。向量嵌入（Vector Embeddings）技术是语义搜索的基石，它将文本（无论是单词、句子还是整个文档）转换成高维空间中的稠密向量，使得语义上相近的文本在向量空间中也彼此接近。

然而，即便有了强大的向量嵌入，用户提交的原始查询往往是简短的、模糊的，或者未能充分表达其复杂意图。直接将一个短查询嵌入并与文档向量进行比较，其效果可能不如将一个更详细、更具上下文的文档嵌入。

这就是“Hypothetical Document Embedding”（HyDE）思想的由来。HyDE的核心理念是：当用户提出一个简短查询时，我们可以利用一个大型语言模型（LLM）根据这个查询生成一个假想的、理想化的、能回答该查询的“伪文档”。这个伪文档通常比原始查询更详细，包含了更多上下文信息和潜在的答案要点。然后，我们将这个伪文档转换为向量，并用它来搜索向量数据库。由于伪文档包含了更丰富的语义信息，它能更好地捕捉用户意图，从而在向量空间中找到更相关的真实文档。

HyDE的初步流程：

1. 用户输入一个简短查询 Q。
2. LLM基于 Q 生成一个假想的文档 D_hyde。
3. 将 D_hyde 嵌入成向量 V_hyde。
4. 在向量数据库中搜索与 V_hyde 最相似的真实文档 D_real_1, D_real_2, ...。

这个基础HyDE模型已经显著提升了语义搜索的效果。但今天，我们要将这个概念推向一个新的高度：迭代优化。我们不满足于一次性的伪文档生成，而是要构建一个反馈循环，让伪文档能够根据搜索结果进行自我修正和优化，直到它能够完美地代表我们正在寻找的信息，并达到与向量库中目标文档的理想匹配度。这正是我们“循环节点”的核心任务。

# 导入必要的库
import os
import numpy as np
from typing import List, Dict, Any, Tuple
from abc import ABC, abstractmethod

# 模拟大型语言模型 (LLM) 和嵌入模型
# 在实际应用中，这里会集成OpenAI, Hugging Face, Cohere等API或本地模型

class MockLLM:
    """
    模拟一个大型语言模型，用于生成伪文档和进行文档细化。
    """
    def generate(self, prompt: str, max_tokens: int = 200) -> str:
        """
        根据prompt生成文本。
        """
        print(f"DEBUG: LLM生成请求 -> {prompt[:100]}...")
        # 简单模拟，实际LLM会进行复杂推理
        if "伪文档" in prompt:
            return f"这是一个关于 '{prompt.split('查询是：')[1].split('.')[0].strip()}' 的详细伪文档。它涵盖了相关概念、背景信息以及潜在的解决方案或答案要点。例如，它可能会讨论到... [根据查询内容扩展具体细节，模拟LLM的丰富性]。"
        elif "优化" in prompt and "检索结果" in prompt:
            return f"根据检索到的相关信息，对原始查询和伪文档进行了优化。新的伪文档更聚焦于... [结合检索结果进行修正，模拟LLM的归纳和重写能力]。这包括了... [进一步细节]。"
        else:
            return "这是一个通用的LLM生成结果。"

class MockEmbeddingModel:
    """
    模拟一个文本嵌入模型，将文本转换为向量。
    """
    def encode(self, texts: List[str]) -> np.ndarray:
        """
        将文本列表编码为向量。
        """
        print(f"DEBUG: 嵌入模型编码请求 -> {len(texts)} 条文本")
        # 模拟生成随机向量，实际模型会生成有意义的语义向量
        if isinstance(texts, str):
            texts = [texts]
        # 假设向量维度为768
        return np.random.rand(len(texts), 768)

# 初始化模拟模型
mock_llm = MockLLM()
mock_embedding_model = MockEmbeddingModel()

# 示例：首次HyDE生成
initial_query = "高效的项目管理方法"
hyde_prompt = f"请根据以下查询，生成一个详细的伪文档，该文档旨在全面回答此查询。查询是：{initial_query}。请包含核心概念、关键步骤和常见挑战等内容。"
initial_hyde_doc = mock_llm.generate(hyde_prompt)
print(f"n--- 初始伪文档生成完成 ---")
print(initial_hyde_doc)

第二章：向量数据库：语义空间的基石

在我们深入迭代优化之前，必须先理解我们进行匹配操作的“场”——向量数据库。向量数据库是专门为存储、管理和高效检索高维向量数据而设计的系统。它们是语义搜索、推荐系统、图像识别等领域的核心基础设施。

2.1 嵌入模型与向量空间

任何文本在进入向量数据库之前，都需要经过嵌入模型（Embedding Model）的处理。嵌入模型（例如BERT, RoBERTa, Sentence Transformers系列模型，或OpenAI的text-embedding-ada-002）将文本映射到一个固定长度的数字列表，即向量。这些向量具有一个关键特性：语义上相似的文本，它们对应的向量在多维空间中会彼此靠近；语义上不相关的文本，它们的向量则会相距较远。

2.2 向量数据库的工作原理

向量数据库不仅仅是存储向量的键值对。它们的核心能力在于高效的“近似最近邻”（Approximate Nearest Neighbor, ANN）搜索。当给定一个查询向量时，向量数据库能够在海量的向量中迅速找到与其“距离”最近（即最相似）的K个向量。

常见的ANN算法包括：

• HNSW (Hierarchical Navigable Small World): 构建多层图结构，在搜索时从粗粒度层级逐步细化到细粒度层级，实现高效搜索。
• IVF (Inverted File Index): 将向量空间划分为多个聚类，每个聚类有一个质心。搜索时，只在查询向量附近的几个聚类中进行精确搜索。
• LSH (Locality Sensitive Hashing): 通过哈希函数将相似的向量映射到相同的哈希桶中，从而减少比较的向量数量。

2.3 常见的向量数据库

市面上有多种优秀的向量数据库，各有特点：

表2.1：主流向量数据库对比

特性/数据库	ChromaDB	Faiss (Meta)	Pinecone	Weaviate	Milvus
类型	轻量级，Python友好	开源库，高性能	托管服务	开源，GraphQL API	开源，分布式
部署	本地，或容器化	库集成，需自行存储	云端（SaaS）	本地，容器化，云端	本地，容器化，云端
查询语言	Python/JS SDK	Python/C++ API	Python/JS SDK	GraphQL/REST API	Python/Java/Go SDK
高级功能	元数据过滤，增量索引	纯粹向量搜索	元数据过滤，命名空间	RAG集成，语义搜索	元数据过滤，流式导入
易用性	极高	中等，需处理存储	高	中等	中等
性能	中等	极高（单机）	很高	高	很高（分布式）

在本讲座的示例中，为了简化环境配置并聚焦核心逻辑，我们将使用ChromaDB作为我们的向量数据库实现。ChromaDB以其易用性和本地部署能力，成为快速原型开发和学习的理想选择。

# 导入ChromaDB
import chromadb
from chromadb.utils import embedding_functions

# 为了简化，我们使用一个自定义的（模拟的）嵌入函数
# 实际场景中，你会使用 SentenceTransformers 提供的嵌入函数
# embedding_function = embedding_functions.SentenceTransformerEmbeddingFunction(model_name="all-MiniLM-L6-v2")

class CustomEmbeddingFunction(embedding_functions.EmbeddingFunction):
    """
    一个包装我们模拟嵌入模型的ChromaDB兼容嵌入函数。
    """
    def __init__(self, mock_embedding_model):
        self.mock_embedding_model = mock_embedding_model

    def __call__(self, texts: embedding_functions.Documents) -> embedding_functions.Embeddings:
        # ChromaDB期望的输入是Documents (List[str])，输出是Embeddings (List[List[float]])
        # 我们的mock_embedding_model.encode返回np.ndarray，需要转换为List[List[float]]
        return self.mock_embedding_model.encode(texts).tolist()

# 初始化ChromaDB客户端
client = chromadb.Client() # 默认使用内存客户端，也可以指定路径 chromadb.PersistentClient(path="/path/to/db")

# 初始化我们的自定义嵌入函数
custom_ef = CustomEmbeddingFunction(mock_embedding_model)

# 创建或获取一个集合 (Collection)
collection_name = "document_refinement_collection"
try:
    collection = client.get_or_create_collection(
        name=collection_name,
        embedding_function=custom_ef # 指定自定义嵌入函数
    )
except Exception as e:
    print(f"Error getting/creating collection: {e}")
    # 如果是持久化客户端，可能需要清理旧数据或确保路径可写
    client = chromadb.Client() # 回退到内存客户端
    collection = client.get_or_create_collection(
        name=collection_name,
        embedding_function=custom_ef
    )

# 填充模拟的真实文档到向量数据库
# 这些文档是我们希望通过伪文档匹配到的目标
documents_to_add = [
    "项目管理中的敏捷方法论，包括Scrum和Kanban的实践与应用。",
    "有效的沟通技巧在团队协作和项目成功中的关键作用。",
    "风险管理策略：识别、评估、缓解项目风险的全面指南。",
    "如何利用OKR（目标与关键成果）框架驱动团队绩效和战略对齐。",
    "软件开发生命周期（SDLC）的各个阶段及其最佳实践。",
    "通过时间管理和优先级排序技术提高个人工作效率。",
    "领导力发展与团队激励，建立高绩效团队的核心要素。",
    "数据分析在商业决策中的应用，从数据到洞察的转化。",
    "云计算架构的演进与主流服务提供商（AWS, Azure, GCP）比较。",
    "用户体验（UX）设计原则与用户研究方法。"
]
document_ids = [f"doc_{i}" for i in range(len(documents_to_add))]

# 添加文档到ChromaDB
if collection.count() == 0: # 避免重复添加
    collection.add(
        documents=documents_to_add,
        ids=document_ids
    )
    print(f"n--- 向量数据库已填充 {len(documents_to_add)} 个文档 ---")
else:
    print(f"n--- 向量数据库已存在 {collection.count()} 个文档 (跳过填充) ---")

# 示例：搜索操作
def search_vector_db(query_vector: np.ndarray, top_k: int = 3) -> List[Dict[str, Any]]:
    """
    在ChromaDB中执行搜索。
    query_vector: 待查询的向量 (np.ndarray of shape (1, 768))
    top_k: 返回最相似的文档数量
    """
    if query_vector.ndim == 1:
        query_vector = query_vector.reshape(1, -1) # 确保是二维数组，ChromaDB期望List[List[float]]

    # ChromaDB的query方法期望的是List[List[float]]
    results = collection.query(
        query_embeddings=query_vector.tolist(),
        n_results=top_k,
        include=['documents', 'distances'] # 返回文档内容和距离
    )

    # 将ChromaDB的结果转换为更友好的格式
    formatted_results = []
    if results and results['documents'] and results['distances']:
        for i in range(len(results['documents'][0])):
            formatted_results.append({
                "document": results['documents'][0][i],
                "distance": results['distances'][0][i] # ChromaDB返回的是L2距离
            })
    return formatted_results

# 计算相似度（通常用余弦相似度，ChromaDB默认L2距离，需要转换或理解其含义）
# L2距离越小表示越相似。为了统一概念，我们仍将它视为某种“匹配度”
# 或者，如果需要余弦相似度，我们可以在外部手动计算
def cosine_similarity(vec1: np.ndarray, vec2: np.ndarray) -> float:
    """计算两个向量的余弦相似度。"""
    dot_product = np.dot(vec1, vec2)
    norm_vec1 = np.linalg.norm(vec1)
    norm_vec2 = np.linalg.norm(vec2)
    if norm_vec1 == 0 or norm_vec2 == 0:
        return 0.0
    return dot_product / (norm_vec1 * norm_vec2)

# 注意：ChromaDB的距离是L2距离。为了与“匹配度”概念保持一致（越大越好），
# 我们可以将其转换为一个介于0-1之间的“相似度”分数，或者直接使用ChromaDB的距离并反向理解。
# 对于L2距离，0表示完全相同，距离越大越不相似。
# 我们可以简单地用 1 / (1 + distance) 来表示一个粗略的相似度，或者直接用-distance。
def convert_l2_to_similarity_score(l2_distance: float) -> float:
    """
    将ChromaDB的L2距离转换为一个简化的0-1相似度分数，越大越相似。
    这是一种启发式转换，不是严格的数学定义，但有助于理解。
    """
    return 1.0 / (1.0 + l2_distance)

# 示例搜索
initial_hyde_vector = mock_embedding_model.encode([initial_hyde_doc])
search_results = search_vector_db(initial_hyde_vector[0], top_k=3)

print("n--- 初始伪文档搜索结果 ---")
for res in search_results:
    similarity_score = convert_l2_to_similarity_score(res['distance'])
    print(f"文档: {res['document'][:50]}..., 距离: {res['distance']:.4f}, 相似度: {similarity_score:.4f}")

第三章：构建循环优化节点：核心算法与实现

现在，我们来到了整个系统的核心——循环优化节点。这个节点将不断迭代，生成、评估、修正伪文档，直到达到我们预设的匹配度阈值。这类似于一个智能体在与向量数据库进行“对话”，通过每次的检索结果来“学习”并修正自己的表达，以更精确地描述其意图。

3.1 循环节点的核心逻辑

一个循环优化节点（RefinementLoopNode）将封装以下关键步骤：

1. 初始化伪文档： 从原始查询开始，通过LLM生成第一个版本的伪文档。
2. 向量化与搜索： 将当前伪文档向量化，并在向量数据库中执行搜索，获取最相关的真实文档。
3. 匹配度计算： 评估当前伪文档与检索结果（特别是排名第一的文档）之间的匹配度。
4. 收敛判断： 检查匹配度是否达到阈值，或是否已达到最大迭代次数。
5. 伪文档优化： 如果未收敛，根据检索结果和原始查询，再次通过LLM生成一个更优的伪文档。
6. 循环： 返回步骤2，重复上述过程。

图3.1：伪文档迭代优化循环节点流程图

+-------------------+
|   初始查询 (Q)    |
+---------+---------+
          |
          v
+---------+---------+
|  LLM生成初始伪文档  |
|  (Initial D_hyde) |
+---------+---------+
          |
          v
+-------------------+
|   循环开始 (Max_Iterations)    |
+---------+---------+
          |
          v
+---------+---------+
|    向量化 D_hyde   |
|     (V_hyde)      |
+---------+---------+
          |
          v
+---------+---------+
|  向量数据库搜索 (Top-K) |
|   (Search Results) |
+---------+---------+
          |
          v
+---------+---------+
|  计算匹配度 (Similarity) |
|  (D_hyde vs Top-1 Result) |
+---------+---------+
          |
          v
+---------+---------+
|   是否达到阈值或Max_Iter?  |
|   (Converged?)    |
+---------+---------+
    | 是           | 否
    v              |
+---------+---------+   |
|   输出最终D_hyde   |   v
|  和检索结果       | +---------+---------+
+-------------------+ |  LLM优化D_hyde (Refined D_hyde) |
                      |  (基于原始查询和检索结果) |
                      +---------+---------+
                                |
                                v
                           (回到循环开始)

3.2 伪文档的初始化与生成

第一次生成伪文档相对简单，只需将原始查询作为LLM的输入。关键在于设计一个良好的Prompt，引导LLM生成一个全面且有代表性的文档。

# 封装HyDE生成逻辑
def generate_hyde_document(query: str, llm_model: MockLLM) -> str:
    """
    根据原始查询，利用LLM生成一个详细的伪文档。
    """
    prompt = f"请根据以下查询，生成一个详细的伪文档，该文档旨在全面回答此查询。请包含核心概念、关键步骤、潜在的解决方案或答案要点、以及相关的背景信息。查询是：'{query}'。文档应具有一定的广度和深度。"
    return llm_model.generate(prompt)

3.3 向量化与匹配度计算

这一步我们已经通过mock_embedding_model和search_vector_db函数初步实现。匹配度计算是迭代优化的核心反馈信号。我们通常会关注检索结果中排名最靠前的文档与当前伪文档的相似度。

# 匹配度计算：获取伪文档向量和最佳匹配文档向量，计算余弦相似度
def calculate_match_degree(
    hyde_doc: str,
    search_results: List[Dict[str, Any]],
    embedding_model: MockEmbeddingModel
) -> Tuple[float, str]:
    """
    计算当前伪文档与最佳检索结果之间的匹配度。
    返回匹配度分数和最佳匹配文档的内容。
    """
    if not search_results:
        return 0.0, ""

    # 获取最佳匹配文档
    best_match_doc_content = search_results[0]['document']

    # 将伪文档和最佳匹配文档都嵌入成向量
    vectors = embedding_model.encode([hyde_doc, best_match_doc_content])
    hyde_vector = vectors[0]
    best_match_vector = vectors[1]

    # 计算余弦相似度
    similarity = cosine_similarity(hyde_vector, best_match_vector)

    return similarity, best_match_doc_content

3.4 伪文档的优化策略

这是整个循环节点中最具创造性和挑战性的部分。如何根据检索到的信息来改进伪文档？有几种策略：

• RAG（Retrieval-Augmented Generation）式优化： 将原始查询、当前伪文档以及检索到的Top-K文档作为上下文，输入给LLM，要求它生成一个更准确、更完善的新伪文档。LLM可以从检索结果中提取关键信息、修正错误或补充遗漏。
• 聚焦式优化： 分析检索结果，识别出与原始查询最相关的关键词或概念，然后指示LLM在生成新伪文档时更加侧重这些方面。
• 反思式优化： 让LLM“反思”为什么当前伪文档未能达到更高匹配度，并根据其“理解”来改进。这可能需要更复杂的Prompt工程。

在本示例中，我们将采用RAG式的优化策略，因为它最直接且效果显著。

def refine_hyde_document(
    original_query: str,
    current_hyde_doc: str,
    search_results: List[Dict[str, Any]],
    llm_model: MockLLM
) -> str:
    """
    根据原始查询和最新的搜索结果，利用LLM优化伪文档。
    """
    # 提取搜索结果中的文档内容
    retrieved_docs_text = "n".join([f"  - {res['document'][:200]}..." for res in search_results])

    # 构建优化Prompt
    # 引导LLM根据检索到的信息来修正和扩展伪文档
    prompt = f"""
    你的任务是优化一个伪文档，使其能更准确、更全面地回答原始查询。
    原始查询是：'{original_query}'。
    当前的伪文档是：'{current_hyde_doc}'。
    以下是从向量数据库中检索到的最相关的文档片段：
    {retrieved_docs_text}

    请根据这些检索到的信息，重新生成一个更完善、更精确的伪文档。
    新的伪文档应该：
    1. 整合检索结果中的关键概念和细节。
    2. 修正当前伪文档中可能存在的偏差或不足。
    3. 保持与原始查询的高度相关性，但表达应更加丰富和具体。
    4. 移除重复信息，并确保逻辑流畅。
    """

    return llm_model.generate(prompt)

3.5 收敛判断与循环终止

循环必须有终止条件，以防止无限迭代。常见的终止条件包括：

• 匹配度阈值： 当伪文档与最佳真实文档的匹配度（例如余弦相似度）达到或超过某个预设值时。
• 最大迭代次数： 即使未达到匹配度阈值，也在达到最大迭代次数后强制终止。
• 匹配度无显著提升： 如果连续几轮迭代，匹配度提升幅度小于一个微小值，也可以认为已收敛。

def is_converged(
    current_similarity: float,
    previous_similarity: float,
    similarity_threshold: float,
    min_improvement: float,
    iteration: int,
    max_iterations: int
) -> bool:
    """
    判断迭代是否收敛。
    """
    if current_similarity >= similarity_threshold:
        print(f"收敛：当前相似度 {current_similarity:.4f} 达到或超过阈值 {similarity_threshold:.4f}。")
        return True

    if iteration >= max_iterations:
        print(f"收敛：达到最大迭代次数 {max_iterations}。")
        return True

    # 如果相似度没有显著提升
    if (current_similarity - previous_similarity) < min_improvement and iteration > 1: # 第一次迭代不算
        print(f"收敛：相似度提升不显著 ({current_similarity:.4f} vs {previous_similarity:.4f})。")
        return True

    return False

第四章：系统集成与端到端实践

现在，我们将以上所有组件整合到一个RefinementLoopNode类中，展示一个完整的端到端工作流。

class RefinementLoopNode:
    """
    实现伪文档迭代优化过程的循环节点。
    """
    def __init__(
        self,
        llm: MockLLM,
        embedding_model: MockEmbeddingModel,
        vector_db_collection: chromadb.api.models.Collection.Collection, # ChromaDB collection
        similarity_threshold: float = 0.85, # 目标相似度阈值
        max_iterations: int = 5,           # 最大迭代次数
        top_k_search: int = 3,             # 每次搜索返回的top K文档
        min_improvement: float = 0.01      # 最小相似度提升，低于此值认为收敛
    ):
        self.llm = llm
        self.embedding_model = embedding_model
        self.vector_db_collection = vector_db_collection
        self.similarity_threshold = similarity_threshold
        self.max_iterations = max_iterations
        self.top_k_search = top_k_search
        self.min_improvement = min_improvement

    def _generate_hyde(self, query: str) -> str:
        """生成初始伪文档"""
        prompt = f"请根据以下查询，生成一个详细的伪文档，该文档旨在全面回答此查询。请包含核心概念、关键步骤、潜在的解决方案或答案要点、以及相关的背景信息。查询是：'{query}'。文档应具有一定的广度和深度。"
        return self.llm.generate(prompt)

    def _refine_hyde(
        self,
        original_query: str,
        current_hyde_doc: str,
        search_results: List[Dict[str, Any]]
    ) -> str:
        """根据搜索结果优化伪文档"""
        retrieved_docs_text = "n".join([f"  - {res['document'][:200]}..." for res in search_results])
        prompt = f"""
        你的任务是优化一个伪文档，使其能更准确、更全面地回答原始查询。
        原始查询是：'{original_query}'。
        当前的伪文档是：'{current_hyde_doc}'。
        以下是从向量数据库中检索到的最相关的文档片段：
        {retrieved_docs_text}

        请根据这些检索到的信息，重新生成一个更完善、更精确的伪文档。
        新的伪文档应该：
        1. 整合检索结果中的关键概念和细节。
        2. 修正当前伪文档中可能存在的偏差或不足。
        3. 保持与原始查询的高度相关性，但表达应更加丰富和具体。
        4. 移除重复信息，并确保逻辑流畅。
        """
        return self.llm.generate(prompt)

    def _search_vector_db(self, query_vector: np.ndarray) -> List[Dict[str, Any]]:
        """在ChromaDB中执行搜索"""
        if query_vector.ndim == 1:
            query_vector = query_vector.reshape(1, -1)

        results = self.vector_db_collection.query(
            query_embeddings=query_vector.tolist(),
            n_results=self.top_k_search,
            include=['documents', 'distances']
        )

        formatted_results = []
        if results and results['documents'] and results['distances']:
            for i in range(len(results['documents'][0])):
                formatted_results.append({
                    "document": results['documents'][0][i],
                    "distance": results['distances'][0][i]
                })
        # 按照距离升序排序 (L2距离越小越相似)
        formatted_results.sort(key=lambda x: x['distance'])
        return formatted_results

    def _calculate_match_degree(
        self,
        hyde_doc: str,
        best_match_doc_content: str
    ) -> float:
        """计算伪文档与最佳匹配文档的余弦相似度"""
        vectors = self.embedding_model.encode([hyde_doc, best_match_doc_content])
        hyde_vector = vectors[0]
        best_match_vector = vectors[1]
        return cosine_similarity(hyde_vector, best_match_vector)

    def run(self, original_query: str) -> Dict[str, Any]:
        """
        运行伪文档迭代优化循环。
        返回最终的伪文档、匹配度、迭代次数和最佳检索结果。
        """
        current_hyde_doc = self._generate_hyde(original_query)
        current_similarity = 0.0
        previous_similarity = 0.0
        best_retrieved_docs = []

        print(f"n--- 开始伪文档迭代优化，原始查询：'{original_query}' ---")

        for iteration in range(1, self.max_iterations + 1):
            print(f"n--- 迭代 {iteration}/{self.max_iterations} ---")
            print(f"当前伪文档 ({len(current_hyde_doc)}字): {current_hyde_doc[:100]}...")

            # 步骤1：向量化当前伪文档
            hyde_vector = self.embedding_model.encode([current_hyde_doc])[0]

            # 步骤2：在向量数据库中搜索
            search_results = self._search_vector_db(hyde_vector)
            best_retrieved_docs = search_results # 保存本次最佳结果

            if not search_results:
                print("未找到任何相关文档，无法进行优化。")
                break

            # 步骤3：计算匹配度
            best_match_content = search_results[0]['document']
            current_similarity = self._calculate_match_degree(current_hyde_doc, best_match_content)

            print(f"最佳匹配文档 ({len(best_match_content)}字): {best_match_content[:100]}...")
            print(f"当前与最佳匹配文档的相似度: {current_similarity:.4f}")

            # 步骤4：收敛判断
            if is_converged(
                current_similarity,
                previous_similarity,
                self.similarity_threshold,
                self.min_improvement,
                iteration,
                self.max_iterations
            ):
                print(f"循环终止：在第 {iteration} 轮迭代后收敛。")
                break

            previous_similarity = current_similarity

            # 步骤5：优化伪文档
            current_hyde_doc = self._refine_hyde(original_query, current_hyde_doc, search_results)

            if iteration == self.max_iterations:
                print(f"达到最大迭代次数 {self.max_iterations}，强制终止。")

        print("n--- 优化过程结束 ---")
        return {
            "final_hyde_document": current_hyde_doc,
            "final_similarity": current_similarity,
            "iterations_run": iteration,
            "best_retrieved_documents": best_retrieved_docs
        }

# --- 运行端到端示例 ---
if __name__ == "__main__":
    # 清理旧的ChromaDB数据 (如果是内存客户端则无需，持久化客户端可能需要)
    # client.delete_collection(name=collection_name) # 如果需要重新初始化数据

    # 确保collection已填充
    if collection.count() == 0:
        collection.add(
            documents=documents_to_add,
            ids=document_ids
        )
        print(f"n--- 向量数据库已填充 {len(documents_to_add)} 个文档 ---")
    else:
        print(f"n--- 向量数据库已存在 {collection.count()} 个文档 ---")

    refinement_node = RefinementLoopNode(
        llm=mock_llm,
        embedding_model=mock_embedding_model,
        vector_db_collection=collection,
        similarity_threshold=0.95, # 提高阈值，观察多轮迭代
        max_iterations=10,
        top_k_search=5,
        min_improvement=0.005
    )

    # 示例查询1
    query_1 = "团队协作与沟通效率的提升策略"
    result_1 = refinement_node.run(query_1)
    print("n-----------------------------------------------------")
    print(f"最终伪文档:n{result_1['final_hyde_document'][:300]}...")
    print(f"最终匹配度: {result_1['final_similarity']:.4f}")
    print(f"运行迭代次数: {result_1['iterations_run']}")
    print("最佳检索文档:")
    for doc in result_1['best_retrieved_documents']:
        print(f"  - {doc['document'][:100]}... (距离: {doc['distance']:.4f})")
    print("-----------------------------------------------------")

    # 示例查询2
    query_2 = "如何有效管理大型IT项目风险"
    result_2 = refinement_node.run(query_2)
    print("n-----------------------------------------------------")
    print(f"最终伪文档:n{result_2['final_hyde_document'][:300]}...")
    print(f"最终匹配度: {result_2['final_similarity']:.4f}")
    print(f"运行迭代次数: {result_2['iterations_run']}")
    print("最佳检索文档:")
    for doc in result_2['best_retrieved_documents']:
        print(f"  - {doc['document'][:100]}... (距离: {doc['distance']:.4f})")
    print("-----------------------------------------------------")

第五章：性能评估、挑战与未来展望

伪文档迭代优化是一个强大的范式，但其效用和效率并非没有考量。

5.1 性能评估

评估这种迭代优化方法的性能，我们需要关注以下几个方面：

• 检索准确性（Recall & Precision）： 最终检索结果集是否包含所有相关的文档（Recall），以及其中有多少是真正相关的（Precision）。迭代优化应该能够提升这两个指标，尤其是在处理模糊查询时。
• 平均倒数排名（Mean Reciprocal Rank, MRR）： 衡量第一个相关文档在结果中的排名。优化后的伪文档理论上能将最相关的文档推到更高的位置。
• 收敛速度： 达到阈值所需的迭代次数。过多的迭代会增加成本和延迟。
• 计算成本： 每次迭代都需要调用LLM（生成/优化伪文档）和嵌入模型（向量化），这可能涉及API费用和计算资源。

5.2 面临的挑战

尽管潜力巨大，伪文档迭代优化也面临一些挑战：

• LLM的“幻觉”问题： LLM在生成伪文档时可能会产生不准确或虚构的信息，这会误导后续的搜索和优化过程。Prompt工程和LLM的选择至关重要。
• 计算资源与延迟： 多轮LLM调用和向量搜索会显著增加系统的延迟和成本。在实时系统中，这可能是一个瓶颈。需要权衡优化深度与用户体验。
• 优化策略的复杂性： 如何有效利用检索结果来优化伪文档是一个开放的研究问题。当前RAG式优化是基础，更高级的方法可能涉及强化学习或更精细的语义分析。
• 阈值和超参数调优： 相似度阈值、最大迭代次数、最小提升幅度等参数的设置对系统性能有直接影响，需要根据具体应用场景进行细致调优。
• 初始伪文档质量： 如果初始伪文档质量很差，即使经过多轮迭代也可能难以收敛到理想状态。

5.3 未来展望

伪文档迭代优化技术拥有广阔的未来发展空间：

• 多模态HyDE： 将文本查询扩展到图像、音频等多种模态，生成多模态的“伪内容”进行检索。例如，根据文本描述生成假想的图片，然后用图片向量搜索。
• 结合强化学习： 利用强化学习代理来学习最佳的伪文档优化策略，根据实际检索效果给予奖励，从而实现更智能的迭代。
• 自适应迭代： 根据查询的复杂性、当前匹配度等因素，动态调整迭代策略和参数，而不是固定地进行N轮迭代。
• 集成到更复杂的Agent系统： 伪文档迭代优化可以作为更大型AI Agent系统中的一个核心“思考”模块，帮助Agent更好地理解和执行复杂任务。
• 知识图谱与语义增强： 结合知识图谱，在生成和优化伪文档时引入更结构化的知识，提高语义的精确性。

迭代，求精，直至洞见

今天，我们共同探索了“Hypothetical Document Refinement”这一引人入胜的技术。它通过将大型语言模型的生成能力与向量数据库的高效检索能力相结合，并辅以智能的迭代优化循环，为我们提供了一种前所未有的方式来理解和满足复杂的信息需求。从最初模糊的查询，到最终与目标语义高度匹配的伪文档，这个过程不仅仅是技术的堆叠，更是一种对人类认知模式的模拟——通过不断地提问、收集反馈、修正理解，最终逼近真相。

这种迭代求精的范式，将极大地提升语义搜索的准确性和鲁棒性，为RAG系统、智能客服、知识管理乃至更广阔的AI应用场景开启新的可能性。我们所展示的代码示例，是这一宏大愿景的冰山一角，它提供了一个坚实的基础，供各位在各自的领域中进一步探索和创新。感谢大家的聆听！