基本检索器

🧩 LangChain 中 Retriever 的接口是什么？和 VectorStore 的关系？

Retriever 接口 在 LangChain 中，Retriever 是一个抽象基类（BaseRetriever），定义了统一的文档检索规范。其核心接口非常简单：

• get_relevant_documents(query: str) -> List[Document]：同步方法，接收一个查询字符串，返回与之相关的文档列表。

• aget_relevant_documents(query: str) -> List[Document]：异步版本，用于并发场景。

任何实现该接口的对象都可以作为一个 Retriever 被其他链或工具使用。这种设计符合依赖倒置原则，使得检索逻辑可以灵活替换，而调用方无需关心底层的具体实现（是向量搜索、BM25、还是混合搜索）。

Retriever 与 VectorStore 的关系 VectorStore 是 LangChain 中用于存储和检索向量（Embeddings）的组件，它本身并不直接暴露 Retriever 接口。但 VectorStore 提供了 as_retriever() 方法，该方法会返回一个 VectorStoreRetriever 实例。VectorStoreRetriever 内部封装了 VectorStore，并实现了 BaseRetriever 接口。

所以关系是：VectorStore 是存储引擎，Retriever 是检索接口。VectorStore 通过 as_retriever() 将自己包装成一个 Retriever 供外部使用。 这种分层使得我们可以轻松地给向量检索器增加过滤、配置搜索参数等。

VectorStore (例如 Chroma, FAISS)
    └── as_retriever() → VectorStoreRetriever (实现了 BaseRetriever)
                              └── get_relevant_documents(query)

🛠️ 写一个 VectorStoreRetriever 的创建代码，设置 search_kwargs（如 k 值、score_threshold）。

下面以 Chroma 向量数据库为例，创建 VectorStoreRetriever 并通过 search_kwargs 设置返回文档数量 k 和相似度阈值 score_threshold。

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 假设已经构建好 embeddings 和 Chroma 实例
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(
    collection_name="my_knowledge_base",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

# 创建 Retriever，并设置搜索参数
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",  # 搜索类型：带分数阈值
    search_kwargs={
        "k": 5,                     # 返回最相似的5个文档
        "score_threshold": 0.7      # 只返回相似度大于0.7的文档
    }
)

说明：

• search_type 除了 "similarity_score_threshold"，还有 "similarity"（默认，返回 Top-K）和 "mmr"（最大边际相关性，用于增加多样性）。

• search_kwargs 中的参数会根据不同的 VectorStore 后端略有差异，但 k 是通用的。部分后端（如 FAISS）还支持 fetch_k（MMR 中初始获取的数量）。

• 如果不设置 score_threshold，则返回最相似的 k 个文档，不进行分数过滤。

---

📊 如何通过配置让 Retriever 返回相似度分数？

默认情况下，VectorStoreRetriever 的 get_relevant_documents 只返回 Document 对象，这些对象的 metadata 中默认不包含相似度分数。要获取分数，有以下几种方法：

方法一：使用 similarity_search_with_score（不通过 Retriever，直接调用 VectorStore）

docs_with_scores = vectorstore.similarity_search_with_score(query, k=5)
for doc, score in docs_with_scores:
    print(f"Score: {score:.4f}, Content: {doc.page_content[:100]}")

方法二：通过自定义 Retriever 在 metadata 中注入分数 我们可以继承 VectorStoreRetriever 并重写 _get_relevant_documents 方法，将分数存入 metadata。

from langchain.vectorstores.base import VectorStoreRetriever
from typing import List
from langchain.schema import Document

class ScoreRetriever(VectorStoreRetriever):
    def _get_relevant_documents(self, query: str, *, run_manager=None) -> List[Document]:
        docs_with_scores = self.vectorstore.similarity_search_with_score(
            query, k=self.search_kwargs.get("k", 4)
        )
        for doc, score in docs_with_scores:
            doc.metadata["score"] = score
        # 也可以在这里根据 score_threshold 过滤
        threshold = self.search_kwargs.get("score_threshold")
        if threshold is not None:
            return [doc for doc, score in docs_with_scores if score >= threshold]
        return [doc for doc, _ in docs_with_scores]

方法三：结合 similarity_score_threshold 搜索类型 如果使用 search_type="similarity_score_threshold"，LangChain 内部会调用 similarity_search_with_relevance_scores，该方法返回的文档的 metadata 中已自动包含 score 字段（部分向量库实现支持）。但并非所有 VectorStore 都实现了该方法。Chroma 和 FAISS 支持。

retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"k": 5, "score_threshold": 0.7}
)
docs = retriever.get_relevant_documents("什么是量子计算？")
# 文档的 metadata 中可能已包含 'score'（取决于VectorStore实现）

结论：要求稳定获取分数，推荐自定义 Retriever，或直接使用 VectorStore 的 similarity_search_with_score。

---

🧠 什么是 MultiQueryRetriever？它内部如何生成多个查询？主要解决什么问题？

MultiQueryRetriever 是 LangChain 提供的一种高级检索器，它通过自动生成多个不同视角的查询，来克服单一查询可能因措辞不佳而导致的检索不全面问题。它本身不替代底层检索器，而是在其上增加了一个“查询增强”层。

内部工作机制：

1. 接收用户原始查询。

2. 使用 LLM（默认用 ChatOpenAI 等）根据一个预设的 Prompt，生成若干个（默认3个）语义相似但表述不同的查询。

3. 将这多个查询分别发送给底层 Retriever，各自获取相关文档。

4. 将所有查询的结果进行合并去重，得到最终的文档集合。

主要解决的问题：

• 查询多样性不足：向量检索对查询的措辞敏感，单个查询可能遗漏重要信息。

• 语义覆盖不足：用户问题可能包含模糊或抽象的表述，通过改写为多个具体查询，可以覆盖更多相关文档。

• 检索召回率提升：尤其适用于长文档或专业领域，不同的提问角度能够激活不同的语义特征。

简单比喻：就像你向图书馆员问同一个问题，但他会从不同角度（书名、关键词、作者）帮你搜索，然后把所有结果汇总给你。

---

🔧 使用 MultiQueryRetriever 时，如何保证多个查询的多样性？内部 prompt 可以定制吗？

保证多样性的关键在于 LLM 生成的查询是否足够发散。 MultiQueryRetriever 通过以下几点来保证：

• Prompt 设计：默认的 Prompt 会明确要求 LLM 生成多个“不同角度”或“不同表述”的查询。例如，它会指示模型“生成3个不同版本的查询，以便从向量数据库中检索到更全面的文档”。

• LLM 的温度（Temperature）：在调用 LLM 生成查询时，可以设置较高的温度（例如 0.7），以增加随机性和多样性。

• 定制 Prompt：这是最直接的方法。你可以完全自定义 Prompt，加入领域特定的指令、示例（few-shot），或者要求生成包含同义词、相关概念的查询。

自定义 Prompt 示例：

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain.prompts import PromptTemplate

# 自定义的查询生成Prompt
custom_prompt = PromptTemplate.from_template(
    """你是一个专业的信息检索专家。请根据用户的问题，生成{num_queries}个不同角度、不同措辞的搜索查询，以便全面覆盖相关文档。
    要求：
    - 每个查询应与原始问题语义相关，但使用不同的表达方式、同义词或下位概念。
    - 输出为JSON列表，如：["查询1", "查询2", ...]

    用户问题：{question}

    生成的查询列表（JSON格式）："""
)

retriever_from_llm = MultiQueryRetriever.from_llm(
    retriever=base_retriever,
    llm=ChatOpenAI(temperature=0.7),
    prompt=custom_prompt,   # 传入自定义prompt
    num_queries=5           # 生成5个查询
)

通过这种方式，你可以精确控制查询的多样性和格式。

---

🔗 MultiQueryRetriever 生成多个查询后，如何合并各查询的检索结果？有哪些合并策略？

MultiQueryRetriever 在得到多个查询各自的文档列表后，需要进行合并。LangChain 的默认实现采用的是一种简单的去重合并策略。

默认合并流程：

1. 将各个查询返回的文档列表展平（flatten）为一个大的文档列表。

2. 基于文档的唯一标识（通常是 page_content 的哈希值，或文档对象的ID）进行去重，保留第一次出现的文档（顺序影响最终排序）。

3. 返回去重后的文档列表。

默认策略的局限性：

• 不考虑不同查询的优先级，所有查询的文档一视同仁。

• 去重后不进行重新排序，可能导致最相关的文档淹没在列表中。

自定义合并策略： 你可以通过子类化 MultiQueryRetriever 并重写 _get_relevant_documents 或 _combine_documents 方法来实现更高级的合并：

• 加权合并：为每个查询返回的文档赋予不同权重（例如，原始查询的结果权重更高），然后按累加权重排序。

• RRF（Reciprocal Rank Fusion）：一种常用的多路召回融合算法，对每个文档在不同查询中的排名取倒数求和，再按得分排序。

• 基于相似度分数的重排序：如果底层检索器返回了相似度分数，可以按分数降序排列。

• 使用 Cross-Encoder 重排序：将合并后的文档与原始问题一起输入一个 Cross-Encoder 模型，重新打分并排序。

例如，实现一个简单的重排序合并：

from langchain.retrievers.multi_query import MultiQueryRetriever

class RerankMultiQueryRetriever(MultiQueryRetriever):
    def _combine_documents(self, doc_lists: list):
        # 展平并去重
        unique_docs = {}
        for doc in (d for sublist in doc_lists for d in sublist):
            unique_docs[doc.page_content] = doc
        docs = list(unique_docs.values())
        # 这里可以调用你的重排序模型对docs重新打分排序
        # 假设按某种逻辑排序
        return docs

在实际项目中，如果需要更精细的控制，常使用 EnsembleRetriever 或自定义管线。

---

🔒 在多用户场景下，MultiQueryRetriever 是否可能引入隐私问题？

是的，可能引入间接隐私风险。 核心问题不在于 MultiQueryRetriever 本身，而在于它利用 LLM 来改写查询。这个改写过程可能会泄露用户原始查询中的敏感信息，或者生成包含敏感推断的查询。

具体风险包括：

• 原始查询泄露给 LLM 供应商：如果使用外部 LLM（如 OpenAI）来生成多个查询，那么用户的原始查询就会被发送到云端 API。如果查询中包含个人身份信息（PII）、医疗健康信息、企业机密等，就违反了数据隐私合规（如 GDPR、HIPAA）。

• 查询扩展可能暴露隐含信息：LLM 生成的查询可能包含对用户偏好的推断。例如，用户查询“便宜的离婚律师”，LLM 可能生成“低收入法律援助”，这间接推断出用户的经济状况。

• 查询日志留存：多个改写后的查询可能比原始查询更具体，这些记录如果被保存或审计，可能增加用户画像的风险。

如何缓解：

• 使用本地 LLM：采用本地部署的开源模型（如 Llama 3）进行查询改写，数据不出域。

• 脱敏处理：在将查询发送给 LLM 前，用正则或 NER 模型脱敏，替换掉人名、电话号码等敏感实体，生成查询后再映射回去（但这可能影响查询质量）。

• 最小权限与审计：确保 LLM 供应商的 API 配置为不记录查询内容（例如 OpenAI API 的商业客户可申请零数据保留）。

• 提示词约束：在 Prompt 中明确要求 LLM 只生成基于公开知识库的通用查询，避免生成包含任何个人信息的查询。

总之，在涉及多用户和敏感数据时，要么使用本地模型，要么在进入 MultiQueryRetriever 前做好数据清洗。

---

🗜️ 解释 ContextualCompressionRetriever 的工作流程：它由哪两部分组成？

ContextualCompressionRetriever 是一种先检索后压缩的检索器增强，旨在从检索到的文档中提取出与查询真正相关的片段，丢弃冗余信息，从而让 LLM 能更聚焦、更高效地生成答案。

组成部分：

• Base Retriever（基础检索器）：负责根据查询返回一个初步的文档列表。可以是任何实现了 BaseRetriever 的对象，如 VectorStoreRetriever。

• Compressor（压缩器）：负责对基础检索器返回的文档进行“压缩”。压缩器可以是任何实现了 BaseDocumentCompressor 的对象。

工作流程：

1. 用户查询传入。

2. 基础检索器根据查询获取一组相关文档（可能包含很多无关段落）。

3. 压缩器逐文档处理，对于每个文档，根据查询提取出相关的句子或片段，或者对整个文档进行过滤（如果完全无关则丢弃）。

4. 压缩后的文档（可能变短，也可能数量减少）被返回给用户。

形象比喻：基础检索器像图书馆的管理员，帮你搬来一堆可能相关的书；压缩器则像研究助理，帮你从这些书中只复印下真正对你有用的段落。

这种设计尤其适合长文档检索，可以避免把整个长文档塞进 LLM 上下文，从而节省 token 并提高答案准确性。

---

🧐 ContextualCompressionRetriever 中的压缩器（Compressor）通常是什么？可以是 LLM 吗？

常见的压缩器类型：

• LLMChainExtractor：使用 LLM 从文档中抽取与查询相关的核心句子。它会调用 LLM，传入文档内容和查询，要求 LLM 输出提取后的内容。这是最强大但也最昂贵的压缩器。

• LLMChainFilter：使用 LLM 判断文档是否与查询相关。返回“YES”或“NO”，从而过滤掉无关文档（只做筛选，不提取内容）。成本比 Extractor 稍低。

• EmbeddingsFilter：基于嵌入向量计算文档与查询的相似度，保留相似度高于阈值的文档或片段。不需要调用 LLM，速度快，但可能不如 LLM 精准。

• 自定义 Compressor：你可以实现 BaseDocumentCompressor 接口，例如结合一个轻量级的自然语言推理（NLI）模型来判断相关性，或使用规则进行过滤。

可以是 LLM 吗？ 是的，LLMChainExtractor 和 LLMChainFilter 都是基于 LLM 的压缩器。它们通过构造 Prompt 让 LLM 完成压缩工作。因为 LLM 能理解语义，所以压缩质量往往很高，但缺点是会增加延迟和 API 费用。在实际项目中，常结合多种压缩器，例如先用 EmbeddingsFilter 快速粗筛，再用 LLMChainExtractor 精细提取，以达到成本和效果的平衡。

---

📝 写一个使用 LLMChainExtractor 作为压缩器的示例，它如何从文档中抽取关键内容？

下面代码演示了如何创建一个 ContextualCompressionRetriever，使用 LLMChainExtractor 从文档中抽取与查询直接相关的内容。

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.chat_models import ChatOpenAI

# 1. 假设我们已经有一个基础检索器 base_retriever（例如 VectorStoreRetriever）
# base_retriever = ...

# 2. 创建 LLM 实例（使用支持 function calling 的模型效果更好，但不是必须）
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 3. 创建压缩器（Extractor）
compressor = LLMChainExtractor.from_llm(llm)

# 4. 构建上下文压缩检索器
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever
)

# 5. 使用检索器
query = "量子计算对制药行业的影响是什么？"
compressed_docs = compression_retriever.get_relevant_documents(query)

# 打印结果
for i, doc in enumerate(compressed_docs):
    print(f"--- 压缩后的文档片段 {i+1} ---")
    print(doc.page_content)
    print()

内部工作原理： 当调用 get_relevant_documents(query) 时，ContextualCompressionRetriever 会：

1. 调用 base_retriever.get_relevant_documents(query) 获取初始文档列表。

2. 对每个文档，调用 LLMChainExtractor，其内部会构建类似以下的 Prompt：

你是一名信息提取专家。请根据用户的问题，从以下文档中提取出相关的关键信息。
如果文档中没有任何相关信息，请输出“无”。

用户问题：{query}

文档内容：{document}

提取的相关信息：

1. LLM 返回的内容（可能是原文的摘录，也可能是一些关键句子的重组）会替换原来的文档内容，形成压缩后的 Document 对象。

2. 过滤掉那些被 LLM 判定为“无”相关信息的文档。

3. 将压缩后的文档列表返回。

通过这种方式，原本可能长达几千字的文档被压缩成了几十字的核心信息，大大减少了后续 LLM 分析时的 token 消耗，同时提高了答案的准确率，因为无关噪音被移除了。

⚖️ LLMChainFilter 和 LLMChainExtractor 有何不同？分别适用于什么场景？

两者都是 LangChain 中基于 LLM 的文档压缩器，但它们在处理粒度、输出内容、计算成本和应用场景上有本质区别。

查看内嵌表格

🛠️ 场景选择指南：

• 使用 LLMChainFilter：当你的底层 LLM 窗口足够大，只是想快速丢弃无关文档以降低后续处理成本时。例如在问答题中，检索到 20 个文档，只需过滤掉完全不沾边的，保留 5 个可能相关的，然后直接丢给生成链。

• 使用 LLMChainExtractor：当你需要严格控制 token 消耗，或者文档本身很长（如技术手册、论文），你想从中提取出直接回答问题所需的那两三句话。这在长文档问答中非常有效。

💡 工程实践：常常组合使用。先用快速的 EmbeddingsFilter 或 LLMChainFilter 粗筛，再用 LLMChainExtractor 对留下的少数文档精细提取，兼顾速度与精度。

---

⏱️ 使用 ContextualCompressionRetriever 时，如果压缩步骤很耗时，你如何降低延迟？

压缩步骤耗时主要是因为对每个文档都要调用一次（或多次）LLM，这是典型的 IO 密集型等待。降低延迟的策略可以从算法、并发、缓存、模型选择四个维度展开。

1. 并发执行 默认情况下，LangChain 的压缩器是串行处理文档的。我们可以通过实现异步压缩（acompress_documents）并使用 asyncio.gather 同时处理所有文档，将总延迟从 N * 单文档耗时 降为 max(单文档耗时)。

# 自定义异步压缩器，或者使用支持异步的 LLM
async def acompress_documents(docs, query):
    tasks = [llm.acall(prompt) for doc in docs]
    results = await asyncio.gather(*tasks)
    # 后处理 results

分层过滤，减少需要 LLM 处理的文档数量 不直接对原始检索结果进行昂贵的 LLM 压缩。先使用速度极快的 EmbeddingsFilter（基于向量相似度）丢弃低分文档，只保留前 5-10 个送入 LLMChainExtractor。或者在检索阶段就设置严格的 score_threshold 和 k。

1. 模型降级与量化

2. 使用更小、更快的模型：对于压缩任务，gpt-3.5-turbo 通常足够，甚至可以用本地部署的 7B 模型（如 Llama-3-8B）经过量化后专门做压缩。

3. 批处理：如果使用 OpenAI API，可以将多个文档的压缩请求打包到一个批次中（如果自定义 Prompt 支持），减少网络往返。

4. 结果缓存

对于重复的 (query, doc) 组合，利用缓存（如 Redis）存储压缩结果。LangChain 的 LLM 缓存可以部分实现这一点，但更精确的是自己实现一个文档级别的缓存。

1. 优化 Prompt 压缩 Prompt 越简洁，LLM 生成越快。要求 LLM 直接输出“相关句子”，避免冗长的解释，并设置 max_tokens 限制。

2. 预计算与离线压缩

如果知识库是静态的，可以预先用 LLM 对每个文档进行摘要或关键句提取，生成“压缩版文档”，并存入向量库。检索时直接命中压缩版，无需实时调用 LLM。这是将延迟从在线转移到离线的根本方法。

---

🔍 在 RAG 中，为什么要先检索后压缩？压缩器会不会丢失重要信息？

为什么要先检索后压缩？

RAG 面临的核心矛盾是：我们希望 LLM 能基于尽可能多的相关上下文回答问题，但 LLM 的上下文窗口有限且 token 有成本。检索器的任务是从海量文档中高召回地捞出一批候选文档（可能包含噪声），压缩器则负责精炼这些文档，只提取与当前问题相关的核心信息，从而在有限的窗口内注入更高的信息密度。

如果反过来（先压缩后检索），就需要对全库文档进行压缩，这在计算上不可能。因此，“检索—压缩—生成”是一种高效的流水线：用廉价方法（向量检索）快速缩小范围，再用昂贵方法（LLM）精细过滤，实现成本与效果的帕累托最优。

压缩器会不会丢失重要信息？

会，这是压缩的本质代价。 任何有损压缩都存在遗漏关键事实的风险，尤其是当 LLM 误解了查询意图或文档内容时。为了尽量降低风险，可以采取以下措施：

1. 保留高置信度文档的原貌：对得分极高（例如相似度 > 0.9）的文档，可以选择不压缩，直接保留原文，避免误伤。

2. 多级压缩：先用 LLMChainFilter 确保文档与问题相关，再用 LLMChainExtractor 抽取。Filter 的误判率更低，能减少后续 Extractor 的压力。

3. 引入验证机制：压缩后，可以再用一个轻量级 NLI 模型检查提取的句子是否忠实于原文，或者是否包含了回答问题的必要信息。

4. 保留来源元数据：即使内容被压缩，也要保留文档的原始 ID 或链接，方便用户回查原文。

5. 调整压缩 Prompt：在 Prompt 中明确要求“保留所有数字、日期、人名等关键事实”，可以降低丢失风险。

💡 核心原则：压缩器的目标是提高信噪比，不是替代原文。 在关键任务中，可以将压缩后的内容与原始文档一同展示给用户，由人工判断。

---

🛡️ 如果检索到 20 个文档，压缩后只保留 3 个相关片段，这个过程如何保证不丢失关键事实？

这是一个典型的信息完整性保障问题。完全保证 100% 不丢失是不可能的，但可以通过冗余机制、多路召回、后验证来将风险降至可接受范围。

1. 冗余压缩（多策略提取）

不依赖单一的压缩器。可以同时运行两种压缩策略：一个用 LLM 提取关键句，另一个用简单的关键词匹配算法（如 Rake）提取高权重词所在的句子。将两者的结果并集合并，即使 LLM 漏了，关键词算法可能命中。

1. 分块与去重

检索到的 20 个文档可能包含重复信息。压缩前先进行语义去重，确保每个文档贡献独特的信息。压缩时，可以要求 LLM 对每个文档输出包含的独特事实列表，然后进行事实级别的合并，保留所有不重复的事实。

1. 关键事实验证

在压缩后，用原始查询和压缩后的片段，再向 LLM 提问：“这些片段是否包含回答该问题所需的所有关键信息？如果没有，请指出缺失的部分。” 这是一种自我反思式验证，虽然增加一次调用，但能及时发现丢失。

1. 保留原始文档的引用 压缩后，不丢弃原始文档。将压缩后的内容作为“线索”提供给最终的生成链，同时在 Prompt 中附上完整的文档列表（或者文档编号）。要求 LLM 在回答时，如果发现压缩内容不足以回答，则自行查阅原始文档。这正是 MapReduce 或 Refine 链的思路。

2. 设置保守的压缩率 不要追求极致的压缩比。如果 20 个文档总共 10K tokens，而你的 LLM 窗口有 32K，完全可以直接全部塞进去而不压缩。只在超限时才压缩。可以用 ContextualCompressionRetriever 配合 LLMChainExtractor，但为 Extractor 设置 max_tokens 较大，允许输出较多内容。

🔑 最终防线：在生成答案后，使用检索到的完整文档作为“标准答案”，用另一个 LLM 检查答案的事实性（Grounding）。这是最可靠的保证。

---

🧩 能否将多种检索器组合成一个？LangChain 中 EnsembleRetriever 如何配置权重？

可以，这正是 EnsembleRetriever 的用途。 它可以将多个不同类型的检索器（如向量检索、BM25、自定义检索器）的结果融合，统一返回。

配置权重的方法： 创建 EnsembleRetriever 时，可以通过 weights 参数为每个检索器指定权重。权重的高低会影响最终排序，但不会影响文档是否被返回（除非结合截断）。

from langchain.retrievers import EnsembleRetriever

ensemble_retriever = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.7, 0.3],   # 向量检索权重0.7，BM25权重0.3
)

权重的作用机制：

EnsembleRetriever 默认使用 Reciprocal Rank Fusion (RRF) 算法进行融合。权重的配置在 LangChain 当前版本中，实际上是通过在 RRF 公式中引入加权因子来实现的。加权 RRF 公式为：

其中 wrwr 是第 rr 个检索器的权重，kk 是平滑常数（默认60），rankr(d)rankr(d) 是文档 dd 在检索器 rr 返回结果中的排名。权重越大，该检索器对最终得分的贡献越大。

如果你不希望使用 RRF 而想自定义融合逻辑，可以重写 EnsembleRetriever 的 _get_relevant_documents 方法，例如简单地合并去重。

---

🔗 EnsembleRetriever 内部如何对多个检索器的结果进行融合？默认是 RRF（倒数排名融合）吗？

是的，默认采用 RRF（Reciprocal Rank Fusion），但代码中实现的是加权 RRF。融合流程如下：

1. 并行调用：同时（或顺序）调用所有子检索器的 get_relevant_documents，获取各自的排序文档列表。

2. 计算 RRF 得分：

3. 遍历每个检索器的结果，对于文档 dd 记录它在各检索器中的排名 rankrank（从0开始或从1开始，内部处理）。

4. 使用公式 score(d) += weight / (k + rank) 累积得分。k 默认60，用来平滑低排名的差距。

5. 按得分降序排序：所有文档按 RRF 得分从高到低排序。

6. 返回最终列表：返回排序后的文档列表（通常限制 total_k）。

RRF 的优点是不需要各检索器输出原始的相似度分数（因为不同检索器的分数尺度不一致，无法直接比较），只依赖排名，因此非常通用。

扩展：你也可以通过 c=60 参数调整 RRF 的平滑常数。更大的 c 使得排名靠后的文档得分衰减更慢，增加多样性。

---

🛠️ 你可以自定义一个简单的 Retriever 吗？比如基于关键词匹配的 retriever。

自定义 Retriever 非常简单，只需继承 BaseRetriever 并实现 _get_relevant_documents 方法。以下是一个基于 TF-IDF 关键词匹配的 Retriever 示例：

from langchain.schema import BaseRetriever, Document
from typing import List
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

class TFIDFRetriever(BaseRetriever):
    documents: List[Document]
    k: int = 4
    vectorizer: TfidfVectorizer = None
    tfidf_matrix: any = None

    def __init__(self, documents: List[Document], k=4):
        super().__init__(documents=documents, k=k)
        self.vectorizer = TfidfVectorizer()
        corpus = [doc.page_content for doc in documents]
        self.tfidf_matrix = self.vectorizer.fit_transform(corpus)

    def _get_relevant_documents(self, query: str) -> List[Document]:
        query_vec = self.vectorizer.transform([query])
        scores = cosine_similarity(query_vec, self.tfidf_matrix).flatten()
        # 获取分数最高的k个索引
        top_indices = scores.argsort()[-self.k:][::-1]
        return [self.documents[i] for i in top_indices]

使用方式：

docs = [Document(page_content="量子计算是一种..."), ...]
tfidf_retriever = TFIDFRetriever(documents=docs, k=5)
results = tfidf_retriever.get_relevant_documents("什么是量子计算？")

你也可以用 BM25（通过 rank_bm25 库）来实现，效果通常更好。只要遵循 BaseRetriever 的接口，就可以和 LangChain 的其他组件无缝集成。

---

💾 如何为 Retriever 添加缓存，避免对相同查询重复检索？

缓存可以大幅降低检索延迟和成本。实现方式有多种：

方案一：使用 LangChain 的缓存抽象（最简单） LangChain 提供了 BaseCache 用于缓存 LLM 调用，但 Retriever 不直接受其管辖。不过，我们可以自定义一个带缓存的 Retriever 包装器。

from langchain.schema import BaseRetriever, Document
from typing import List
from functools import lru_cache

class CachedRetriever(BaseRetriever):
    base_retriever: BaseRetriever

    def __init__(self, base_retriever):
        super().__init__(base_retriever=base_retriever)

    @lru_cache(maxsize=128)
    def _get_relevant_documents(self, query: str) -> List[Document]:
        return self.base_retriever.get_relevant_documents(query)

使用 lru_cache 是内存缓存，重启即消失。对于持久化，可以用 Redis：

import redis
import pickle

class RedisCachedRetriever(BaseRetriever):
    base_retriever: BaseRetriever
    redis_client: redis.Redis

    def __init__(self, base_retriever, redis_client):
        super().__init__(base_retriever=base_retriever, redis_client=redis_client)

    def _get_relevant_documents(self, query: str) -> List[Document]:
        key = f"retriever_cache:{hash(query)}"
        cached = self.redis_client.get(key)
        if cached:
            return pickle.loads(cached)
        docs = self.base_retriever.get_relevant_documents(query)
        self.redis_client.setex(key, 3600, pickle.dumps(docs))  # 1小时过期
        return docs

方案二：使用 VectorStore 自带的缓存

某些向量数据库（如 Redis、Pinecone）本身就支持查询缓存。可以在数据库配置中开启，无需在应用层实现。

注意事项：

• 缓存键的设计要唯一，可以包含查询字符串的规范化（如去空格、小写）。

• 设置合理的过期时间，避免知识库更新后缓存未失效。

• 对于需要高实时性的场景，可以采用“写后失效”策略：当文档更新时，主动清除相关缓存。

---

⚡ Retriever 的 invoke 和 ainvoke 方法在异步环境下如何提升吞吐？

在 LangChain 中，invoke 是同步方法，ainvoke 是异步方法（返回 awaitable）。在异步 Web 服务（如 FastAPI）中，使用 ainvoke 可以避免阻塞事件循环，从而显著提升并发吞吐量。

提升吞吐的关键：

• 不阻塞主线程：异步方法在等待 IO（如网络请求、数据库查询）时会让出 CPU，允许处理其他请求。

• 批量并发：可以同时发起多个异步检索任务，用 asyncio.gather 等待所有结果，然后统一返回。

示例：

import asyncio
from langchain.retrievers import ContextualCompressionRetriever

# 假设 compression_retriever 支持 ainvoke
async def retrieve_multiple(queries: list[str]):
    tasks = [compression_retriever.ainvoke(query) for query in queries]
    results = await asyncio.gather(*tasks)
    return results

如果底层的 VectorStore（如 Chroma）不支持异步，ainvoke 内部会通过 asyncio.to_thread 在线程池中执行同步方法，虽不如原生异步高效，但仍能避免阻塞主线程。

优化建议：

• 确保整个调用链都使用异步：从 LLM 到 HTTP 客户端，尽可能使用 AsyncOpenAI 等异步客户端。

• 对于自定义 Retriever，实现 _aget_relevant_documents 方法以利用异步数据库驱动（如 asyncpg、aioredis）。

• 使用连接池并限制最大并发数，防止下游服务过载。

通过全面异步化，一个 4 核的 API 服务器可以轻松支撑上千 QPS 的检索请求。

---

📊 在一个生产级 RAG 系统中，你如何监控 Retriever 的性能（如召回率、延迟）？

监控目标：

• 召回率（Recall）：检索到的相关文档数 / 总共相关文档数。衡量检索器找全的能力。

• 精确率（Precision）：检索到的相关文档数 / 检索到的总文档数。衡量检索器找准的能力。

• 平均延迟（Latency）：P50、P95、P99 的检索耗时。

• 缓存命中率：缓存使用效果。

• 错误率：检索器异常比例。

实施策略：

1. 建立评估集 准备一组代表性的查询和对应的标准答案文档 ID。这个黄金数据集可以来自用户反馈（点击、点赞）或专家标注。定期（如每天）离线评估检索器的召回率、MRR 等指标。

2. 在线埋点与指标收集 在 Retriever 的包装器中，记录每次调用的耗时、返回文档数量、检索源等。将这些指标输出到监控系统（如 Prometheus + Grafana）。示例：

import time
from prometheus_client import Histogram, Counter

REQUEST_TIME = Histogram('retriever_latency_seconds', '...')
DOCS_RETURNED = Counter('retriever_docs_returned', '...')

class MonitoredRetriever(BaseRetriever):
    def _get_relevant_documents(self, query):
        start = time.time()
        docs = self.base_retriever.get_relevant_documents(query)
        REQUEST_TIME.observe(time.time() - start)
        DOCS_RETURNED.inc(len(docs))
        return docs

1. 延迟监控与告警
2. 设置延迟阈值（如 P95 > 500ms），超过即触发告警。

3. 按检索器类型、文档库分区观察延迟，定位慢查询。

4. 召回率在线近似评估 通过用户行为间接评估：如果用户对 RAG 回答点了“踩”或追问，可能意味着检索未命中关键文档。将这些会话记录下来，定期人工分析。也可以使用 LLM 自动比较检索结果和理想答案，计算事实覆盖度。

5. 日志审计 记录未命中的查询（返回 0 文档），分析是否是知识库缺失还是检索器失效。

工具推荐：

• Prometheus + Grafana（指标可视化）

• LangSmith（LangChain 官方，可跟踪链的每一步）

• 自建 ELK 日志分析

通过这套监控体系，你可以持续优化检索器，确保 RAG 系统的高质量运行。

---