企业官网建设流程全解析

做 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. 实战：写一个接自家知识库的混合召回 Retriever

3.1 场景设计：真实项目里的知识库往往不是单一存储

拿我自己做过的一个案例来说。团队有一个内部知识库，内容分成三块：历史工单、FAQ、内部文档。物理上它们分布在三个地方：

• 工单和 FAQ 的元数据、权限字段在 MySQL 里
• 正文经过 Embedding 后存在本地 FAISS 向量库
• 团队还有一个基于 Elasticsearch 的全文检索引擎，已经跑了好几年，很多搜索逻辑都在那边

当时的诉求是：RAG 项目要基于这套知识库做问答，但每次检索不能只看向量相似度，还要兼顾精确关键词、权限隔离和工单优先级。那直接用as_retriever()显然不行，我就在 LangChain 里自定义了一个TicketKnowledgeRetriever。

整体流程大致是这样：

query ├── 向量召回 (FAISS, top=30) ├── 关键词召回 (HTTP 搜索服务, top=20) └── 合并去重、加权排序 └── MySQL 权限过滤、补全元数据 └── 返回 top=5 的 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, k=self.top_k * 6 ) docs = [] for doc, score in docs_with_scores: d = Document( page_content=doc.page_content, metadata=dict(doc.metadata), # 拷贝，别污染原对象 ) d.metadata["_score"] = float(score) d.metadata["_source"] = "vector" docs.append(d) run_manager.on_text(f"vector 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}, timeout=10, ) resp.raise_for_status() results = resp.json().get("hits", []) docs = [] for item in results: doc = Document( page_content=item["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(f"keyword 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(key=lambda d: d.metadata["_score"], reverse=True) 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=..., metadata=dict(doc.metadata))重新构造了对象，这是顺手的事，但能避免很多诡异的 bug。

3.4 写完怎么验证这个 Retriever 真的可用

自测不需要很复杂，写一个小函数把召回结果打出来就够：

def debug_retriever(retriever, query: str): docs = retriever.invoke(query) print(f"query: {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": f"Bearer {self.token}"}, timeout=10, ) resp.raise_for_status() data = resp.json() documents = [] for item in data.get("documents", []): documents.append( Document( page_content=item["content"], metadata={ "title": item["title"], "url": item["url"], "_source": "wiki", }, ) ) return documents

对接外部接口时，我强烈建议不要做“拿回全部正文再自己重排”这种事。外部系统往往已经是成熟的知识库，有自己的相关性排序逻辑，你接入时先看它返回的字段能不能满足Document的基本要求，满足就直接包装。

4.2 超时、重试与限流必须一起设计

外部接口和本地向量库不一样，它可能慢、可能挂、可能有并发限制。Retriever 一旦成为 RAG 链路里的常规调用，你就得把它当成一个服务调用去治理。

超时是最基本的：httpx里可以直接指定timeout=10，别让它无限等。接着是重试：对 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(timeout=10) 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_content=item["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_compressor=compressor, base_retriever=my_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_factory=Lock) 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_content=d.page_content, metadata=dict(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 不会把它们当成出入参字段，适合放内部状态。多实例部署时这个内存缓存各存各的，如果想要更强一致，可以把缓存放到 Redis，key 用 query，value 存文档 id 列表，配合按 id 查询文档的接口，能省很多体量。

6. 常见问题排查速查表，以及上线前必做的三件事

6.1 报错与现象速查表

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 问题的时候，一半时间都能省在它身上。