1. 层次化文档索引 #

层次化文档索引（Hierarchical Document Indexing）是一种提升大文档块检索效果的常用策略。它主要思想是：将文档首先分割成较大的“父块”（parent chunks），然后每个父块再次细分为较小的“子块”（child chunks）。在索引和检索过程中，实际建立索引和相似度计算的是子块，但最终返回内容时给出的是其对应的父块（即较大的原始内容片段）。这样做的好处主要包括：

1. 更细粒度的匹配：通过对子块进行向量化检索，系统能更准确地定位相关信息，即便这一信息只出现在文档的小部分。
2. 给用户更有上下文的结果：检索逻辑是用子块匹配，但最终返回父块内容，这可以让用户获得更完整的上下文，不会因为文本被切得太碎而丢失重要信息。
3. 高效的索引与召回：子块作为向量存储数量不会过于庞大（因为只在父块级别分割的基础上再细切），可以在召回精度和存储消耗之间取得平衡。
4. 避免响应不连贯：直接段落甚至句子的切分可能导致召回结果缺乏语境，通过父块返回可以实现信息整合，提升问答的相关性和可读性。

实现一般分为两步：

• 建索引时，原始文档先被切成父块，每个父块再被切为多个子块。每个子块都指向其父块ID，用子块的向量表示进行索引，并记录父块的原文内容及其ID。
• 检索时，对查询做向量化，与所有子块做相似度匹配。得到相关子块后，通过子块的父ID查找对应的父块内容作为最终返回结果。

这种索引方式特别适合对长文档、章节性强的文本、复杂报告等做语义检索与问答，能极大提升用户获取信息的有效性和体验。

1.1 HierarchicalDocumentIndexing.py #

# 层次化文档索引整体说明
"""
层次化文档索引

层次化文档索引是一种RAG优化方案，其核心思想是：
1. 将大文档块（parent chunks）分解为更小的块（child chunks）
2. 对小块进行向量化处理，因为embedding模型对短文本的语义提取效果更佳
3. 检索时使用小块的embedding进行相似度计算
4. 返回时返回对应的大文档块内容，提供更完整的上下文

优势：
- 提高检索精度：短文本更容易与用户问题产生匹配
- 保持上下文完整性：返回大块内容，避免信息碎片化
"""

# 导入uuid模块，用于生成唯一的id
import uuid
# 导入类型注解List、Dict、Any
from typing import List, Dict, Any
# 导入pydantic配置工具，允许BaseRetriever包含任意类型字段
from pydantic import ConfigDict
# 从langchain_text_splitters导入字符级文本切分工具
from langchain_text_splitters import CharacterTextSplitter
# 从langchain_core导入抽象检索器BaseRetriever
from langchain_core.retrievers import BaseRetriever
# 从langchain_core导入文档数据结构Document
from langchain_core.documents import Document
# 导入自定义嵌入对象
from embeddings import embeddings
# 导入获取向量库的工具
from vector_store import get_vector_store

# 定义层次化文档索引器类
class HierarchicalDocumentIndexer:
    # 类说明文档字符串
    """层次化文档索引器：将大文档块分解为小块进行索引，检索时用小块匹配，返回时用大块内容"""

    # 初始化函数
    def __init__(
        self,
        embeddings,
        vector_store,
        parent_chunk_size: int = 300,
        child_chunk_size: int = 100,
        child_chunk_overlap: int = 0
    ):
        # 保存embedding模型
        self.embeddings = embeddings
        # 保存向量存储数据库
        self.vector_store = vector_store
        # 父文档块的长度
        self.parent_chunk_size = parent_chunk_size
        # 子文档块的长度
        self.child_chunk_size = child_chunk_size
        # 用于记录父块id对应的原始文本内容
        self.parent_chunks: Dict[str, str] = {}
        # 实例化子块文本分割器
        self.child_splitter = CharacterTextSplitter(
            separator="",
            chunk_size=child_chunk_size,
            chunk_overlap=child_chunk_overlap,
        )

    # 文档索引方法，将大块分为子块并载入数据库
    def index_documents(self, texts: List[str], metadatas: List[Dict[str, Any]] = None):
        # 用于存储所有子块文本
        child_texts = []
        # 用于存储所有子块的元数据
        child_metadatas = []
        # 打印索引启动信息
        print(f"\n开始索引 {len(texts)} 个文档...")
        print(f"配置：Parent块大小={self.parent_chunk_size}字符, Child块大小={self.child_chunk_size}字符\n")
        # 遍历每个文档及其关联元数据
        for doc_idx, (parent_chunk, metadata) in enumerate(zip(texts, metadatas)):
            # 为每个父块生成唯一id
            parent_id = str(uuid.uuid4())
            # 保存父块内容到映射表
            self.parent_chunks[parent_id] = parent_chunk
            # 使用子块分割工具对父块切分
            child_chunks = self.child_splitter.split_text(parent_chunk)
            # 输出当前文档分块情况
            print(f"文档 {doc_idx + 1}: Parent块({len(parent_chunk)}字符) -> {len(child_chunks)}个Child块")
            # 遍历所有子块
            for child_idx, child_chunk in enumerate(child_chunks):
                # 构造子块元数据信息，包括父块信息
                child_metadata = {
                    **metadata,              # 父块的元数据
                    "parent_chunk_id": parent_id,         # 对应的父块ID
                    "child_chunk_index": child_idx,       # 当前子块在父块中的序号
                    "parent_chunk_length": len(parent_chunk), # 父块长度
                    "child_chunk_length": len(child_chunk),   # 当前子块长度
                }
                # 收集子块内容
                child_texts.append(child_chunk)
                # 收集子块元数据
                child_metadatas.append(child_metadata)
        # 输出切分完成和即将存储的信息
        print(f"\n总共生成 {len(child_texts)} 个Child块，开始向量化并存储...")
        # 向量化并添加所有子块到向量数据库
        self.vector_store.add_texts(child_texts, child_metadatas)
        # 索引流程结束提示
        print("索引完成！\n")

    # 检索接口方法
    def retrieve(self, query: str, k: int = 4) -> List[Document]:
        # 使用子块在向量库中进行相似度检索，实际topK数量为k*3用以提高多样性，带分数
        child_results = self.vector_store.similarity_search_with_score(query, k=k * 3)
        # 记录已添加的父块id集合，防止重复返回
        parent_ids_seen = set()
        # 存储需要返回的父块文档对象及其分数
        parent_docs_with_scores = []
        # 遍历检索到的子块文档对象及其分数
        for child_doc, distance in child_results:
            # 获取父块的唯一id
            parent_id = child_doc.metadata.get("parent_chunk_id")
            # 如果该父块还未计入结果，则添加（去重）
            if parent_id and parent_id not in parent_ids_seen:
                parent_ids_seen.add(parent_id)
                # 获取父块完整内容
                parent_content = self.parent_chunks.get(parent_id)
                # 父块内容存在才处理
                if parent_content:
                    # 构造父块文档对象并补充元数据
                    parent_doc = Document(
                        page_content=parent_content,    # 父块内容
                        metadata={
                            **child_doc.metadata,       # 原有子块元数据
                            "retrieved_from_child": True,   # 标记是由子块检索得到
                            "parent_chunk_id": parent_id,    # 父块id
                            "score": float(distance)  # 使用原始分数（距离，越小越相似）
                        }
                    )
                    # 添加到结果列表
                    parent_docs_with_scores.append((parent_doc, distance))
                    # 如果结果数达标则提前返回
                    if len(parent_docs_with_scores) >= k:
                        break
        # 按分数排序（距离越小越相似），取前k个
        parent_docs_with_scores.sort(key=lambda x: x[1])
        # 返回聚合的父块文档对象列表（只返回Document，不返回分数）
        return [doc for doc, _ in parent_docs_with_scores[:k]]

# 定义层次化检索器类，便于链式调用等用法
class HierarchicalRetriever(BaseRetriever):
    """层次化检索器（继承自BaseRetriever，可用于LangChain链）"""
    # 用于检索的索引器对象
    indexer: HierarchicalDocumentIndexer
    # 返回文档的数量，默认4
    k: int = 4
    # 配置允许任意类型成员
    model_config = ConfigDict(arbitrary_types_allowed=True)
    # 实现LangChain检索接口，供链调用
    def _get_relevant_documents(self, query: str, **kwargs) -> List[Document]:
        # 支持外部通过kwargs调整k，默认为类内k
        k = kwargs.get("k", self.k)
        # 调用索引器进行实际检索
        return self.indexer.retrieve(query, k=k)

# 获取向量存储实例
vector_store = get_vector_store(
    persist_directory="chroma_db",
    collection_name="hierarchical"
)

# 创建层次化文档索引器对象
indexer = HierarchicalDocumentIndexer(
    embeddings=embeddings,
    vector_store=vector_store,
    parent_chunk_size=300,      # 父块最大字符数
    child_chunk_size=100,       # 子块最大字符数
    child_chunk_overlap=0       # 子块无重叠
)

# 声明示例文档（每行为一个领域大文档块）
documents = [
    # 人工智能领域文档
    "人工智能（Artificial Intelligence，AI）是计算机科学的一个分支，旨在创建能够执行通常需要人类智能的任务的系统。"
    "机器学习是人工智能的核心技术之一，它使计算机能够从数据中学习，而无需明确编程。"
    "深度学习是机器学习的一个子集，使用人工神经网络来模拟人脑的工作方式。"
    "自然语言处理（NLP）是AI的另一个重要领域，专注于使计算机能够理解和生成人类语言。",
    # 区块链领域文档
    "区块链技术是一种分布式账本技术，通过密码学方法确保数据的安全性和不可篡改性。"
    "比特币是第一个成功应用区块链技术的加密货币，它解决了数字货币的双重支付问题。"
    "以太坊是一个支持智能合约的区块链平台，允许开发者在其上构建去中心化应用（DApps）。"
    "智能合约是自动执行的合约，其条款直接写入代码中，无需第三方中介。",
    # 量子计算领域文档
    "量子计算是一种利用量子力学现象进行计算的新兴技术，具有巨大的计算潜力。"
    "量子比特（qubit）是量子计算的基本单位，与经典比特不同，它可以同时处于0和1的叠加态。"
    "量子纠缠是量子计算的关键特性，允许量子比特之间建立特殊的关联关系。"
    "量子计算在密码学、药物发现和优化问题等领域具有潜在的应用前景。",
]

# 对所有示例文档进行切块并索引到向量库
indexer.index_documents(
    texts=documents,
    metadatas=[
        {"topic": "人工智能", "category": "科技"},
        {"topic": "区块链", "category": "科技"},
        {"topic": "量子计算", "category": "科技"},
    ]
)

# 创建层次化检索器对象
retriever = HierarchicalRetriever(indexer=indexer, k=2)
# 指定一个用户查询
query = "区块链的应用"
# 检索相关文档
docs = retriever.invoke(query)
# 遍历打印检索出的文档内容前200字及分数
for i, doc in enumerate(docs, 1):
    print(f"\n文档 {i} (分数: {doc.metadata.get('score', 'N/A'):.4f}):")
    print(f"内容: {doc.page_content[:200]}...")

1.2 embeddings.py #

"""
Embedding模型配置

统一管理项目中使用的Embedding模型配置
"""

from langchain_huggingface import HuggingFaceEmbeddings

# 初始化Embedding模型
model_path = "C:/Users/Administrator/.cache/modelscope/hub/models/sentence-transformers/all-MiniLM-L6-v2"
embeddings = HuggingFaceEmbeddings(
    model_name=model_path,
    model_kwargs={"device": "cpu"}
)

1.3 vector_store.py #

vector_store.py

"""
向量存储配置

统一管理项目中使用的向量存储配置
"""

from langchain_chroma import Chroma
from embeddings import embeddings

# 初始化Chroma向量存储（通用配置）
# 注意：不同用途可以使用不同的 persist_directory 和 collection_name
def get_vector_store(persist_directory: str, collection_name: str):
    """
    获取向量存储实例

    Args:
        persist_directory: 持久化目录路径
        collection_name: 集合名称

    Returns:
        Chroma向量存储实例
    """
    return Chroma(
        persist_directory=persist_directory,
        embedding_function=embeddings,
        collection_name=collection_name,
        collection_metadata={"hnsw:space": "cosine"}
    )

# 层次化文档索引整体说明
"""
层次化文档索引

层次化文档索引是一种RAG优化方案，其核心思想是：
1. 将大文档块（parent chunks）分解为更小的块（child chunks）
2. 对小块进行向量化处理，因为embedding模型对短文本的语义提取效果更佳
3. 检索时使用小块的embedding进行相似度计算
4. 返回时返回对应的大文档块内容，提供更完整的上下文

优势：
- 提高检索精度：短文本更容易与用户问题产生匹配
- 保持上下文完整性：返回大块内容，避免信息碎片化
"""
from langchain_core.retrievers import BaseRetriever
from embeddings import embeddings
from vector_store import get_vector_store
from langchain_text_splitters import CharacterTextSplitter
import uuid
from langchain_core.documents import Document


class HierarchicalDocumentIndexer:
    def __init__(
        self,
        embeddings,
        vector_store,
        parent_chunk_size,
        child_chunk_size,
        child_chunk_overlap,
    ):
        self.embeddings = embeddings
        self.vector_store = vector_store
        self.parent_chunk_size = parent_chunk_size
        self.child_chunk_size = child_chunk_size
        self.child_chunk_overlap = child_chunk_overlap
        # 实例化子块文本分割器
        self.child_splitter = CharacterTextSplitter(
            separator="",  # 分割用空串分代表就是按字符分
            chunk_size=child_chunk_size,  # 100 就意味着分割后每一个子块都是100个字符
            chunk_overlap=child_chunk_overlap,
        )

    def index_documents(self, texts, metadatas):
        # 存储所有的子块文本
        child_texts = []
        # 存储所有的子块的元数据
        child_metadatas = []
        # 遍历每个父文档块
        for doc_idx, (parent_chunk, metadata) in enumerate(zip(texts, metadatas)):
            # 生成parent_id (UUID)
            parent_id = str(uuid.uuid4())
            # 使用子块分割工具对父块进行分割,返回子块列表
            child_chunks = self.child_splitter.split_text(parent_chunk)
            # 遍历每个子块
            for child_idx, child_chunk in enumerate(child_chunks):
                # 构造子块元数据信息，包括父块信息
                child_metadata = {
                    **metadata,  # 继承父块的元数据
                    "parent_chunk_id": parent_id,  # 对应的父块的ID
                    "child_chunk_index": child_idx,  # 对庆的子块在父块中的序号或者索引
                    "parent_chunk_length": len(parent_chunk),  # 父块的长度
                    "child_chunk_length": len(child_chunk),  # 子块的长度
                }
                # 收集子块文本和元数据
                child_texts.append(child_chunk)
                child_metadatas.append(child_metadata)
        # 向量化并存储所有子块
        self.vector_store.add_texts(child_texts, child_metadatas)
        print("索引创建完成")

    def retrieve(self, query, k):
        # 使用子块在向量数据库中进行相似度检索，实际的topk的值为k*3以提高多样性，带分数返回
        # 使用子块进行相似度检索 返回6个候选子块(k*3)
        child_results = self.vector_store.similarity_search_with_score(query, k=k * 3)
        # 遍历检索结果，去重父块
        # 记录已经添加的父块的ID的集合，防止重复返回
        parent_ids_seen = set()
        # 存储需要返回的父块文档对象及其分数
        parent_docs_with_scores = []
        # 遍历检索到的子块及其分数
        for child_doc, distance in child_results:
            # 获取这个子块对应的父块ID
            parent_id = child_doc.metadata.get("parent_chunk_id")
            # 如果父块还没未入结果，则添加
            if parent_id and parent_id not in parent_ids_seen:
                # 把父ID添加集合中
                parent_ids_seen.add(parent_id)
                # 通过父文档ID从向量数据库中检索所有子文档
+               all_child_docs = self.vector_store.get(
+                   where={"parent_chunk_id": parent_id}
+               )
                # 如果没有找到子文档，跳过
+               if not all_child_docs.get("documents"):
+                   continue
                # 获取所有子文档及其元数据
+               child_documents = all_child_docs.get("documents", [])
+               child_metadatas = all_child_docs.get("metadatas", [])
                # 创建(索引, 子文档内容, 元数据)的列表，用于排序
+               child_items = [
+                   (meta.get("child_chunk_index", 0), doc, meta)
+                   for doc, meta in zip(child_documents, child_metadatas)
+               ]
                # 按照child_chunk_index排序
+               child_items.sort(key=lambda x: x[0])
                # 拼接所有子文档内容重建父文档
+               parent_content = "".join([item[1] for item in child_items])
                # 获取第一个子文档的元数据作为基础元数据
+               base_metadata = child_items[0][2] if child_items else {}
                # 构建父块文档对象并补充元数据
+               parent_doc = Document(
+                   page_content=parent_content,
+                   metadata={
+                       **base_metadata,  # 原有的子块元数据
+                       "retrieved_from_child": True,  # 标记这个父块由它的子块间接检索得到
+                       "parent_chunk_id": parent_id,
+                       "score": float(distance),
+                   },
+               )
                # 在这里相当于把小文本的得到直接赋给大文件得分了
+               parent_docs_with_scores.append((parent_doc, distance))
                # 如果结果数达标则提前返回
+               if len(parent_docs_with_scores) >= k:
+                   break
        # 按分数排序，距离越小越相似
        parent_docs_with_scores.sort(key=lambda x: x[1])
        return [doc for doc, _ in parent_docs_with_scores[:k]]


class HierarchicalRetriever(BaseRetriever):
    indexer: HierarchicalDocumentIndexer
    k: int = 5

    def _get_relevant_documents(self, query, **kwargs):
        k = kwargs.get("k", self.k)
        return self.indexer.retrieve(query, k=k)


# 声明示例文档（每行为一个领域大文档块）
# documents 就是我们把文本分割后的结果
documents = [
    # 人工智能领域文档
    "人工智能（Artificial Intelligence，AI）是计算机科学的一个分支，旨在创建能够执行通常需要人类智能的任务的系统。"
    "机器学习是人工智能的核心技术之一，它使计算机能够从数据中学习，而无需明确编程。"
    "深度学习是机器学习的一个子集，使用人工神经网络来模拟人脑的工作方式。"
    "自然语言处理（NLP）是AI的另一个重要领域，专注于使计算机能够理解和生成人类语言。",
    # 区块链领域文档
    "区块链技术是一种分布式账本技术，通过密码学方法确保数据的安全性和不可篡改性。"
    "比特币是第一个成功应用区块链技术的加密货币，它解决了数字货币的双重支付问题。"
    "以太坊是一个支持智能合约的区块链平台，允许开发者在其上构建去中心化应用（DApps）。"
    "智能合约是自动执行的合约，其条款直接写入代码中，无需第三方中介。",
    # 量子计算领域文档
    "量子计算是一种利用量子力学现象进行计算的新兴技术，具有巨大的计算潜力。"
    "量子比特（qubit）是量子计算的基本单位，与经典比特不同，它可以同时处于0和1的叠加态。"
    "量子纠缠是量子计算的关键特性，允许量子比特之间建立特殊的关联关系。"
    "量子计算在密码学、药物发现和优化问题等领域具有潜在的应用前景。",
]
# 1.获取向量数据库
vector_store = get_vector_store(
    persist_directory="./chroma_db", collection_name="hierarchical"
)
indexer = HierarchicalDocumentIndexer(
    embeddings=embeddings,  # 向量工具
    vector_store=vector_store,  # 向量数据库
    parent_chunk_size=300,  # 父块的最大字符数
    child_chunk_size=100,  # 子块的最大字符数
    child_chunk_overlap=0,  # 子块的重叠数为0
)
# 对所有的示例文档进行切块并索引到向量数据库中
indexer.index_documents(
    texts=documents,
    metadatas=[
        {"topic": "人工智能", "category": "科技"},
        {"topic": "区块链", "category": "科技"},
        {"topic": "量子计算", "category": "科技"},
    ],
)

retriever = HierarchicalRetriever(indexer=indexer, k=2)
query = "区块链的应用"
docs = retriever.invoke(query)
for i, doc in enumerate(docs, 1):
    print(
        f"文档{i}(分数:{doc.metadata.get('score','N/A'):.4f}):{doc.page_content[:50]}..."
    )

1.4 执行过程 #

1.4.1 核心思想 #

层次化文档索引采用“小块检索，大块返回”：

• 将大文档块（parent chunks）切分为更小的子块（child chunks）
• 对子块进行向量化并存储
• 检索时用子块匹配，返回对应的父块内容

1.4.2 执行流程 #

阶段一：初始化

# 1. 获取向量存储实例
vector_store = get_vector_store(...)

# 2. 创建层次化文档索引器
indexer = HierarchicalDocumentIndexer(
    embeddings=embeddings,
    vector_store=vector_store,
    parent_chunk_size=300,    # 父块300字符
    child_chunk_size=100,     # 子块100字符
    child_chunk_overlap=0     # 无重叠
)

初始化时：

• 创建 CharacterTextSplitter 用于切分
• 初始化 parent_chunks 字典用于存储父块内容

阶段二：文档索引

indexer.index_documents(
    texts=documents,           # 3个父文档块
    metadatas=[...]           # 对应的元数据
)

索引过程：

1. 遍历每个父文档块
2. 为每个父块生成唯一 UUID
3. 将父块内容存入 parent_chunks 字典
4. 使用 child_splitter 将父块切分为子块
5. 为每个子块构建元数据（包含 parent_chunk_id）
6. 将所有子块向量化并存入向量数据库

示例：一个300字符的父块可能被切分为3个100字符的子块。

阶段三：文档检索

retriever = HierarchicalRetriever(indexer=indexer, k=2)
docs = retriever.invoke("区块链的应用")

检索过程：

1. retriever.invoke() 调用 _get_relevant_documents()
2. 调用 indexer.retrieve(query, k=2)
3. 在向量库中用子块进行相似度检索（取 k×3=6 个候选）
4. 遍历结果，通过 parent_chunk_id 去重
5. 从 parent_chunks 获取父块完整内容
6. 构建 Document 对象并记录分数
7. 按分数排序，返回前 k 个父块文档

1.4.3 类图 #

1.4.4 时序图 #

1.4.4.1 索引阶段时序图 #

1.4.4.2 检索阶段时序图 #

1.4.5 关键设计要点 #

1. 两级索引结构

父块 (Parent Chunk) - 300字符
  ├── 子块1 (Child Chunk) - 100字符 → 向量化存储
  ├── 子块2 (Child Chunk) - 100字符 → 向量化存储
  └── 子块3 (Child Chunk) - 100字符 → 向量化存储

2. 检索策略

• 检索阶段：使用子块进行相似度匹配（精度更高）
• 返回阶段：返回父块完整内容（上下文更完整）
• 去重机制：通过 parent_ids_seen 确保每个父块只返回一次
• 多样性保证：检索 k×3 个子块，提高结果多样性

3. 元数据设计

子块元数据包含：

• parent_chunk_id：关联父块
• child_chunk_index：子块序号
• parent_chunk_length：父块长度
• child_chunk_length：子块长度
• 继承父块的原始元数据（如 topic、category）

4. 优势

• 检索精度：短文本匹配更准确
• 上下文完整：返回父块，避免信息碎片
• 灵活性：可调整父块和子块大小
• 兼容性：继承 BaseRetriever，可集成到 LangChain 链中

该设计在检索精度和上下文完整性之间取得平衡，适用于需要精确匹配且保持完整上下文的场景。

2. 摘要化索引 #

摘要化索引简介

在大规模检索增强生成（RAG）场景中，直接对原始文档块进行向量化可能导致以下问题：

• 文档块较大时，向量表达容易稀释关键信息，影响检索相关性；
• 用户的查询往往偏向抽象、概括或主题性的内容，而原文块细节较杂；
• 需要在“召回能力”和“语义浓缩”之间平衡。

摘要化索引的核心思想：

以文档块为单位，先用LLM对内容做自动摘要，将摘要作为检索的向量化对象；检索时只与摘要embedding进行相似度匹配，命中后返回原始文档块，兼顾语义相关性与完整性。

主要优势

• 检索精度提升：LLM摘要能浓缩关键信息，使相关性检索更聚焦。
• 兼容抽象性查询：当用户提问较抽象/概括性问题时，摘要更有可能与之高度匹配。
• 召回完整上下文：最终返还原始文档块，便于后续生成链拿到完整语境。
• 灵活自动扩展：可平滑兼容主题多样的大语料索引。

流程与实现要点

1. 分块
   先将大文档按预设字符数切分为若干块（如300字符/块）。

2. 摘要生成
   对每个文档块自动调用LLM生成简明摘要（可自定义摘要提示词模板，突出主题核心）。

3. 摘要向量化存储
   将每条摘要进行embedding，存入向量库（如Chroma），同时存储摘要的元信息（含原文块id等）。

4. 检索阶段
   针对用户查询，计算查询embedding并召回向量空间中最相关的摘要（top-k）。

5. 原文块召回
   检索命中的摘要可反查原始文档块内容，便于后续长上下文问答或生成。

典型应用场景

• 长文档、高冗余知识库的智能问答系统
• 支持多主题、多专业领域的RAG大模型产品
• 复杂业务知识抽象检索与聚合
• 面向事实、经验总结、原理性问询等任务

2.1 AbstractedIndex.py #

"""
摘要索引（Abstracted Index）

摘要索引是一种RAG优化方案，其核心思想是：
1. 对于大文档块（如300字符），使用大语言模型生成摘要
2. 对摘要进行向量化处理，建立索引
3. 检索时使用摘要的embedding进行相似度匹配
4. 返回原始的大文档块内容，提供完整上下文

优势：
- 提高检索精度：摘要包含文档的核心语义，更容易匹配概括性查询
- 匹配抽象需求：用户有时需要概括性经验或抽象信息，而非具体细节
- 保持上下文完整性：返回原始文档块，避免信息丢失
"""

# 导入uuid用于生成唯一ID
import uuid
# 导入类型标注相关模块
from typing import List, Dict, Any, Optional
# 导入pydantic用于配置
from pydantic import ConfigDict
# 导入字符切分器
from langchain_text_splitters import CharacterTextSplitter
# 导入基础检索器
from langchain_core.retrievers import BaseRetriever
# 导入文档类
from langchain_core.documents import Document
# 导入基础LLM类型
from langchain_core.language_models import BaseLanguageModel
# 导入提示词模板
from langchain_core.prompts import PromptTemplate
# 导入自定义llm对象
from llm import llm
# 导入自定义嵌入对象
from embeddings import embeddings
# 导入获取向量库的工具
from vector_store import get_vector_store

# 定义摘要索引器类
class AbstractedDocumentIndexer:
    # 摘要索引器：使用LLM生成文档摘要，对摘要向量化，检索时用摘要匹配，返回原始文档
    """摘要索引器：使用LLM生成文档摘要，对摘要向量化，检索时用摘要匹配，返回原始文档"""

    # 初始化方法
    def __init__(
        self,
        llm: BaseLanguageModel,
        embeddings,
        vector_store,
        chunk_size: int = 300,
        summary_prompt_template: Optional[str] = None
    ):
        # 初始化摘要索引器，包括llm、嵌入模型、向量库、分块大小、摘要提示词模板
        """
        初始化摘要索引器

        Args:
            llm: 大语言模型，用于生成摘要
            embeddings: 嵌入模型，用于生成向量
            vector_store: 向量存储，用于存储和检索向量
            chunk_size: 文档块大小（字符数）
            summary_prompt_template: 摘要生成提示词模板，如果为None则使用默认模板
        """
        # 保存大语言模型
        self.llm = llm
        # 保存嵌入模型
        self.embeddings = embeddings
        # 保存向量库
        self.vector_store = vector_store
        # 保存文档块大小
        self.chunk_size = chunk_size

        # 存储原始文档块的内容（doc_id -> content）
        self.original_chunks: Dict[str, str] = {}

        # 判断是否传入了自定义的摘要生成提示词模板
        if summary_prompt_template is None:
            # 没有则使用默认的中文摘要提示词模板
            self.summary_prompt_template = PromptTemplate(
                input_variables=["text"],
                template="请为以下文档生成一个简洁的摘要，突出核心内容和关键信息：\n\n{text}"
            )
        else:
            # 否则使用用户传入的模板
            self.summary_prompt_template = PromptTemplate(
                input_variables=["text"],
                template=summary_prompt_template
            )

        # 创建按字符切分的文档切分器，不重叠
        self.text_splitter = CharacterTextSplitter(
            separator="",
            chunk_size=chunk_size,
            chunk_overlap=0
        )

    # 索引文档的方法
    def index_documents(self, texts: List[str], metadatas: List[Dict[str, Any]] = None):
        # 为每个文档块生成摘要并向量化存储
        """索引文档：为每个文档块生成摘要并向量化存储"""
        # 用于保存所有摘要文本
        summary_texts = []
        # 用于保存所有摘要文本的元数据
        summary_metadatas = []

        # 打印开始索引文档数
        print(f"\n开始索引 {len(texts)} 个文档...")
        # 打印当前分块配置
        print(f"配置：文档块大小={self.chunk_size}字符\n")

        # 遍历所有原始文档和对应元数据
        for doc_idx, (text, metadata) in enumerate(zip(texts, metadatas)):
            # 生成唯一文档ID
            doc_id = str(uuid.uuid4())
            # 保存文档块内容到original_chunks中
            self.original_chunks[doc_id] = text

            # 打印当前正在生成摘要的提示
            print(f"文档 {doc_idx + 1}: 正在生成摘要（原始长度: {len(text)} 字符）...")
            # 基于模板生成摘要的prompt
            prompt = self.summary_prompt_template.format(text=text)
            # 用LLM生成摘要内容
            summary = self.llm.invoke(prompt)

            # 如果返回对象有content属性，则取其内容，否则转为字符串
            if hasattr(summary, 'content'):
                summary_text = summary.content.strip()
            else:
                summary_text = str(summary).strip()

            # 打印摘要生成完成
            print(f"摘要生成完成（摘要长度: {len(summary_text)} 字符）")

            # 构造摘要的元数据，包含原始ID、长度、摘要长度、索引类型及原有元数据
            summary_metadata = {
                **metadata,#原有元数据
                "original_chunk_id": doc_id,#原始文档块ID
                "original_chunk_length": len(text),#原始文档块长度
                "summary_length": len(summary_text),#摘要长度
                "index_type": "abstracted"#索引类型
            }

            # 保存摘要和摘要元数据
            summary_texts.append(summary_text)
            summary_metadatas.append(summary_metadata)


        # 所有摘要生成完成后进行批量向量化并保存到向量库
        print(f"\n总共生成 {len(summary_texts)} 个摘要，开始向量化并存储...")
        self.vector_store.add_texts(summary_texts, summary_metadatas)
        print("索引完成！\n")

    # 检索文档的方法
    def retrieve(self, query: str, k: int = 4) -> List[Document]:
        # 检索文档：使用摘要进行相似度计算，返回原始文档块（带分数）
        """检索文档：使用摘要进行相似度计算，返回原始文档块，分数存储在metadata中"""
        # 利用向量库的similarity_search_with_score方法召回k个相关摘要及分数
        summary_results = self.vector_store.similarity_search_with_score(query, k=k)

        # 构建原始文档的结果列表
        original_docs = []

        # 遍历召回的每个摘要及其分数
        for summary_doc, distance in summary_results:
            # 获取与该摘要相关的原始文档块ID
            doc_id = summary_doc.metadata.get("original_chunk_id")

            # 如果找到该ID
            if doc_id:
                # 取回原始内容
                original_content = self.original_chunks.get(doc_id)

                # 如果能取得内容则包装为Document对象
                if original_content:
                    original_doc = Document(
                        page_content=original_content,#原始内容
                        metadata={
                            **summary_doc.metadata,#摘要元数据
                            "retrieved_from_summary": True,#是否从摘要中检索
                            "original_chunk_id": doc_id,#原始文档块ID
                            "score": float(distance)  # 使用原始分数（距离，越小越相似）
                        }#原始文档块元数据
                    )
                    # 加入结果列表
                    original_docs.append(original_doc)

        # 返回所有原始文档块（包含分数）
        return original_docs


# 定义摘要检索器，继承自LangChain的BaseRetriever
class AbstractedRetriever(BaseRetriever):
    # 摘要检索器（继承自BaseRetriever，可用于LangChain链）
    """摘要检索器（继承自BaseRetriever，可用于LangChain链）"""

    # 索引器实例
    indexer: AbstractedDocumentIndexer
    # 默认top-k检索数量
    k: int = 4

    # 允许任意类型的配置
    model_config = ConfigDict(arbitrary_types_allowed=True)

    # LangChain检索器核心方法
    def _get_relevant_documents(self, query: str, **kwargs) -> List[Document]:
        # 获取检索的top-k数量
        k = kwargs.get("k", self.k)
        # 调用indexer的检索方法返回相关文档
        return self.indexer.retrieve(query, k=k)

# 获取一个Chroma的向量存储实例，指定路径和集合名
vector_store = get_vector_store(
    persist_directory="chroma_db",
    collection_name="abstracted"
)

# 创建摘要索引器实例，传入llm、embeddings、向量库和分块大小
indexer = AbstractedDocumentIndexer(
    llm=llm,
    embeddings=embeddings,
    vector_store=vector_store,
    chunk_size=300
)

# 示例文档，三类主题，每类文档为单条超长字符串
documents = [
    # 人工智能相关文档块
    "人工智能（Artificial Intelligence，AI）是计算机科学的一个分支，旨在创建能够执行通常需要人类智能的任务的系统。"
    "机器学习是人工智能的核心技术之一，它使计算机能够从数据中学习，而无需明确编程。"
    "深度学习是机器学习的一个子集，使用人工神经网络来模拟人脑的工作方式。"
    "自然语言处理（NLP）是AI的另一个重要领域，专注于使计算机能够理解和生成人类语言。",
    # 区块链相关文档块
    "区块链技术是一种分布式账本技术，通过密码学方法确保数据的安全性和不可篡改性。"
    "比特币是第一个成功应用区块链技术的加密货币，它解决了数字货币的双重支付问题。"
    "以太坊是一个支持智能合约的区块链平台，允许开发者在其上构建去中心化应用（DApps）。"
    "智能合约是自动执行的合约，其条款直接写入代码中，无需第三方中介。",
    # 量子计算相关文档块
    "量子计算是一种利用量子力学现象进行计算的新兴技术，具有巨大的计算潜力。"
    "量子比特（qubit）是量子计算的基本单位，与经典比特不同，它可以同时处于0和1的叠加态。"
    "量子纠缠是量子计算的关键特性，允许量子比特之间建立特殊的关联关系。"
    "量子计算在密码学、药物发现和优化问题等领域具有潜在的应用前景。",
]

# 对示例文档进行摘要索引构建，每个文档带有不同的主题元数据
indexer.index_documents(
    texts=documents,
    metadatas=[
        {"topic": "人工智能", "category": "科技"},
        {"topic": "区块链", "category": "科技"},
        {"topic": "量子计算", "category": "科技"},
    ]
)

# 新建一个摘要检索器实例，只返还最相关的2个结果
retriever = AbstractedRetriever(indexer=indexer, k=2)
# 示例检索问题
query = "机器学习是人工智能的核心技术之一，它使计算机能够从数据中学习，而无需明确编程"
# 执行检索获得相关文档（带分数）
docs = retriever.invoke(query)
# 遍历检索结果并打印内容的前200个字符及分数
for i, doc in enumerate(docs, 1):
    print(f"\n文档 {i} (分数: {doc.metadata.get('score', 'N/A'):.4f}):")
    print(f"内容: {doc.page_content[:200]}...")

2.2 llm.py #

"""
LLM配置

统一管理项目中使用的LLM配置
"""

from langchain_deepseek import ChatDeepSeek

# 初始化DeepSeek LLM
# 需要设置环境变量 DEEPSEEK_API_KEY，或直接在代码中传入 api_key 参数
llm = ChatDeepSeek(
    model="deepseek-chat",
    temperature=0.7
)

2.3 执行过程 #

2.3.1 核心思想 #

摘要索引采用“摘要检索，原文返回”：

• 对大文档块使用 LLM 生成摘要
• 对摘要进行向量化并建立索引
• 检索时用摘要的 embedding 进行相似度匹配
• 返回原始文档块，保持上下文完整

2.3.2 执行流程 #

阶段一：初始化

# 1. 获取向量存储实例
vector_store = get_vector_store(
    persist_directory="chroma_db",
    collection_name="abstracted"
)

# 2. 创建摘要索引器
indexer = AbstractedDocumentIndexer(
    llm=llm,                    # 用于生成摘要
    embeddings=embeddings,      # 用于向量化
    vector_store=vector_store,  # 向量存储
    chunk_size=300              # 文档块大小
)

初始化时：

• 创建默认摘要提示词模板（如未提供自定义模板）
• 创建 CharacterTextSplitter 用于文档切分
• 初始化 original_chunks 字典用于存储原始文档

阶段二：文档索引

indexer.index_documents(
    texts=documents,           # 3个文档块
    metadatas=[...]           # 对应的元数据
)

索引过程：

1. 遍历每个原始文档块
2. 为每个文档生成唯一 UUID（doc_id）
3. 将原始文档内容存入 original_chunks 字典
4. 使用 LLM 生成摘要：
   • 基于提示词模板构建 prompt
   • 调用 llm.invoke(prompt) 生成摘要
   • 提取摘要文本内容
5. 为摘要构建元数据（包含 original_chunk_id、长度信息等）
6. 收集所有摘要文本和元数据
7. 批量向量化并存入向量数据库

示例：一个300字符的文档块可能生成一个50-100字符的摘要。

阶段三：文档检索

retriever = AbstractedRetriever(indexer=indexer, k=2)
docs = retriever.invoke("机器学习是人工智能的核心技术...")

检索过程：

1. retriever.invoke() 调用 _get_relevant_documents()
2. 调用 indexer.retrieve(query, k=2)
3. 在向量库中使用摘要进行相似度检索（返回 k 个最相关的摘要）
4. 遍历检索结果：
   • 从摘要元数据中获取 original_chunk_id
   • 从 original_chunks 字典中获取原始文档内容
   • 构建 Document 对象，包含原始内容和分数
5. 返回原始文档列表（按相似度排序）

2.3.3 类图 #

2.3.4 时序图 #

2.3.4.1 索引阶段时序图 #

2.3.4.2 检索阶段时序图 #

2.3.5 关键设计要点 #

1. 摘要生成流程

原始文档块 (300字符)
    ↓
LLM生成摘要 (50-100字符)
    ↓
向量化存储
    ↓
检索时：摘要匹配 → 返回原始文档

2. 数据结构设计

• original_chunks 字典：{doc_id: original_content}

  • 用于存储原始文档内容
  • 通过 original_chunk_id 关联摘要和原文

• 摘要元数据包含：

  • original_chunk_id：关联原始文档
  • original_chunk_length：原始文档长度
  • summary_length：摘要长度
  • index_type: "abstracted"：索引类型标识
  • 继承原始文档的元数据（如 topic、category）

3. 检索策略

• 检索阶段：使用摘要进行相似度匹配
  • 摘要包含核心语义，适合概括性查询
  • 匹配抽象需求，而非具体细节
• 返回阶段：返回原始文档块
  • 保持上下文完整性
  • 避免信息丢失

2.3.6 与层次化索引的对比 #

2.3.7 优势与应用场景 #

优势：

• 提高检索精度：摘要包含核心语义，更容易匹配概括性查询
• 匹配抽象需求：适合需要概括性经验或抽象信息的场景
• 保持上下文完整性：返回原始文档块，避免信息丢失

适用场景：

• 知识库问答：用户询问概念、原理等抽象问题
• 文档检索：需要匹配文档主题而非具体细节
• 经验总结：检索概括性经验或最佳实践

该设计在抽象语义匹配和上下文完整性之间取得平衡，适用于需要概括性检索的场景。

3. 问题化索引 #

背景与动机

在传统的RAG（Retrieval-Augmented Generation）流程中，常用的文档检索方式是将文本分块（chunking），再将这些原始内容向量化后存入向量库。在查询阶段，将用户的问题与原始文本分块做语义匹配。如果用户问题和chunk内容表达方式差异较大，即使意思很接近，也容易“错过”重要内容。

例如：

• 用户问题：什么是机器学习？
• 文本块内容：机器学习是人工智能的核心技术之一，它使计算机能够从数据中学习，而无需明确编程。

如果直接用内容向量去比对，很容易分数不理想。

为了解决自然语言表达形态差异导致的检索精准度问题，可以引入“假设性问题（Hypothetical Questions）”作为索引桥梁。

方法简介

HQI的核心流程如下：

1. 文档块转换为问题（Question化）：

   • 对每个分块内容，调用LLM，让其生成若干个基于原文、覆盖关键信息的、能在当前文档内直接答复的问题。
   • 这些“假设性问题”表达方式与用户真正检索时提出的问题往往更接近（例如都会问“什么是…？”、“…的原理是什么？”“…有哪些应用？”等）。

2. 假设性问题向量化与存储：

   • 将生成的问题而不是原始内容进行向量化，和原有分块的元信息一起存入向量库。

3. 查询时用“问题vs问题”做语义比对：

   • 用户的问题直接与索引库中的所有“假设性问题”比对，相似度更高、语义距离更短。

4. 返回原始文档内容：

   • 一旦检索到高相关的假设性问题，即可定位原始内容块，实现信息的完整召回。

HQI的优势

• 问题表达适配查询： 假设性问题的形态与用户输入自然地对齐，检索的“意图对齐”能力得到数倍提升。
• 保障原文完整性： 只用问题做索引，真正返回内容时仍然提供原始chunk，信息粒度可控。
• 灵活性强： 支持为每个文档生成多个问题，更好覆盖不同“问法、提问风格”。

典型流程举例

以“人工智能”相关文档块为例：

• 文档块原文：

  人工智能（Artificial Intelligence，AI）是计算机科学的一个分支，旨在创建能够执行通常需要人类智能的任务的系统。机器学习是人工智能的核心技术之一，它使计算机能够从数据中学习，而无需明确编程。

• 由LLM自动生成的不超过3条假设性问题：

  1. 什么是人工智能？
  2. 机器学习在人工智能中的作用是什么？
  3. 人工智能系统为什么能够执行人类智能任务？

• 存储到向量库的单位是这些问题，其元数据会记录是由哪一块内容生成。

• 检索时，如果用户问：“什么是机器学习？”，此问题就和其中的第二个假设性问题有较高相似度，从而精准召回对应内容。

实践建议

• 推荐每个文档块生成2~5个覆盖面的高质量问题。
• 问题生成prompt要明确要求“答案须能在该分块找到”，保证检索后不会出现“找不到答案片段”的尴尬情况。
• 假设性问题的内容只做索引，不直接返回给用户；返回结果时尽量展现完整原文块内容，以供进一步问答或总结。

3.1. HypotheticalQuestionIndex.py #

60.HypotheticalQuestionIndex.py

# 假设性问题索引（Hypothetical Question Index）
"""
假设性问题索引（Hypothetical Question Index）

假设性问题索引是一种RAG优化方案，其核心思想是：
1. 对于给定的文档块，使用大语言模型生成若干个假设性问题
2. 确保这些问题的答案都能在原文中找到
3. 对假设性问题进行向量化处理，建立索引
4. 检索时使用假设性问题的embedding与用户问题进行匹配
5. 返回对应的原始文档内容

优势：
- 问题与问题之间的语义匹配更加直接和准确
- 能够更好地匹配用户的问题形式查询
- 保持上下文完整性：返回原始文档块，避免信息丢失
"""

# 导入uuid用于生成唯一ID
import uuid
# 导入类型注解相关模块
from typing import List, Dict, Any, Optional
# 导入pydantic用于配置
from pydantic import ConfigDict
# 导入字符切分器
from langchain_text_splitters import CharacterTextSplitter
# 导入基础检索器
from langchain_core.retrievers import BaseRetriever
# 导入文档类
from langchain_core.documents import Document
# 导入基础LLM类型
from langchain_core.language_models import BaseLanguageModel
# 导入提示词模板
from langchain_core.prompts import PromptTemplate
# 导入自定义llm对象
from llm import llm
# 导入自定义嵌入对象
from embeddings import embeddings
# 导入获取向量库的工具
from vector_store import get_vector_store
# 导入正则表达式用于提取问题
import re

# 定义假设性问题索引器类
class HypotheticalQuestionIndexer:
    """假设性问题索引器：使用LLM生成假设性问题，对问题向量化，检索时用问题匹配，返回原始文档"""

    # 初始化方法
    def __init__(
        self,
        llm: BaseLanguageModel,
        embeddings,
        vector_store,
        chunk_size: int = 300,
        num_questions: int = 3,
        question_prompt_template: Optional[str] = None
    ):
        """
        初始化假设性问题索引器

        Args:
            llm: 大语言模型，用于生成假设性问题
            embeddings: 嵌入模型，用于生成向量
            vector_store: 向量存储，用于存储和检索向量
            chunk_size: 文档块大小（字符数）
            num_questions: 为每个文档块生成的假设性问题数量
            question_prompt_template: 问题生成提示词模板，如果为None则使用默认模板
        """
        # 保存大语言模型
        self.llm = llm
        # 保存嵌入模型
        self.embeddings = embeddings
        # 保存向量库
        self.vector_store = vector_store
        # 保存文档块大小
        self.chunk_size = chunk_size
        # 保存每个文档块生成的问题数量
        self.num_questions = num_questions

        # 存储原始文档块内容（doc_id -> content）
        self.original_chunks: Dict[str, str] = {}

        self.question_prompt_template = PromptTemplate(
                input_variables=["text", "num_questions"],
                template="请为以下文档生成{num_questions}个假设性问题，确保这些问题的答案都能在文档中找到。每个问题单独一行，以问号结尾。\n\n文档内容：\n{text}\n\n问题："
        )
        # 创建按字符切分的文档切分器，不重叠
        self.text_splitter = CharacterTextSplitter(
            separator="",#分隔符
            chunk_size=chunk_size,#文档块大小
            chunk_overlap=0#文档块重叠
        )

    # 索引文档的方法
    def index_documents(self, texts: List[str], metadatas: List[Dict[str, Any]] = None):
        """索引文档：为每个文档块生成假设性问题并向量化存储"""
        # 用于保存所有问题文本
        question_texts = []
        # 用于保存所有问题的元数据
        question_metadatas = []

        # 打印开始索引文档数
        print(f"\n开始索引 {len(texts)} 个文档...")
        # 打印当前配置
        print(f"配置：文档块大小={self.chunk_size}字符, 每个文档生成{self.num_questions}个问题\n")

        # 遍历所有原始文档和对应元数据
        for doc_idx, (text, metadata) in enumerate(zip(texts, metadatas)):
            # 生成唯一文档ID
            doc_id = str(uuid.uuid4())
            # 保存文档块内容到original_chunks中
            self.original_chunks[doc_id] = text

            # 打印当前正在生成问题的提示
            print(f"文档 {doc_idx + 1}: 正在生成假设性问题（原始长度: {len(text)} 字符）...")
            # 基于模板生成问题的prompt
            prompt = self.question_prompt_template.format(text=text, num_questions=self.num_questions)
            # 用LLM生成问题内容
            questions_response = self.llm.invoke(prompt)

            # 如果返回对象有content属性，则取其内容，否则转为字符串
            questions_text = questions_response.content.strip()

            # 解析生成的问题（按行分割，提取包含问号的问题）
            questions = self._parse_questions(questions_text)
            # 限制问题数量为num_questions
            questions = questions[:self.num_questions]

            # 打印问题生成数量
            print(f"问题生成完成（生成了 {len(questions)} 个问题）")

            # 为每个问题创建索引
            for question_idx, question in enumerate(questions):
                # 构造问题的元数据，包含原始ID、长度、问题索引、索引类型及原有元数据
                question_metadata = {
                    **metadata,  # 原有元数据
                    "original_chunk_id": doc_id,  # 原始文档块ID
                    "original_chunk_length": len(text),  # 原始文档块长度
                    "question_index": question_idx,  # 问题索引
                    "question_length": len(question),  # 问题长度
                    "index_type": "hypothetical_question"  # 索引类型
                }

                # 保存问题和问题元数据
                question_texts.append(question)
                question_metadatas.append(question_metadata)

        # 所有问题生成完成后进行批量向量化并保存到向量库
        print(f"\n总共生成 {len(question_texts)} 个问题，开始向量化并存储...")
        self.vector_store.add_texts(question_texts, question_metadatas)
        print("索引完成！\n")

    # 解析LLM生成的问题文本，提取问题列表
    def _parse_questions(self, questions_text: str) -> List[str]:
        """
        解析LLM生成的问题文本，提取问题列表

        Args:
            questions_text: LLM返回的问题文本

        Returns:
            问题列表
        """
        # 初始化问题列表
        questions = []
        # 按行分割
        lines = questions_text.split('\n')

        # 遍历每一行
        for line in lines:
            # 去除首尾空白
            line = line.strip()
            # 跳过空行
            if not line:
                continue
            # 移除行首编号（如"1. "、"1、"等）
            line = re.sub(r'^\d+[\.、]\s*', '', line)
            # 如果包含问号，认为是问题
            if '？' in line or '?' in line:
                questions.append(line)

        # 返回问题列表
        return questions

    # 检索文档的方法
    def retrieve(self, query: str, k: int = 4) -> List[Document]:
        """检索文档：使用假设性问题进行相似度计算，返回原始文档块（带分数）"""
        # 利用向量库的similarity_search_with_score方法召回k*3个相关问题及分数（提高覆盖率）
        question_results = self.vector_store.similarity_search_with_score(query, k=k * 3)

        # 用于记录每个文档的最佳分数（距离越小越相似）
        doc_scores: Dict[str, float] = {}  # doc_id -> best_score
        # 用于存储每个文档的Document对象
        doc_docs_map: Dict[str, Document] = {}  # doc_id -> document

        # 遍历召回的每个问题及其分数
        for question_doc, distance in question_results:
            # 获取与该问题相关的原始文档块ID
            doc_id = question_doc.metadata.get("original_chunk_id")

            # 如果存在文档块ID
            if doc_id:
                # 获取原始内容
                original_content = self.original_chunks.get(doc_id)
                # 如果能取得内容
                if original_content:
                    # 如果文档首次出现或分数更好（距离更小），则更新
                    if doc_id not in doc_scores or distance < doc_scores[doc_id]:
                        # 更新文档分数
                        doc_scores[doc_id] = float(distance)
                        # 创建Document对象
                        doc_docs_map[doc_id] = Document(
                            page_content=original_content,  # 原始内容
                            metadata={
                                **question_doc.metadata,  # 问题元数据
                                "retrieved_from_question": True,  # 是否从问题中检索
                                "original_chunk_id": doc_id,  # 原始文档块ID
                                "score": float(distance)  # 使用原始分数（距离，越小越相似）
                            }
                        )

        # 将所有文档按分数升序排序，取前k个
        sorted_docs = sorted(doc_scores.items(), key=lambda x: x[1])[:k]
        # 返回Document对象组成的列表
        return [doc_docs_map[doc_id] for doc_id, _ in sorted_docs]


# 定义假设性问题检索器，继承自LangChain的BaseRetriever
class HypotheticalQuestionRetriever(BaseRetriever):
    """假设性问题检索器（继承自BaseRetriever，可用于LangChain链）"""

    # 索引器实例
    indexer: HypotheticalQuestionIndexer
    # 默认top-k检索数量
    k: int = 4

    # 允许任意类型的配置
    model_config = ConfigDict(arbitrary_types_allowed=True)

    # LangChain检索器核心方法
    def _get_relevant_documents(self, query: str, **kwargs) -> List[Document]:
        # 获取检索的top-k数量
        k = kwargs.get("k", self.k)
        # 调用indexer的检索方法返回相关文档
        return self.indexer.retrieve(query, k=k)


# 获取向量存储实例
vector_store = get_vector_store(
    persist_directory="chroma_db",
    collection_name="hypothetical_question"
)

# 创建假设性问题索引器实例，传入llm、embeddings、向量库和相关配置
indexer = HypotheticalQuestionIndexer(
    llm=llm,#大语言模型
    embeddings=embeddings,#嵌入模型
    vector_store=vector_store,#向量库
    chunk_size=300,#文档块大小
    num_questions=3,#每个文档块生成的问题数量
)

# 示例文档，三类主题，每类文档为单条超长字符串
documents = [
    # 人工智能相关文档块
    "人工智能（Artificial Intelligence，AI）是计算机科学的一个分支，旨在创建能够执行通常需要人类智能的任务的系统。"
    "机器学习是人工智能的核心技术之一，它使计算机能够从数据中学习，而无需明确编程。"
    "深度学习是机器学习的一个子集，使用人工神经网络来模拟人脑的工作方式。"
    "自然语言处理（NLP）是AI的另一个重要领域，专注于使计算机能够理解和生成人类语言。",

    # 区块链相关文档块
    "区块链技术是一种分布式账本技术，通过密码学方法确保数据的安全性和不可篡改性。"
    "比特币是第一个成功应用区块链技术的加密货币，它解决了数字货币的双重支付问题。"
    "以太坊是一个支持智能合约的区块链平台，允许开发者在其上构建去中心化应用（DApps）。"
    "智能合约是自动执行的合约，其条款直接写入代码中，无需第三方中介。",

    # 量子计算相关文档块
    "量子计算是一种利用量子力学现象进行计算的新兴技术，具有巨大的计算潜力。"
    "量子比特（qubit）是量子计算的基本单位，与经典比特不同，它可以同时处于0和1的叠加态。"
    "量子纠缠是量子计算的关键特性，允许量子比特之间建立特殊的关联关系。"
    "量子计算在密码学、药物发现和优化问题等领域具有潜在的应用前景。",
]

# 对示例文档进行假设性问题索引构建，每个文档带有不同的主题元数据
indexer.index_documents(
    texts=documents,
    metadatas=[
        {"topic": "人工智能", "category": "科技"},
        {"topic": "区块链", "category": "科技"},
        {"topic": "量子计算", "category": "科技"},
    ]
)

# 新建一个假设性问题检索器实例，设置返回最相关的2个结果
retriever = HypotheticalQuestionRetriever(indexer=indexer, k=2)
# 定义示例检索问题
query = "什么是机器学习？"
# 执行检索，获得相关文档（带分数）
docs = retriever.invoke(query)
# 遍历检索结果并打印内容前200个字符及分数
for i, doc in enumerate(docs, 1):
    print(f"\n文档 {i} (分数: {doc.metadata.get('score', 'N/A'):.4f}):")
    print(f"内容: {doc.page_content[:200]}...")

3.2 执行过程 #

3.2.1 核心思想 #

假设性问题索引采用“问题匹配问题”：

• 为每个文档块使用 LLM 生成多个假设性问题
• 确保这些问题的答案都在原文中
• 对假设性问题进行向量化并建立索引
• 检索时用假设性问题的 embedding 与用户问题匹配
• 返回对应的原始文档内容

3.2.2 执行流程 #

阶段一：初始化

# 1. 获取向量存储实例
vector_store = get_vector_store(
    persist_directory="chroma_db",
    collection_name="hypothetical_question"
)

# 2. 创建假设性问题索引器
indexer = HypotheticalQuestionIndexer(
    llm=llm,                    # 用于生成问题
    embeddings=embeddings,      # 用于向量化
    vector_store=vector_store,  # 向量存储
    chunk_size=300,            # 文档块大小
    num_questions=3            # 每个文档生成3个问题
)

初始化时：

• 创建默认问题生成提示词模板（如未提供自定义模板）
• 创建 CharacterTextSplitter 用于文档切分
• 初始化 original_chunks 字典用于存储原始文档

阶段二：文档索引

indexer.index_documents(
    texts=documents,           # 3个文档块
    metadatas=[...]           # 对应的元数据
)

索引过程：

1. 遍历每个原始文档块
2. 为每个文档生成唯一 UUID（doc_id）
3. 将原始文档内容存入 original_chunks 字典
4. 使用 LLM 生成假设性问题：
   • 基于提示词模板构建 prompt（包含文档内容和问题数量）
   • 调用 llm.invoke(prompt) 生成问题文本
   • 提取问题文本内容
5. 解析生成的问题：
   • 调用 _parse_questions() 方法
   • 按行分割，移除编号，提取包含问号的问题
   • 限制问题数量为 num_questions
6. 为每个问题构建元数据（包含 original_chunk_id、问题索引等）
7. 收集所有问题文本和元数据
8. 批量向量化并存入向量数据库

示例：一个300字符的文档块可能生成3个假设性问题，如：

• "什么是机器学习？"
• "机器学习如何工作？"
• "机器学习有哪些应用？"

阶段三：文档检索

retriever = HypotheticalQuestionRetriever(indexer=indexer, k=2)
docs = retriever.invoke("什么是机器学习？")

检索过程：

1. retriever.invoke() 调用 _get_relevant_documents()
2. 调用 indexer.retrieve(query, k=2)
3. 在向量库中使用假设性问题进行相似度检索（返回 k×3=6 个相关问题，提高覆盖率）
4. 遍历检索结果，进行文档聚合：
   • 从问题元数据中获取 original_chunk_id
   • 从 original_chunks 字典中获取原始文档内容
   • 使用 doc_scores 和 doc_docs_map 记录每个文档的最佳分数
   • 如果同一文档出现多次，保留最佳分数（距离最小）
5. 按分数排序，取前 k 个文档
6. 返回原始文档列表（按相似度排序）

3.2.3 类图 #

3.2.4 时序图 #

3.2.4.1 索引阶段时序图 #

3.2.4.2 检索阶段时序图 #

3.2.5 关键设计要点 #

问题生成流程

原始文档块 (300字符)
    ↓
LLM生成假设性问题 (3个问题)
    ↓
解析问题（提取、去编号）
    ↓
向量化存储（每个问题单独索引）
    ↓
检索时：问题匹配问题 → 返回原始文档

数据结构设计

• original_chunks 字典：{doc_id: original_content}

  • 存储原始文档内容
  • 通过 original_chunk_id 关联问题和原文

• 问题元数据包含：

  • original_chunk_id：关联原始文档
  • question_index：问题在文档中的索引（0, 1, 2...）
  • question_length：问题长度
  • original_chunk_length：原始文档长度
  • index_type: "hypothetical_question"：索引类型标识
  • 继承原始文档的元数据（如 topic、category）

检索策略

• 检索阶段：使用假设性问题进行相似度匹配
  • 问题与问题之间的语义匹配更直接
  • 检索 k×3 个问题，提高覆盖率
• 聚合阶段：文档去重与最佳分数选择
  • 同一文档可能被多个问题匹配到
  • 使用 doc_scores 记录每个文档的最佳分数（距离最小）
  • 使用 doc_docs_map 存储对应的 Document 对象
• 返回阶段：按分数排序，返回前 k 个原始文档

问题解析逻辑

_parse_questions() 方法的关键步骤：

1. 按行分割问题文本
2. 去除首尾空白，跳过空行
3. 移除行首编号（如 "1. "、"1、"等）
4. 提取包含问号（？或?）的行作为问题

示例解析：

输入：
"1. 什么是机器学习？
2. 机器学习如何工作？
3. 机器学习有哪些应用？"

输出：
["什么是机器学习？", "机器学习如何工作？", "机器学习有哪些应用？"]

3.2.6 与摘要索引的对比 #

3.2.7 优势与应用场景 #

优势：

• 问题与问题之间的语义匹配更直接和准确
• 能更好地匹配用户的问题形式查询
• 保持上下文完整性：返回原始文档块，避免信息丢失
• 提高覆盖率：每个文档生成多个问题，增加匹配机会

适用场景：

• 问答系统：用户以问题形式查询
• 知识库检索：需要匹配问题语义
• 教育场景：学生提问，系统检索相关内容
• FAQ系统：匹配常见问题

该设计通过“问题匹配问题”的方式，在问题形式查询场景下提供更准确的语义匹配，适用于需要直接匹配问题语义的 RAG 应用。

4. hnsw:space #

4.1. 核心原则 #

所有hnsw:space设置下，similarity_search_with_score返回的score都是距离值（distance），不是相似度（similarity）。
因此：score越小表示越相似，score越大表示越不相似。

4.2. 三种空间类型详细对比 #

4.2.1. L2（欧几里得距离） - 默认值 #

collection_metadata={"hnsw:space": "l2"}

示例：

• score = 0.0 → 完全相同
• score = 1.414 → 中等相似
• score = 10.0 → 很不相似

4.2.2. IP（内积距离） #

collection_metadata={"hnsw:space": "ip"}

示例（假设向量已归一化）：

• score = -0.98 → 非常相似（内积=0.98）
• score = 0.0 → 正交/无关（内积=0）
• score = 0.5 → 不相似（内积=-0.5）

4.2.3. Cosine（余弦距离） - 文本嵌入最常用 #

collection_metadata={"hnsw:space": "cosine"}

示例：

• score = 0.0 → 完全相同（余弦相似度=1.0）
• score = 0.3 → 很相似（余弦相似度=0.7）
• score = 1.0 → 正交/无关（余弦相似度=0.0）
• score = 1.7 → 很不相似（余弦相似度=-0.7）
• score = 2.0 → 完全相反（余弦相似度=-1.0）

4.3. 关系转换公式 #

从距离转换到相似度：

# L2距离转相似度（需要最大值假设）
def l2_to_similarity(distance, max_distance=10):
    return 1 - (distance / max_distance)

# IP距离转相似度
def ip_to_similarity(distance):
    return -distance  # 因为 distance = -内积

# Cosine距离转相似度
def cosine_to_similarity(distance):
    return 1 - distance  # 因为 distance = 1 - 余弦相似度

4.4. 选择指南 #

4.4.1. 根据向量类型选择： #

4.4.2. 根据阈值过滤 #

# Cosine空间的常用阈值
if distance < 0.3:     # 很相似（相似度>0.7）
    pass
elif distance < 0.5:   # 一般相似（相似度>0.5）
    pass
elif distance < 0.7:   # 弱相关（相似度>0.3）
    pass

# L2空间的阈值取决于向量维度
# IP空间的阈值通常围绕0设定

4.5. 实践代码 #

# 导入Chroma模块用于构建向量数据库
from langchain_chroma import Chroma

# 导入HuggingFaceEmbeddings用于加载embedding模型
from langchain_huggingface import HuggingFaceEmbeddings

# 指定本地embedding模型的路径
model_path = "C:/Users/Administrator/.cache/modelscope/hub/models/sentence-transformers/all-MiniLM-L6-v2"

# 加载指定路径的模型，并指定运行在CPU上
embeddings = HuggingFaceEmbeddings(
    model_name=model_path,
    model_kwargs={"device": "cpu"}
)

# 创建Chroma向量存储，指定持久化目录、embedding函数、集合名称和集合元数据
store = Chroma(
    persist_directory="chroma_db",
    embedding_function=embeddings,
    collection_name="hierarchical",
    collection_metadata={"hnsw:space": "cosine"}
)

# 定义检索的查询文本
query = "区块链的应用"

# 定义返回的top_k相似结果数量
top_k = 3

# 调用相似性搜索方法，返回结果及其分数
results = store.similarity_search_with_score(query, k=top_k)

# 获取当前集合的空间类型（如余弦、欧式等），默认为'cosine'
space = store._collection.metadata.get('hnsw:space', 'cosine')


# 打印查询内容
print(f"查询: '{query}'")

# 提示分数的含义，分数越小越相似
print(f"分数解释: 值越小越相似")

# 遍历结果，逐条输出内容及分数解释
for i, (doc, score) in enumerate(results):
    # 根据空间类型选择分数的解释方式
    if space == 'cosine':
        # 余弦空间：1-score即为相似度
        similarity = 1 - score
        interpretation = f"相似度: {similarity:.3f}"
    elif space == 'ip':
        # 内积空间：取负数为相似度
        similarity = -score
        interpretation = f"内积: {similarity:.3f}"
    else:  # l2
        # 欧式距离直接显示
        interpretation = f"欧几里得距离: {score:.3f}"
    # 打印结果序号
    print(f"结果 {i+1}:")
    # 打印分数和解释
    print(f"  分数: {score:.4f} ({interpretation})")
    # 打印相应文档的前60个字符内容
    print(f"  内容: {doc.page_content[:60]}...")
    print()

# 给出距离/相似度的阈值建议说明
print("=== 阈值建议 ===")
if space == 'cosine':
    # 针对余弦距离，输出典型阈值区间
    print("余弦距离阈值建议:")
    print("  <0.3 : 高度相关")
    print("  0.3-0.5 : 相关")
    print("  0.5-0.7 : 弱相关")
    print("  >0.7 : 可能不相关")
elif space == 'l2':
    # 针对欧式距离，说明阈值依赖实际数据
    print("L2距离: 阈值依赖向量维度和数据分布")

4.6. 重要注意事项 #

1. 存储与检索一致性：创建索引时的hnsw:space设置必须与查询时一致
2. 向量归一化：
   • Cosine：必须归一化（Chroma自动处理）
   • IP：强烈建议归一化
   • L2：不需要归一化
3. 性能差异：
   • Cosine和L2：计算复杂度相似
   • IP：通常稍快，但需要归一化保证效果
4. 跨空间比较：不同空间的分数不能直接比较数值大小

4.7. 总结要点 #

最终建议：对于文本搜索和RAG应用，优先使用hnsw:space: "cosine"，因为它：

1. 专门为方向比较设计
2. 分数范围固定[0,2]，易于设定阈值
3. 与文本语义匹配直觉一致
4. 大多数文本嵌入模型已优化支持