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检索可见范围Alicealice自己的 chunk 共享 chunk绝不包含 Bob 的Bobbob自己的 chunk 共享 chunk绝不包含 Alice 的管理员None全部语料admin view是所有受限视图的超集关键设计点user_idNone不是报错而是不设作用域等价于管理员视图。正因为作用域丢失会退化成管理员视图而不是抛异常示例代码一律使用**断言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 向量与生成模型因此首先需要设置环境变量OPENAI_API_KEY嵌入型后端LanceDB、Chroma、Qdrant随 Python 进程内嵌运行无需额外启动服务服务型后端需先启动对应服务仓库提供了现成脚本位于 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.shMilvus特殊bash standalone_embed.sh start启动 standalone 服务器——因为 Milvus Lite本地文件 uri在搜索读取路径上会丢弃标量字段导致检索内容为空必须用真实服务MongoDB特殊docker run -d -p 27017:27017 mongodb/mongodb-atlas-local:latest——普通 MongoDB 没有$vectorSearch需要 Atlas-Local 容器云后端Pinecone 需要PINECONE_API_KEYUpstash 需要UPSTASH_VECTOR_REST_URL与UPSTASH_VECTOR_REST_TOKEN且索引维度必须为1536SingleStore 与 Couchbase 需要各自的凭据环境变量。另外注意Redis 与 Valkey 都绑定 6379 端口同一时间只能运行其中一个。所有示例每次启动都会先 drop 自己的集合/表因此重复运行是安全的。统一的三段式示例骨架17 个示例文件虽然后端不同但骨架完全一致可以对照阅读写入阶段knowledge.ainsert(name..., text_content..., user_id...)插入私有文档不带user_id的插入成为共享内容作用域检索阶段分别以user_idalice、user_idbob、user_idNone调用knowledge.asearch(querysalary, ...)打印结果并用断言校验——Alice 视图必须包含自己的180,000与共享的January 1必须不含215,000Bob 视图对称管理员视图必须三者全含且是 Alice 视图的超集Agent 中介检索阶段构建Agent(user_idalice, knowledgeknowledge, search_knowledgeTrue)让它回答What is Bobs 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: Alices agent retrieved Bobs salary. The owner was dropped between the run context and the vector DB, so retrieval ran unscoped (user_idNone, the admin view). )先断言检索确实返回了文档防止空结果让隔离检查假通过再断言不该出现的内容没有出现。Agent 的user_id会进入运行上下文run contextKnowledge 在生成检索工具时通过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 NULLlance_db.pyuser_id列.where(user_id X OR user_id IS NULL, prefilterTrue)保证 top-K 只在允许的行内排名chroma_db.py每个用户一个 collection{base}__{user_id}base collection 即共享桶按距离合并两次检索结果qdrant_db.py关键词索引的user_idpayload 字段is_tenantTrueshould匹配 空值milvus_db.py非空user_id标量字段无主 chunk 使用__shared__哨兵值mongo_db.py顶层user_id字段声明为向量索引的 filter 字段$vectorSearch前用$match {$in: [X, null]}预过滤weaviate_db.pyuser_id文本属性where中OR is_noneopensearch_db.pyuser_idkeyword 字段termORmust_not existsredis_db.pyhash 上的user_idTAG 字段FT.SEARCH内过滤无主 chunk 用__shared__哨兵 tagvalkey_db.py同 Redisuser_idTAG 字段 __shared__哨兵 tagclickhouse_db.py非空String列共享内容用哨兵cassandra_db.pyuser_id元数据无主 chunk 用__shared__哨兵couchbase_db.py关键词索引的 FTSuser_id字段__shared__哨兵singlestore_db.py可空user_id列WHERE user_id X OR user_id IS NULLsurreal_db.pyuser_id字段专用的$scope_user_id绑定参数pinecone_db.py向量 metadata 中的user_id$or [{$eq: X}, {$exists: false}]过滤upstash_db.pymetadata 中的user_iduser_id X OR HAS NOT FIELD user_id可以看到各后端的实现策略可分为三类可空列 双条件 ORPgVector、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 postgresqlpsycopg://ai:ailocalhost:5532/ai TABLE_NAME per_user_isolation_demo vector_db PgVector(table_nameTABLE_NAME, db_urldb_url) if vector_db.exists(): vector_db.drop() vector_db.create() knowledge Knowledge( nameper_user_demo, descriptionPer-user RAG isolation demo (PgVector), vector_dbvector_db, )写入三类内容Alice 私有、Bob 私有、无属主共享await knowledge.ainsert(namealice_salary, text_contentALICE_SALARY, user_idalice) await knowledge.ainsert(namebob_salary, text_contentBOB_SALARY, user_idbob) # The last upload has no user_id, which makes it shared with everyone. await knowledge.ainsert(namecompany_holidays, text_contentHOLIDAYS)作用域检索与断言alice_view await knowledge.asearch(querysalary, user_idalice) 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 Alices scoped view assert 215,000 not in alice_text, Isolation broken: Alices scoped view leaked Bobs salary admin_view await knowledge.asearch(querysalary, user_idNone) admin_text .join(d.content for d in admin_view) for expected in (180,000, 215,000, January 1): assert expected in admin_text, fAdmin 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 users viewAgent 中介检索Agent 携带user_idalice模型用OpenAIResponses(idgpt-5.5)开启search_knowledgeTrue指令约束模型只依据检索到的知识作答alice_agent Agent( nameAlices Assistant, modelOpenAIResponses(idgpt-5.5), knowledgeknowledge, search_knowledgeTrue, user_idalice, instructions[ Answer questions using ONLY the knowledge you can retrieve., If you dont know, say so - do not invent salary figures., ], markdownTrue, ) response await alice_agent.arun(What is Bobs salary?) # 断言 references检索返回的文档而不是模型生成的文字运行方式其他示例同理换成对应文件名即可.venvs/demo/bin/python cookbook/07_knowledge/04_advanced/07_per_user_isolation/pgvector_db.py各后端的实现差异与注意事项LanceDBlance_db.py过滤条件必须带prefilterTrue让向量搜索的 top-K 只在允许的行内排名否则隔离会在候选集层面失效使用uv pip install lancedb pyarrow内嵌运行。Chromachroma_db.pydrop()会连同该 base 名称派生的所有 per-user collection 一起删除每次运行前 drop 是为了清掉上一轮遗留的属主 collection。Qdrantqdrant_db.pyuser_id需建立关键词索引is_tenantTrue运行结束后要显式await vector_db.async_close()——因为async_client是惰性属性提前 close 会导致重建且永不关闭的新客户端。Milvusmilvus_db.pyuser_id字段非空因此共享 chunk 必须写入__shared__哨兵务必使用 standalone 服务器而非 Milvus Lite。MongoDBmongo_db.py初始化时给wait_until_index_ready_in_seconds300因为 Atlas-Local 的向量索引在后台构建建表走同步create()而非async_create()后者的就绪轮询在 Atlas-Local 上会卡住写入后await asyncio.sleep(10)因为$vectorSearch读取的是后台索引新写入的数据不能立即被检索到连接串需带?directConnectiontrue。Redis / Valkeyredis_db.py、valkey_db.pyRedisDb(index_name..., redis_url..., search_typeSearchType.vector)共享内容用__shared__哨兵 TAG在FT.SEARCH内完成过滤两者都占 6379 端口不能同时运行。Pinecone / Upstashpinecone_db.py、upstash_db.py属主写在向量 metadata 中分别用$exists: false与HAS NOT FIELD user_id表达无属主即共享Upstash 索引维度固定为 1536。隔离语义在 Agno 源码中的落点从源码层面可以确认整条链路的设计意图写入侧Knowledge.ainsert(..., user_id...)把属主挂在内容Content上user_id作为显式参数流动不会写进meta_dataknowledge.py#L2000 附近的注释明确说明并通过strict_user_id_kwarg决定是否把user_id传给后端的insert/upsert检索侧Knowledge.search/asearch把user_id作为独立参数转发给vector_db.search/async_searchknowledge.py#L903-L908后端各自实现过滤原语Agent 侧Agent 的user_id进入运行上下文Knowledge 检索工具从run_context读取属主并再次传入searchknowledge.py#L5123、knowledge.py#L5250过滤器 DSL 与user_id分离user_id不属于过滤器 DSL它独立传送给向量库knowledge.py#L850-L862这保证了同一个filters对象可以在不同属主之间复用而不会串号失败模式隔离是安全默认还是开放默认由后端决定——示例展示的语义是user_idNone即管理员视图因此代码库用断言兜底任何一环丢失属主都会在测试期暴露而不是在生产期静默泄漏。小结一份知识库、每个用户私有视图在 Agno 中只需两件事写入时给Knowledge.ainsert传user_id检索时给Knowledge.asearch传同样的user_id。共享内容不传属主即可user_idNone则是管理员全量视图。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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 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检索可见范围Alicealice自己的 chunk 共享 chunk绝不包含 Bob 的Bobbob自己的 chunk 共享 chunk绝不包含 Alice 的管理员None全部语料admin view是所有受限视图的超集关键设计点user_idNone不是报错而是不设作用域等价于管理员视图。正因为作用域丢失会退化成管理员视图而不是抛异常示例代码一律使用**断言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 向量与生成模型因此首先需要设置环境变量OPENAI_API_KEY嵌入型后端LanceDB、Chroma、Qdrant随 Python 进程内嵌运行无需额外启动服务服务型后端需先启动对应服务仓库提供了现成脚本位于 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.shMilvus特殊bash standalone_embed.sh start启动 standalone 服务器——因为 Milvus Lite本地文件 uri在搜索读取路径上会丢弃标量字段导致检索内容为空必须用真实服务MongoDB特殊docker run -d -p 27017:27017 mongodb/mongodb-atlas-local:latest——普通 MongoDB 没有$vectorSearch需要 Atlas-Local 容器云后端Pinecone 需要PINECONE_API_KEYUpstash 需要UPSTASH_VECTOR_REST_URL与UPSTASH_VECTOR_REST_TOKEN且索引维度必须为1536SingleStore 与 Couchbase 需要各自的凭据环境变量。另外注意Redis 与 Valkey 都绑定 6379 端口同一时间只能运行其中一个。所有示例每次启动都会先 drop 自己的集合/表因此重复运行是安全的。统一的三段式示例骨架17 个示例文件虽然后端不同但骨架完全一致可以对照阅读写入阶段knowledge.ainsert(name..., text_content..., user_id...)插入私有文档不带user_id的插入成为共享内容作用域检索阶段分别以user_idalice、user_idbob、user_idNone调用knowledge.asearch(querysalary, ...)打印结果并用断言校验——Alice 视图必须包含自己的180,000与共享的January 1必须不含215,000Bob 视图对称管理员视图必须三者全含且是 Alice 视图的超集Agent 中介检索阶段构建Agent(user_idalice, knowledgeknowledge, search_knowledgeTrue)让它回答What is Bobs 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: Alices agent retrieved Bobs salary. The owner was dropped between the run context and the vector DB, so retrieval ran unscoped (user_idNone, the admin view). )先断言检索确实返回了文档防止空结果让隔离检查假通过再断言不该出现的内容没有出现。Agent 的user_id会进入运行上下文run contextKnowledge 在生成检索工具时通过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 NULLlance_db.pyuser_id列.where(user_id X OR user_id IS NULL, prefilterTrue)保证 top-K 只在允许的行内排名chroma_db.py每个用户一个 collection{base}__{user_id}base collection 即共享桶按距离合并两次检索结果qdrant_db.py关键词索引的user_idpayload 字段is_tenantTrueshould匹配 空值milvus_db.py非空user_id标量字段无主 chunk 使用__shared__哨兵值mongo_db.py顶层user_id字段声明为向量索引的 filter 字段$vectorSearch前用$match {$in: [X, null]}预过滤weaviate_db.pyuser_id文本属性where中OR is_noneopensearch_db.pyuser_idkeyword 字段termORmust_not existsredis_db.pyhash 上的user_idTAG 字段FT.SEARCH内过滤无主 chunk 用__shared__哨兵 tagvalkey_db.py同 Redisuser_idTAG 字段 __shared__哨兵 tagclickhouse_db.py非空String列共享内容用哨兵cassandra_db.pyuser_id元数据无主 chunk 用__shared__哨兵couchbase_db.py关键词索引的 FTSuser_id字段__shared__哨兵singlestore_db.py可空user_id列WHERE user_id X OR user_id IS NULLsurreal_db.pyuser_id字段专用的$scope_user_id绑定参数pinecone_db.py向量 metadata 中的user_id$or [{$eq: X}, {$exists: false}]过滤upstash_db.pymetadata 中的user_iduser_id X OR HAS NOT FIELD user_id可以看到各后端的实现策略可分为三类可空列 双条件 ORPgVector、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 postgresqlpsycopg://ai:ailocalhost:5532/ai TABLE_NAME per_user_isolation_demo vector_db PgVector(table_nameTABLE_NAME, db_urldb_url) if vector_db.exists(): vector_db.drop() vector_db.create() knowledge Knowledge( nameper_user_demo, descriptionPer-user RAG isolation demo (PgVector), vector_dbvector_db, )写入三类内容Alice 私有、Bob 私有、无属主共享await knowledge.ainsert(namealice_salary, text_contentALICE_SALARY, user_idalice) await knowledge.ainsert(namebob_salary, text_contentBOB_SALARY, user_idbob) # The last upload has no user_id, which makes it shared with everyone. await knowledge.ainsert(namecompany_holidays, text_contentHOLIDAYS)作用域检索与断言alice_view await knowledge.asearch(querysalary, user_idalice) 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 Alices scoped view assert 215,000 not in alice_text, Isolation broken: Alices scoped view leaked Bobs salary admin_view await knowledge.asearch(querysalary, user_idNone) admin_text .join(d.content for d in admin_view) for expected in (180,000, 215,000, January 1): assert expected in admin_text, fAdmin 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 users viewAgent 中介检索Agent 携带user_idalice模型用OpenAIResponses(idgpt-5.5)开启search_knowledgeTrue指令约束模型只依据检索到的知识作答alice_agent Agent( nameAlices Assistant, modelOpenAIResponses(idgpt-5.5), knowledgeknowledge, search_knowledgeTrue, user_idalice, instructions[ Answer questions using ONLY the knowledge you can retrieve., If you dont know, say so - do not invent salary figures., ], markdownTrue, ) response await alice_agent.arun(What is Bobs salary?) # 断言 references检索返回的文档而不是模型生成的文字运行方式其他示例同理换成对应文件名即可.venvs/demo/bin/python cookbook/07_knowledge/04_advanced/07_per_user_isolation/pgvector_db.py各后端的实现差异与注意事项LanceDBlance_db.py过滤条件必须带prefilterTrue让向量搜索的 top-K 只在允许的行内排名否则隔离会在候选集层面失效使用uv pip install lancedb pyarrow内嵌运行。Chromachroma_db.pydrop()会连同该 base 名称派生的所有 per-user collection 一起删除每次运行前 drop 是为了清掉上一轮遗留的属主 collection。Qdrantqdrant_db.pyuser_id需建立关键词索引is_tenantTrue运行结束后要显式await vector_db.async_close()——因为async_client是惰性属性提前 close 会导致重建且永不关闭的新客户端。Milvusmilvus_db.pyuser_id字段非空因此共享 chunk 必须写入__shared__哨兵务必使用 standalone 服务器而非 Milvus Lite。MongoDBmongo_db.py初始化时给wait_until_index_ready_in_seconds300因为 Atlas-Local 的向量索引在后台构建建表走同步create()而非async_create()后者的就绪轮询在 Atlas-Local 上会卡住写入后await asyncio.sleep(10)因为$vectorSearch读取的是后台索引新写入的数据不能立即被检索到连接串需带?directConnectiontrue。Redis / Valkeyredis_db.py、valkey_db.pyRedisDb(index_name..., redis_url..., search_typeSearchType.vector)共享内容用__shared__哨兵 TAG在FT.SEARCH内完成过滤两者都占 6379 端口不能同时运行。Pinecone / Upstashpinecone_db.py、upstash_db.py属主写在向量 metadata 中分别用$exists: false与HAS NOT FIELD user_id表达无属主即共享Upstash 索引维度固定为 1536。隔离语义在 Agno 源码中的落点从源码层面可以确认整条链路的设计意图写入侧Knowledge.ainsert(..., user_id...)把属主挂在内容Content上user_id作为显式参数流动不会写进meta_dataknowledge.py#L2000 附近的注释明确说明并通过strict_user_id_kwarg决定是否把user_id传给后端的insert/upsert检索侧Knowledge.search/asearch把user_id作为独立参数转发给vector_db.search/async_searchknowledge.py#L903-L908后端各自实现过滤原语Agent 侧Agent 的user_id进入运行上下文Knowledge 检索工具从run_context读取属主并再次传入searchknowledge.py#L5123、knowledge.py#L5250过滤器 DSL 与user_id分离user_id不属于过滤器 DSL它独立传送给向量库knowledge.py#L850-L862这保证了同一个filters对象可以在不同属主之间复用而不会串号失败模式隔离是安全默认还是开放默认由后端决定——示例展示的语义是user_idNone即管理员视图因此代码库用断言兜底任何一环丢失属主都会在测试期暴露而不是在生产期静默泄漏。小结一份知识库、每个用户私有视图在 Agno 中只需两件事写入时给Knowledge.ainsert传user_id检索时给Knowledge.asearch传同样的user_id。共享内容不传属主即可user_idNone则是管理员全量视图。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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考