Agno 多用户 RAG 隔离实战指南：一份知识库、按 user_id 实现每个用户私有视图

Agno 多用户 RAG 隔离实战指南：一份知识库、按 user_id 实现每个用户私有视图

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno

本指南以 Agno 仓库 cookbook 中 per_user_isolation 示例集 为核心，讲解如何在一份共享知识库之上，通过Knowledge.asearch(user_id=...)与Knowledge.ainsert(..., user_id=...)实现「每个用户看到自己的私有文档 + 全体共享文档」的隔离模型，并完整覆盖 PgVector、LanceDB、Chroma、Qdrant、Milvus、MongoDB、Weaviate、OpenSearch、Redis、Valkey、ClickHouse、Cassandra、Couchbase、SingleStore、SurrealDB、Pinecone、Upstash 共 17 种向量后端的隔离原语。读完本文，你既能拿到可直接复制运行的示例代码与断言逻辑，也能理解user_id从 Agent 运行上下文到向量检索底层过滤条件的完整传递链路。

场景设定：Alice、Bob 与一份共享知识库

整个示例集围绕同一个业务场景展开：公司内部 RAG 系统里存放员工薪酬与公司节假日信息，但必须做到按人隔离。

• Alice 上传了一份私有文档（她的薪酬$180,000，每年三月评审）；
• Bob 上传了一份私有文档（他的薪酬$215,000，每年六月评审）；
• 第三个上传没有指定user_id，因此它是共享内容（公司节假日：1 月 1 日、7 月 4 日、12 月 25 日闭园），全员可见。

隔离语义由三个角色构成：

调用方	user_id	检索可见范围
Alice	"alice"	自己的 chunk + 共享 chunk，绝不包含 Bob 的
Bob	"bob"	自己的 chunk + 共享 chunk，绝不包含 Alice 的
管理员	None	全部语料（admin view，是所有受限视图的超集）

关键设计点：user_id=None不是"报错"，而是"不设作用域"，等价于管理员视图。正因为"作用域丢失"会退化成管理员视图而不是抛异常，示例代码一律使用**断言（assert）**来验证隔离是否成立，而不是依赖异常来暴露问题——这是本示例集在工程上最值得借鉴的地方。

在 Agno 源码 libs/agno/agno/knowledge/knowledge.py 中，Knowledge.search()的user_id参数被定义为"转发给vector_db.search()的属主作用域，None表示搜索全部"，并通过strict_user_id_kwarg(self.vector_db.search, user_id)按后端能力有选择地传递；asearch异步版本走完全相同的路径（knowledge.py#L920-L968）。也就是说，Knowledge 层负责统一携带user_id，真正的隔离语义由每个向量后端各自的过滤原语实现——这正是 17 个示例文件存在的原因。

运行前置条件

示例统一使用 OpenAI 向量与生成模型，因此首先需要：

1. 设置环境变量OPENAI_API_KEY；
2. 嵌入型后端（LanceDB、Chroma、Qdrant）随 Python 进程内嵌运行，无需额外启动服务；
3. 服务型后端需先启动对应服务，仓库提供了现成脚本（位于 cookbook/scripts）：./cookbook/scripts/run_pgvector.sh、run_weaviate.sh、run_opensearch.sh、run_redis.sh、run_valkey.sh、run_clickhouse.sh、run_cassandra.sh、run_couchbase.sh、run_surrealdb.sh、run_singlestore.sh；
4. Milvus特殊：bash standalone_embed.sh start启动 standalone 服务器——因为 Milvus Lite（本地文件 uri）在搜索读取路径上会丢弃标量字段，导致检索内容为空，必须用真实服务；
5. MongoDB特殊：docker run -d -p 27017:27017 mongodb/mongodb-atlas-local:latest——普通 MongoDB 没有$vectorSearch，需要 Atlas-Local 容器；
6. 云后端：Pinecone 需要PINECONE_API_KEY；Upstash 需要UPSTASH_VECTOR_REST_URL与UPSTASH_VECTOR_REST_TOKEN，且索引维度必须为1536；SingleStore 与 Couchbase 需要各自的凭据环境变量。

另外注意：Redis 与 Valkey 都绑定 6379 端口，同一时间只能运行其中一个。所有示例每次启动都会先 drop 自己的集合/表，因此重复运行是安全的。

统一的三段式示例骨架

17 个示例文件虽然后端不同，但骨架完全一致，可以对照阅读：

1. 写入阶段：knowledge.ainsert(name=..., text_content=..., user_id=...)插入私有文档；不带user_id的插入成为共享内容；
2. 作用域检索阶段：分别以user_id="alice"、user_id="bob"、user_id=None调用knowledge.asearch(query="salary", ...)，打印结果并用断言校验——Alice 视图必须包含自己的180,000与共享的January 1，必须不含215,000；Bob 视图对称；管理员视图必须三者全含，且是 Alice 视图的超集；
3. Agent 中介检索阶段：构建Agent(user_id="alice", knowledge=knowledge, search_knowledge=True)，让它回答"What is Bob's salary?"，然后断言RunOutput.references中的检索结果不包含 Bob 的薪酬。

第三个阶段是示例集的技术精华：不在模型生成的文字上做断言，而是在检索返回的references上做断言。以 pgvector_db.py 为例，通过response.references逐层取回检索到的文档内容并拼接校验：

retrieved = " ".join( item["content"] for ref in (response.references or []) for item in (ref.references or []) if isinstance(item, dict) and item.get("content") ) assert retrieved, "Retrieval returned no documents..." assert "215,000" not in retrieved, ( "Isolation broken: Alice's agent retrieved Bob's salary. The owner was " "dropped between the run context and the vector DB, so retrieval ran " "unscoped (user_id=None, the admin view)." )

先断言"检索确实返回了文档"（防止空结果让隔离检查"假通过"），再断言"不该出现的内容没有出现"。Agent 的user_id会进入运行上下文（run context），Knowledge 在生成检索工具时通过getattr(run_context, "user_id", None)取出属主并传给search（见 knowledge.py#L5123 与 knowledge.py#L5250）；一旦这条链路断裂，user_id变成None，检索就会退化为管理员视图、泄漏所有用户的数据——这正是示例用断言而不是异常来兜底的原因。

17 种向量后端的隔离原语全景

下表是各示例文件及其底层隔离机制（继承自 README.md 并补充了示例文件中的实现细节）：

文件	隔离原语
pgvector_db.py	可空的user_id列，WHERE user_id = X OR user_id IS NULL
lance_db.py	user_id列，.where("user_id = X OR user_id IS NULL", prefilter=True)，保证 top-K 只在允许的行内排名
chroma_db.py	每个用户一个 collection（{base}__{user_id}），base collection 即共享桶，按距离合并两次检索结果
qdrant_db.py	关键词索引的user_idpayload 字段（is_tenant=True），should匹配 + 空值
milvus_db.py	非空user_id标量字段，无主 chunk 使用__shared__哨兵值
mongo_db.py	顶层user_id字段（声明为向量索引的 filter 字段），$vectorSearch前用$match {$in: [X, null]}预过滤
weaviate_db.py	user_id文本属性，where中OR is_none
opensearch_db.py	user_idkeyword 字段，termORmust_not exists
redis_db.py	hash 上的user_idTAG 字段，FT.SEARCH内过滤，无主 chunk 用__shared__哨兵 tag
valkey_db.py	同 Redis：user_idTAG 字段 +__shared__哨兵 tag
clickhouse_db.py	非空String列，共享内容用""哨兵
cassandra_db.py	user_id元数据，无主 chunk 用__shared__哨兵
couchbase_db.py	关键词索引的 FTSuser_id字段，__shared__哨兵
singlestore_db.py	可空user_id列，WHERE user_id = X OR user_id IS NULL
surreal_db.py	user_id字段，专用的$scope_user_id绑定参数
pinecone_db.py	向量 metadata 中的user_id，$or [{$eq: X}, {$exists: false}]过滤
upstash_db.py	metadata 中的user_id，user_id = X OR HAS NOT FIELD user_id

可以看到，各后端的实现策略可分为三类：

• 可空列 + 双条件 OR（PgVector、SingleStore、LanceDB、MongoDB、Weaviate、OpenSearch、Pinecone、Upstash 等）：共享内容就是"没有属主"（IS NULL/$exists: false/must_not exists/is_none），过滤条件写成"等于我 或 无属主"；
• 哨兵值标记共享（Milvus、Redis、Valkey、ClickHouse、Cassandra、Couchbase）：后端字段不允许空值（如 Milvus 的非空标量字段、Redis 的 TAG），于是用__shared__（或 ClickHouse 的空串""）作为"无主"的显式标记；
• 按用户拆分存储（Chroma）：每个用户一个 collection，天然物理隔离，检索时合并调用者 collection 与 base 共享 collection，按距离统一排序。

端到端实例：PgVector 上的完整流程

以 pgvector_db.py 为例走一遍完整代码（其余示例的差异仅在于向量库初始化和隔离原语，检索与断言部分完全同构）。

连接与初始化：指定连接串与表名，启动时先drop()再create()，确保建出带user_id属主列的表——对隔离功能上线前的旧表做作用域检索会抛错，Knowledge 层会把异常转为空结果：

db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai" TABLE_NAME = "per_user_isolation_demo" vector_db = PgVector(table_name=TABLE_NAME, db_url=db_url) if vector_db.exists(): vector_db.drop() vector_db.create() knowledge = Knowledge( name="per_user_demo", description="Per-user RAG isolation demo (PgVector)", vector_db=vector_db, )

写入三类内容：Alice 私有、Bob 私有、无属主（共享）：

await knowledge.ainsert(name="alice_salary", text_content=ALICE_SALARY, user_id="alice") await knowledge.ainsert(name="bob_salary", text_content=BOB_SALARY, user_id="bob") # The last upload has no user_id, which makes it shared with everyone. await knowledge.ainsert(name="company_holidays", text_content=HOLIDAYS)

作用域检索与断言：

alice_view = await knowledge.asearch(query="salary", user_id="alice") alice_text = " ".join(d.content for d in alice_view) assert "180,000" in alice_text, "Alice cannot retrieve her own document" assert "January 1" in alice_text, "Shared content is unreachable from Alice's scoped view" assert "215,000" not in alice_text, "Isolation broken: Alice's scoped view leaked Bob's salary" admin_view = await knowledge.asearch(query="salary", user_id=None) admin_text = " ".join(d.content for d in admin_view) for expected in ("180,000", "215,000", "January 1"): assert expected in admin_text, f"Admin view is missing {expected}" assert all(d.content in admin_text for d in alice_view), "Admin view has to be a superset of a scoped user's view"

Agent 中介检索：Agent 携带user_id="alice"，模型用OpenAIResponses(id="gpt-5.5")，开启search_knowledge=True，指令约束模型只依据检索到的知识作答：

alice_agent = Agent( name="Alice's Assistant", model=OpenAIResponses(id="gpt-5.5"), knowledge=knowledge, search_knowledge=True, user_id="alice", instructions=[ "Answer questions using ONLY the knowledge you can retrieve.", "If you don't know, say so - do not invent salary figures.", ], markdown=True, ) response = await alice_agent.arun("What is Bob's salary?") # 断言 references（检索返回的文档），而不是模型生成的文字

运行方式（其他示例同理，换成对应文件名即可）：

.venvs/demo/bin/python cookbook/07_knowledge/04_advanced/07_per_user_isolation/pgvector_db.py

各后端的实现差异与注意事项

• LanceDB（lance_db.py）：过滤条件必须带prefilter=True，让向量搜索的 top-K 只在允许的行内排名，否则隔离会在候选集层面失效；使用uv pip install lancedb pyarrow，内嵌运行。
• Chroma（chroma_db.py）：drop()会连同该 base 名称派生的所有 per-user collection 一起删除；每次运行前 drop 是为了清掉上一轮遗留的属主 collection。
• Qdrant（qdrant_db.py）：user_id需建立关键词索引（is_tenant=True），运行结束后要显式await vector_db.async_close()——因为async_client是惰性属性，提前 close 会导致重建且永不关闭的新客户端。
• Milvus（milvus_db.py）：user_id字段非空，因此共享 chunk 必须写入__shared__哨兵；务必使用 standalone 服务器而非 Milvus Lite。
• MongoDB（mongo_db.py）：初始化时给wait_until_index_ready_in_seconds=300，因为 Atlas-Local 的向量索引在后台构建；建表走同步create()而非async_create()（后者的就绪轮询在 Atlas-Local 上会卡住）；写入后await asyncio.sleep(10)，因为$vectorSearch读取的是后台索引，新写入的数据不能立即被检索到；连接串需带?directConnection=true。
• Redis / Valkey（redis_db.py、valkey_db.py）：RedisDb(index_name=..., redis_url=..., search_type=SearchType.vector)，共享内容用__shared__哨兵 TAG，在FT.SEARCH内完成过滤；两者都占 6379 端口，不能同时运行。
• Pinecone / Upstash（pinecone_db.py、upstash_db.py）：属主写在向量 metadata 中，分别用$exists: false与HAS NOT FIELD user_id表达"无属主即共享"；Upstash 索引维度固定为 1536。

隔离语义在 Agno 源码中的落点

从源码层面可以确认整条链路的设计意图：

1. 写入侧：Knowledge.ainsert(..., user_id=...)把属主挂在内容（Content）上，user_id作为显式参数流动，不会写进meta_data（knowledge.py#L2000 附近的注释明确说明），并通过strict_user_id_kwarg决定是否把user_id传给后端的insert/upsert；
2. 检索侧：Knowledge.search/asearch把user_id作为独立参数转发给vector_db.search/async_search（knowledge.py#L903-L908），后端各自实现过滤原语；
3. Agent 侧：Agent 的user_id进入运行上下文，Knowledge 检索工具从run_context读取属主并再次传入search（knowledge.py#L5123、knowledge.py#L5250）；
4. 过滤器 DSL 与user_id分离：user_id不属于过滤器 DSL，它独立传送给向量库（knowledge.py#L850-L862），这保证了同一个filters对象可以在不同属主之间复用而不会串号；
5. 失败模式：隔离是"安全默认"还是"开放默认"由后端决定——示例展示的语义是user_id=None即管理员视图，因此代码库用断言兜底，任何一环丢失属主都会在测试期暴露，而不是在生产期静默泄漏。

小结

一份知识库、每个用户私有视图，在 Agno 中只需两件事：写入时给Knowledge.ainsert传user_id，检索时给Knowledge.asearch传同样的user_id。共享内容不传属主即可，user_id=None则是管理员全量视图。17 个示例文件用完全相同的业务场景与断言逻辑，把"如何隔离"翻译成每一种主流向量后端各自的过滤原语——从 SQL 风格的可空列双条件，到 Milvus/Redis 的__shared__哨兵，再到 Chroma 的按用户分 collection——并特别强调用RunOutput.references验证 Agent 检索链路，确保user_id在"Agent 运行上下文 → Knowledge → 向量库"的全链路中不丢失。这套模式可直接迁移到任何需要多租户 RAG 隔离的生产场景。

【免费下载链接】agnoBuild, run, and manage agent platforms.项目地址: https://gitcode.com/GitHub_Trending/ag/agno