内置检索器与自定义检索器：把检索策略工程化

LangChain 提供了很多现成 Retriever，用来接入向量库、搜索引擎、百科、文档系统、数据库或第三方服务。但生产 RAG 不会永远停留在“向量库 top_k”。当数据源、权限、过滤、排序、业务规则变复杂时，就需要自定义 Retriever。

内置 Retriever 的边界

内置 Retriever 适合快速接入外部数据源，例如向量库检索、Wikipedia 搜索、Elasticsearch、Weaviate 混合检索、Zep 记忆检索等。它们的好处是封装了连接、参数、返回格式，可以直接接入链路。

但要注意三点。

第一，不同 Retriever 的维护状态不同。有些较早实现的组件可能没有完整接入 Runnable 能力，使用前需要确认是否支持 invoke()、batch()、运行时配置和回调。

第二，内置 Retriever 的参数并不完全统一。某些支持 filter，某些支持 score_threshold，某些只支持 top_k。不能因为接口名字一样，就默认能力一致。

第三，业务权限通常不是内置 Retriever 能自动解决的。多租户、部门权限、文档状态、用户角色这些过滤条件，需要在应用层明确注入。

什么时候自定义 Retriever

自定义 Retriever 适合以下场景。

数据源不是标准向量库，例如企业内部搜索接口、工单系统、CRM、对象存储、日志系统、API 网关、知识图谱或 SQL 查询结果。

检索逻辑包含业务规则，例如先按租户过滤，再按知识库状态过滤，再按关键词命中扩展，最后做向量 rerank。

需要统一观测和降级，例如主库超时后切备用库，或向量库没有命中时转关键词搜索。

需要返回特殊 metadata，例如原系统文档 ID、权限范围、引用地址、片段得分、业务类型、版本号。

自定义 Retriever 示例

最小实现方式是继承 BaseRetriever，声明字段，并实现 _get_relevant_documents()。下面示例用最朴素的字符串包含关系来检索，重点是接口形态。

from typing import List

from langchain_core.callbacks import CallbackManagerForRetrieverRun
from langchain_core.documents import Document
from langchain_core.retrievers import BaseRetriever


class CustomRetriever(BaseRetriever):
    """自定义检索器。"""

    documents: list[Document]
    k: int

    def _get_relevant_documents(
        self,
        query: str,
        *,
        run_manager: CallbackManagerForRetrieverRun,
    ) -> List[Document]:
        """根据 query 获取相关文档列表。"""
        matching_documents = []
        for document in self.documents:
            if len(matching_documents) >= self.k:
                return matching_documents
            if query.lower() in document.page_content.lower():
                matching_documents.append(document)
        return matching_documents


documents = [
    Document(page_content="笨笨是一只很喜欢睡觉的猫咪", metadata={"page": 1}),
    Document(page_content="我喜欢在夜晚听音乐，这让我感到放松。", metadata={"page": 2}),
    Document(page_content="猫咪在窗台上打盹，看起来非常可爱。", metadata={"page": 3}),
    Document(page_content="掌握新技能是每个人都应该追求的目标。", metadata={"page": 4}),
    Document(page_content="我最喜欢的食物是意大利面，尤其是番茄酱的那种。", metadata={"page": 5}),
    Document(page_content="昨晚我做了一个奇怪的梦，梦见自己在太空飞行。", metadata={"page": 6}),
    Document(page_content="我的手机突然关机了，让我有些焦虑。", metadata={"page": 7}),
    Document(page_content="阅读是我每天都会做的事情，我觉得很充实。", metadata={"page": 8}),
    Document(page_content="他们一起计划了一次周末的野餐，希望天气能好。", metadata={"page": 9}),
    Document(page_content="我的狗喜欢追逐球，看起来非常开心。", metadata={"page": 10}),
]

retriever = CustomRetriever(documents=documents, k=3)

retriever_documents = retriever.invoke("猫")
print(retriever_documents)
print(len(retriever_documents))

这个例子会返回包含“猫”的文档。实际生产里不会只做字符串包含，而是会接入搜索服务、数据库、向量库、reranker 或业务 API。

实现细节

documents 和 k 被声明成类字段，是因为 BaseRetriever 基于 Pydantic 风格的数据模型管理字段。实例化时传入 documents=...、k=... 即可。

_get_relevant_documents() 的返回值必须是 Document 列表。不要返回字符串、字典或数据库行对象。外部系统返回的数据需要转换成 Document(page_content=..., metadata=...)。

run_manager 用于回调和追踪。简单示例里可以不用它，但生产中可以通过回调记录查询、候选数量、耗时、异常等信息。

示例中 len(matching_documents) >= self.k 用于控制返回数量。如果写成 > self.k，可能多返回一条，这是实现时常见的小错误。

异步实现

如果底层数据源是异步 API，可以实现 _aget_relevant_documents()。

from langchain_core.callbacks import AsyncCallbackManagerForRetrieverRun

class AsyncCustomRetriever(BaseRetriever):
    documents: list[Document]
    k: int

    async def _aget_relevant_documents(
        self,
        query: str,
        *,
        run_manager: AsyncCallbackManagerForRetrieverRun,
    ) -> list[Document]:
        matches = []
        for document in self.documents:
            if len(matches) >= self.k:
                return matches
            if query.lower() in document.page_content.lower():
                matches.append(document)
        return matches

如果只实现同步方法，很多情况下 LangChain 可以通过线程方式兼容异步调用，但高并发系统最好提供真正的异步实现。

Retriever 与工具调用

不是所有外部数据查询都必须做 Retriever。对于实时搜索、数据库查询、订单状态、天气、库存这类强实时或强结构化任务，工具调用可能比 Retriever 更合适。

Retriever 更适合返回文档片段，交给 LLM 组织答案。工具更适合执行明确动作或精确查询。复杂 Agent 系统里两者可以同时存在：Retriever 提供知识上下文，工具提供实时操作。

最新版 LangChain 用法提示

当前自定义检索器仍然建议继承 langchain_core.retrievers.BaseRetriever，实现 _get_relevant_documents()，并通过 invoke() 调用。

内置 Retriever 分散在 langchain_community、具体集成包和其他生态包中。新项目不要只看旧导入路径，要确认对应包的当前文档和安装方式。

如果检索逻辑只是一个函数，也可以用 RunnableLambda 快速封装。但只要它在语义上是“query -> documents”，并且会长期维护，继承 BaseRetriever 更清晰。

拓展

可以做组合 Retriever：向量检索、关键词检索、业务 API 检索并行执行，然后合并、去重、rerank。合并时按 source_id、chunk_id、version 去重，避免同一证据重复进入 Prompt。

可以做路由 Retriever：先判断 query 类型，再选择不同数据源。例如 API 文档问题走接口知识库，故障问题走运维知识库，产品问题走产品手册。

可以做权限 Retriever：把用户身份、租户、角色、知识库范围注入过滤条件，保证检索层不返回越权文档。

常见问题

自定义 Retriever 可以直接返回字符串吗？

不建议。标准返回值是 Document 列表。字符串会破坏后续引用、metadata、rerank 和格式化链路。

为什么已经有 VectorStoreRetriever 还要自定义？

VectorStoreRetriever 适合标准向量检索。自定义 Retriever 用于业务过滤、外部系统、组合检索、权限控制和特殊排序。

自定义 Retriever 是否必须支持异步？

不是必须，但如果底层是网络服务或高并发系统，建议实现异步方法。

面试题

如何实现一个 LangChain 自定义 Retriever？

继承 BaseRetriever，声明需要的字段，实现 _get_relevant_documents()，把外部数据转换成 Document 列表，并通过 invoke() 调用。

Retriever 和 Tool 的区别是什么？

Retriever 侧重检索文档上下文，Tool 侧重执行动作或精确查询。RAG 问答常用 Retriever，实时业务操作更适合 Tool。

自定义 Retriever 的生产关注点有哪些？

权限过滤、超时、重试、观测、去重、排序、metadata 规范、异常降级和返回数量控制。

生产问题排查

问题	常见原因	处理方式
返回类型导致链路报错	返回了字符串或字典	统一返回 Document 列表
多返回一条	k 判断条件写错	使用 >= self.k 并增加单测
检索耗时不稳定	外部 API 无超时或串行调用	设置超时、并发、缓存和降级
越权返回文档	权限过滤没在检索层执行	Retriever 内强制注入权限条件
后续无法展示来源	metadata 缺少 source_id 或 URL	统一 metadata 字段规范