Agno 知识系统生产化实践指南：多源 RAG、生命周期管理、多租户隔离与安全加固

导读

本文围绕 Agno 仓库中 cookbook/07_knowledge/03_production 目录下的一组"生产模式"示例展开，面向要把知识库从 Demo 推向真实服务的开发者。读完本文，你将掌握四类生产级技能：用一次批量调用摄入 PDF、网页与纯文本等多源内容并支撑混合检索；用内容数据库 + skip_if_exists 管理知识的插入、去重、删除与状态；在多租户共享向量库场景下用 isolate_vector_search 实现租户级数据隔离；以及通过 allowed_hosts 白名单加固 URL 抓取类 Reader，防范 SSRF 攻击。

示例总览与前置条件

该目录下共 5 个生产模式示例，对应 README 中的示例矩阵：

文件	演示主题
01_multi_source_rag.py	一次批量调用同时从本地文件、远程 URL 与纯文本加载内容
02_knowledge_lifecycle.py	插入、已存在则跳过、删除与状态跟踪
03_multi_tenant.py	借助 isolate_vector_search 实现按租户隔离知识
04_error_handling.py	幂等插入、批量错误处理与入库后的可用性验证
05_ssrf_allowed_hosts.py	用 allowed_hosts 限制 URL 抓取型 Reader 的目标主机以预防 SSRF

运行这些示例有两个前置条件：

1. 启动 Qdrant 向量库：执行仓库脚本 cookbook/scripts/run_qdrant.sh（多数示例默认连接 http://localhost:6333）；
2. 配置环境变量 OPENAI_API_KEY——示例中的 Embedder 使用 OpenAI text-embedding-3-small，Agent 模型使用 gpt-5.2。

唯一的例外是 SSRF 示例 05_ssrf_allowed_hosts.py：它选用进程内运行的 LanceDB（持久化到本地 tmp/lancedb_ssrf_demo 目录），因此无需启动任何外部服务。

典型的运行方式（以 01 为例）：

.venvs/demo/bin/python cookbook/07_knowledge/03_production/01_multi_source_rag.py

多源 RAG：一次调用摄入 PDF、网页与文本

真实生产环境中，Agent 的知识往往同时来自 PDF、网页、文本片段乃至数据库。01 示例演示了如何把不同类型的来源写入同一个知识库，核心手段是 ainsert_many（异步批量插入）。

knowledge = Knowledge(
    vector_db=Qdrant(
        collection="multi_source_rag",
        url=qdrant_url,
        search_type=SearchType.hybrid,
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)

await knowledge.ainsert_many(
    [
        {
            "name": "Candidate Resume",
            "path": "cookbook/07_knowledge/testing_resources/cv_1.pdf",
            "metadata": {"source": "resume", "department": "engineering"},
        },
        {
            "name": "Thai Recipes",
            "url": "https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
            "metadata": {"source": "web", "topic": "cooking"},
        },
        {
            "name": "Company Policy",
            "text_content": "All employees must complete security training annually. ...",
            "metadata": {"source": "internal", "topic": "policy"},
        },
    ]
)

可见每个"来源字典"以 name 命名内容，并通过互斥的关键字之一指明来源形态：

• path：本地文件路径（上例指向仓库内的真实 PDF 测试资源 cv_1.pdf）；
• url：可公开访问的远程 URL（Reader 会自动抓取并按文件类型解析）；
• text_content：直接内联的纯文本，适合注入内部规则、策略等碎片化内容；
• 三者之上还可叠加 metadata 字典，用于为后续元数据过滤与来源追踪打标。

需要指出的是，knowledge.py 中 insert_many / ainsert_many 实际支持两种调用形态：既支持上面这种"传入内容字典列表"的形态，也支持 paths=[...]、urls=[...]、text_contents=[...] 的关键字形态；两种形态都会逐条转发到单条 insert / ainsert 处理。每个字典可独立覆盖 upsert、skip_if_exists、user_id 等参数。示例随后用同一个 Agent 跨不同主题查询（简历技能、报销政策），验证多来源内容确实能在一个知识库内被统一检索。

知识生命周期：contents 数据库与幂等管理

生产环境中知识需要随时间演进：重复内容应跳过、过期内容应删除、已摄入内容的当前状态应可追踪。02 示例演示了完整的生命周期管理，关键是为 Knowledge 配置 contents_db（内容数据库）：

knowledge = Knowledge(
    name="Lifecycle Demo",
    vector_db=Qdrant(collection="lifecycle_demo", ...),
    contents_db=SqliteDb(db_file="tmp/agent.db"),
)

从源码实现看，Knowledge 同时接受 contents_db 与其读写别名 content_db（二者必须指向同一数据库对象，否则抛 ValueError）。内容数据库记录"哪些内容已被摄入、当前状态与元数据"，是去重与追踪的基础。

第一步：首次插入

await knowledge.ainsert(
    name="Recipes",
    url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
)

第二步：已存在则跳过

await knowledge.ainsert(
    name="Recipes",
    url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
    skip_if_exists=True,   # 内容哈希一致时不会重复处理
)

关键机制在于内容指纹。在 insert 的实现 中，每条内容都会先通过 _build_content_hash 生成内容哈希 content_hash，再用该哈希派生稳定 ID（generate_id(content.content_hash)）。当 skip_if_exists=True 且同名/同内容哈希的记录已存在时，插入被直接跳过，避免对文档重复切分与重复调用 Embedding API。

第三步：按名称删除向量

await knowledge.aremove_vectors_by_name("Recipes")

与之对应，Knowledge 类暴露了同步的 remove_vectors_by_name(name) 方法，按内容名称从向量库中删除对应向量。README 与示例代码共同提示：完整生命周期还应包括内容变更后的"重新索引（re-index）"，这正是 contents_db 记录哈希与状态的用武之地——你可以据此判断哪些内容真的变了、需要重建向量。

多租户隔离：共享向量库中的 isolate_vector_search

当多个 Knowledge 实例共享同一个向量 collection（常见于不同用户或部门共用一套基础设施，但只应检索自己的文档），需要靠 isolate_vector_search 保证隔离。03 示例中两个租户共享同一个 Qdrant 实例：

vector_db = Qdrant(
    collection="multi_tenant",
    url=qdrant_url,
    search_type=SearchType.hybrid,
    embedder=OpenAIEmbedder(id="text-embedding-3-small"),
)

tenant_a_knowledge = Knowledge(
    name="Tenant A",
    vector_db=vector_db,
    isolate_vector_search=True,
)
tenant_b_knowledge = Knowledge(
    name="Tenant B",
    vector_db=vector_db,
    isolate_vector_search=True,
)

两条语义必须理解清楚（README 原文要点）：

• isolate_vector_search=False（默认）：搜索向量库中的全部向量；
• isolate_vector_search=True：只搜索被打上当前实例 name 标签的向量。

重要前提：启用隔离后，存量数据若没有 linked_to 元数据，将无法被检索到，必须重新索引（re-index）以补齐该元数据。

源码层面对这一语义的实现非常直观：字段默认值为 False，是为了向后兼容既有存量数据（见 knowledge.py 的注释）；在检索路径上，_inject_instance_scope_filter 会在 isolate_vector_search=True 且 name 存在时，自动向用户传入的过滤器注入 linked_to=<instance name> 条件——无论过滤器为空、是字典还是 FilterExpr 列表，都会正确合并，从而保证"每个实例只搜到自己的数据"。仓库的单元测试 test_knowledge_isolation.py 对该隔离行为有覆盖验证。示例通过两个 Agent 分别询问"我们用哪个数据库""我们用什么云厂商"，直观展示各自只命中自己的文档。

错误处理与容错模式

04 示例针对生产中最常见的三类故障给出了可复用模式：内容加载失败、向量库连接异常、大批量文档部分失败。

# PATTERN 1: 幂等插入 —— 可安全重复调用，不会重复处理已存在内容
await knowledge.ainsert(
    name="Recipes",
    url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf",
    skip_if_exists=True,
)

# PATTERN 2: 批量插入 + 逐条错误日志 —— 单条失败不阻断整批
for source in sources:
    try:
        await knowledge.ainsert(**source)
        print("Inserted: %s" % source["name"])
    except Exception as e:
        logger.error("Failed to insert %s: %s", source["name"], e)

# PATTERN 3: 摄入后验证 —— 用真实查询确认知识可被 Agent 检索
agent.print_response("What do you know?", stream=True)

模式 1 依赖前面介绍的内容哈希机制实现幂等；模式 2 采用"逐条 try/except + 结构化日志"的方式，让一批中的合法来源正常入库，失败来源既不会中断整批，也能留下可定位的错误记录；模式 3 则是"摄入后验证"的经典手法：不只看写入是否成功，还要用一次真实检索确认内容真正进入了可被 Agent 使用的检索链路。

值得注意的是，错误处理并非只有示例脚本这一层。框架侧在 Knowledge 上额外提供了两个针对 Embedding 故障的调优参数：max_embedding_retries（默认 0，即失败不重试）与 embedding_retry_backoff（默认 1.0 秒，之后每次等待翻倍）。默认关闭重试是刻意的工程取舍——重试意味着整篇文档被重新 Embedding，大文件在后段失败会重复计费，且并发 worker 重试时往往正好撞上触发限流的同一速率瓶颈。仓库亦有配套的失败处理测试（如 test_embedding_failure_handling.py）印证这类场景。

SSRF 加固：用 allowed_hosts 约束 URL 抓取型 Reader

最后一个生产模式聚焦安全。凡是会抓取任意 URL 的知识 Reader——WebsiteReader、FirecrawlReader、DoclingReader、LLMsTxtReader、WebSearchReader——都支持一个可选启用（opt-in）的 allowed_hosts 参数，用于把对外请求限制在主机名白名单内。

生产环境必须关注它有两个直接原因（README 原文要点）：

1. AgentOS 暴露了 POST /knowledge/content 接口，它接受一个 URL 并调度后台抓取。若不加白名单，攻击者可以利用该入口探测/访问内部服务；
2. 白名单同时作用于每一次重定向目标（通过 httpx 的 request 事件钩子实现），因此一个被允许的主机无法用 3xx 跳转把请求"甩"到内网地址。

05 示例以 WebsiteReader 演示四种行为：

# CASE 1: 命中白名单 -> 正常摄入
reader = WebsiteReader(max_depth=1, max_links=5, allowed_hosts=["docs.agno.com"])
await knowledge.ainsert(
    name="Agno Docs",
    url="https://docs.agno.com/introduction",
    reader=reader,
)

# CASE 2: 不在白名单 -> 直接拒绝，根本不发出请求
for ssrf_target in (
    "http://127.0.0.1:8000/admin",
    "http://10.0.0.5/internal",
    "http://169.254.169.254/latest/meta-data/iam/security-credentials/",
):
    documents = reader.read(ssrf_target)
    print(f"  {ssrf_target} -> {len(documents)} documents (refused)")

示例点名的三类典型 SSRF 目标是：本机回环服务（127.0.0.1）、RFC1918 内网网段（如 10.0.0.5）、以及云元数据端点 169.254.169.254（攻击者常借其窃取 IAM 临时凭据）。

# CASE 3: 不设置 allowed_hosts -> 完全放行（历史默认行为，无任何策略）
permissive_reader = WebsiteReader(max_depth=1, max_links=2)
print(f"  allowed_hosts is {permissive_reader.allowed_hosts}")

白名单在源码中的落地方式

白名单并非仅作用于"最初提交的 URL"。其实现集中在 libs/agno/agno/knowledge/reader/utils/url_validation.py：

• validate_allowed_hosts：校验入参，若误传单个字符串会抛出 TypeError 并提示应传列表；随后统一转小写；
• is_host_allowed：用 urlparse 提取主机名做大小写不敏感、精确匹配（不隐式包含子域）的比较；allowed_hosts 为 None 时放行所有主机；
• make_redirect_guard / make_async_redirect_guard：为 httpx 构造请求事件钩子。由于 httpx 在跟随 3xx 重定向时对每个出站请求都会触发 request 事件钩子，配合 follow_redirects=True，合法同主机重定向可正常工作，而跳向白名单外主机的请求会在真正发出前被拦截并抛错；allowed_hosts 为 None 时返回 None，即默认放行路径完全不挂钩子。

从调用侧看，各 URL 抓取 Reader 均在读取/抓取前调用 is_host_allowed 做前置校验，例如 firecrawl_reader.py（拒绝抓取）、docling_reader.py（拒绝读取）、llms_txt_reader.py（校验后再挂重定向守卫）。

同样的开关也存在于其他 URL 抓取 Reader，示例代码给出了对应构造方式：

# FirecrawlReader(api_key=..., allowed_hosts=["docs.agno.com"])
# DoclingReader(allowed_hosts=["docs.agno.com"])
# LLMsTxtReader(allowed_hosts=["docs.agno.com"])
# WebSearchReader(allowed_hosts=["docs.agno.com", "github.com"])

小结与延伸阅读

这 5 个生产模式覆盖了知识系统从"能用"到"好用、可靠、安全"的关键路径：多源批量摄入解决内容形态碎片化；skip_if_exists + contents_db 解决重复摄入与状态追踪；isolate_vector_search 在共享存储上落地租户隔离；逐条容错与摄入后验证提升鲁棒性；allowed_hosts 则在系统暴露 URL 抓取能力时筑起 SSRF 防线。

建议按序研读源码继续深入：核心类与方法在 libs/agno/agno/knowledge/knowledge.py（重点看 insert/insert_many、_inject_instance_scope_filter、remove_vectors_by_name 与 contents_db 参数），URL 校验机制在 url_validation.py，隔离行为的单测覆盖可参考 test_knowledge_isolation.py。若想从基础开始补齐知识库的构建知识，可回到 cookbook/07_knowledge 的入门与构建块章节循序渐进。