1. 多租户 RAG 到底在解决什么问题做过企业级知识库的人都有一个共识单用户场景下的 RAG 是玩具多租户场景下的 RAG 才是产品。我最早接触 RAG 是在一个内部文档问答项目里当时所有文档放在一个向量库里谁都能搜到所有内容跑得挺欢。后来要把它交付给多个部门共用问题立刻炸了——A 部门的合同模板被 B 部门的人搜出来了C 员工的绩效文档出现在 D 员工的检索结果里。这不是技术 bug这是架构缺陷。多租户 RAG 的核心命题只有一句话让每个用户只检索到自己的知识库。听起来简单做起来涉及数据隔离、检索过滤、权限校验、索引设计、性能优化一整条链路。热搜词里出现的“dify 社区版 1.10 多租户”“ruoyi 多租户漏洞”“citus 多租户”这些本质上都是同一个问题的不同侧面——当一套 RAG 系统要服务多个互不可见的租户时怎么保证数据不串、性能不塌、权限不漏。这篇文章适合三类人看正在把单租户 RAG 改造成多租户的工程师、准备从零搭建企业知识库的架构师、以及被“检索串数据”问题折磨过的开发者。我会从架构设计讲到代码实现从向量库选型讲到 user_id 过滤的坑把这条链路上能踩的坑基本都铺开讲一遍。读完你至少能拿到一套可直接复现的多租户 RAG 方案而不是停留在“加个 where 条件”的想当然层面。先明确一个概念边界。多租户 RAG 里的“租户”可以是企业、部门、团队也可以是单个用户。粒度不同隔离策略完全不同。企业级隔离通常走独立 collection 或独立库用户级隔离走元数据过滤。热搜词里的“user_id”就是用户级隔离的核心字段。选哪种粒度取决于你的数据敏感度和规模后面会详细拆。2. 多租户隔离的三种架构路线与选型逻辑2.1 物理隔离独立库或独立 collection最粗暴也最安全的方案每个租户一套独立的向量库实例或独立的 collection。租户 A 的数据在 collection_a租户 B 的数据在 collection_b检索时根本不会跨 collection 查询从物理层面杜绝了数据串扰。这种方案的优势非常明显隔离性最强一个租户的索引损坏不影响其他人权限模型简单不需要在查询里拼过滤条件性能可预测每个租户的检索延迟互不干扰。缺点是资源开销大向量库的连接数、内存占用随租户数量线性增长。我实测过Milvus 单实例撑几百个 collection 没问题但上千个 collection 后元数据管理会开始吃力。适合场景租户数量在几十到几百、数据敏感度极高比如法务、医疗、金融、租户之间绝对不能有任何数据可见性交叉。热搜词里的“himmpat 专利检索网站”这类场景就属于典型的高敏感度专利数据往往涉及商业机密物理隔离是首选。2.2 逻辑隔离共享 collection 元数据过滤这是目前最主流的方案也是 dify、langchain4j 这类框架默认推荐的路线。所有租户的数据存在同一个 collection 里每条向量记录带一个tenant_id或user_id元数据字段检索时在查询条件里加上filter: {user_id: xxx}。这种方案资源利用率高一套索引服务所有租户运维成本低。但它的风险点在于过滤条件一旦漏写或写错就是全量数据泄露。热搜词里“ruoyi 多租户漏洞”大概率就是这类问题——某个查询接口忘了带租户过滤条件导致越权访问。逻辑隔离的安全性完全依赖于代码的严谨性没有任何物理兜底。适合场景租户数量大上千甚至上万、数据敏感度中等、团队有完善的代码审查和测试覆盖。SaaS 类产品基本都走这条路。2.3 混合隔离分层设计实际生产里我用得最多的是混合方案。把租户分成两类VIP 租户或高敏感租户走物理隔离普通租户走逻辑隔离。或者按数据层级分租户的私有知识库走独立 collection租户间的共享知识库走公共 collection 加过滤。这种方案兼顾了安全和成本但架构复杂度最高需要一套租户路由层来决定查询打到哪个 collection。我一般会在网关层做这个路由根据请求里的 tenant_id 查配置表拿到该租户的存储策略再转发到对应的检索服务。三种方案的对比我整理成表格方便你直接对照选型维度物理隔离逻辑隔离混合隔离隔离强度最强依赖代码可调资源开销高低中运维复杂度中低高适用租户规模几十到几百上千到上万任意典型风险资源耗尽过滤漏写路由错误代表场景专利、医疗SaaS 知识库大型企业平台选型的核心判断依据是数据泄露的代价有多大。如果泄露一次就是事故别犹豫物理隔离。如果泄露代价可控且有完善的监控告警逻辑隔离的性价比最高。3. 核心实现user_id 过滤链路的关键细节3.1 数据写入阶段就要绑定租户身份很多人以为多租户隔离是检索时的事其实从数据写入那一刻就开始了。每条文档切片在入库时必须把user_id或tenant_id写进元数据而且要保证这个字段不可被后续操作篡改。以 LangChain 的写法为例切分文档后给每个 chunk 附加元数据from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document def build_tenant_documents(raw_text, user_id, doc_id): splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , ] ) chunks splitter.split_text(raw_text) documents [] for idx, chunk in enumerate(chunks): documents.append(Document( page_contentchunk, metadata{ user_id: user_id, doc_id: doc_id, chunk_index: idx, source: f{user_id}/{doc_id} } )) return documents这里有个细节值得说user_id我建议用字符串而不是整型。整型在某些向量库的过滤表达式里会有类型推断问题字符串更稳。另外source字段拼上 user_id 前缀方便后续排查问题时快速定位归属。注意元数据字段一旦写入不要提供任何允许用户修改user_id的接口。我见过一个项目更新文档的接口允许传入完整 metadata结果有人把别人的 user_id 改成自己的直接实现了数据窃取。写入和更新接口必须从服务端会话里取 user_id绝不信任客户端传参。3.2 检索阶段的过滤表达式写法检索时加过滤条件不同向量库的语法不一样这是最容易出错的地方。我逐个说。Milvus 的过滤表达式search_params {metric_type: COSINE, params: {nprobe: 10}} results collection.search( data[query_vector], anns_fieldembedding, paramsearch_params, limit10, exprfuser_id {user_id}, output_fields[doc_id, chunk_index, source] )注意expr里的字符串要用双引号包裹user_id 本身如果含特殊字符要转义。我踩过一次坑user_id 里带了单引号表达式直接语法错误检索报 500。Qdrant 的过滤写法from qdrant_client.models import Filter, FieldCondition, MatchValue search_result client.search( collection_nameknowledge_base, query_vectorquery_vector, query_filterFilter( must[FieldCondition(keyuser_id, matchMatchValue(valueuser_id))] ), limit10 )Qdrant 的过滤是结构化的比字符串表达式安全不容易注入。如果你的向量库支持结构化过滤优先用结构化写法。Chroma 的 where 条件results collection.query( query_embeddings[query_vector], n_results10, where{user_id: user_id} )Chroma 的 where 支持$eq、$in、$and等操作符多租户场景下如果要做租户组共享可以用$in传一组 user_id。3.3 过滤字段的索引优化逻辑隔离方案里user_id过滤是每次查询都要执行的如果向量库没给这个字段建索引查询会退化成全量扫描后过滤性能灾难。Milvus 需要显式创建标量索引from pymilvus import Collection, utility collection Collection(knowledge_base) collection.create_index( field_nameuser_id, index_params{index_type: INVERTED} )INVERTED 索引适合低基数字段比如几百个租户如果 user_id 基数极高上万个用户可以考虑 STL_SORT。Qdrant 会自动为 payload 字段建索引但建议显式声明client.create_payload_index( collection_nameknowledge_base, field_nameuser_id, field_schemakeyword )实测下来加了标量索引后带 user_id 过滤的检索延迟从 800ms 降到 60ms 左右差距非常明显。这个优化在多租户场景里是必做的不是可选项。4. 完整实操从零搭一套多租户 RAG 检索服务4.1 环境准备与依赖安装我用 Python 生态来演示向量库选 Qdrant结构化过滤友好本地部署简单嵌入模型用 BGE-M3中文效果好支持多语言框架用 LangChain 做编排。pip install langchain langchain-community qdrant-client \ sentence-transformers fastapi uvicorn pydanticQdrant 本地起一个docker run -d --name qdrant \ -p 6333:6333 -p 6334:6334 \ -v $(pwd)/qdrant_storage:/qdrant/storage \ qdrant/qdrant:latest启动后访问 6333 端口能看到 Web UI说明服务正常。生产环境建议用集群模式单机版适合开发和中小规模。4.2 租户感知的向量库封装我习惯把多租户逻辑封装成一个类所有检索入口强制走这个类避免有人绕过过滤直接查库。from qdrant_client import QdrantClient from qdrant_client.models import ( Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue ) from sentence_transformers import SentenceTransformer import uuid class TenantAwareVectorStore: def __init__(self, collection_nameknowledge_base): self.client QdrantClient(hostlocalhost, port6333) self.collection_name collection_name self.encoder SentenceTransformer(BAAI/bge-m3) self._ensure_collection() def _ensure_collection(self): collections [c.name for c in self.client.get_collections().collections] if self.collection_name not in collections: self.client.create_collection( collection_nameself.collection_name, vectors_configVectorParams(size1024, distanceDistance.COSINE) ) self.client.create_payload_index( collection_nameself.collection_name, field_nameuser_id, field_schemakeyword ) def add_documents(self, user_id, doc_id, chunks): vectors self.encoder.encode(chunks, normalize_embeddingsTrue) points [] for idx, (chunk, vector) in enumerate(zip(chunks, vectors)): points.append(PointStruct( idstr(uuid.uuid4()), vectorvector.tolist(), payload{ user_id: user_id, doc_id: doc_id, chunk_index: idx, content: chunk } )) self.client.upsert( collection_nameself.collection_name, pointspoints ) return len(points) def search(self, user_id, query, top_k5): query_vector self.encoder.encode(query, normalize_embeddingsTrue) results self.client.search( collection_nameself.collection_name, query_vectorquery_vector.tolist(), query_filterFilter( must[FieldCondition( keyuser_id, matchMatchValue(valueuser_id) )] ), limittop_k, with_payloadTrue ) return [ { content: hit.payload[content], doc_id: hit.payload[doc_id], score: hit.score } for hit in results ]这个封装的关键点在于search方法的第一个参数强制是user_id调用方无法省略。这比在业务代码里到处拼过滤条件安全得多。我见过太多项目在 service 层拼过滤结果某个新接口忘了加直接出事。4.3 FastAPI 接口层与身份注入接口层要做的事很简单从认证信息里拿 user_id传给向量库封装绝不接受客户端传的 user_id。from fastapi import FastAPI, Depends, HTTPException, Header from pydantic import BaseModel from typing import List app FastAPI() store TenantAwareVectorStore() class SearchRequest(BaseModel): query: str top_k: int 5 class SearchResponse(BaseModel): results: List[dict] def get_current_user(authorization: str Header(...)) - str: # 实际项目里这里解析 JWT 或查 session # 演示用简单映射 token_map { token_alice: user_alice, token_bob: user_bob } user_id token_map.get(authorization) if not user_id: raise HTTPException(status_code401, detailinvalid token) return user_id app.post(/search, response_modelSearchResponse) def search(req: SearchRequest, user_id: str Depends(get_current_user)): results store.search(user_iduser_id, queryreq.query, top_kreq.top_k) return SearchResponse(resultsresults)注意get_current_user这个依赖注入它是整个多租户安全的守门人。所有涉及数据访问的接口都必须挂这个依赖user_id 从 token 解析客户端无法伪造。热搜词里“ruoyi 多租户漏洞”的根因往往就是某些接口没挂这个依赖或者挂了但内部逻辑又用了客户端传的 tenant_id。4.4 参数选择与性能调优几个关键参数我说明一下选择依据。chunk_size500是中文场景的经验值。中文一个字符约等于 1.5 个 token500 字符大概 750 token留足上下文空间。如果你的文档是技术手册这类信息密度高的可以降到 300如果是小说、新闻这类叙事性的可以升到 800。top_k5是检索返回条数。多租户场景下我建议不要设太大因为每个租户的数据量有限返回太多会引入噪声。实测 top_k 在 3 到 8 之间效果最好超过 10 后准确率反而下降。嵌入维度 1024 是 BGE-M3 的默认输出。如果你用 OpenAI 的 text-embedding-3-small维度是 1536需要相应调整VectorParams的 size。维度不匹配会直接报错这是新手常踩的坑。Qdrant 的 HNSW 索引参数m和ef_construct默认值分别是 16 和 100中小规模够用。如果租户数据量超过百万级可以把m调到 32ef_construct调到 200检索精度会提升但索引构建时间变长。5. 常见问题与排查技巧实录5.1 检索结果串租户的排查路径这是最严重的问题一旦出现必须立刻定位。排查顺序我总结成一张表排查点检查方法常见原因接口层看接口是否挂了身份依赖新接口漏挂过滤条件打印实际查询的 filter变量为空导致过滤失效元数据抽查库里的 payload写入时 user_id 为空索引确认标量索引存在索引未建导致过滤失效缓存检查是否有跨租户缓存缓存 key 没带 user_id我遇到过一次诡异的情况过滤条件明明写了但结果还是串。最后发现是 user_id 变量在某个分支里是空字符串而 Qdrant 的MatchValue(value)会匹配所有空值记录等于没过滤。修复方法是在封装层加断言def search(self, user_id, query, top_k5): if not user_id or not user_id.strip(): raise ValueError(user_id must not be empty) # ... 后续逻辑这个断言救过我两次强烈建议加上。5.2 性能随租户增长而下降逻辑隔离方案在租户数量增长后检索性能会下降因为 HNSW 索引要在大集合里搜索。优化手段有几个。第一给 user_id 建标量索引前面说过了这是基础。第二用 Qdrant 的hnsw_ef参数控制搜索精度和速度的平衡值越小越快但精度越低多租户场景下可以适当调低。第三如果单租户数据量很小比如几百条可以考虑用暴力搜索代替 HNSW反而更快。第四也是最有效的按租户分片。Qdrant 支持 collection 分片把不同租户的数据路由到不同分片检索时只查对应分片。这个需要在写入时指定 shard_key配置稍复杂但性能提升明显。5.3 租户删除与数据清理租户注销后数据必须彻底清理否则既占空间又有泄露风险。逻辑隔离方案下删除租户数据要按 user_id 过滤删除def delete_tenant_data(self, user_id): self.client.delete( collection_nameself.collection_name, points_selectorFilter( must[FieldCondition( keyuser_id, matchMatchValue(valueuser_id) )] ) )注意这个操作是异步的Qdrant 会在后台执行。如果租户数据量大删除可能耗时较久。我建议在业务层做软删除标记先让租户不可见再异步清理物理数据。提示删除操作一定要加二次确认和审计日志。我见过误删整个 collection 的事故原因是删除接口的过滤条件传了空值结果匹配了所有记录。删除接口的 user_id 必须做非空校验且建议加一个confirmtrue参数强制调用方确认。5.4 多轮对话场景下的租户上下文保持热搜词里有“rag 多轮对话怎么设计”这在多租户场景下有个额外注意点对话历史也必须按租户隔离。如果对话历史存在 Redis 里key 必须带 user_iddef get_conversation_key(user_id, session_id): return fconv:{user_id}:{session_id}我见过一个项目对话历史 key 只用 session_id结果两个用户碰巧 session_id 相同前端生成的随机数碰撞互相看到了对方的对话。这种问题概率低但一旦发生就是事故key 里带上 user_id 是零成本的保险。5.5 嵌入模型的租户隔离误区有人会想能不能给每个租户用不同的嵌入模型来实现隔离技术上可行但没必要而且会带来模型加载的内存开销。嵌入模型是共享的隔离靠的是向量库的元数据过滤不是模型。这个认知要摆正否则会走弯路。6. 我踩过的坑和几条实战建议第一个坑是过滤字段类型不一致。写入时 user_id 是字符串某次查询时从数据库读出来是整型Qdrant 的 MatchValue 类型不匹配过滤直接失效返回了全量数据。修复方法是在封装层统一做类型转换所有 user_id 强制转字符串。第二个坑是批量写入时元数据丢失。用 LangChain 的add_documents批量写入时如果 documents 列表里混入了没有 user_id 元数据的文档这些文档会变成“无主数据”任何租户都搜不到但也清理不掉。我的做法是在写入前做一次校验过滤掉元数据不完整的文档并记录告警。第三个坑是测试环境用同一个 user_id。开发时图省事所有测试都用user_test结果上线后才发现多租户逻辑有问题。后来我强制要求测试用例必须至少用两个不同的 user_id且要有一个“越权检索”的测试用例验证 A 用户搜不到 B 用户的数据。几条实战建议把 user_id 过滤做成向量库封装的强制参数而不是可选参数从 API 设计上杜绝漏写。加一个定时任务定期扫描库里 user_id 为空或格式异常的记录及时清理。监控每个租户的检索 QPS 和延迟异常租户单独告警防止某个租户的数据量拖垮整体性能。上线前做一次渗透测试专门尝试用各种方式越权检索包括改 token、改参数、构造特殊字符等。这套方案我在两个项目里落地过一个服务几百个企业租户一个服务上万个个人用户跑下来都挺稳。核心就一句话隔离不是加个过滤条件那么简单而是从写入、存储、检索、删除全链路的身份贯穿。任何一环松了整个隔离就形同虚设。
多租户RAG架构实战：从隔离方案到user_id过滤全链路
1. 多租户 RAG 到底在解决什么问题做过企业级知识库的人都有一个共识单用户场景下的 RAG 是玩具多租户场景下的 RAG 才是产品。我最早接触 RAG 是在一个内部文档问答项目里当时所有文档放在一个向量库里谁都能搜到所有内容跑得挺欢。后来要把它交付给多个部门共用问题立刻炸了——A 部门的合同模板被 B 部门的人搜出来了C 员工的绩效文档出现在 D 员工的检索结果里。这不是技术 bug这是架构缺陷。多租户 RAG 的核心命题只有一句话让每个用户只检索到自己的知识库。听起来简单做起来涉及数据隔离、检索过滤、权限校验、索引设计、性能优化一整条链路。热搜词里出现的“dify 社区版 1.10 多租户”“ruoyi 多租户漏洞”“citus 多租户”这些本质上都是同一个问题的不同侧面——当一套 RAG 系统要服务多个互不可见的租户时怎么保证数据不串、性能不塌、权限不漏。这篇文章适合三类人看正在把单租户 RAG 改造成多租户的工程师、准备从零搭建企业知识库的架构师、以及被“检索串数据”问题折磨过的开发者。我会从架构设计讲到代码实现从向量库选型讲到 user_id 过滤的坑把这条链路上能踩的坑基本都铺开讲一遍。读完你至少能拿到一套可直接复现的多租户 RAG 方案而不是停留在“加个 where 条件”的想当然层面。先明确一个概念边界。多租户 RAG 里的“租户”可以是企业、部门、团队也可以是单个用户。粒度不同隔离策略完全不同。企业级隔离通常走独立 collection 或独立库用户级隔离走元数据过滤。热搜词里的“user_id”就是用户级隔离的核心字段。选哪种粒度取决于你的数据敏感度和规模后面会详细拆。2. 多租户隔离的三种架构路线与选型逻辑2.1 物理隔离独立库或独立 collection最粗暴也最安全的方案每个租户一套独立的向量库实例或独立的 collection。租户 A 的数据在 collection_a租户 B 的数据在 collection_b检索时根本不会跨 collection 查询从物理层面杜绝了数据串扰。这种方案的优势非常明显隔离性最强一个租户的索引损坏不影响其他人权限模型简单不需要在查询里拼过滤条件性能可预测每个租户的检索延迟互不干扰。缺点是资源开销大向量库的连接数、内存占用随租户数量线性增长。我实测过Milvus 单实例撑几百个 collection 没问题但上千个 collection 后元数据管理会开始吃力。适合场景租户数量在几十到几百、数据敏感度极高比如法务、医疗、金融、租户之间绝对不能有任何数据可见性交叉。热搜词里的“himmpat 专利检索网站”这类场景就属于典型的高敏感度专利数据往往涉及商业机密物理隔离是首选。2.2 逻辑隔离共享 collection 元数据过滤这是目前最主流的方案也是 dify、langchain4j 这类框架默认推荐的路线。所有租户的数据存在同一个 collection 里每条向量记录带一个tenant_id或user_id元数据字段检索时在查询条件里加上filter: {user_id: xxx}。这种方案资源利用率高一套索引服务所有租户运维成本低。但它的风险点在于过滤条件一旦漏写或写错就是全量数据泄露。热搜词里“ruoyi 多租户漏洞”大概率就是这类问题——某个查询接口忘了带租户过滤条件导致越权访问。逻辑隔离的安全性完全依赖于代码的严谨性没有任何物理兜底。适合场景租户数量大上千甚至上万、数据敏感度中等、团队有完善的代码审查和测试覆盖。SaaS 类产品基本都走这条路。2.3 混合隔离分层设计实际生产里我用得最多的是混合方案。把租户分成两类VIP 租户或高敏感租户走物理隔离普通租户走逻辑隔离。或者按数据层级分租户的私有知识库走独立 collection租户间的共享知识库走公共 collection 加过滤。这种方案兼顾了安全和成本但架构复杂度最高需要一套租户路由层来决定查询打到哪个 collection。我一般会在网关层做这个路由根据请求里的 tenant_id 查配置表拿到该租户的存储策略再转发到对应的检索服务。三种方案的对比我整理成表格方便你直接对照选型维度物理隔离逻辑隔离混合隔离隔离强度最强依赖代码可调资源开销高低中运维复杂度中低高适用租户规模几十到几百上千到上万任意典型风险资源耗尽过滤漏写路由错误代表场景专利、医疗SaaS 知识库大型企业平台选型的核心判断依据是数据泄露的代价有多大。如果泄露一次就是事故别犹豫物理隔离。如果泄露代价可控且有完善的监控告警逻辑隔离的性价比最高。3. 核心实现user_id 过滤链路的关键细节3.1 数据写入阶段就要绑定租户身份很多人以为多租户隔离是检索时的事其实从数据写入那一刻就开始了。每条文档切片在入库时必须把user_id或tenant_id写进元数据而且要保证这个字段不可被后续操作篡改。以 LangChain 的写法为例切分文档后给每个 chunk 附加元数据from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.schema import Document def build_tenant_documents(raw_text, user_id, doc_id): splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , ] ) chunks splitter.split_text(raw_text) documents [] for idx, chunk in enumerate(chunks): documents.append(Document( page_contentchunk, metadata{ user_id: user_id, doc_id: doc_id, chunk_index: idx, source: f{user_id}/{doc_id} } )) return documents这里有个细节值得说user_id我建议用字符串而不是整型。整型在某些向量库的过滤表达式里会有类型推断问题字符串更稳。另外source字段拼上 user_id 前缀方便后续排查问题时快速定位归属。注意元数据字段一旦写入不要提供任何允许用户修改user_id的接口。我见过一个项目更新文档的接口允许传入完整 metadata结果有人把别人的 user_id 改成自己的直接实现了数据窃取。写入和更新接口必须从服务端会话里取 user_id绝不信任客户端传参。3.2 检索阶段的过滤表达式写法检索时加过滤条件不同向量库的语法不一样这是最容易出错的地方。我逐个说。Milvus 的过滤表达式search_params {metric_type: COSINE, params: {nprobe: 10}} results collection.search( data[query_vector], anns_fieldembedding, paramsearch_params, limit10, exprfuser_id {user_id}, output_fields[doc_id, chunk_index, source] )注意expr里的字符串要用双引号包裹user_id 本身如果含特殊字符要转义。我踩过一次坑user_id 里带了单引号表达式直接语法错误检索报 500。Qdrant 的过滤写法from qdrant_client.models import Filter, FieldCondition, MatchValue search_result client.search( collection_nameknowledge_base, query_vectorquery_vector, query_filterFilter( must[FieldCondition(keyuser_id, matchMatchValue(valueuser_id))] ), limit10 )Qdrant 的过滤是结构化的比字符串表达式安全不容易注入。如果你的向量库支持结构化过滤优先用结构化写法。Chroma 的 where 条件results collection.query( query_embeddings[query_vector], n_results10, where{user_id: user_id} )Chroma 的 where 支持$eq、$in、$and等操作符多租户场景下如果要做租户组共享可以用$in传一组 user_id。3.3 过滤字段的索引优化逻辑隔离方案里user_id过滤是每次查询都要执行的如果向量库没给这个字段建索引查询会退化成全量扫描后过滤性能灾难。Milvus 需要显式创建标量索引from pymilvus import Collection, utility collection Collection(knowledge_base) collection.create_index( field_nameuser_id, index_params{index_type: INVERTED} )INVERTED 索引适合低基数字段比如几百个租户如果 user_id 基数极高上万个用户可以考虑 STL_SORT。Qdrant 会自动为 payload 字段建索引但建议显式声明client.create_payload_index( collection_nameknowledge_base, field_nameuser_id, field_schemakeyword )实测下来加了标量索引后带 user_id 过滤的检索延迟从 800ms 降到 60ms 左右差距非常明显。这个优化在多租户场景里是必做的不是可选项。4. 完整实操从零搭一套多租户 RAG 检索服务4.1 环境准备与依赖安装我用 Python 生态来演示向量库选 Qdrant结构化过滤友好本地部署简单嵌入模型用 BGE-M3中文效果好支持多语言框架用 LangChain 做编排。pip install langchain langchain-community qdrant-client \ sentence-transformers fastapi uvicorn pydanticQdrant 本地起一个docker run -d --name qdrant \ -p 6333:6333 -p 6334:6334 \ -v $(pwd)/qdrant_storage:/qdrant/storage \ qdrant/qdrant:latest启动后访问 6333 端口能看到 Web UI说明服务正常。生产环境建议用集群模式单机版适合开发和中小规模。4.2 租户感知的向量库封装我习惯把多租户逻辑封装成一个类所有检索入口强制走这个类避免有人绕过过滤直接查库。from qdrant_client import QdrantClient from qdrant_client.models import ( Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue ) from sentence_transformers import SentenceTransformer import uuid class TenantAwareVectorStore: def __init__(self, collection_nameknowledge_base): self.client QdrantClient(hostlocalhost, port6333) self.collection_name collection_name self.encoder SentenceTransformer(BAAI/bge-m3) self._ensure_collection() def _ensure_collection(self): collections [c.name for c in self.client.get_collections().collections] if self.collection_name not in collections: self.client.create_collection( collection_nameself.collection_name, vectors_configVectorParams(size1024, distanceDistance.COSINE) ) self.client.create_payload_index( collection_nameself.collection_name, field_nameuser_id, field_schemakeyword ) def add_documents(self, user_id, doc_id, chunks): vectors self.encoder.encode(chunks, normalize_embeddingsTrue) points [] for idx, (chunk, vector) in enumerate(zip(chunks, vectors)): points.append(PointStruct( idstr(uuid.uuid4()), vectorvector.tolist(), payload{ user_id: user_id, doc_id: doc_id, chunk_index: idx, content: chunk } )) self.client.upsert( collection_nameself.collection_name, pointspoints ) return len(points) def search(self, user_id, query, top_k5): query_vector self.encoder.encode(query, normalize_embeddingsTrue) results self.client.search( collection_nameself.collection_name, query_vectorquery_vector.tolist(), query_filterFilter( must[FieldCondition( keyuser_id, matchMatchValue(valueuser_id) )] ), limittop_k, with_payloadTrue ) return [ { content: hit.payload[content], doc_id: hit.payload[doc_id], score: hit.score } for hit in results ]这个封装的关键点在于search方法的第一个参数强制是user_id调用方无法省略。这比在业务代码里到处拼过滤条件安全得多。我见过太多项目在 service 层拼过滤结果某个新接口忘了加直接出事。4.3 FastAPI 接口层与身份注入接口层要做的事很简单从认证信息里拿 user_id传给向量库封装绝不接受客户端传的 user_id。from fastapi import FastAPI, Depends, HTTPException, Header from pydantic import BaseModel from typing import List app FastAPI() store TenantAwareVectorStore() class SearchRequest(BaseModel): query: str top_k: int 5 class SearchResponse(BaseModel): results: List[dict] def get_current_user(authorization: str Header(...)) - str: # 实际项目里这里解析 JWT 或查 session # 演示用简单映射 token_map { token_alice: user_alice, token_bob: user_bob } user_id token_map.get(authorization) if not user_id: raise HTTPException(status_code401, detailinvalid token) return user_id app.post(/search, response_modelSearchResponse) def search(req: SearchRequest, user_id: str Depends(get_current_user)): results store.search(user_iduser_id, queryreq.query, top_kreq.top_k) return SearchResponse(resultsresults)注意get_current_user这个依赖注入它是整个多租户安全的守门人。所有涉及数据访问的接口都必须挂这个依赖user_id 从 token 解析客户端无法伪造。热搜词里“ruoyi 多租户漏洞”的根因往往就是某些接口没挂这个依赖或者挂了但内部逻辑又用了客户端传的 tenant_id。4.4 参数选择与性能调优几个关键参数我说明一下选择依据。chunk_size500是中文场景的经验值。中文一个字符约等于 1.5 个 token500 字符大概 750 token留足上下文空间。如果你的文档是技术手册这类信息密度高的可以降到 300如果是小说、新闻这类叙事性的可以升到 800。top_k5是检索返回条数。多租户场景下我建议不要设太大因为每个租户的数据量有限返回太多会引入噪声。实测 top_k 在 3 到 8 之间效果最好超过 10 后准确率反而下降。嵌入维度 1024 是 BGE-M3 的默认输出。如果你用 OpenAI 的 text-embedding-3-small维度是 1536需要相应调整VectorParams的 size。维度不匹配会直接报错这是新手常踩的坑。Qdrant 的 HNSW 索引参数m和ef_construct默认值分别是 16 和 100中小规模够用。如果租户数据量超过百万级可以把m调到 32ef_construct调到 200检索精度会提升但索引构建时间变长。5. 常见问题与排查技巧实录5.1 检索结果串租户的排查路径这是最严重的问题一旦出现必须立刻定位。排查顺序我总结成一张表排查点检查方法常见原因接口层看接口是否挂了身份依赖新接口漏挂过滤条件打印实际查询的 filter变量为空导致过滤失效元数据抽查库里的 payload写入时 user_id 为空索引确认标量索引存在索引未建导致过滤失效缓存检查是否有跨租户缓存缓存 key 没带 user_id我遇到过一次诡异的情况过滤条件明明写了但结果还是串。最后发现是 user_id 变量在某个分支里是空字符串而 Qdrant 的MatchValue(value)会匹配所有空值记录等于没过滤。修复方法是在封装层加断言def search(self, user_id, query, top_k5): if not user_id or not user_id.strip(): raise ValueError(user_id must not be empty) # ... 后续逻辑这个断言救过我两次强烈建议加上。5.2 性能随租户增长而下降逻辑隔离方案在租户数量增长后检索性能会下降因为 HNSW 索引要在大集合里搜索。优化手段有几个。第一给 user_id 建标量索引前面说过了这是基础。第二用 Qdrant 的hnsw_ef参数控制搜索精度和速度的平衡值越小越快但精度越低多租户场景下可以适当调低。第三如果单租户数据量很小比如几百条可以考虑用暴力搜索代替 HNSW反而更快。第四也是最有效的按租户分片。Qdrant 支持 collection 分片把不同租户的数据路由到不同分片检索时只查对应分片。这个需要在写入时指定 shard_key配置稍复杂但性能提升明显。5.3 租户删除与数据清理租户注销后数据必须彻底清理否则既占空间又有泄露风险。逻辑隔离方案下删除租户数据要按 user_id 过滤删除def delete_tenant_data(self, user_id): self.client.delete( collection_nameself.collection_name, points_selectorFilter( must[FieldCondition( keyuser_id, matchMatchValue(valueuser_id) )] ) )注意这个操作是异步的Qdrant 会在后台执行。如果租户数据量大删除可能耗时较久。我建议在业务层做软删除标记先让租户不可见再异步清理物理数据。提示删除操作一定要加二次确认和审计日志。我见过误删整个 collection 的事故原因是删除接口的过滤条件传了空值结果匹配了所有记录。删除接口的 user_id 必须做非空校验且建议加一个confirmtrue参数强制调用方确认。5.4 多轮对话场景下的租户上下文保持热搜词里有“rag 多轮对话怎么设计”这在多租户场景下有个额外注意点对话历史也必须按租户隔离。如果对话历史存在 Redis 里key 必须带 user_iddef get_conversation_key(user_id, session_id): return fconv:{user_id}:{session_id}我见过一个项目对话历史 key 只用 session_id结果两个用户碰巧 session_id 相同前端生成的随机数碰撞互相看到了对方的对话。这种问题概率低但一旦发生就是事故key 里带上 user_id 是零成本的保险。5.5 嵌入模型的租户隔离误区有人会想能不能给每个租户用不同的嵌入模型来实现隔离技术上可行但没必要而且会带来模型加载的内存开销。嵌入模型是共享的隔离靠的是向量库的元数据过滤不是模型。这个认知要摆正否则会走弯路。6. 我踩过的坑和几条实战建议第一个坑是过滤字段类型不一致。写入时 user_id 是字符串某次查询时从数据库读出来是整型Qdrant 的 MatchValue 类型不匹配过滤直接失效返回了全量数据。修复方法是在封装层统一做类型转换所有 user_id 强制转字符串。第二个坑是批量写入时元数据丢失。用 LangChain 的add_documents批量写入时如果 documents 列表里混入了没有 user_id 元数据的文档这些文档会变成“无主数据”任何租户都搜不到但也清理不掉。我的做法是在写入前做一次校验过滤掉元数据不完整的文档并记录告警。第三个坑是测试环境用同一个 user_id。开发时图省事所有测试都用user_test结果上线后才发现多租户逻辑有问题。后来我强制要求测试用例必须至少用两个不同的 user_id且要有一个“越权检索”的测试用例验证 A 用户搜不到 B 用户的数据。几条实战建议把 user_id 过滤做成向量库封装的强制参数而不是可选参数从 API 设计上杜绝漏写。加一个定时任务定期扫描库里 user_id 为空或格式异常的记录及时清理。监控每个租户的检索 QPS 和延迟异常租户单独告警防止某个租户的数据量拖垮整体性能。上线前做一次渗透测试专门尝试用各种方式越权检索包括改 token、改参数、构造特殊字符等。这套方案我在两个项目里落地过一个服务几百个企业租户一个服务上万个个人用户跑下来都挺稳。核心就一句话隔离不是加个过滤条件那么简单而是从写入、存储、检索、删除全链路的身份贯穿。任何一环松了整个隔离就形同虚设。
