各位编程专家、架构师和对检索增强生成（RAG）充满热情的同仁们，大家下午好！
今天，我们将深入探讨一个在构建高级 RAG 系统中至关重要的技术：Multi-Vector Retriever，特别是它如何通过利用文档的摘要、标题和原始文本进行多维索引，从而显著提升检索的准确性和效率。在当前这个大型语言模型（LLM）飞速发展的时代，如何高效、精准地为 LLM 提供高质量的外部知识，是决定其性能上限的关键。传统的向量检索方法已经取得了巨大的成功，但它们并非没有局限。我们将从这些局限出发，一步步揭示 Multi-Vector Retriever 的强大之处。
一、传统向量检索的局限性：为何我们需要更智能的方案？
在探讨 Multi-Vector Retriever 之前，我们首先回顾一下当前 RAG 系统中最常见的检索范式：单模态向量检索。
其基本流程是：
- 文档分割 (Chunking)：将原始文档分割成固定大小或语义相关的文本块（chunks）。
- 向量嵌入 (Embedding)：使用预训练的嵌入模型（如 BERT, OpenAI Embeddings, BGE 等）将每个文本块转换成高维向量。
- 向量存储 (Vector Store)：将这些向量及其对应的文本块存储在向量数据库（如 Chroma, Pinecone, Weaviate, Milvus 等）中。
- 检索 (Retrieval)：当用户提出查询时，将查询也转换成向量，然后在向量数据库中搜索与查询向量最相似的文本块向量。
- 增强生成 (Augmented Generation)：将检索到的相关文本块作为上下文，连同用户查询一起发送给 LLM，指导其生成答案。
这种方法简单有效，为 RAG 带来了革命性的进展。然而，它也面临着一些固有的挑战：
1. 粒度不匹配 (Granularity Mismatch)
- 过小的文本块：如果文本块太小，它可能缺乏足够的上下文来完整表达一个概念。在检索时，这样的块可能无法被准确匹配，或者即便匹配到，也无法提供给 LLM 足够的信息。
- 过大的文本块：如果文本块太大，它可能包含太多无关信息，稀释了关键内容的密度。这不仅增加了嵌入模型的负担，也可能导致检索到的信息不够聚焦，甚至引入噪音，影响 LLM 的推理。此外，LLM 的上下文窗口有限，过大的块会迅速耗尽可用空间。
2. 信息密度与上下文丢失的矛盾
- 我们希望检索到的文本块既能包含足够多的细节，又能避免冗余。在单模态检索中，这是一个艰难的权衡。选择一个理想的
chunk_size和chunk_overlap往往需要大量的实验和领域知识。 - 一个文档的标题、摘要和正文在信息密度和抽象层次上是截然不同的。标题是文档的最高层概括，摘要提供了更详细但仍精炼的上下文，而正文则包含了所有细节。单模态检索通常只能选择其中一种粒度进行索引，无法兼顾。
3. 检索效率与准确性的挑战
- 当查询非常宽泛或非常具体时，单一粒度的向量检索可能表现不佳。例如，一个宽泛的查询可能需要先找到相关的主题或文档，然后再深入细节；一个具体的查询则需要直接定位到精确的信息点。单一向量类型难以同时满足这两种需求。
- 在某些情况下，用户查询可能更像一个“意图”或“主题”，而不是一个具体的关键词。此时，用文档的完整内容去匹配，效率可能不高，而用文档的摘要或标题去匹配，效果可能更好。
这些局限性促使我们思考：能否利用文档不同层次的语义信息，来构建一个更智能、更鲁棒的检索系统？答案便是 Multi-Vector Retriever。
二、揭示 Multi-Vector Retriever 范式
Multi-Vector Retriever 的核心思想是：不再将一个逻辑文档仅仅表示为一个向量，而是为同一个逻辑文档创建多个不同粒度、不同语义层次的向量表示。这些向量可以对应文档的不同部分（如标题、摘要、段落、表格），或者对同一内容的多种抽象（如关键词列表、问题-答案对、子文档）。
通过为每个逻辑文档生成多组向量，我们可以在检索时根据查询的性质，选择最合适的向量类型进行匹配，或者通过组合不同类型的向量匹配结果，来获得更全面、更精准的检索上下文。
为何这如此强大？
想象一下图书馆。如果你想找一本关于“人工智能在医疗健康领域应用”的书，你可能会：
- 首先在书架上浏览书名（Title）快速筛选。
- 找到几本看起来相关的书后，翻阅它们的内容提要（Summary）来了解主要内容。
- 最终，如果你找到了感兴趣的书籍，你会阅读正文（Original Text）的特定章节来获取详细信息。
Multi-Vector Retriever 正是将这种人类的认知过程数字化和自动化。它允许我们的检索系统以类似的方式“理解”和“导航”文档集合。
在今天的讲座中，我们将重点聚焦于 Multi-Vector Retriever 的一种高级实践：利用摘要 (Summaries)、标题 (Titles) 和原始文本块 (Original Text Chunks) 这三个维度来索引同一文档。
三、高级实践：摘要、标题和原始文本的多维索引
我们将一个逻辑文档视为一个整体，并为它生成三种不同粒度的表示，每种表示都将拥有自己的向量嵌入。
1. 标题向量：快速定位宏观主题
- 目的：标题是文档最精炼的概括，它通常能准确捕捉文档的核心主题或领域。将标题嵌入向量空间，可以用于快速、粗粒度地筛选出与用户查询在宏观主题上相关的文档。
- 场景：当用户查询是一个宽泛的主题（例如，“介绍机器学习的最新进展”），或者用户希望快速了解哪些文档涵盖了某个特定领域时，对标题向量进行检索非常有效。它有助于过滤掉大量不相关的文档，将检索范围缩小到可能包含相关信息的文档集。
- 优势：
- 高效率：标题通常很短，嵌入和检索速度快。
- 高召回率（主题层面）：即使查询的措辞与正文细节不符，只要与标题主题相关，也能被召回。
- 节省LLM上下文：初步筛选后，可以避免将大量不相关的长文本送入LLM。
- 挑战：标题信息量有限，无法提供具体细节。
2. 摘要向量：提供精炼上下文
- 目的：摘要是对文档或文档某个重要部分的凝练总结。它比标题包含更多的上下文信息，但又比原始文本更简洁，去除了冗余细节。将摘要嵌入向量空间，可以用于检索那些在语义内容上与查询高度相关的文档，并提供一个中等粒度的上下文。
- 场景：当用户查询需要了解某个概念的总体解释、某个事件的概要、或者某个问题的解决方案概览时，对摘要向量进行检索非常有用。它能够提供比标题更丰富的背景信息，同时又避免了原始文本的冗长。
- 优势：
- 平衡性：在信息密度和简洁性之间取得了良好平衡。
- 语义丰富：能够捕获文档的核心论点和关键信息。
- 抗噪性强：LLM 生成的摘要通常是抽象的，能够更好地应对查询措辞的多样性。
- 挑战：摘要的质量高度依赖于生成它的 LLM 模型，生成成本相对较高。
3. 原始文本块向量：获取精确细节
- 目的：原始文本块是文档的实际内容片段。它们是提供最详细、最精确信息的来源。将这些文本块嵌入向量空间，用于在已经通过标题和摘要筛选出的文档中，进一步定位到与查询最直接相关的具体细节。
- 场景：一旦通过标题和摘要确定了相关文档，用户可能需要深入了解某个特定的定义、具体步骤、数据点或代码示例。此时，原始文本块是不可替代的。
- 优势：
- 信息最完整：包含所有原始细节。
- 精确性高：能够直接回答非常具体的查询。
- 直接用于LLM：作为最终的上下文输入给LLM。
- 挑战：
- 粒度选择困难：需要精心设计分块策略。
- 冗余信息：可能包含与查询不直接相关的部分。
- 上下文窗口限制：需要仔细管理送入LLM的块数量。
这三种向量的协同作用
Multi-Vector Retriever 的强大之处在于其协同工作机制。它不仅仅是简单地创建三种向量，而是设计一个智能的检索流程，让它们在不同的阶段发挥作用。
一个典型的协同检索流程可能如下：
- 用户查询：例如，“如何在LangChain中实现自定义工具？”
- 第一阶段：主题过滤（利用标题向量）
- 将查询嵌入向量。
- 在标题向量存储中搜索最相似的标题。
- 召回与这些标题关联的逻辑文档 ID。这一步快速缩小了搜索范围到“LangChain”、“自定义工具”等主题。
- 第二阶段：上下文理解（利用摘要向量）
- 针对第一阶段召回的逻辑文档 ID，获取它们的摘要。
- 将查询再次与这些摘要的向量进行匹配，或者直接利用摘要的文本内容进行二次筛选（例如，使用交叉编码器进行重排序）。
- 这一步能更深入地判断哪些文档在内容上真正相关，而不仅仅是标题相关。
- 第三阶段：细节提取（利用原始文本块向量）
- 对于第二阶段筛选出的高度相关的逻辑文档，检索其所有或部分原始文本块。
- 将查询向量与这些原始文本块的向量进行匹配，找出最相关的具体文本片段。
- 或者，将所有相关文档的原始文本块作为候选集，并使用更强大的重排序模型（如交叉编码器）对这些块进行排序，选出Top-K个最相关的块。
- 最终上下文：将重排序后得到的原始文本块作为最终上下文，连同用户查询一起发送给LLM。
通过这种多阶段、多粒度的检索策略，我们能够兼顾检索的广度（通过标题和摘要快速定位相关主题）和深度（通过原始文本块提供精确细节），显著提升 RAG 系统的性能。
四、架构深潜与实现策略
实现 Multi-Vector Retriever 需要精心设计数据预处理、索引和检索流程。我们将以 Python 语言，结合 LangChain 框架和 Chroma 向量数据库为例，详细讲解其实现。
1. 数据预处理：构建多维表示
这是整个流程的基础，也是最耗时和资源密集的部分。
1.1. 文档加载与标识
首先，我们需要加载原始文档，并为每个逻辑文档分配一个唯一的 ID。
from langchain_community.document_loaders import TextLoader
from langchain_core.documents import Document
def load_documents(file_paths):
"""
加载文档并分配唯一ID。
每个Document将有一个metadata['source']和metadata['document_id']。
"""
documents = []
for i, file_path in enumerate(file_paths):
loader = TextLoader(file_path, encoding="utf-8")
doc = loader.load()[0] # 假设每个文件只生成一个Document
doc.metadata["document_id"] = f"doc_{i}"
doc.metadata["source"] = file_path
documents.append(doc)
return documents
# 示例：
# raw_docs = load_documents(["path/to/doc1.txt", "path/to/doc2.txt"])
# print(raw_docs[0].metadata)1.2. 标题提取与生成
如果文档本身没有明确的标题字段，我们可以通过几种方式生成：
- 手动提取：如果文档结构规范，可以从第一行或特定标签中提取。
- LLM 生成：让 LLM 根据文档内容生成一个简洁的标题。这在文档结构不规范时非常有用。
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
def generate_title_with_llm(llm, document_content):
"""
使用LLM为文档生成标题。
"""
prompt = ChatPromptTemplate.from_messages([
("system", "你是一个专业的文档标题生成器。请为以下文档内容生成一个简洁、准确的标题，不超过10个字。"),
("human", "{document_content}")
])
chain = prompt | llm
title = chain.invoke({"document_content": document_content}).content
return title.strip()
# 示例：
# llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# doc_content = raw_docs[0].page_content
# title = generate_title_with_llm(llm, doc_content)
# print(f"Generated Title: {title}")1.3. 摘要生成
摘要生成通常需要 LLM 的帮助，因为它们能够进行抽象总结。
from langchain_core.prompts import PromptTemplate
from langchain.chains.summarize import load_summarize_chain
def generate_summary_with_llm(llm, document_content):
"""
使用LLM为文档生成摘要。
"""
prompt_template = """请对以下文档内容进行总结，提供一个详细且富有信息量的摘要，长度在150-300字之间，涵盖主要观点和关键信息：
"{text}"
摘要:"""
prompt = PromptTemplate(template=prompt_template, input_variables=["text"])
# 简单模式下，直接用prompt和LLM生成
# 更复杂的文档可能需要map_reduce或stuffing策略
summary_chain = load_summarize_chain(llm, chain_type="stuff", prompt=prompt)
summary = summary_chain.invoke({"input_documents": [Document(page_content=document_content)]})
return summary["output_text"].strip()
# 示例：
# llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# doc_content = raw_docs[0].page_content
# summary = generate_summary_with_llm(llm, doc_content)
# print(f"Generated Summary: {summary}")1.4. 原始文本分块
使用 RecursiveCharacterTextSplitter 是一个常见的选择，它能根据多种分隔符递归地分割文本。
from langchain.text_splitter import RecursiveCharacterTextSplitter
def split_document_into_chunks(document_content, chunk_size=1000, chunk_overlap=200):
"""
将文档内容分割成固定大小的文本块。
"""
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap,
separators=["nn", "n", " ", ""],
length_function=len,
is_separator_regex=False,
)
chunks = text_splitter.create_documents([document_content])
return chunks
# 示例：
# doc_content = raw_docs[0].page_content
# chunks = split_document_into_chunks(doc_content)
# print(f"Number of chunks: {len(chunks)}")
# print(chunks[0].page_content[:100])1.5. 结构化处理与关联
这一步是关键，我们需要将生成的标题、摘要和原始文本块与它们的原始逻辑文档关联起来。
def prepare_multi_vector_documents(raw_documents, llm, chunk_size=1000, chunk_overlap=200):
"""
为每个原始文档准备标题、摘要和文本块。
"""
all_title_docs = []
all_summary_docs = []
all_chunk_docs = []
for raw_doc in raw_documents:
doc_id = raw_doc.metadata["document_id"]
doc_content = raw_doc.page_content
# 1. 生成标题
title = generate_title_with_llm(llm, doc_content)
title_doc = Document(
page_content=title,
metadata={
"document_id": doc_id,
"type": "title",
"source": raw_doc.metadata["source"]
}
)
all_title_docs.append(title_doc)
# 2. 生成摘要
summary = generate_summary_with_llm(llm, doc_content)
summary_doc = Document(
page_content=summary,
metadata={
"document_id": doc_id,
"type": "summary",
"source": raw_doc.metadata["source"]
}
)
all_summary_docs.append(summary_doc)
# 3. 分割原始文本块
chunks = split_document_into_chunks(doc_content, chunk_size, chunk_overlap)
for i, chunk in enumerate(chunks):
chunk.metadata.update({
"document_id": doc_id,
"type": "chunk",
"chunk_id": f"{doc_id}_chunk_{i}", # 独立的chunk ID
"source": raw_doc.metadata["source"]
})
all_chunk_docs.append(chunk)
return all_title_docs, all_summary_docs, all_chunk_docs
# 示例：
# raw_docs = load_documents(["data/doc1.txt", "data/doc2.txt"]) # 假设有data目录和txt文件
# llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# titles, summaries, chunks = prepare_multi_vector_documents(raw_docs, llm)
# print(f"Titles count: {len(titles)}")
# print(f"Summaries count: {len(summaries)}")
# print(f"Chunks count: {len(chunks)}")
# print(titles[0].metadata, summaries[0].metadata, chunks[0].metadata)2. 索引阶段：构建多向量存储
现在，我们有了不同类型的文档表示。为了实现高效检索，我们需要将它们存储在向量数据库中。最直观的方法是为每种类型的向量创建一个独立的向量存储。
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings
def create_vector_stores(titles, summaries, chunks, embedding_model, persist_directory="./chroma_db"):
"""
为标题、摘要和文本块创建独立的Chroma向量存储。
"""
# 标题向量存储
title_vectorstore = Chroma.from_documents(
documents=titles,
embedding=embedding_model,
collection_name="titles",
persist_directory=persist_directory + "/titles"
)
title_vectorstore.persist()
print("Title vector store created and persisted.")
# 摘要向量存储
summary_vectorstore = Chroma.from_documents(
documents=summaries,
embedding=embedding_model,
collection_name="summaries",
persist_directory=persist_directory + "/summaries"
)
summary_vectorstore.persist()
print("Summary vector store created and persisted.")
# 文本块向量存储
chunk_vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embedding_model,
collection_name="chunks",
persist_directory=persist_directory + "/chunks"
)
chunk_vectorstore.persist()
print("Chunk vector store created and persisted.")
return title_vectorstore, summary_vectorstore, chunk_vectorstore
# 示例：
# embedding_model = OpenAIEmbeddings(model="text-embedding-ada-002")
# title_vs, summary_vs, chunk_vs = create_vector_stores(titles, summaries, chunks, embedding_model)关键概念：ParentDocumentRetriever
LangChain 提供了一个非常方便的抽象 ParentDocumentRetriever，它正是为这种多粒度检索设计的。它允许你存储两种类型的文档：
- 子文档 (Child Documents)：这些是用于检索的较小粒度文档（例如，摘要、标题、小文本块）。
- 父文档 (Parent Documents)：这些是当子文档被检索到时，你希望返回给 LLM 的较大粒度文档（例如，原始的完整文本块，或者更长的语义块）。
ParentDocumentRetriever 的核心在于它维护了一个映射，从子文档到其对应的父文档。当我们检索到子文档时，它会自动返回对应的父文档。
在这个高级实践中，我们的摘要和标题可以被视为“子文档”，用于检索。而当它们被检索到时，我们最终希望获取的是对应的原始文本块。因此，我们可以将原始文本块存储在一个 BaseStore 中，并让 ParentDocumentRetriever 协调摘要/标题向量存储和原始文本块存储。
让我们重新组织一下索引阶段，使用 ParentDocumentRetriever 的思想。
使用 ParentDocumentRetriever 进行索引：
为了更好地利用 ParentDocumentRetriever，我们需要将原始文本块存储在一个文档存储 (Document Store) 中，而不是直接在向量存储中。向量存储只存储摘要和标题的向量，但这些向量的元数据会包含一个指向其父文档（即原始文本块）的 ID。
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore # 也可以是DuckDBStore, RedisStore等
def create_parent_document_retriever(raw_documents, llm, embedding_model, persist_directory="./chroma_db_parent"):
"""
使用ParentDocumentRetriever构建多向量存储。
我们用摘要作为查询目标，检索后返回原始文本块。
"""
# 1. 定义子文档（用于检索）和父文档（用于返回给LLM）的分割器
# 摘要作为子文档，不需分割
# 原始文本块作为父文档，我们希望检索到摘要后，返回其对应的原始文本块
# 对于原始文本块，我们可能也需要一个分割器来定义“父文档”的粒度。
# 这里我们简化一下：让摘要作为查询目标，然后返回原始文档的整个内容或者更大的语义块。
# 更精细的做法是：摘要作为查询，返回对应的原始文本的“段落”或“章节”。
# 定义用于存储父文档（即原始的、更大的文本块）的文本分割器
# 这里我们假设“父文档”就是原始文档的语义块，可能比用于检索的摘要要长
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200) # 父文档块可以更大
# 定义用于存储子文档（这里我们用摘要）的文本分割器
# 摘要不需要再分割，所以我们直接将摘要作为一个整体作为子文档
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=0) # 或者不用分割器，直接用整个摘要
# 2. 创建文档存储（存储原始文本块）
doc_store = InMemoryStore() # 内存存储，生产环境应使用持久化存储
# 3. 创建向量存储（存储摘要的向量）
# 注意：这里的向量存储只存储“子文档”（即摘要）的向量
vectorstore = Chroma(
collection_name="parent_document_summaries",
embedding_function=embedding_model,
persist_directory=persist_directory + "/parent_summaries_vs"
)
# 4. 创建 ParentDocumentRetriever
retriever = ParentDocumentRetriever(
vectorstore=vectorstore,
docstore=doc_store,
child_splitter=child_splitter, # 用于分割子文档，这里可以不传，如果子文档本身就是完整摘要
parent_splitter=parent_splitter, # 用于分割父文档，并存储到doc_store
search_kwargs={"k": 3} # 检索摘要时返回的top-k
)
# 5. 准备文档并添加到 retriever
# 对于每个原始文档，我们生成摘要，然后将摘要作为子文档，原始文本（或其语义块）作为父文档。
# 这部分需要更精细地处理，因为ParentDocumentRetriever的add_documents方法期望的是父文档。
# 它的内部逻辑会为父文档生成子文档并索引。
# 更直接的Multi-Vector Retriever实现方式（不完全依赖ParentDocumentRetriever的自动子文档生成）：
# 这种情况下，我们自己管理标题、摘要和原始文本块的生成和存储。
# 我们回到之前多向量存储的思路，因为ParentDocumentRetriever虽然方便，
# 但它的child_splitter通常用于将一个父文档分割成多个子文档，
# 而我们这里是“标题/摘要”作为子文档，对应“原始文本块”作为父文档，
# 这种关系需要更手动地管理。
# 所以，我们采用创建多个独立向量存储的方法，并手动管理其间的关联。
# 这也是许多高级RAG系统更灵活的做法。
# ParentDocumentRetriever更适合于“将大文档切成小块用于检索，但返回大块作为上下文”的场景。
# --------------------------------------------------------------------------
# 重新审视，我们如何用ParentDocumentRetriever来模拟我们的需求？
# 方案1: 将原始文本作为"Parent Document"，将摘要作为其对应的"Child Document"。
# retriever.add_documents() 方法接受的是“父文档”。我们需要确保父文档的元数据能关联到其摘要。
# LangChain的ParentDocumentRetriever实际上是为以下场景设计的：
# 1. 你有一个大文档 (parent document)。
# 2. 你想把这个大文档切割成许多小块 (child documents) 进行索引。
# 3. 当检索到小块时，返回整个大文档或大文档的语义块。
#
# 我们的需求是：
# 1. 查询标题向量 -> 获取相关文档ID -> 获取其原始文本块。
# 2. 查询摘要向量 -> 获取相关文档ID -> 获取其原始文本块。
#
# 这意味着我们的“子文档”是标题或摘要，而“父文档”是原始文本块。
# LangChain的ParentDocumentRetriever可以这样用：
# `vectorstore` 存储的是子文档（摘要）的向量。
# `docstore` 存储的是父文档（原始文本块）的文本内容。
# 当我们 `add_documents` 时，传入的是父文档，但 `ParentDocumentRetriever` 会根据 `child_splitter`
# 和 `parent_splitter` 来生成子文档和父文档，并分别存储。
#
# 为了满足我们的特定需求，我们可以利用 `ParentDocumentRetriever` 的底层机制，
# 但需要手动控制子文档（摘要/标题）的生成和与父文档（原始文本块）的关联。
# --------------------------------------------------------------------------
# 最终决定：采用手动管理多个向量存储的方法，因为它更灵活，更能直接表达“标题/摘要是独立于原始文本块的表示”这一概念。
# ParentDocumentRetriever 实际上更侧重于 chunking strategy，而不是多种独立表示的索引。
# 之前的 `create_vector_stores` 函数是正确的方向。
return create_vector_stores(titles, summaries, chunks, embedding_model, persist_directory)
# 实际使用时，我们还是会调用 create_vector_stores 来创建三个独立的向量存储。
# 但为了保持代码的完整性和逻辑性，我们可以在索引阶段，将原始文本块也存储在一个In-Memory store中，
# 方便后续通过 document_id 快速查找。
# 重新组织索引和存储逻辑，以支持通过 document_id 查找原始文本块
def index_multi_vector_documents(titles, summaries, chunks, embedding_model, persist_directory="./chroma_db_multi_vector"):
"""
为标题、摘要和文本块创建独立的Chroma向量存储，并存储原始文本块以备检索。
"""
# 初始化一个存储所有原始文本块的内存存储，以便通过 document_id 查找
# 这里的 chunk_store 存储的是完整的 Document 对象，包括其 page_content 和 metadata
chunk_store = InMemoryStore()
# 将所有文本块添加到 chunk_store
# 注意：这里的 key 应该是 document_id，以便我们能通过 document_id 检索到该文档的所有块
# 但一个 document_id 对应多个 chunk，所以我们需要一个列表
# 更好的做法是，chunk_store 存储完整的 chunk_id -> chunk_content 映射
# 这样，当检索到某个 document_id 时，我们可以通过这个 ID 过滤出所有相关的 chunk_id，然后从 chunk_store 中获取。
# 另一种简单方法是，直接让 chunk_vectorstore 存储完整的块，并利用其检索能力。
# 我们将采用之前 `create_vector_stores` 的思路，并确保每个 Document 都有 `document_id`。
# 当从 title_vs 或 summary_vs 检索到 document_id 后，我们再用这个 document_id 去 chunk_vs 中过滤。
# 标题向量存储
title_vectorstore = Chroma.from_documents(
documents=titles,
embedding=embedding_model,
collection_name="titles",
persist_directory=persist_directory + "/titles"
)
title_vectorstore.persist()
print("Title vector store created and persisted.")
# 摘要向量存储
summary_vectorstore = Chroma.from_documents(
documents=summaries,
embedding=embedding_model,
collection_name="summaries",
persist_directory=persist_directory + "/summaries"
)
summary_vectorstore.persist()
print("Summary vector store created and persisted.")
# 文本块向量存储
chunk_vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embedding_model,
collection_name="chunks",
persist_directory=persist_directory + "/chunks"
)
chunk_vectorstore.persist()
print("Chunk vector store created and persisted.")
return title_vectorstore, summary_vectorstore, chunk_vectorstore
# 示例：
# embedding_model = OpenAIEmbeddings(model="text-embedding-ada-002")
# title_vs, summary_vs, chunk_vs = index_multi_vector_documents(titles, summaries, chunks, embedding_model)3. 检索阶段：协同查询与上下文组装
检索阶段是 Multi-Vector Retriever 的核心逻辑所在。我们将结合 LLM 和三个向量存储，实现多阶段的协同检索。
from typing import List, Dict, Any, Set
from langchain_core.documents import Document
from langchain.retrievers import MultiQueryRetriever # 也可以用，但我们这里实现更定制化流程
class MultiVectorRetrievalSystem:
def __init__(self, llm, embedding_model, title_vectorstore, summary_vectorstore, chunk_vectorstore):
self.llm = llm
self.embedding_model = embedding_model
self.title_vectorstore = title_vectorstore
self.summary_vectorstore = summary_vectorstore
self.chunk_vectorstore = chunk_vectorstore
def _get_unique_document_ids(self, documents: List[Document]) -> Set[str]:
"""从检索到的文档中提取唯一的 document_id。"""
return {doc.metadata["document_id"] for doc in documents if "document_id" in doc.metadata}
def retrieve(self, query: str, k_titles=5, k_summaries=3, k_chunks_per_doc=2) -> List[Document]:
"""
执行多阶段检索。
1. 基于标题向量进行初步筛选。
2. 基于摘要向量进行二次筛选和上下文理解。
3. 从原始文本块中提取最终细节。
"""
print(f"n--- Multi-Vector Retrieval for Query: '{query}' ---")
# 阶段1: 标题向量检索 - 粗粒度过滤
print(f"Phase 1: Retrieving top {k_titles} titles...")
relevant_titles = self.title_vectorstore.similarity_search(query, k=k_titles)
if not relevant_titles:
print("No relevant titles found.")
return []
candidate_doc_ids_from_titles = self._get_unique_document_ids(relevant_titles)
print(f"Candidate document IDs from titles: {candidate_doc_ids_from_titles}")
# 阶段2: 摘要向量检索 - 精炼上下文
print(f"Phase 2: Retrieving top {k_summaries} summaries from candidate documents...")
relevant_summaries = []
# 优化：只对标题阶段筛选出的文档ID对应的摘要进行检索
# 由于Chroma不支持直接按document_id过滤后进行向量搜索，我们需要先检索所有摘要，再根据document_id过滤。
# 或者更高效的做法是：如果向量数据库支持，直接在检索时加入metadata过滤。
# 这里为了演示，我们先进行 broad search，再后处理。
# 更好的方法是：如果Chroma支持，可以这样：
# summary_retriever = self.summary_vectorstore.as_retriever(
# search_kwargs={"k": k_summaries, "filter": {"document_id": {"$in": list(candidate_doc_ids_from_titles)}}}
# )
# relevant_summaries = summary_retriever.invoke(query)
#
# 鉴于Chroma的filter通常用于精确匹配，而不是在similarity_search后筛选，
# 我们模拟一个两步走：先 broad search summaries，再按 document_id 过滤。
all_potential_summaries = self.summary_vectorstore.similarity_search(query, k=k_summaries * 2) # 检索多一点，以防过滤后数量不足
for summary_doc in all_potential_summaries:
if summary_doc.metadata.get("document_id") in candidate_doc_ids_from_titles:
relevant_summaries.append(summary_doc)
if len(relevant_summaries) >= k_summaries: # 确保只取k_summaries个
break
if not relevant_summaries:
print("No relevant summaries found for candidate documents.")
return []
final_relevant_doc_ids = self._get_unique_document_ids(relevant_summaries)
print(f"Final relevant document IDs after summary filtering: {final_relevant_doc_ids}")
# 阶段3: 原始文本块检索 - 获取具体细节
print(f"Phase 3: Retrieving top {k_chunks_per_doc} chunks for each final relevant document...")
final_context_chunks = []
for doc_id in final_relevant_doc_ids:
# 检索特定文档ID的所有文本块，并基于查询进行相似度排序
# 同样，Chroma的filter是用于过滤集合，而不是相似性搜索的结果。
# 我们需要先检索，再过滤。
# 更好的方式是直接利用 vectorstore 的 retriever 接口的 filter 参数。
# 假设Chroma的filter可以在similarity_search中使用
# Example: filtered_chunks = self.chunk_vectorstore.similarity_search(
# query, k=k_chunks_per_doc, filter={"document_id": doc_id}
# )
# 如果Chroma不支持直接在similarity_search中进行复杂过滤，
# 我们需要先检索出所有与该doc_id相关的chunks，然后对它们进行二次排序。
# 方法一：先用 filter 找到所有属于该 document_id 的块，然后对这些块进行相似性搜索 (或直接取top-k)
# Chroma的get方法可以按ID获取，但我们不知道所有chunk ID
# 所以，更通用的做法是：检索所有相关块，然后通过元数据过滤和重排序。
# 让我们使用 `as_retriever` 方法，它通常支持 `filter` 参数。
chunk_retriever_for_doc = self.chunk_vectorstore.as_retriever(
search_kwargs={
"k": k_chunks_per_doc,
"filter": {"document_id": doc_id}
}
)
chunks_for_this_doc = chunk_retriever_for_doc.invoke(query)
final_context_chunks.extend(chunks_for_this_doc)
print(f" - Retrieved {len(chunks_for_this_doc)} chunks for document ID: {doc_id}")
if not final_context_chunks:
print("No relevant chunks found for the final relevant documents.")
return []
# 阶段4 (可选): 重排序
# 对所有检索到的原始文本块进行最终的重排序，以确保最相关的块排在前面。
# 可以使用交叉编码器或LLM进行重排序。这里我们简化，假设相似度搜索已经足够。
# 如果需要，可以引入 Cohere Rerank 或 BGE Rerank 等模型。
print(f"Total {len(final_context_chunks)} chunks selected as final context.")
return final_context_chunks
def generate_response(self, query: str) -> str:
"""
使用检索到的上下文生成LLM响应。
"""
retrieved_docs = self.retrieve(query)
if not retrieved_docs:
return "对不起，我未能找到相关信息来回答您的问题。"
context = "nn".join([doc.page_content for doc in retrieved_docs])
prompt_template = ChatPromptTemplate.from_messages([
("system", "你是一个知识渊博的助手。请根据提供的上下文信息，简洁、准确地回答用户的问题。如果上下文没有提供足够的信息，请说明你无法回答。"),
("human", "上下文信息:n{context}nn用户问题: {query}")
])
chain = prompt_template | self.llm
response = chain.invoke({"context": context, "query": query}).content
return response
# 示例使用流程：
# 1. 准备数据文件（例如，data/doc1.txt, data/doc2.txt）
# 2. 加载原始文档
# raw_docs = load_documents(["data/doc1.txt", "data/doc2.txt"])
# 3. 初始化LLM和嵌入模型
# llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
# embedding_model = OpenAIEmbeddings(model="text-embedding-ada-002")
# 4. 预处理生成标题、摘要、文本块
# titles, summaries, chunks = prepare_multi_vector_documents(raw_docs, llm)
# 5. 索引到向量存储
# title_vs, summary_vs, chunk_vs = index_multi_vector_documents(titles, summaries, chunks, embedding_model)
# 6. 初始化检索系统
# retriever_system = MultiVectorRetrievalSystem(llm, embedding_model, title_vs, summary_vs, chunk_vs)
# 7. 执行查询
# query = "请解释一下LangChain中Agent的工作原理。"
# response = retriever_system.generate_response(query)
# print("n--- LLM Response ---")
# print(response)代码说明和进一步考虑：
- Chroma 的
filter参数：在similarity_search中直接使用filter参数可以显著提高效率。例如self.chunk_vectorstore.similarity_search(query, k=k_chunks_per_doc, filter={"document_id": doc_id})。这比先检索所有再在内存中过滤要高效。 - 重排序 (Re-ranking)：在
retrieve方法的最后，对final_context_chunks进行重排序是非常有益的。使用交叉编码器（如sentence-transformers/cross-encoder-ms-marco-MiniLM-L-6-v2）可以根据查询和每个块的语义相关性进行更精确的排序。这能确保送入 LLM 的上下文是最相关的，即使向量相似度检索的结果略有偏差。# 简单的重排序示例 (伪代码) # from sentence_transformers import CrossEncoder # reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2') # pairs = [(query, chunk.page_content) for chunk in final_context_chunks] # scores = reranker.predict(pairs) # # # 将分数与块关联并排序 # scored_chunks = sorted(zip(scores, final_context_chunks), key=lambda x: x[0], reverse=True) # final_context_chunks = [chunk for score, chunk in scored_chunks[:top_n_after_rerank]] - 缓存：摘要生成和嵌入是耗时操作。在生产环境中，应考虑对这些结果进行缓存。
- 错误处理：上述代码为简化起见，没有包含全面的错误处理。在实际应用中，需要增加对 LLM 调用失败、向量存储连接问题等的处理。
五、高级考量与最佳实践
1. 选择合适的LLM进行摘要生成
- 模型大小与成本：更小的模型（如
gpt-3.5-turbo）可能成本更低、速度更快，但摘要质量可能不如大型模型（如gpt-4）。根据对摘要质量的要求和预算进行权衡。 - 抽象性与忠实性：确保 LLM 生成的摘要既能捕捉核心思想（抽象性），又能忠实于原文内容（忠实性），避免“幻觉”。
- 提示工程：精心设计的提示词对于生成高质量摘要至关重要。可以尝试 Few-shot Learning 或 Chain-of-Thought Prompting 来提高摘要质量。
2. 精心设计分块策略
- 语义分块：传统的固定大小分块可能破坏语义边界。尝试使用更智能的分块策略，例如基于标题、段落、章节进行分割，或者使用 LangChain 的
SemanticChunker(需要cohere或openaiembedding)。 - 重叠度：适当的
chunk_overlap有助于保留上下文，避免信息在块边界处丢失。 - 父子关系：如果使用
ParentDocumentRetriever，要明确父文档和子文档的粒度。
3. 稳健的元数据设计
document_id：每个逻辑文档必须有一个唯一的document_id，这是将不同类型的向量关联起来的关键。type字段：明确标识每个向量代表的是标题、摘要还是原始文本块，方便检索时进行过滤和决策。source字段：保留原始文档的来源信息，便于追溯。chunk_id：对于原始文本块，也需要唯一的chunk_id，如果需要独立检索或引用某个具体块。
4. 重排序的重要性
- 即使是最好的向量检索，也可能返回一些相关性不高的文档。重排序阶段使用更复杂的模型（如交叉编码器）对检索到的文档进行二次评估和排序，能够显著提升送入 LLM 上下文的质量，从而提高最终答案的准确性。
5. 扩展性与性能优化
- 多向量存储：随着文档数量的增长，需要考虑向量数据库的扩展性。例如，使用分布式向量数据库。
- 批处理：在生成摘要和嵌入时，利用批处理（batch processing）可以显著提高效率。
- 索引更新策略：当原始文档更新时，需要有策略地重新生成摘要、标题和嵌入，并更新向量存储。
6. 评估与迭代
- 离线评估：使用 RAGAS、Retrieval-Augmented Generation Benchmark (RGB) 等工具评估检索和生成质量。关注召回率、准确率、上下文相关性、忠实度等指标。
- 在线 A/B 测试：部署到生产环境后，通过 A/B 测试比较不同检索策略的实际用户体验。
7. 成本考量
- LLM 费用：摘要生成和潜在的重排序可能涉及多次 LLM 调用，会增加成本。
- 存储费用：存储多个向量（标题、摘要、文本块）会占用更多向量数据库空间。
- 计算费用：生成这些向量需要计算资源。
六、挑战与未来方向
Multi-Vector Retriever 虽然强大，但并非没有挑战，同时它也为未来的创新提供了广阔空间。
挑战：
- 复杂性管理：系统包含更多的组件（多个向量存储、LLM 调用、复杂的检索逻辑），增加了开发和维护的复杂性。
- 延迟优化：多阶段检索意味着多次向量数据库查询和可能的 LLM 调用，可能增加检索延迟。需要精心设计并行化和缓存策略。
- 数据同步：当原始文档更新时，如何高效、准确地更新所有相关的标题、摘要和文本块的向量，并保持数据一致性，是一个实际的工程挑战。
- 抽象与细节的平衡：如何确定最佳的标题长度、摘要粒度、文本块大小，以及如何在不同查询类型下动态选择最佳的检索策略，仍是需要深入研究的领域。
未来方向：
- 自适应检索：根据用户查询的类型、意图或历史行为，动态地调整检索策略。例如，对于事实性查询，可能直接跳到原始文本块；对于探索性查询，可能优先使用摘要。
- 多模态融合：将 Multi-Vector Retriever 的思想扩展到多模态数据。例如，为包含图片、表格、代码的文档生成图片描述向量、表格摘要向量、代码功能描述向量，并进行多模态检索。
- 更智能的摘要与分块：利用更先进的 LLM 和技术（如 LLM-based Agents）来生成更具洞察力的摘要，或者实现真正的语义分块，确保每个块都包含一个完整的语义单元。
- 图谱与向量的结合：将 Multi-Vector Retriever 与知识图谱相结合，利用图谱的结构化知识来增强检索的准确性和可解释性。
- 实时更新：研究更高效的增量索引和实时更新机制，以适应不断变化的数据源。
结语
Multi-Vector Retriever 是一种强大的范式，它通过为文档创建多层次的语义表示，极大地增强了 RAG 系统的检索能力。通过巧妙地利用文档的标题、摘要和原始文本块，我们能够构建一个更加智能、鲁棒且能够提供更精确上下文的检索系统。这不仅提升了 LLM 的问答质量，也为用户带来了更优质的信息获取体验。在不断演进的 AI 领域，对这些高级检索技术的掌握和创新，将是我们构建下一代智能应用的关键。
感谢大家的聆听！