多租户 RAG 与Agent系统的生产实践中，最致命的事故莫过于数据串租，系统将租户 B 的私有数据作为背景知识，回答了租户 A 的提问。针对这个问题，本文将深入分析串租发生的根源，并展示如何利用 Milvus 的 Partition Key 能力 进行物理隔离，同时引入 AutoRAG 自动评测框架，实现一整套完整的多租隔离验证机制。
一、串租是怎么发生的
串租的根本原因，通常有两种：第一，运行时风险（缺少租户过滤的物理边界）：查询时如果缺少严格的租户过滤，由于向量空间的连续性，如果 A 和 B 的文档语义相似，又没有 tenant_id 过滤，检索结果就可能跨租户混排。第二，验证缺失风险（缺乏持续监控）：即使代码中加了过滤逻辑，也不能默认它一直有效。模型、数据、索引、检索参数、Pipeline 配置都会变。每次变更后，如果没有自动化评测，就很难知道边界是否还在继续生效。也是因此，解决以上问题，需要我们从检索执行层（Milvus ）以及评测流程（AutoRAG)两手抓起。方法论总结如下：
二、物理隔离层：如何用好Milvus 的Partition Key
在多租户场景中，本文重点使用 Milvus 的 Partition Key。更多多租户场景的Milvus实战，可以参考 Milvus多租户实践：你的技术选型扛得住一夜爆火吗？在 Partition Key模式下，将 tenant_id 字段设为分区键后，Milvus 会在写入时对该值做 Hash 路由，数据落到对应物理分区；只要查询时携带过滤表达式，系统就会先收敛到对应分区，再做向量相似度搜索。目前，单个 Collection 支持最多 4096 个物理分区（默认 16 个），足以覆盖绝大多数多租户规模。
值得注意的是，Partition Key 不是权限系统，它不会自动替你判断当前用户属于哪个租户。真正的安全边界仍然依赖业务层从认证上下文中取出 tenantid，并在每次查询时强制注入过滤条件。也就是说：如果写入时有 tenantid 可以哈希路由到对应物理分区，不带 tenant_id 过滤时，会默认访问所有分区里的数据。
三、校验层：AutoRAG 如何做多租的自动化验证
AutoRAG 是一个 RAG 流水线自动评测与优化框架，其 核心架构分为三层：
数据准备层（Data Creation）：Parser 解析文档 → Chunker 切块 → QA Creator 生成评测集，输出标准的 corpus.parquet 和 qa.parquet。优化核心层（RAG Optimization）：通过 YAML 串联 Query Expansion、Retrieval、Reranker、Filter、Prompt Maker 和 Generator 等节点，并自动枚举最优组合：部署层（Deployment）：评测产出的最优 Pipeline 可直接部署为 Runner（代码调用）、REST API 或 Gradio Web UI。基于以上架构，AutoRAG有两个核心能力：第一个是 Pipeline 优化。用户可以在 YAML 里声明候选模块，比如检索器、重排器、生成器。AutoRAG 会自动枚举组合，评测不同配置，并找出效果最好的 Pipeline。第二个是可重复评测。评测流程配置好之后，每次换模型、改数据、调参数，都需要重新执行，并用同一套指标横向对比。本文主要用到的是第二个能力：把“多租户场景下检索结果是否可信”变成可以重复运行的评测流程。另外，值得一提的是，Milvus 在 AutoRAG 的 Retrieval 节点中是 原生一等公民。AutoRAG 的 vectordb 配置里，db_type: milvus 开箱即用，不需要任何适配代码。在 AutoRAG 的评测流水线里，Milvus 也可以直接作为检索后端参与评测，没有额外集成成本。
四、教程：从零构建多租户隔离与验证流水线
Step 1：准备环境 python3 -m venv .venvsource .venv/bin/activatepip install -U pippip install "autorag>=0.3" "pymilvus>=2.4.0" "openai" "pandas" "pyarrow"export OPENAIAPIKEY=sk-… #自行准备OpenAIAPIKEYexport MILVUSURI=http://127.0.0.1:19530 # Milvus Standalone 服务地址export MILVUSTOKEN="root:Milvus"部署 Milvus# 下载docker-compose.ymlwget https://github.com/milvus-io/milvus/releases/download/v2.6.8/milvus-standalone-docker-compose.yml -O docker-compose.yml# 启动Milvus（检查端口映射：19530:19530）docker-compose up -d# 验证服务启动docker ps | grep milvus# 应该看到3个容器：milvus-standalone, milvus-etcd, milvus-minio Step 2：准备 AutoRAG 标准格式数据 为了精准验证隔离是否生效，我们需要设计一种“相同提问（ q1 和 q2 是 文字完全相同的查询）、不同租户、不同答案”的高难度测试集。如果系统隔离失效，全库检索必然会将两个租户的答案混淆。
⚠️ 说明1：AutoRAG 对输入字段有严格的命名约定（如docid、retrievalgt等），写错会导致解析报错。
说明 2：以下 Step 2-4 中的 Python 代码块，各自保存为对应的.py文件后，在激活的虚拟环境中用python3 文件名.py执行。
文件 必须字段 常见错误写法
corpus.parquetdoc_id、contents、metadata❌ 写成text，或缺少metadata
qa.parquetqid、query、retrievalgt、generationgt❌ 写成question/answer，或缺少retrieval_gt
retrievalgt是检索标注字段，记录每条问题期望命中的docid列表，AutoRAG 用它计算 Recall / Precision。没有这个字段，评测无法运行。
import osimport pandas as pdos.makedirs("./data", existok=True)corpus = pd.DataFrame([ {"docid": "a-1", "contents": "A租户的报销规则：差旅上限为内部标准。", "metadata": {"tenantid": "tenanta"}, "tenantid": "tenanta"}, {"docid": "a-2", "contents": "A租户合同模板要求法务审批。", "metadata": {"tenantid": "tenanta"}, "tenantid": "tenanta"}, {"docid": "b-1", "contents": "B租户的报销规则：海外差旅需要二级审批。", "metadata": {"tenantid": "tenantb"}, "tenantid": "tenantb"}, {"docid": "b-2", "contents": "B租户合同模板要求采购会签。", "metadata": {"tenantid": "tenantb"}, "tenantid": "tenantb"},])qa = pd.DataFrame([ { "qid":"q1", "query": "报销规则里差旅审批要求是什么？", "retrievalgt": [["a-1"]], # List[List[str]]：期望命中的 docid 集合 "generationgt": ["A租户内部标准。"], # List[str]：可接受的参考答案 "tenantid": "tenanta", }, { "qid": "q2", "query": "报销规则里差旅审批要求是什么？", # 与 q1 文字完全相同的查询 "retrievalgt": [["b-1"]], "generationgt": ["B租户海外差旅需二级审批。"], "tenantid": "tenantb", },])corpus.toparquet("./data/corpus.parquet", index=False)qa.toparquet("./data/qa.parquet",index=False)Step 3：创建 Collection 并设置 Partition Key import osfrom pymilvus import MilvusClient, DataTypeclient = MilvusClient( uri=os.getenv("MILVUSURI", "http://127.0.0.1:19530"), token=os.getenv("MILVUSTOKEN", ""),)COLLECTION = "kbmultitenantpk"if client.hascollection(COLLECTION): client.dropcollection(COLLECTION)schema = client.createschema(autoid=False, enabledynamicfield=False)schema.addfield("pk",DataType.VARCHAR, isprimary=True,maxlength=64)schema.addfield("tenantid", DataType.VARCHAR, ispartitionkey=True, maxlength=64)schema.addfield("docid", DataType.VARCHAR, maxlength=64)schema.addfield("contents", DataType.VARCHAR, maxlength=2048)# text-embedding-3-small 默认输出1536维度schema.addfield("embedding", DataType.FLOATVECTOR, dim=1536)idx = client.prepareindexparams()idx.addindex(fieldname="embedding", indextype="AUTOINDEX", metrictype="COSINE")client.createcollection( collectionname=COLLECTION, schema=schema, indexparams=idx, numpartitions=16, # Partition Key 模式下的物理分区数，默认 16，最大 4096)print(f"✅ Collection '{COLLECTION}' created，Partition Key → tenantid")Step 4：生成 Embedding，写入 Milvus 这一步是 数据进入检索层的实际入口，也是 tenantid 被绑定到向量上的时机。import osimport pandas as pdfrom openai import OpenAIfrom pymilvus import MilvusClientopenaiclient = OpenAI()client = MilvusClient( uri=os.getenv("MILVUSURI", "http://127.0.0.1:19530"), token=os.getenv("MILVUSTOKEN", ""),)COLLECTION = "kbmultitenantpk"def embed(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]: resp = openaiclient.embeddings.create(input=texts, model=model) return [item.embedding for item in resp.data]corpusdf= pd.readparquet("./data/corpus.parquet")embeddings = embed(corpusdf["contents"].tolist())rows = [ { "pk": row["docid"], "tenantid": row["tenantid"],# Partition Key 字段，决定物理路由 "docid": row["docid"], "contents": row["contents"], "embedding": emb, } for (, row), emb in zip(corpusdf.iterrows(), embeddings)]client.insert(collectionname=COLLECTION, data=rows)client.flush(collectionname=COLLECTION)print(f"✅ Inserted {len(rows)} documents into Milvus")Step 5：配置 AutoRAG，执行评测 说明：AutoRAG 的 YAML 解析基于标准 PyYAML，不会自动展开${ENVVAR} 形式的环境变量。运行下方脚本先生成含真实值的配置文件，再执行评测命令。import osmilvusuri = os.getenv("MILVUSURI", "http://127.0.0.1:19530")milvustoken = os.getenv("MILVUSTOKEN", "")collectionname = os.getenv("AUTORAGCOLLECTION", "kbautorageval")config = f"""vectordb: – name: milvustenantstore dbtype: milvus embeddingmodel: openaiembed3small collectionname: {collectionname} uri: {milvusuri} token: {milvustoken}nodelines: – nodelinename: retrievenodeline nodes: – nodetype: semanticretrieval strategy: metrics: [retrievalrecall, retrievalprecision, retrievalf1] topk: 5 modules: – moduletype: vectordb vectordb: milvustenantstore"""os.makedirs("./config", existok=True)with open("./config/autoragmilvustenant.yaml", "w") as f: f.write(config.strip())print("✅ Config written to ./config/autoragmilvustenant.yaml")（这里要先讲清楚 AutoRAG 在测什么。AutoRAG 这一步主要测租户内部的检索质量：在某个租户自己的语料范围内，Recall、Precision、F1 是否达标。它不是在证明 Partition Key 的隔离边界。YAML 里没有配置 tenant 过滤，AutoRAG 会搜全库。）接着，在终端中执行以下 Shell 脚本，切分数据集并跑通自动化评测：# 按租户拆分评测集，保证评测数据不跨租户污染python3 -'EOF'import pandas as pdqa= pd.readparquet('./data/qa.parquet')corpus = pd.readparquet('./data/corpus.parquet')for tid in ["tenanta", "tenantb"]: qa[qa["tenantid"]== tid].toparquet(f"./data/qa{tid}.parquet", index=False) corpus[corpus["tenantid"] == tid].toparquet(f"./data/corpus{tid}.parquet", index=False)EOF# 分别对两个租户执行评测，结果落到各自的 benchmark 目录# 注意：每个租户使用独立 collection，避免评测数据相互污染for TENANT in tenanta tenantb; do AUTORAGCOLLECTION=kbautorageval${TENANT} python3 step5writeconfig.py autorag evaluate –config ./config/autoragmilvustenant.yaml –qadatapath ./data/qa${TENANT}.parquet –corpusdatapath ./data/corpus${TENANT}.parquet –projectdir ./benchmark/${TENANT}done 评测完成后，可以在 benchmark/tenanta/*/retrievenodeline/semantic_retrieval/summary.csv 中看到量化的检索质量。在此标准测试下，租户内部的检索表现优秀：
retrieval_recall=1.0
retrieval_precision=0.5
retrieval_f1=0.6666666666666666
这说明在当前评测集内，检索能命中目标文档。但它还不能单独证明不会串租。隔离边界需要下一步直接查询 Milvus 来验证。
五、直接查 Milvus，验证 tenant 过滤是否生效
AutoRAG 评测产出的 Recall / Precision / F1 反映的是租户内部的检索质量。但在执行 AutoRAG CLI 评测时，为了兼容其底层机制、避免评测时的状态复用导致跨租户污染，我们 在评测期为不同租户初始化独立的评测 Collection，以此确保评测结论的绝对纯净和可信。但要验证 隔离是否生效，我们必须在单 Collection 架构下进行双重对撞测试。用同一条查询，分别携带 tenanta 和 tenantb 的过滤条件直接测试 Milvus，确认结果集没有任何交叉，同时对比去掉过滤后的混排结果。import osfrom openai import OpenAIfrom pymilvus import MilvusClientCOLLECTION = "kbmultitenantpk"client = MilvusClient( uri=os.getenv("MILVUSURI", "http://127.0.0.1:19530"), token=os.getenv("MILVUSTOKEN", ""),)openaiclient = OpenAI()def embed(texts: list[str], model: str = "text-embedding-3-small") -> list[list[float]]: resp = openaiclient.embeddings.create(input=texts, model=model) return [item.embedding for item in resp.data]query = "报销规则里差旅审批要求是什么？"queryvector = embed([query])[0]# ✅ 带 tenant 条件查询for tid in ["tenanta", "tenantb"]: results = client.search( collectionname=COLLECTION, data=[queryvector], filter=f'tenantid == "{tid}"', limit=5, outputfields=["docid", "tenantid", "contents"], ) print(f"=== 查询租户: {tid} ===") for hit in results[0]: e = hit["entity"] print(f" doc={e['docid']} tenant={e['tenantid']} score={hit['distance']:.4f}") print(f" → {e['contents'][:40]}…")# ❌ 无过滤，语义相似度跨租户返回print("=== ⚠️ 无 tenant 过滤（危险示范）===")resultsnf = client.search( collectionname=COLLECTION, data=[queryvector], limit=5, outputfields=["docid", "tenantid", "contents"],)for hit in resultsnf[0]: e = hit["entity"] print(f" doc={e['docid']} tenant={e['tenantid']} score={hit['distance']:.4f}")输出类似：=== 查询租户: tenanta === doc=a-1 tenant=tenanta score=0.6015 → A租户的报销规则：差旅上限为内部标准。… doc=a-2 tenant=tenanta score=0.3933 → A租户合同模板要求法务审批。…=== 查询租户: tenantb === doc=b-1 tenant=tenantb score=0.6914 → B租户的报销规则：海外差旅需要二级审批。… doc=b-2 tenant=tenantb score=0.2637 → B租户合同模板要求采购会签。…=== ⚠️ 无 tenant 过滤（危险示范）=== doc=b-1 tenant=tenantb score=0.6914 ← 两个租户的文档混排 doc=a-1 tenant=tenanta score=0.6015 doc=a-2 tenant=tenanta score=0.3933 doc=b-2 tenant=tenant_b score=0.2637 结论一目了然：带过滤的查询，两边结果 严格互无交集，物理隔离完全生效；而不带过滤时，两个租户的数据立刻发生混排，证明串租风险确实存在，存储层的 Partition Key 是非常有必要存在的。
六、上线前的两道核心校验
要将这套方案推进到生产环境，业务层还必须增加两道校验 1. 写入时强校验 tenantid，字段缺失直接拒绝 def validateandinsert(doc: dict): if not doc.get("tenantid"): raise ValueError( f"docid={doc.get('docid')} 缺少 tenantid，拒绝入库。" "不允许事后补填——无tenantid 的向量进入集合后无法补救。" ) client.insert(collectionname=COLLECTION, data=[doc])依赖“约定大家都会填”是串租的根源之一。缺字段时的静默写入比报错更危险。2. 查询时 tenantid 必须来自认证上下文，不接受客户端传参#❌ 错误：相信客户端传进来的值，可以被伪造tenantid = request.params.get("tenantid")filterexpr = f'tenantid == "{tenantid}"'# ✅ 正确：从服务端验证过的Token 中提取，不可伪造tenantid = authtoken.claims["tenantid"]filterexpr = f'tenantid == "{tenantid}"'results = client.search( collectionname=COLLECTION, data=[queryvector], filter=filterexpr, # 过滤条件由系统注入，不经过客户端 limit=topk, outputfields=["docid", "contents"],)作为后台网关，检索所使用的 tenantid 必须来自 服务端解析验证后的 Token 上下文（如 JWT），严禁接收客户端直接传参（如 POST /search?tenant_id=xxx），防止黑客通过篡改参数进行越权水平攻击。
七、写在最后
在实践中，我们建议将多租户的设计与校验分为三层
写入层：没有 tenant_id 的数据拒绝入库；
检索层：用 Milvus Partition Key 执行 tenant_id 过滤和分区收敛；
验证层：用 AutoRAG 评测租户内检索质量，再用直接 Milvus 查询验证结果不交叉。
这样，多租户隔离就不再只是代码里的一个约定，而是一套可以重复运行、可以对比结果、可以接入 CI 的工程检查。
