做 RAG 做得越久越会发现 Retriever 这一层才是整个系统的天花板。模型不行可以换Prompt 不行可以调但如果在召回环节拿回来的文档本身就是错的后面所有工作都只是在把错误的上下文包装得更完整。很多教程会直接甩给你vector_store.as_retriever()一行代码但那套做法是演示用的不是生产环境用的。真到接自己知识库的时候你躲不开权限过滤、多路召回、外部搜索接口这些需求这时候自定义 Retriever 接口就是必须走的一步。这篇文章我会把自定义 Retriever 接口这件事拆开讲接口本质是什么、核心方法怎么写、怎么接自己的知识库、怎么在 RAG 链路里组合使用最后是我的踩坑记录。适合两类人一类是已经跑通基础 RAG、想把它往生产里推的开发者另一类是刚接触 LangChain 生态、对检索层还停留在“调 as_retriever”阶段的初学者。1. Retriever 接口的本质为什么这一层决定了 RAG 的天花板1.1 先搞明白 Retriever 在整条链路里的位置RAG 系统的线上链路其实很简单用户提一个问题系统先去知识库里找相关文档再把“问题 文档”一起交给大模型生成答案。这个过程一般写成query - retrieve - augment - generate其中 retrieve 就是 Retriever 在做的事augment 则是把检索结果塞进 Prompt。Retriever 的地位很特殊。LLM 本身不具备“临时读知识库”的能力它只能看到你塞给它的上下文。Retriever 决定了哪些内容能进入这个上下文也就直接决定了答案的上限。换句话说只要检索结果有问题后面的 Prompt 设计和模型调优就都是在打补丁。我给新人打过一个比方Retriever 就是一个图书管理员。你问他一个问题他去书库给你抱来一摞书然后你基于这摞书写回答。默认的向量检索管理员只会按“语义相似度”这个单一规则找书但实际知识库往往有权限分区、有特殊分类法、有不同介质的内容。这时候你就得给他培训一套专属找书流程让他按照你的规则去取书。自定义 Retriever就是在做这件事。这里有一个很关键的接口约定Retriever 的输入是一个查询字符串query输出是一个文档列表List[Document]。至于中间是查向量库、查 Elasticsearch、调外部 API还是先做一轮查询改写接口不关心。这就是“接口”的意义——它把“要什么”和“怎么做”分开了。1.2 默认的 as_retriever 什么时候会明显失灵VectorStore.as_retriever()能省事是因为它帮你把向量库的相似度搜索包装成了 Retriever 协议。但它在下面这些场景里会很吃力知识库根本不在向量库里。比如公司内部 Wiki 只有一套 HTTP 查询接口业务数据在 MySQL 里历史工单在 Elasticsearch 里。你不能为了接 RAG 硬把全量数据搬进向量库搬完还有权限同步问题。需要多路召回而不是单路语义检索。向量检索擅长语义相近但关键词精确命中也很重要。比如型号ABC-123这种字符串语义相似度不一定排前面精确检索反而更稳定。需要权限过滤和结构化约束。知识库文档不是所有人都有权限看Retriever 必须在召回阶段就过滤掉无权内容而不是让 LLM 在回答时规避。权限落到检索层比落到生成层靠谱得多。需要检索日志、埋点、评测。默认 retriever 的召回过程是黑盒出了问题很难定位。自定义之后每个环节都可以打日志、算指标。知识库里不只有文本。比如你存了一批产品图片、截图、视频Document 的page_content可以是文字说明文件路径或 URL 放在metadata里后续需要时再加载图片交给多模态模型。这个逻辑默认的 as_retriever 也给不了。一句话总结as_retriever适合原型验证自定义 Retriever 才适合生产交付。你要接的是“自己的知识库”不是教程里那个 demo 用的玩具库。2. BaseRetriever 接口到底约束了你什么2.1 你真正要实现的方法其实只有一个LangChain 里自定义 Retriever 的正确姿势是继承BaseRetriever。这个基类本身已经实现了invoke、batch、stream、ainvoke这些 Runnable 接口能力你不需要重写它们。你需要做的是实现一个私有方法_get_relevant_documents。这是同步检索的核心逻辑如果你有异步场景再补一个_aget_relevant_documents。from langchain_core.callbacks import CallbackManagerForRetrieverRun from langchain_core.documents import Document from langchain_core.retrievers import BaseRetriever class MyRetriever(BaseRetriever): top_k: int 5 def _get_relevant_documents( self, query: str, *, run_manager: CallbackManagerForRetrieverRun, ) - list[Document]: # 在这里实现你自己的检索逻辑 # 最终只要返回 List[Document] 即可 return []注意两个细节。第一run_manager是 keyword-only 参数前面有个*这是基类签名约束少写了都会报错。第二这里的方法名带下划线。在 LangChain 0.2/0.3 版本里公开入口是get_relevant_documents它会负责封装回调管理然后在内部调用你这个_get_relevant_documents如果你直接覆盖get_relevant_documents会破坏回调机制还会收到 deprecation warning。2.2 理解 Document、run_manager 和异步三件事Document是检索结果的标准容器有两个核心字段page_content放正文文本metadata放附加信息比如文档标题、来源、score、权限标识。我之前说过page_content不只是能放纯文本——图片路径、音频文件地址、SQL 查询结果都可以放关键是你在生成阶段怎么用这些信息。但多数情况下它就是字符串metadata 里建议只放可序列化的基础类型别放复杂对象否则后面做缓存、做序列化都会冒出来找你。run_manager是回调管理器。你在检索过程中可以通过它记录日志、发送文本事件、标注中间结果。调试时很管用生产里也可以用它接监控。比如你在_get_relevant_documents里中途查完向量库、准备调外部 API 时可以run_manager.on_text(vector recall done)这些事件会被注册到链路的回调里。异步方法。如果你的知识库是外部 HTTP 服务或者你不想让异步 RAG 链路被同步阻塞就实现_aget_relevant_documents。它的签名和同步版本几乎一样只是 run_manager 类型换成AsyncCallbackManagerForRetrieverRun方法体里用await调用异步客户端。2.3 实现 BaseRetriever 之后你白捡了哪些能力BaseRetriever继承自 Runnable这意味着定义好了_get_relevant_documents你的自定义检索器就自动具备了一整套运行时能力invoke(query)同步调用返回List[Document]ainvoke(query)异步调用batch([q1, q2])批量调用stream(query)流式输出结果as_tool()在 agent 场景中把 Retriever 暴露成工具让模型自己决定什么时候调用知识库这一点是很多人没意识到的。你以为你在写一个“检索函数”实际上你在接入 RAG 生态的标准协议。这是为什么我一直强调自定义 Retriever 不是让你绕开框架而是用框架的原生方式扩展边界。3. 实战写一个接自家知识库的混合召回 Retriever3.1 场景设计真实项目里的知识库往往不是单一存储拿我自己做过的一个案例来说。团队有一个内部知识库内容分成三块历史工单、FAQ、内部文档。物理上它们分布在三个地方工单和 FAQ 的元数据、权限字段在 MySQL 里正文经过 Embedding 后存在本地 FAISS 向量库团队还有一个基于 Elasticsearch 的全文检索引擎已经跑了好几年很多搜索逻辑都在那边当时的诉求是RAG 项目要基于这套知识库做问答但每次检索不能只看向量相似度还要兼顾精确关键词、权限隔离和工单优先级。那直接用as_retriever()显然不行我就在 LangChain 里自定义了一个TicketKnowledgeRetriever。整体流程大致是这样query ├── 向量召回 (FAISS, top30) ├── 关键词召回 (HTTP 搜索服务, top20) └── 合并去重、加权排序 └── MySQL 权限过滤、补全元数据 └── 返回 top5 的 Document 列表注意这个流程完全是在 Retriever 内部完成的对上层链路透明。你直接retriever.invoke(如何配置工单超时时间)拿到的就是已经过滤好、排好序的文档列表。3.2 按接口协议写一个混合召回 Retriever下面是我实际项目里的简化版本核心逻辑都在from typing import Any import httpx from langchain_core.callbacks import CallbackManagerForRetrieverRun from langchain_core.documents import Document from langchain_core.retrievers import BaseRetriever class TicketKnowledgeRetriever(BaseRetriever): 接公司知识库的自定义 Retriever向量 关键词 权限过滤。 vector_store: Any # 本地向量库支持 similarity_search_with_score search_service_url: str # 团队已有全文搜索服务 mysql_conn: Any # 数据库连接用于权限过滤和元数据补全 top_k: int 5 score_threshold: float 0.3 owner_filter: str | None None # 当前用户或空 def _get_relevant_documents( self, query: str, *, run_manager: CallbackManagerForRetrieverRun ) - list[Document]: # 1. 向量召回 vec_docs self._vector_recall(query, run_manager) # 2. 关键词召回 kw_docs self._keyword_recall(query, run_manager) # 3. 合并去重、加权重排 merged self._merge_and_rerank(vec_docs, kw_docs, query) return merged[: self.top_k] def _vector_recall(self, query: str, run_manager) - list[Document]: docs_with_scores self.vector_store.similarity_search_with_score( query, kself.top_k * 6 ) docs [] for doc, score in docs_with_scores: d Document( page_contentdoc.page_content, metadatadict(doc.metadata), # 拷贝别污染原对象 ) d.metadata[_score] float(score) d.metadata[_source] vector docs.append(d) run_manager.on_text(fvector recall: {len(docs)} docs) return docs def _keyword_recall(self, query: str, run_manager) - list[Document]: resp httpx.post( f{self.search_service_url}/search, json{query: query, size: self.top_k * 4}, timeout10, ) resp.raise_for_status() results resp.json().get(hits, []) docs [] for item in results: doc Document( page_contentitem[content], metadata{ doc_id: item[id], title: item.get(title, ), _score: float(item.get(score, 0.0)), _source: keyword, }, ) docs.append(doc) run_manager.on_text(fkeyword recall: {len(docs)} docs) return docs def _merge_and_rerank(self, vec_docs, kw_docs, query: str) - list[Document]: fused: dict[str, Document] {} for doc in vec_docs kw_docs: doc_id doc.metadata.get(doc_id) or doc.metadata.get(source, ) if doc_id in fused: old fused[doc_id] old.metadata[_score] max(old.metadata[_score], doc.metadata[_score]) if old.metadata[_source] ! doc.metadata[_source]: old.metadata[_hit_both] True else: fused[doc_id] doc candidates list(fused.values()) # 双路命中的文档加权类似 RRF 的思路 for doc in candidates: if doc.metadata.get(_hit_both): doc.metadata[_score] 1.0 candidates.sort(keylambda d: d.metadata[_score], reverseTrue) candidates [d for d in candidates if d.metadata[_score] self.score_threshold] return self._enrich_from_db(candidates) def _enrich_from_db(self, docs: list[Document]) - list[Document]: 模拟从 MySQL 补全标题、过滤无权限文档。 result [] for doc in docs: # 这里实际会走 SQL: SELECT title, permission FROM doc_meta WHERE doc_id ? permission public if permission private and self.owner_filter is None: continue doc.metadata.setdefault(title, 工单/FAQ标题) result.append(doc) return result代码看起来不短但拆开其实就四步向量召回、关键词召回、合并排序、权限过滤。这个 Retriever 对上层来说依然只是输入一个 query、返回一个List[Document]。你把top_k、score_threshold这些参数声明成类字段实例化的时候可以传还天然支持 Pydantic 校验。3.3 这个实现里最容易踩的 3 个细节第一metadata 里的分数一定要转成 float。很多向量库返回的相似度分数是numpy.float32直接塞进 metadata 后后续序列化、缓存、打印都会出问题。不只是分数凡是往 metadata 里放的数值都建议统一转成 Python 原生类型。第二doc_id 的去重约定要提前定好。我在这里用doc_id或source做合并键但实际项目里同一个文档在不同来源可能用不同 ID。你需要在索引构建阶段就给每个文档分配一个全局唯一的doc_id并且保证从向量库、从搜索服务、从 MySQL 查回来时带的 ID 是同一个。这个约定不做后面多路召回越多重复文档越严重。第三从向量库取出来的 Document 一定要拷贝。很多向量库保存的是索引内的文档对象你直接在原对象上改 metadata会影响后续复用。上面的代码里我用Document(page_content..., metadatadict(doc.metadata))重新构造了对象这是顺手的事但能避免很多诡异的 bug。3.4 写完怎么验证这个 Retriever 真的可用自测不需要很复杂写一个小函数把召回结果打出来就够def debug_retriever(retriever, query: str): docs retriever.invoke(query) print(fquery: {query}, hits: {len(docs)}) for i, doc in enumerate(docs, 1): print(i, doc.metadata.get(_source), round(doc.metadata.get(_score, 0), 4), doc.metadata.get(title))我习惯先看三件事召回数量是否稳定、双路命中的文档占比、低分噪声有没有被过滤掉。然后挑 20 条真实业务 query 人工看 top5 的相关性。别急着上大模型评估人工先看一遍最快暴露问题。4. 对接外部知识库 API超时、重试与异步必须一起考虑4.1 典型场景你的知识库只有一个 HTTP 接口第二种常见情况是你的知识库根本不在本地而是公司某个内部系统提供的 HTTP 接口。比如说团队 Wiki 平台、企业网盘索引、自研搜索引擎它们只开放了一组 HTTP API文档内容和权限校验都在对方那边。这时候自定义 Retriever 反而更简单因为核心逻辑变成了一次 HTTP 请求。伪代码是这样import httpx from langchain_core.documents import Document class WikiApiRetriever(BaseRetriever): api_base: str token: str top_k: int 5 def _get_relevant_documents(self, query, *, run_manager): resp httpx.get( f{self.api_base}/search, params{keyword: query, size: self.top_k}, headers{Authorization: fBearer {self.token}}, timeout10, ) resp.raise_for_status() data resp.json() documents [] for item in data.get(documents, []): documents.append( Document( page_contentitem[content], metadata{ title: item[title], url: item[url], _source: wiki, }, ) ) return documents对接外部接口时我强烈建议不要做“拿回全部正文再自己重排”这种事。外部系统往往已经是成熟的知识库有自己的相关性排序逻辑你接入时先看它返回的字段能不能满足Document的基本要求满足就直接包装。4.2 超时、重试与限流必须一起设计外部接口和本地向量库不一样它可能慢、可能挂、可能有并发限制。Retriever 一旦成为 RAG 链路里的常规调用你就得把它当成一个服务调用去治理。超时是最基本的httpx里可以直接指定timeout10别让它无限等。接着是重试对 5 开头的状态码或者网络超时做有限次数的重试累计三次还不行就放弃让链路上抛异常或者走降级。我一般会用装饰器或者简单的循环实现不会引入太重的重试库关键是控制总耗时。限流这块容易被忽视。RAG 服务一般不会只服务一个用户当多个请求同时进来每个请求又触发 Retriever 去调外部 API瞬间就打爆了。最简单的做法是在 Retriever 内部放一个信号量import threading _semaphore threading.Semaphore(10) # 同一时间最多 10 个检索请求 def _get_relevant_documents(self, query, *, run_manager): with _semaphore: return self._call_external_service(query)降级策略也建议提前设计。我的习惯是外部搜索服务不可用时自动退化成只走本地向量召回而不是让整个 RAG 链路报错。这样至少用户还能拿到答案只是质量会差一些。4.3 异步方法 _aget_relevant_documents 怎么写如果你的 RAG 链路本身是异步的同步检索会把事件循环堵住。实现异步版本很简单方法名前面加一个a内部用协程import httpx from langchain_core.callbacks import AsyncCallbackManagerForRetrieverRun class AsyncWikiApiRetriever(BaseRetriever): api_base: str top_k: int 5 async def _aget_relevant_documents( self, query: str, *, run_manager: AsyncCallbackManagerForRetrieverRun, ) - list[Document]: async with httpx.AsyncClient(timeout10) as client: resp await client.get( f{self.api_base}/search, params{keyword: query, size: self.top_k}, ) resp.raise_for_status() data resp.json() return [ Document(page_contentitem[content], metadata{title: item[title]}) for item in data.get(documents, []) ]写异步版本时有个隐含约定如果同步和异步都实现了LangChain 在异步调用时优先走_aget_relevant_documents不会意外调用同步版本阻塞事件循环。我建议对外部 API 的场景无论如何都补上异步实现哪怕你当前链路是同步的以后改造也方便。5. 把自定义 Retriever 接进 RAG 链路这 3 个组合技巧最实用5.1 LCEL 接线把 retriever 当作一个 Runnable 用写好自定义 Retriever 之后接入 RAG 链路比你想象的简单。LangChain 的 LCEL 表达式语言支持把 Retriever 直接放到一个并行字典里from langchain_core.runnables import RunnablePassthrough, RunnableLambda from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser def format_docs(docs): return \n\n.join( f[{doc.metadata.get(title, 未知标题)}]\n{doc.page_content} for doc in docs ) prompt ChatPromptTemplate.from_template( 请基于以下背景资料回答用户问题。资料不足时请直接说明。\n\n 背景资料\n{context}\n\n 用户问题{question} ) chain ( { context: retriever | RunnableLambda(format_docs), question: RunnablePassthrough(), } | prompt | llm | StrOutputParser() ) answer chain.invoke(工单超时时间在哪里配置)这里有一个很多人踩过的坑如果不加RunnableLambda(format_docs)context传到 Prompt 里就是一个List[Document]ChatPromptTemplate 会把它格式化成page_content... metadata...这种难看的字符串既占 token 又不好读。加一个format_docs把文档列表拼接成可读文本再进 Prompt输出的质量和稳定性都会好很多。5.2 组合技巧用压缩 Retriever 做二次裁剪自定义 Retriever 负责召回但召回结果不一定每一段都值得进入上下文。比较经典的组合是把自己写的 Retriever 外包给ContextualCompressionRetriever让它对召回文档做二次筛选和压缩from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor compressor LLMChainExtractor.from_llm(llm) compression_retriever ContextualCompressionRetriever( base_compressorcompressor, base_retrievermy_custom_retriever, )这样每次检索时自定义 Retriever 先召回一批候选文档压缩器再用 LLM 判断哪些片段和 query 真正相关裁掉无关片段。代价是增加了一次 LLM 调用延迟和成本都会涨。我的建议是只在知识库文档偏长、或者召回噪声偏大的场景使用如果召回结果已经足够精准就别让它进场。5.3 组合技巧给检索加缓存避免重复打外部服务外部 API 或者重量级检索的成本很高同一个 query 短时间内被多次检索是常见浪费。可以给自定义 Retriever 包一层简单的内存缓存import time from threading import Lock from pydantic import PrivateAttr from langchain_core.documents import Document from langchain_core.retrievers import BaseRetriever class CachedRetriever(BaseRetriever): retriever: BaseRetriever ttl: int 300 _cache: dict PrivateAttr(default{}) _lock: Lock PrivateAttr(default_factoryLock) def _get_relevant_documents(self, query, *, run_manager): now time.time() with self._lock: cached self._cache.get(query) if cached and cached[0] now: # 拷贝返回避免调用方修改污染缓存 return [ Document(page_contentd.page_content, metadatadict(d.metadata)) for d in cached[1] ] docs self.retriever.invoke(query) with self._lock: self._cache[query] (now self.ttl, docs) return docs注意_cache和_lock用PrivateAttr声明Pydantic 不会把它们当成出入参字段适合放内部状态。多实例部署时这个内存缓存各存各的如果想要更强一致可以把缓存放到 Rediskey 用 queryvalue 存文档 id 列表配合按 id 查询文档的接口能省很多体量。6. 常见问题排查速查表以及上线前必做的三件事6.1 报错与现象速查表现象可能原因处理方式覆盖get_relevant_documents后出现 deprecation warningLangChain 0.2 改用_get_relevant_documents把逻辑迁移到_get_relevant_documents保留公开入口让基类管理回调run_manager参数缺失或位置不对基类签名要求 keyword-only写成*之后的 keyword-only 参数返回结果在 LCEL 里变成奇怪的字符串没把List[Document]转成文本用RunnableLambda(format_docs)拼接metadata 里的分数无法 JSON 序列化存了numpy.float32等非原生类型统一float()转换外部 API 慢导致整个链路超时没设置请求超时和重试加timeout、重试和降级开关异步链路卡顿只实现了同步_get_relevant_documents补_aget_relevant_documents同一个文档反复出现多路召回没有统一去重建立全局doc_id约定在合并时去重LLM 答案涉及无权限内容权限过滤放在了生成层而非检索层在 Retriever 内做权限过滤从源头去掉6.2 上线前必做的三件事如果要用在生产环境我建议在做完基础验证之后立刻做这三件事缺一不可。一是给检索层加日志。记录每次检索的 query、召回 doc_id、分数分布、命中的来源。不要嫌日志量大检索日志是 RAG 问题排查的第一手证据。很多“答错了”的问题最后追根溯源都是召回阶段没召回对没有日志你根本无从查起。二是建一个小而准的评测集。我一般是挑 50 到 100 条真实业务 query每条标注 2 到 5 个期望命中的文档 ID。改完 Retriever 之后跑一遍评测看召回率有没有变化。这个评测集不用很庞大但必须能反映业务真实分布比一百条自嗨的通用问题有用得多。三是准备好降级开关。外部知识库服务、向量库都可能出故障。给检索链路的每个下游依赖做一个开关外部服务挂了几秒内切到本地向量召回至少保证系统还能回答基础问题。降级链路也要提前测一遍不要在故障发生时才写逻辑。收尾我对自定义 Retriever 的一点体会这篇文章看起来是在讲一个具体接口实际上想说的东西更底层在 RAG 项目里检索层是全系统性价比最高的改造点。很多时候你调模型、调 Prompt 没效果多半是问题出在召回环节。自定义 Retriever 的价值就在于它把“怎么找文档”这件事完全还给了你你可以把任何检索逻辑——向量、关键词、权限、API、重排——都收进同一个接口里中间的实现细节完全不影响上层。我做过的几个 RAG 项目里凡是“接自己知识库”卡住的基本都不是向量库不会用而是没想明白 Retriever 接口的约定你只管输出List[Document]查询过程全是你的自由。把这个约定踩实了后面接 ES、接企业 Wiki、接自研搜索引擎都是一套模式。最后分享一个我觉得很实用的小技巧可以给自己写的 Retriever 加一个可视化调试函数专门用来打印每次检索的完整链路——query 是什么、召回了哪些 id、每个 id 的 score 和来源。这个函数几乎每天都会被用到排查 RAG 问题的时候一半时间都能省在它身上。