RAG 系列（二十）：企业级 RAG 架构设计

Demo 和生产之间的墙

前面十九篇的代码有一个共同特点：所有人共享同一个向量库，任何问题都能检索到任何文档。

这在 Demo 里没问题，但放到企业环境里会立刻崩溃：

• 公司 A 的数据库能被公司 B 的用户查到
• 财务数据能被普通员工检索到
• HR 政策能被外包人员拿到
• 一个用户疯狂发请求让系统崩溃，全公司都受影响

生产级企业 RAG 需要三层防护：

用户请求 ↓ 限流检查 —— 这个用户还在配额内吗？ ↓ 缓存查询 —— 这个问题有缓存过吗？ ↓ 租户路由 —— 去哪个知识库？ ↓ 权限过滤 —— 在这个知识库里，这个用户能看什么？ ↓ 检索 + 生成 —— 拿到被授权的内容后生成答案 ↓ 写入缓存 —— 存下来下次直接用

本文逐层实现这个架构。

第一层：多租户隔离

策略：每个租户一个 Qdrant Collection

每个客户/部门的知识库存在独立的 Collection 里，Collection 之间物理隔离——你无法在acme_corp里搜到globex_corp的内容，因为这两个 Collection 根本不在同一个向量空间里。

fromlangchain_qdrantimportQdrantVectorStorefromqdrant_clientimportQdrantClientfromqdrant_client.modelsimportDistance,VectorParams qdrant_client=QdrantClient(":memory:")# 生产用 host="qdrant-server"tenant_stores:dict[str,QdrantVectorStore]={}fortenant_id,docsinTENANT_DOCS.items():qdrant_client.create_collection(collection_name=tenant_id,vectors_config=VectorParams(size=1024,distance=Distance.COSINE),)store=QdrantVectorStore(client=qdrant_client,collection_name=tenant_id,embedding=embeddings,)store.add_documents(docs)tenant_stores[tenant_id]=store

路由逻辑也很简单——用户请求带上tenant_id，从tenant_stores里取对应的 Collection：

defget_retriever(tenant_id:str,role:str,k:int=3):iftenant_idnotintenant_stores:raiseValueError(f"Unknown tenant:{tenant_id}")store=tenant_stores[tenant_id]# 下面加权限过滤（见第二层）...

为什么不用同一个 Collection + tenant_id 字段过滤？

技术上可行，但有两个问题：

1. 一旦过滤器出 bug，A 公司的数据就泄露给了 B 公司——一个 Collection 没有硬边界
2. Collection 级别隔离还带来存储层面的独立（删除一个租户的数据只需要 drop 对应 Collection，干净彻底）

对于同一公司内不同部门这种"软隔离"场景，元数据过滤足够；对于不同客户这种"硬隔离"场景，独立 Collection 更安全。

第二层：权限控制

策略：文档携带 access_level，检索时注入 Qdrant 过滤器

每篇文档在 metadata 里声明自己的访问级别：

Document(page_content="年终奖：S级3个月，A级2个月...",metadata={"source":