文章目录
- LangChain 元数据过滤与多租户隔离:别让 A 公司看到 B 公司的数据
- 一、先看一个真实的事故场景
- 二、过滤必须在检索层,不能靠 prompt
- 三、过滤的写法(各向量库不同)
- 3.1 Chroma
- 3.2 PGVector
- 3.3 FAISS(能力有限)
- 四、多租户隔离的标准做法
- 4.1 方案一:metadata 加 tenant_id(推荐)
- 4.2 方案二:物理隔离(collection / 库)
- 4.3 无论哪种,都要写测试
- 五、入库时怎么打标
- 六、过滤与检索的顺序问题
- 七、常见的越权风险点
- 八、小结
LangChain 元数据过滤与多租户隔离:别让 A 公司看到 B 公司的数据
RAG 上线后最严重的事故不是"答不准",
而是**“答了不该答的东西”**——
财务的文档被普通员工搜到,A 客户的资料出现在 B 客户的回答里。这篇讲怎么用 metadata 过滤做隔离,以及为什么过滤必须在检索层做。
一、先看一个真实的事故场景
公司知识库:10 万份文档,涵盖 8 个部门、3 个保密等级 员工问:"今年年终奖怎么算?" 系统召回:薪酬制度(密级=机密,仅 HR 可见) 模型回答:详细列出了各职级的奖金系数技术上系统"工作正常"——检索到了最相关的文档,回答也准确。
但这是一次严重的越权。
问题出在:检索时没有按权限过滤。
二、过滤必须在检索层,不能靠 prompt
常见错误做法:
# ❌ 靠 prompt 约束prompt=""" 你是助手。只回答公开级别的内容, 不要泄露机密信息。 """这完全不可靠,原因:
- 机密内容已经被送进上下文了,模型已经"看到"了;
- prompt 约束可以被绕过(提示注入、多轮诱导);
- 你无法证明它真的没泄露。
正确做法:让不该看的内容根本进不了上下文。
# ✅ 检索时就过滤掉retriever=vectorstore.as_retriever(search_kwargs={"filter":{"security_level":{"$lte":user_level}}})三、过滤的写法(各向量库不同)
3.1 Chroma
retriever=vectorstore.as_retriever(search_kwargs={"k":4,"filter":{"dept":"技术部"},# 等于})# 多条件(Chroma 支持 $and / $or / $in / $gte 等)retriever=vectorstore.as_retriever(search_kwargs={"k":4,"filter":{"$and":[{"dept":{"$in":["技术部","产品部"]}},{"year":{"$gte":2024}},]},})支持的运算符:$eq$ne$in$nin$gt$gte$lt$lte$and$or。
3.2 PGVector
PGVector 的过滤在 SQL 层执行,性能好且可以用复杂条件:
retriever=vectorstore.as_retriever(search_kwargs={"k":4,"filter":{"dept":"技术部","year":2024,},})还可以直接写 SQL 片段:
retriever=vectorstore.as_retriever(search_kwargs={"k":4,"filter":"security_level <= 2 AND dept = '技术部'",})3.3 FAISS(能力有限)
FAISS 原生不支持 metadata 过滤。
LangChain 做了部分模拟(先检索再过滤),但有两个问题:
- 性能差——要检索很多条再筛;
- 可能返回不足——
k=4过滤后可能只剩 1 条。
# FAISS 的模拟过滤:先取 20 条,过滤后取前 4 条retriever=vectorstore.as_retriever(search_kwargs={"k":20,"filter":{"dept":"技术部"}})结论:需要权限过滤的业务,别选 FAISS。用 Chroma 或 PGVector。
四、多租户隔离的标准做法
4.1 方案一:metadata 加 tenant_id(推荐)
# 入库时打标fordocindocs:doc.metadata["tenant_id"]=current_tenant doc.metadata["security_level"]=doc.security_level vectorstore.add_documents(docs)# 检索时强制带上defbuild_retriever(tenant_id:str,user_level:int,k=4):returnvectorstore.as_retriever(search_kwargs={"k":k,"filter":{"tenant_id":tenant_id,# 租户隔离"security_level":{"$lte":user_level},# 权限隔离},})关键:tenant_id 必须来自服务端会话,绝不能来自用户输入。
# ❌ 危险:租户 id 从请求参数来,用户可以伪造@app.post("/ask")asyncdefask(tenant_id:str,question:str):...# ✅ 正确:从登录态解析,用户改不了@app.post("/ask")asyncdefask(req:AskRequest,current_user=Depends(get_current_user)):retriever=build_retriever(current_user.tenant_id,current_user.level)4.2 方案二:物理隔离(collection / 库)
# 每个租户一个 collectionvs=Chroma(collection_name=f"tenant_{tenant_id}",...)| metadata 过滤 | 物理隔离 | |
|---|---|---|
| 隔离强度 | 依赖代码正确性 | 天然隔离 |
| 成本 | 一套索引 | 每租户一套(贵) |
| 适合 | 租户多、数据量小 | 租户少、数据量大、合规要求高 |
金融、医疗这类强合规场景用物理隔离,
一般 SaaS 用 metadata 过滤 + 严格的单元测试就够。
4.3 无论哪种,都要写测试
deftest_tenant_isolation():"""A 租户的检索绝不能返回 B 租户的文档"""retriever_a=build_retriever("tenant_a",level=3)# 用 B 租户特有的关键词去搜results=retriever_a.invoke("B公司的项目代号 XZZ-999")fordinresults:assertd.metadata["tenant_id"]=="tenant_a","越权!"deftest_level_isolation():"""低权限用户不能看到高密级文档"""retriever=build_retriever("tenant_a",level=1)results=retriever.invoke("年终奖系数")fordinresults:assertd.metadata["security_level"]<=1,"越权!"这类测试必须进 CI,它比"答案准不准"重要得多。
五、入库时怎么打标
过滤能力取决于 metadata,所以入库阶段就要把字段埋好:
frompathlibimportPathdefenrich(doc,tenant_id:str,base_dir:str):src=doc.metadata.get("source","")rel=Path(src).relative_to(base_dir)parts=rel.parts doc.metadata["tenant_id"]=tenant_id doc.metadata["dept"]=parts[0]iflen(parts)>1else"通用"doc.metadata["filename"]=rel.name doc.metadata["ext"]=rel.suffix# 密级可以从目录结构或单独的配置表里取doc.metadata["security_level"]=LEVEL_MAP.get(str(rel),1)doc.metadata["indexed_at"]=datetime.now().isoformat()returndoc docs=[enrich(d,tenant_id,"docs/")fordinraw_docs]密级的来源:目录结构(/confidential/)、文件名约定、
或者一份单独的权限配置表。别指望模型判断密级——那是业务逻辑。
六、过滤与检索的顺序问题
一个容易忽略的细节:先过滤还是先向量检索?
| 顺序 | 做法 | 问题 |
|---|---|---|
| 先向量后过滤 | 取 top-20 再筛 | 可能筛完只剩 1 条(召回不足) |
| 先过滤后向量 | 在子集里检索 | 正确,但需要向量库支持 |
Chroma / PGVector 是先过滤后检索(在过滤后的子集里做 ANN),
所以没问题。FAISS 是先检索后过滤,会召回不足——这也是不推荐它的原因之一。
如果只能用 FAISS,补偿办法是把 k 放大:
retriever=vectorstore.as_retriever(search_kwargs={"k":50,"filter":{"tenant_id":tid}})# 然后自己截断到 4 条docs=retriever.invoke(q)[:4]七、常见的越权风险点
| 风险 | 说明 | 防护 |
|---|---|---|
| tenant_id 来自请求参数 | 用户可伪造 | 从登录态取 |
| 拼接 SQL 注入 | 过滤条件用字符串拼接 | 用参数化过滤 |
| 缓存串了 | 缓存 key 没包含 tenant | key 里加 tenant_id |
| Agent 用工具绕过过滤 | 工具直接查库没带过滤 | 工具内部也要传 tenant |
| 引用溯源暴露 | 展示了不该看的来源 | 展示前也要过滤 |
缓存那条特别容易漏:
# ❌ 缓存 key 只有问题,A 公司问过的问题,B 公司会命中cache_key=hash(question)# ✅cache_key=hash(f"{tenant_id}:{user_level}:{question}")八、小结
- 权限过滤必须在检索层做,靠 prompt 约束完全不可靠
(内容已经进上下文了); - 过滤写法因库而异:Chroma 用
$and/$in/$lte等运算符,
PGVector 还能写 SQL 片段,FAISS 不支持(只能模拟,且召回不足); - 多租户两种方案:metadata 加 tenant_id(推荐,成本低)
vs 物理 collection 隔离(强合规场景); - ⚠️tenant_id 必须从服务端登录态取,绝不能来自用户输入;
- 必须写隔离测试并进 CI:“A 租户搜不到 B 租户的文档”;
- 注意隐蔽风险:缓存 key 要包含租户和权限等级。
下一篇讲引用溯源——让答案能点开原文。