☰
LangChain 元数据过滤与多租户隔离:别让 A 公司看到 B 公司的数据
2026/10/12 8:47:08 网站建设 项目流程
个人主页: > for_ever_love__ <(欢迎各位大佬莅临😊)
其他栏目: > 大模型开发从0到1 <
其他栏目: > iOS项目总结大全 <
其他栏目: > 我想学python了 <
其他栏目: > iOS UI <

文章目录

  • 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=""" 你是助手。只回答公开级别的内容, 不要泄露机密信息。 """

这完全不可靠,原因:

  1. 机密内容已经被送进上下文了,模型已经"看到"了;
  2. prompt 约束可以被绕过(提示注入、多轮诱导);
  3. 你无法证明它真的没泄露。

正确做法:让不该看的内容根本进不了上下文。

# ✅ 检索时就过滤掉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 做了部分模拟(先检索再过滤),但有两个问题:

  1. 性能差——要检索很多条再筛;
  2. 可能返回不足——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 没包含 tenantkey 里加 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 要包含租户和权限等级。

下一篇讲引用溯源——让答案能点开原文。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询