做 RAG 做到一定阶段你大概率会开始琢磨一件事怎么把 Retriever 换成自己的。官方内置的向量检索方案确实好用但它默认你面前是一个已经切好块、算好向量、装进向量库的干净知识库。而现实里的知识库往往是散落在公司 Wiki、工单系统、网盘文档、数据库里的各种内容它们没有现成的向量版本甚至只能通过一个内部 HTTP 接口按关键词查。这时候自定义 Retriever 接口就是那个把你自己的知识库接进 RAG 链路的关键插槽。这篇文章我会从接口设计、代码实现讲到踩坑调优目标是把整条路走通一遍。1. 先确认前提官方自带 Retriever 为什么不够你接1.1 自带方案解决的其实是标准场景先看官方组件解决了什么问题。以 LangChain 里最常见的VectorStoreRetriever为例它的检索逻辑简单粗暴把 query 做 embedding然后去向量库做相似度检索返回 Top-K 个 Document。这套流程能成立依赖一个默认条件——你的知识库已经被离线清洗、切块、向量化并且存进了 FAISS、Milvus、Chroma 这类存储里。如果你的知识库恰好就是这个形态那确实没必要自己写 Retriever。我见过不少项目文档丢进解析器切块后灌进向量库检索直接用官方默认效果也能跑。但问题在于恰好这两个字很少成立。真实业务场景里知识库永远比 Demo 乱得多。我之前接过一个内部需求知识库是公司的 Wiki 系统文档实时更新每天有几百篇新内容。如果按传统做法得先把整个 Wiki 定期同步出来、切块、embedding、写增量更新逻辑。麻烦不说用户刚在 Wiki 上改完的内容向量库最快也要几小时后才能查到。这种实时性要求直接靠VectorStoreRetriever是接不住的。1.2 出现这三个信号就该考虑自定义 Retriever我自己判断是否要写自定义 Retriever基本看三个信号第一个信号是知识库只有接口没有文档包。比如你只能通过GET /api/search?q关键词拿到检索结果拿不到原始文档全文或者原始数据分布在十几个系统里根本没有一个统一的离线文件集。这时候与其先同步成文档再向量化不如直接在 Retriever 层对接接口。第二个信号是检索逻辑不是简单的相似度。比如你的知识库需要根据用户所属部门过滤权限比如某些数据必须走数据库的全文索引而不是向量比如答案依赖最新数据必须实时查。这些业务逻辑放进 query 预处理、放进检索后过滤、放进 score 重排全都是在自定义 Retriever 里完成的。第三个信号是你已经在为召回质量头疼。很多人以为 RAG 的瓶颈在生成但实际调下来大部分效果翻车都出在召回。召回不准换再大的模型也白搭。而召回这一步恰恰是最值得你完全掌控的地方。注意不是让你推翻官方方案。更常见的架构是自定义 Retriever 向量检索组合使用后面我会讲到多路召回的做法。先搞清楚自定义 Retriever 能做什么你才能判断什么时候该上。2. Retriever 接口的本质一份只有两道必答题的答卷模板2.1 核心契约输入一个 query输出一批文档不管你是用 LangChain、LlamaIndex 还是自研框架Retriever 的接口思想高度一致给一个查询字符串返回一组最相关的文档。区别只在类名、方法名和文档结构。LangChain 里自定义 Retriever 的模板继承自BaseRetriever核心要实现的是_get_relevant_documents。框架负责调用链、回调、缓存等外围逻辑你只需要回答给定 query返回哪些文本这一道题。from typing import List from langchain_core.retrievers import BaseRetriever from langchain_core.documents import Document from langchain_core.callbacks import CallbackManagerForRetrieverRun class MyKnowledgeRetriever(BaseRetriever): 把内部知识库接口包装成 LangChain Retriever top_k: int 5 def _get_relevant_documents( self, query: str, *, run_manager: CallbackManagerForRetrieverRun, **kwargs, ) - List[Document]: # 这里写你的检索逻辑 ...关键点有三个query是用户输入的最后一次提问文本返回值必须是List[Document]run_manager是用来上报日志、错误和回调事件的不是摆设后面调试全靠它。如果你用的是 LlamaIndex模板略有不同继承BaseRetriever实现_retrieve方法返回List[NodeWithScore]。思想一模一样只是Node换成了Documentscore直接挂在节点上。2.2 同步、异步、流式先想清楚你的运行环境BaseRetriever还提供了异步版本_aget_relevant_documents。很多人会忽略这一点但这是个很实际的工程问题。如果你的 RAG 服务跑在 FastAPI 这类异步框架里而你的知识库接口又是基于httpx.AsyncClient的异步调用那么实现异步版_aget_relevant_documents能让整个请求链路不阻塞事件循环。反过来如果你内部知识库 SDK 只有同步版本那强行异步没有任何意义直接用同步版框架会在线程池里调度。还有一个细节BaseRetriever提供了invoke方法作为统一入口。别自己重写invoke除非你想彻底截断框架的生命周期管理。按照抽象方法去实现让框架在invoke里帮你处理回调、重试和 run 追踪才是正确姿势。2.3 不要忽略回调run_manager 是白嫖的调试工具我之前写自定义 Retriever 时也嫌run_manager碍事后来发现它真香。在_get_relevant_documents内部你可以通过run_manager.on_text()输出调试信息通过run_manager.on_retriever_end(docs)上报结果。配合 LangSmith 或者 OpenAI 的 tracing 工具你能清楚看到每次检索的输入、输出、耗时排查问题效率高很多。# 在检索前后可以这么打点 if run_manager: run_manager.on_text(fquery: {query}, colorgreen) docs ... if run_manager: run_manager.on_retriever_end(docs)这不是必须的但建议写。线上出了问题这几行代码能帮你少掉一半头发。3. 实战把公司 Wiki 系统的搜索接口包装成 Retriever3.1 场景设定Wiki 接口长什么样为了让代码可落地我假设一个典型场景。你所在的公司有一个内部 Wiki提供一个搜索接口不支持向量检索只支持关键词搜索GET http://wiki.internal.example.com/api/search 参数q(查询词)、size(返回条数) 返回 JSON { code: 0, data: { total: 12, items: [ { id: wk-1001, title: 发布流程说明, content: 版本发布前需要准备..., score: 0.87, updated_at: 2025-06-01 10:00:00 } ] } }这种接口非常典型它不知道什么是 embedding但它的关键词搜索通常是数据库全文索引或者 Elasticsearch在特定场景下效果并不差尤其是对专有名词、系统名称、操作步骤这类内容关键词命中往往比向量相似度更准。3.2 核心代码把 HTTP 调用变成 Document 列表接下来就是重头戏实现自定义 Retriever。我直接用requests写同步版本方便你跑通流程。生产环境建议用httpx后面会讲原因。import requests from typing import List, Dict, Any from langchain_core.retrievers import BaseRetriever from langchain_core.documents import Document from langchain_core.callbacks import CallbackManagerForRetrieverRun class WikiRetriever(BaseRetriever): 对接公司 Wiki 检索接口的自定义 Retriever base_url: str http://wiki.internal.example.com/api/search top_k: int 5 timeout: float 5.0 headers: Dict[str, str] {Content-Type: application/json} def _get_relevant_documents( self, query: str, *, run_manager: CallbackManagerForRetrieverRun, **kwargs, ) - List[Document]: params {q: query, size: self.top_k} # 用 session 复用连接池避免每次都新建 TCP 连接 with requests.Session() as session: resp session.get( self.base_url, paramsparams, headersself.headers, timeoutself.timeout, ) resp.raise_for_status() payload resp.json() if payload.get(code) ! 0: raise RuntimeError(fWiki API error: {payload.get(msg)}) items payload.get(data, {}).get(items, []) docs [] for item in items: # 把标题和正文拼成 page_content page_content f{item[title]}\n\n{item[content]} docs.append( Document( page_contentpage_content, metadata{ source: fwiki/{item[id]}, title: item[title], score: item.get(score, 0.0), updated_at: item.get(updated_at, ), }, ) ) return docs这段代码的要点有几个。首先是Document的metadata不要只塞source把你能拿到的元数据都放进去title、updated_at、score全都有用。其次是 page_content 的拼接我把title放到了正文前这样 LLM 在生成时能同时看到标题和正文对回答质量有实际提升。3.3 把自定义 Retriever 挂进 RAG 链路实现完 Retriever怎么用最简单的方式retriever WikiRetriever(top_k5) # 直接用 docs retriever.invoke(如何发起一次版本发布) for d in docs: print(d.metadata[title], d.metadata[score])如果你在用 LangChain 的create_retrieval_chain或create_history_aware_retriever这个自定义 Retriever 可以直接替换原生的向量检索器因为框架只依赖BaseRetriever这个抽象接口。from langchain.chains import create_retrieval_chain from langchain.chains.combine_documents import create_stuff_documents_chain from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) combine_chain create_stuff_documents_chain(llm, prompt) rag_chain create_retrieval_chain(retriever, combine_chain)这一步能跑通说明你的自定义 Retriever 已经成功插入了 RAG 链路。此时你再回头看开头的问题——接自己的知识库本质上就是在这个插槽里写你的检索逻辑。3.4 如果请求参数里要带用户上下文怎么办真实场景里检索接口往往不只有q和size还需要带用户 ID、部门、权限范围。比如 Wiki 接口要求GET /api/search?qxxxuser_idu_10086teamcore_platformLangChain 的_get_relevant_documents有一个**kwargs参数它可以透传调用链里的额外参数。所以你可以这样做def _get_relevant_documents( self, query, *, run_manager, **kwargs ) - List[Document]: # 从调用侧传入 user_id kwargs.get(user_id, default_user) team kwargs.get(team, public) params {q: query, size: self.top_k, user_id: user_id, team: team} ...调用时docs retriever.invoke(发布流程, config{user_id: u_10086, team: core_platform})这个能力很重要因为知识库检索一旦涉及到权限你就必须在检索这层把上下文带进去而不是等检索完再做文档级过滤。后面我会单独展开讲。4. 接私有知识库绕不开的三个非标问题4.1 结构化数据如何把数据库记录变成人话文本很多知识库不只是文档还有数据库里的结构化记录比如工单、需求单、故障记录。这些数据的原始形态是表格行直接丢给 LLM 会非常难用。我的做法是在 Retriever 里把每条记录格式化成一句完整的人话描述而不是原样返回 JSON。比如工单记录有app_name、env、error_type、summary、status我会拼成工单#T-20250601-001状态已解决 应用order-service 环境prod 错误类型TimeoutException 描述订单查询接口在高峰期出现大量超时... 处理方案增加缓存层优化DB连接池...为什么这么做因为 LLM 理解自然语言远比理解散落的 JSON 字段更可靠。JSON 字段之间缺少显式的语义关系而一个完整的句子已经把上下文串联起来了。这一步属于看不见的预处理但提升效果非常明显。4.2 权限过滤检索前过滤别等返回后过滤知识库最麻烦的不是找不到而是不该找的也找到了。如果你在 Retriever 返回之后才做权限过滤假设前面命中了 20 篇其中 15 篇用户没权限你等于浪费了 3/4 的检索资源而且返回结果可能因为过滤后数量不足而变稀。正确做法是在检索请求里带上权限参数从源头过滤。如果你的知识库接口不支持权限参数那至少在 Retriever 内部拿到结果后、封装成 Document 之前做过滤allowed_ids kwargs.get(allowed_doc_ids) # 用户可访问的文档ID集合 if allowed_ids is not None: items [it for it in items if it[id] in allowed_ids]这里还有个容易被忽略的点不要只过滤正文忘了过滤 metadata。如果你把source、author、department这些信息留在了 Document 的 metadata 里检索结果是过滤了但敏感信息仍然可能通过 metadata 泄露给 LLM 或日志系统。稳妥的做法是过滤后重建 Document只保留允许的 metadata 字段。4.3 接口挂了你的 RAG 也就挂了超时和降级策略内部知识库接口的稳定性通常没有你想象的那么高。尤其是那种由别的团队维护、不属于你管控范围的接口随时可能因为上游数据库抖动、服务发版、网络闪断而变慢或报错。Retriever 一旦抛异常整个 RAG 链路就会失败。所以自定义 Retriever 里必须设计容错策略通常分三层超时、重试、降级。超时requests的timeout参数一定要显式设置不设等于无限等待一个慢接口能拖垮你整个服务。重试对 5xx、网络异常这种瞬时错误对同一个 query 重试 1-2 次通常能解决。降级如果重试还失败是让整个请求报错还是返回一个兜底结果我的经验是对 C 端体验要求高的场景宁可返回知识库暂时不可用的提示也不要让 RAG 直接崩溃。对一些低成本场景可以降级到本地关键词匹配或者最近缓存保证有东西可答。for attempt in range(retry_times): try: resp session.get(...) resp.raise_for_status() break except requests.RequestException: if attempt retry_times - 1: # 最后一轮还失败走降级 return self._fallback_search(query)5. 调优实录从能跑到好用我踩过的几个坑5.1 召回结果的 score 不能直接跨源比较这是自定义 Retriever 最常见的坑。当你开始做多路召回比如自定义 Wiki 接口 向量库 数据库全文索引你会发现每一路返回的 score 衡量标准完全不同。Wiki 接口返回的可能是 Elasticsearch 的相关性得分0~3 分之间或更高向量库返回的是余弦相似度通常是 -1 到 1数据库全文索引又可能是词频统计分。直接把这些分数拿来排序是错的。你必须先做归一化。两种常用的方案一个是 Min-Max 归一化把每路分数压到 0~1另一个是 Rank-based 归一化不看具体分数只看排序位置把第 i 名的得分定为1 - i/top_k。后者更稳因为它不依赖分数本身是绝对可比的。举个例子向量检索返回的第一名相似度是 0.82第三名是 0.80差距很小但 Wiki 接口第一名和第二名的得分差了一倍。如果直接按原始分数混合排序Wiki 的结果会永远压过向量检索的结果这明显是错的。用排序位置转换后每一路的高低位次才有统一的比较基础。5.2 RAG 效果不佳时先查召回别急着换模型我见过不少人 RAG 答不好就怪 LLM 不行然后从小模型换到大模型成本上去了效果却没改善。我的经验是大多数问题出在 Retriever 返回的上下文质量太差比如返回了错误的文档、缺少关键信息、文档数目不够、切块太碎导致上下文语义不完整。我调优时有一个固定套路先单独跑 Retriever看召回结果。把用户问题单独丢给 Retriever打印 top-5 的标题和片段人眼判断是不是相关。如果检索出来的文档本身就不对那后面给 LLM 什么提示词都白搭。这一步排查成本极低但能定位 80% 的效果问题。5.3 用日志把每次检索的输入输出完整记录下来线上环境不像本地调试那么直观所以日志就是你唯一的眼睛。我的习惯是在_get_relevant_documents里把 query、参数、返回条数、每条的source和score、耗时都打印出来格式化成结构化日志。import time start time.perf_counter() ... # 检索逻辑 elapsed time.perf_counter() - start if run_manager: run_manager.on_text( fWikiRetriever: query{query!r}, found{len(docs)}, ftime{elapsed:.3f}s, top1{docs[0].metadata[title] if docs else None} )这样一旦线上用户反馈某个问题答得不对你能直接从日志回放用户问了什么、检索到了什么快速判断是 Retriever 的问题还是 LLM 的问题而不是瞎猜。5.4 多模态内容的另类知识库处理有读者问过RAG 知识库能存图片吗这类问题。我的回答是图片可以存但能不能被检索到取决于你怎么描述它。绝大多数 RAG 系统并不具备图片语义检索能力与其期望 Retriever 直接理解图片不如把图片转成可检索的文本描述OCR 提取文字、图片标题、替代文本、附近正文内容一并拼进 Document 的 page_content 里这样图片信息就能被检索和喂给 LLM。如果你要做真正意义上的图片问答那就得走多模态模型路线超出了 Retriever 的职责范围但思路是通的先让 Retriever 能召回图片相关的文本上下文。6. 把自定义 Retriever 当积木多路召回、重排与缓存6.1 组合多个 Retriever混合检索的正确姿势自定义 Retriever 本质上是积木你可以把多个积木拼在一起。业界常说的混合检索Hybrid Search最常见就是向量检索 关键词检索 自定义业务检索的组合各自跑一遍再合并结果。以 LangChain 为例EnsembleRetriever可以帮你做这个事from langchain.retrievers import EnsembleRetriever from langchain_community.vectorstores import FAISS # 一路是向量检索 vector_retriever vectorstore.as_retriever(search_kwargs{k: 5}) # 一路是自己的 Wiki 接口 wiki_retriever WikiRetriever(top_k5) # 还可以一路是 BM25 关键词检索 bm25_retriever BM25Retriever.from_documents(docs) ensemble_retriever EnsembleRetriever( retrievers[vector_retriever, wiki_retriever, bm25_retriever], weights[0.4, 0.4, 0.2], )为什么这么做因为不同知识源的语义空间不同向量检索擅长语义相似关键词检索擅长精确命中Wiki 接口则可能已经内置了业务排序逻辑。三者互补召回质量比任何单一路都稳。要注意weights的配比。我一开始图简单设成均分效果并不好。原因是向量检索和 Wiki 接口的召回重叠率高而 BM25 召回的往往是长尾关键词。后来我通过实际数据分析统计每一路的 de-dup 后独立贡献数调整成 0.4/0.4/0.2效果才好一些。配比没有标准答案必须基于你的数据分布去调。6.2 召回之后加一道 Rerank效果立竿见影如果你预算允许强烈建议在多路召回之后加一个 Reranker。Reranker 通常是一个 cross-encoder 模型把query 文档当成一个整体做相关性打分比单独把 query 和文档分别编码再做向量相似度的双塔模型更准。常用的有bge-reranker-v2-m3也有基于 LLM 的 rerank API。它们的通病是慢、贵所以只在召回阶段先粗筛比如 top-20再用 Reranker 精排取 top-5成本可控。流程就是自定义 Retriever 负责召回候选 → 合并去重 → Reranker 精排 → 把最终 top-k 交给 LLM。这一步往往能让你的 RAG 效果从勉强能用提升到明显好用。6.3 缓存省钱又省延迟但要注意时效性如果你对同一批 query 反复检索会发现 embedding API 的费用和接口延迟在蹭蹭涨。加一层缓存非常管用。主流的缓存维度有两个一个是query 到检索结果的缓存适合高频重复问题另一个是embedding 的缓存避免同样的句子反复调用 embedding 服务。LangChain 有现成的CacheBackedEmbeddings检索结果级缓存需要自己维护。这里有一个时效性问题内部知识库一旦更新缓存的结果就过期了。我的做法是缓存很短比如 5 分钟~30 分钟并且对实时性要求高的知识源比如工单状态、发布状态不做缓存或做极短缓存。千万别为了省成本把用户查到的都是旧信息那才是最大的坑。我自己接过的项目里有一类让我印象很深的地方自定义 Retriever 写起来不难真正难的是你能否把它当成一个检索服务认真对待——权限、稳定性、并发、监控、降级每一样都是工程问题。等这层做扎实了你会发现换知识源、加知识源、调召回策略都变成改配置的事RAG 系统的骨架才算真正立住了。最后分享一个小经验接第一个自定义 Retriever 时我犯过一次过度设计的错——把降级、重排、缓存全堆上了结果排查问题时根本无法定位是哪一层出了问题。后来学乖了第一版只做最朴素的接口调用 文档封装跑通链路后再一层层加上去。你如果正打算写自己的 Retriever我建议也按照这个节奏来先能跑再调优最后再谈架构。这样每一步的效果都能看得见踩坑时也更容易定位问题在哪里。