合并FAISS索引后如何确保所有文档参与RAG摘要生成

问题背景 #

我正在构建基于FAISS向量库的LangChain RAG流水线，合并多个代表单文档的FAISS索引后，通过OpenAI LLM生成摘要或回答。核心代码如下：

# Merge multiple FAISS vectorstores
combined_vectorstore = vectorstores[0]
for vs in vectorstores[1:]:
    combined_vectorstore.merge_from(vs)

# Save and reload the merged index
combined_vectorstore.save_local(merged_path)

vectorstore = FAISS.load_local(
    merged_path, embeddings, allow_dangerous_deserialization=True
)

# Set up retriever with top-k similarity
retriever = vectorstore.as_retriever(
    search_type="similarity", search_kwargs={"k": 5}
)

遇到的问题：
现有两个文档：

• A1：大文档（30页，约200个chunk）
• A2：小文档（1页，约2个chunk）

当输入查询"Generate a summary of all documents"时，检索器返回的Top5相似chunk几乎全来自A1，导致LLM生成的摘要遗漏A2。但针对A2的定向查询能正常返回结果，说明A2已被正确索引。

需求：确保每个文档（尤其是小文档）在合并索引后，至少贡献一个chunk到检索上下文，让摘要覆盖所有文档。

解决方案 #

方法1：按文档分组强制采样 #

核心思路是给每个chunk添加所属文档标识，检索时先按文档分组提取每个文档的最相关chunk，剩余配额再用常规相似性检索填充，确保每个文档至少有一个chunk进入上下文。

步骤1：拆分文档时添加标识

在文档拆分阶段，为每个chunk的元数据添加文档ID：

from langchain.text_splitter import RecursiveCharacterTextSplitter

def split_doc_with_id(doc_content, doc_id):
    splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
    chunks = splitter.split_text(doc_content)
    return [{"page_content": chunk, "metadata": {"doc_id": doc_id}} for chunk in chunks]

# 对A1、A2分别处理
a1_chunks = split_doc_with_id(a1_full_text, "doc_a1")
a2_chunks = split_doc_with_id(a2_full_text, "doc_a2")

步骤2：自定义检索逻辑

重写检索函数，实现按文档强制采样：

def retrieve_with_coverage(query, vectorstore, total_k=5):
    # 获取所有唯一文档ID
    all_metadata = vectorstore.get()["metadatas"]
    unique_doc_ids = list(set(m["doc_id"] for m in all_metadata))
    
    # 每个文档取1个最相关chunk
    per_doc_chunks = []
    for doc_id in unique_doc_ids:
        docs = vectorstore.similarity_search(
            query,
            k=1,
            filter={"doc_id": doc_id}
        )
        if docs:
            per_doc_chunks.extend(docs)
    
    # 补充剩余配额的相似性结果
    remaining_k = max(total_k - len(per_doc_chunks), 0)
    if remaining_k > 0:
        extra_docs = vectorstore.similarity_search(query, k=remaining_k)
        per_doc_chunks.extend(extra_docs)
    
    # 去重并返回前total_k个结果
    unique_chunks = []
    seen_hashes = set()
    for doc in per_doc_chunks:
        chunk_hash = hash(doc.page_content)
        if chunk_hash not in seen_hashes:
            seen_hashes.add(chunk_hash)
            unique_chunks.append(doc)
            if len(unique_chunks) >= total_k:
                break
    return unique_chunks

替换原检索器使用

# 在RAG流水线中替换为自定义检索函数
def custom_retriever(query):
    return retrieve_with_coverage(query, vectorstore, total_k=5)

方法2：保留原始独立索引，分阶段检索 #

不合并索引，直接针对每个原始单文档索引单独检索，再合并结果，从根源保证每个文档都被覆盖。

# vectorstores为保存原始单文档FAISS索引的列表
def retrieve_all_docs(query, vectorstores, k_per_doc=1, total_k=5):
    all_relevant = []
    # 每个文档索引取指定数量的chunk
    for vs in vectorstores:
        docs = vs.similarity_search(query, k=k_per_doc)
        all_relevant.extend(docs)
    
    # 结果不足时补充全局相似性检索（可选）
    if len(all_relevant) < total_k:
        remaining = total_k - len(all_relevant)
        extra_docs = vectorstore.similarity_search(query, k=remaining)
        all_relevant.extend(extra_docs)
    
    # 去重后返回
    unique_docs = []
    seen = set()
    for doc in all_relevant:
        h = hash(doc.page_content)
        if h not in seen:
            seen.add(h)
            unique_docs.append(doc)
            if len(unique_docs) >= total_k:
                break
    return unique_docs

方法3：改用MMR检索策略提升多样性 #

使用LangChain的mmr（最大边际相关性）检索，在相关性和内容多样性之间平衡，降低大文档chunk的垄断概率（无法100%保证覆盖，但能有效改善）：

retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 5, "fetch_k": 20}  # 扩大候选集提升多样性
)

内容的提问来源于stack exchange，提问作者Musab