agno Knowledge 集成实战：从多格式读取器、云存储到向量数据库的完整接入指南

导读

在 agno 中，Knowledge 负责将外部数据（本地文件、网页、云存储对象等）切分、向量化后写入向量数据库，供 Agent 在回答问题时检索引用。本篇技术指南以 cookbook/07_knowledge/05_integrations 为核心，系统讲解三类核心集成——Readers（多格式读取器）、Cloud Storage（云存储内容源） 与 Vector Databases（向量数据库）：既覆盖每类集成支持的具体格式/厂商与配置参数，也结合配套示例脚本说明每行关键代码的作用，并串联 agno 底层模块的调用关系。读完本文，你将能够在不同数据源与不同检索后端之间自由组合，搭建一套可本地原型、可上生产的知识库 + RAG Agent。

一、集成全景与目录速览

集成示例位于 cookbook/07_knowledge/05_integrations，按功能拆分为三个子目录。下表整理自该目录的 README.md，是理解后续内容的地图：

Readers（数据读取器）

文件	支持格式 / 来源
readers/01_documents.py	PDF、DOCX、PPTX、Excel
readers/02_data.py	CSV、JSON
readers/03_web.py	网站、YouTube、ArXiv、Firecrawl

Cloud Storage（云存储内容源）

文件	提供商
cloud/01_aws.py	S3 存储桶
cloud/02_azure.py	Azure Blob Storage
cloud/03_gcp.py	Google Cloud Storage
cloud/04_sharepoint.py	SharePoint

Vector Databases（向量数据库）

文件	数据库
vector_dbs/01_qdrant.py	Qdrant（生产环境推荐）
vector_dbs/02_local.py	ChromaDB + LanceDB（本地开发）
vector_dbs/03_managed.py	Pinecone（托管/生产）
vector_dbs/04_pgvector.py	PgVector（PostgreSQL）
vector_dbs/05_scylladb.py	ScyllaDB

提示：该目录还包含 README 表格之外的同主题扩展示例，如 cloud/02_azure_sas.py、cloud/05_github_dynamic_repo.py、cloud/06_multi_source.py（多源混合），以及 readers/docling 目录下的高性能文档解析器示例，可作进阶阅读。

二、通用前置条件：先满足，再运行

所有集成示例共享同一套运行前提（见 README.md 的 Prerequisites 部分）：

1. 启动 Qdrant：运行仓库脚本 ./cookbook/scripts/run_qdrant.sh（脚本位于 scripts 目录，run_qdrant.sh），它会以 Docker 方式在本地拉起 Qdrant 服务，默认监听 http://localhost:6333。
2. 配置 OPENAI_API_KEY 环境变量：绝大多数示例的嵌入向量与对话模型默认走 OpenAI，其中嵌入模型统一使用 OpenAIEmbedder(id="text-embedding-3-small")，对话模型为 OpenAIResponses(id="gpt-5.2")。
3. 云存储场景：按各脚本头注释设置对应云厂商的凭证环境变量（AWS、Azure AD、GCP 服务账号等）。
4. 托管数据库场景：先安装对应的 Python 驱动包（如 pip install pinecone、pip install chromadb），脚本内部用 try/except ImportError 兜底，未安装时会跳过演示并打印安装提示。

运行方式可直接复用 README 给出的命令模板（使用仓库预设的 demo 虚拟环境）：

.venvs/demo/bin/python cookbook/07_knowledge/05_integrations/readers/01_documents.py

把末尾脚本路径替换成 vector_dbs 或 cloud 下的任意示例即可。每个示例都自带 if __name__ == "__main__" 入口，既能被直接执行，也能作为模块被导入复用。

三、Readers：让不同文件类型自动适配正确的解析器

Readers 负责把非结构化文档与结构化数据读取为纯文本片段。核心机制是 Knowledge 会自动按文件扩展名/URL 后缀识别类型并挑选合适的 reader；你也可以显式传入 reader 实例，以便对解析行为做更精细控制（该行为在两个示例脚本的模块注释中均有说明）。

3.1 文档读取器：PDF / DOCX / PPTX / Excel

readers/01_documents.py 覆盖四类最常见的办公文档：

• PDF：文本抽取，可选 OCR；
• DOCX：Microsoft Word 文档；
• PPTX：PowerPoint 演示文稿；
• Excel：.xlsx 与 .xls 电子表格。

对应的 agno 底层读取器分别位于 agno.knowledge.reader 下的 pdf_reader、docx_reader、pptx_reader 与 excel_reader 模块——示例脚本中显式导入了 ExcelReader，其余三个以注释形式列出，说明它们是"自动检测"链路中的默认选择。

脚本先构建一个挂载了 Qdrant 向量库的 Knowledge 对象，再创建开启 search_knowledge=True 的 Agent：

knowledge = Knowledge(
    vector_db=Qdrant(
        collection="document_readers",
        url="http://localhost:6333",
        search_type=SearchType.hybrid,              # 混合检索：向量 + 关键词
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)

agent = Agent(
    model=OpenAIResponses(id="gpt-5.2"),
    knowledge=knowledge,
    search_knowledge=True,   # 允许 Agent 检索知识库
    markdown=True,
)

随后通过异步 API knowledge.ainsert(...) 三种方式喂入数据，正好覆盖"自动检测"的两种触发场景：

# (1) PDF：根据本地路径的 .pdf 扩展名自动选择 PDFReader
await knowledge.ainsert(name="CV", path="cookbook/07_knowledge/testing_resources/cv_1.pdf")
agent.print_response("What skills does Jordan Mitchell have?", stream=True)

# (2) Excel：显式指定 reader，便于控制解析行为
await knowledge.ainsert(
    name="Products",
    path="cookbook/07_knowledge/testing_resources/sample_products.xlsx",
    reader=ExcelReader(),
)
agent.print_response("What products are listed?", stream=True)

# (3) 远程 PDF：URL 以 .pdf 结尾，同样自动检测
await knowledge.ainsert(name="Recipes", url="https://agno-public.s3.amazonaws.com/recipes/ThaiRecipes.pdf")
agent.print_response("What Thai recipes are available?", stream=True)

可见 ainsert 的三种数据入口设计：name + path（本地文件）、name + text_content（纯文本/原始内容）、name + url（远程资源）。实测用到的两个样例文件都位于仓库的 testing_resources 目录（含 cv_1.pdf、sample_products.xlsx），可直接运行复现。

3.2 数据读取器：CSV / JSON

readers/02_data.py 面向结构化数据，与文档读取器的差异在于处理粒度：CSV 会逐行作为独立文档入库，JSON 则按文件/数组整体处理。示例通过 text_content 直接注入内存中的文本，无需落盘：

from agno.knowledge.reader.csv_reader import CSVReader
from agno.knowledge.reader.json_reader import JSONReader

# CSVReader：将每一行解析成一条独立记录（列名保留在输出中）
await knowledge.ainsert(
    name="Sample Data",
    text_content="name,role,department\nAlice,Engineer,Platform\nBob,Designer,Product\nCarol,Manager,Engineering",
    reader=CSVReader(),
)
agent.print_response("Who works in engineering?", stream=True)

# JSONReader：把整个 JSON 对象作为结构化文档入库
await knowledge.ainsert(
    name="Config",
    text_content='{"app": "acme", "version": "2.0", "features": ["auth", "billing", "analytics"]}',
    reader=JSONReader(),
)
agent.print_response("What features does the app have?", stream=True)

这种设计对"以行为主"的表格数据（员工表、流水明细）和"整体语义"的 JSON 配置/日志非常友好，检索命中后，Agent 能把每一行的字段还原成可回答的结构化依据。另外 README 的表格提到 field-labeled CSV 场景，即输出时把列名作为标签附在每行内容之前，便于大模型理解字段语义。

3.3 网页读取器：Website / YouTube / ArXiv / Firecrawl

readers/03_web.py 把触角伸向 Web，共列出四类来源：WebsiteReader 抓取网页、YouTubeReader 抽取视频字幕、ArxivReader 拉取学术论文、FirecrawlReader 通过 Firecrawl API 做深度网页抓取。演示代码只显式导入了 WebsiteReader，且展示了可配置的爬取参数：

from agno.knowledge.reader.website_reader import WebsiteReader

# max_depth=1 只抓当前页，max_links=5 最多跟随 5 个链接
website_reader = WebsiteReader(max_depth=1, max_links=5)
await knowledge.ainsert(
    name="Agno Docs",
    url="https://docs.agno.com/introduction",
    reader=website_reader,
)
agent.print_response("What is Agno?", stream=True)

其中 max_depth 控制爬取层级、max_links 控制跟随链接数量的上限，是避免无限爬取的关键护栏。脚本后半部分再次演示了"直接 URL 自动检测"——以 .pdf、.md、.txt 等后缀结尾的 URL 会自动匹配对应读取器，无需显式指定。

四、Cloud Storage：把对象存储作为远程内容源

当你不想把文件下载到本地再入库时，可以用 agno 的 remote_content 机制，把云对象存储（bucket / container）配置为 Content Source，再通过 s3_config.file(...)、s3_config.folder(...) 这类方法在 ainsert 时按需指定单文件或整目录。其统一模式在 AWS / Azure / GCP / SharePoint 四个脚本中几乎一致，下面以差异点为主线展开。

4.1 AWS S3

cloud/01_aws.py 支持加载单个文件或递归加载整个前缀（文件夹），支持任何 S3 兼容存储，并为每个文件记录 bucket/key/region 等元数据。脚本开头注释明确了所需环境变量：

AWS_ACCESS_KEY_ID     - AWS access key
AWS_SECRET_ACCESS_KEY - AWS secret key
AWS_REGION            - AWS region（默认 us-east-1）

配置与入库的关键代码如下：

from agno.knowledge.remote_content import S3Config

s3_config = S3Config(
    id="my-bucket",
    name="My S3 Bucket",
    bucket_name=getenv("AWS_S3_BUCKET", "my-bucket"),
    region=getenv("AWS_REGION", "us-east-1"),
)

knowledge = Knowledge(
    name="S3 Knowledge",
    vector_db=Qdrant(collection="s3_knowledge", url="http://localhost:6333"),
    content_sources=[s3_config],        # 声明可用内容源
)

# 单文件：reports/quarterly-report.pdf
await knowledge.ainsert(name="Report", remote_content=s3_config.file("reports/quarterly-report.pdf"))
# 整目录：reports/ 前缀下所有对象
await knowledge.ainsert(name="All Reports", remote_content=s3_config.folder("reports/"))

注意这里 Qdrant 省略了 embedder，说明向量库的嵌入器是可继承默认配置的；而 content_sources=[...] 把远端源预先登记进 Knowledge。检索时可直接调用 knowledge.search(...) 遍历命中文档名称，输出 doc.name。

4.2 Azure Blob Storage

cloud/02_azure.py 使用 Azure AD 应用注册（App Registration）的客户端凭证认证，前置要求是应用具有 Storage Blob Data Reader 角色。所需环境变量：

AZURE_TENANT_ID            - Azure AD tenant ID
AZURE_CLIENT_ID            - 应用注册 client ID
AZURE_CLIENT_SECRET        - 应用注册 client secret
AZURE_STORAGE_ACCOUNT_NAME - 存储账号名
AZURE_CONTAINER_NAME       - 容器名

代码骨架与 S3 相同，只是把 S3Config 换成 AzureBlobConfig，并用 .file("reports/annual-report.pdf")、.folder("documents/") 指定单文件与整个前缀：

from agno.knowledge.remote_content import AzureBlobConfig

azure_blob = AzureBlobConfig(
    id="company-blob",
    name="Company Blob Storage",
    tenant_id=getenv("AZURE_TENANT_ID"),
    client_id=getenv("AZURE_CLIENT_ID"),
    client_secret=getenv("AZURE_CLIENT_SECRET"),
    storage_account=getenv("AZURE_STORAGE_ACCOUNT_NAME"),
    container=getenv("AZURE_CONTAINER_NAME"),
)
knowledge = Knowledge(
    name="Azure Blob Knowledge",
    vector_db=Qdrant(collection="azure_blob_knowledge", url="http://localhost:6333"),
    content_sources=[azure_blob],
)

同目录下的 cloud/02_azure_sas.py 提供了基于 SAS 令牌的替代认证方案，供无法使用 AD 凭证的受限环境参考。

4.3 Google Cloud Storage

cloud/03_gcp.py 面向 GCS，支持服务账号密钥或 Application Default Credentials 两种认证方式：

GOOGLE_APPLICATION_CREDENTIALS - 服务账号密钥文件路径
GCS_BUCKET_NAME               - GCS bucket 名

接入类为 GcsConfig，用法与前两者对称：

from agno.knowledge.remote_content import GcsConfig

gcs_config = GcsConfig(
    id="my-gcs-bucket",
    name="My GCS Bucket",
    bucket_name=getenv("GCS_BUCKET_NAME", "my-bucket"),
)
knowledge = Knowledge(
    name="GCS Knowledge",
    vector_db=Qdrant(collection="gcs_knowledge", url="http://localhost:6333"),
    content_sources=[gcs_config],
)
# gcs_config.file("reports/quarterly.pdf")  /  gcs_config.folder("reports/")

4.4 SharePoint

cloud/04_sharepoint.py 面向企业内部文档库，同样走 Azure AD 客户端凭证，但要求应用注册拥有 Sites.Read.All 权限，并额外提供 SharePoint 主机名：

AZURE_TENANT_ID     - Azure AD tenant ID
AZURE_CLIENT_ID     - 应用注册 client ID
AZURE_CLIENT_SECRET - 应用注册 client secret
SHAREPOINT_HOSTNAME - SharePoint 主机名（如 contoso.sharepoint.com）

对应配置类是 SharePointConfig：

from agno.knowledge.remote_content import SharePointConfig

sharepoint = SharePointConfig(
    id="company-sharepoint",
    name="Company SharePoint",
    tenant_id=getenv("AZURE_TENANT_ID"),
    client_id=getenv("AZURE_CLIENT_ID"),
    client_secret=getenv("AZURE_CLIENT_SECRET"),
    hostname=getenv("SHAREPOINT_HOSTNAME"),
)
knowledge = Knowledge(
    name="SharePoint Knowledge",
    vector_db=Qdrant(collection="sharepoint_knowledge", url="http://localhost:6333"),
    content_sources=[sharepoint],
)
# sharepoint.file("Shared Documents/policy.pdf")  /  sharepoint.folder("Shared Documents/")

4.5 多源与动态源（进阶）

同一目录还提供了扩展场景：cloud/05_github_dynamic_repo.py 演示按需动态拉取 GitHub 仓库内容；cloud/06_multi_source.py 演示在同一 Knowledge 中注册多个内容源并混合检索。当企业数据散落在 S3、Azure、SharePoint 与 GitHub 时，用 content_sources=[...] 一次性声明即可统一建库，不必维护多条入库管线。

五、Vector Databases：从本地开发到生产部署的选型阶梯

向量数据库是 Knowledge 的检索后端。agno 对常见后端做了统一抽象，示例脚本把不同场景的选型逻辑讲得很清楚，按"从易到难、从本地到生产"的顺序排列：

5.1 Qdrant：生产环境推荐（默认）

vector_dbs/01_qdrant.py 明确标注 Qdrant 为生产推荐，理由包括：支持向量检索、关键词检索与混合检索三类模式，支持 reranking，具备丰富的元数据过滤能力，且可云托管或自托管。脚本同时给出了两档配置：

# 基础档：默认向量检索
knowledge_basic = Knowledge(
    vector_db=Qdrant(
        collection="qdrant_basic",
        url="http://localhost:6333",
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
    ),
)

# 进阶档：混合检索 + Cohere 重排
from agno.vectordb.qdrant import SearchType
from agno.knowledge.reranker.cohere import CohereReranker

knowledge_advanced = Knowledge(
    vector_db=Qdrant(
        collection="qdrant_advanced",
        url="http://localhost:6333",
        search_type=SearchType.hybrid,                 # 混合检索
        embedder=OpenAIEmbedder(id="text-embedding-3-small"),
        reranker=CohereReranker(model="rerank-multilingual-v3.0"),  # 语义重排
    ),
)

SearchType.hybrid 同时融合向量相似度与关键词命中，能显著提升专有名词与精确匹配场景的召回；reranker 参数则在召回后对候选文档做二次精排。这一档配置也是第三、四节所有示例使用的默认形态（只有少数示例省略了 embedder，让向量库使用默认嵌入器）。

5.2 ChromaDB + LanceDB：本地开发的无服务器选择

vector_dbs/02_local.py 面向原型验证，两类数据库都不需要独立服务器：

• ChromaDB：内存或持久化存储，pip install chromadb，适合快速原型；
• LanceDB：纯文件存储（无服务器），原生支持混合检索，pip install lancedb。

由于是可选依赖，脚本用 try/except ImportError 包裹导入，未安装时置空并提示安装命令：

try:
    from agno.vectordb.chroma import ChromaDb
    knowledge_chroma = Knowledge(
        vector_db=ChromaDb(
            collection="local_demo",
            embedder=OpenAIEmbedder(id="text-embedding-3-small"),
        ),
    )
except ImportError:
    knowledge_chroma = None
    print("ChromaDB not installed. Run: pip install chromadb")

try:
    from agno.vectordb.lancedb import LanceDb, SearchType
    knowledge_lance = Knowledge(
        vector_db=LanceDb(
            uri="tmp/lancedb",            # 本地文件目录
            table_name="local_demo",
            search_type=SearchType.hybrid,
            embedder=OpenAIEmbedder(id="text-embedding-3-small"),
        ),
    )
except ImportError:
    knowledge_lance = None
    print("LanceDB not installed. Run: pip install lancedb")

ChromaDB 默认运行在内存、LanceDb 则把数据落在 uri 指定的目录，两种模式切换几乎只改一处 vector_db 参数，是典型的"开发环境与生产环境用同一套 Knowledge API"的体现。

5.3 Pinecone：零运维的托管服务

vector_dbs/03_managed.py 面向希望完全托管基础设施的生产场景。Pinecone 提供 serverless 选项、自动扩缩容与高可用、元数据过滤以及 Namespace 多租户隔离。接入类为 PineconeDb，需要 PINECONE_API_KEY：

try:
    from agno.vectordb.pineconedb import PineconeDb

    knowledge_pinecone = Knowledge(
        vector_db=PineconeDb(
            name="knowledge-demo",
            api_key=getenv("PINECONE_API_KEY"),
            embedder=OpenAIEmbedder(id="text-embedding-3-small"),
        ),
    )
except ImportError:
    knowledge_pinecone = None
    print("Pinecone not installed. Run: pip install pinecone")

说明：该目录还包含 vector_dbs/04_pgvector.py，用于把知识库直接放进 PostgreSQL 的 pgvector 扩展中，适合已有 PostgreSQL 基础设施、希望向量与业务数据同库管理的团队（PostgreSQL 相关驱动与运行脚本可参见仓库 run_pgvector.sh）。

5.4 ScyllaDB：复用 Cassandra 集成的高性能实时方案

vector_dbs/05_scylladb.py 演示了 agno 集成体系的一个重要设计：ScyllaDB 可复用现有 Cassandra 集成，通过启动参数开启兼容模式即可接入。ScyllaDB 主打高可用、多区域分布与向量 ANN 检索，适合对读写延迟和吞吐有高要求的场景。

自托管 ScyllaDB 的启动命令（脚本头部注释给出）必须开启 cassio 兼容：

docker run -d --name scylla \
    -p 9042:9042 \
    scylladb/scylla:latest \
    --developer-mode=1 \
    --enable-cassio-compatibility=1

依赖安装为 pip install scylla-driver cassio。接入代码会先连接集群并确保 keyspace 存在，再以 Cassandra 向量库承接知识：

from agno.vectordb.cassandra import Cassandra
from cassandra.cluster import Cluster

cluster = Cluster(["127.0.0.1"], port=9042)
session = cluster.connect()
session.execute("CREATE KEYSPACE IF NOT EXISTS agno_knowledge;")

# 向量维度由嵌入器决定；text-embedding-3-small 支持通过 API 调整输出维度
embedder = OpenAIEmbedder(id="text-embedding-3-small", dimensions=1024)

knowledge = Knowledge(
    vector_db=Cassandra(
        table_name="thai_recipes",
        keyspace="agno_knowledge",
        session=session,
        embedder=embedder,
    ),
)

脚本注释特别提醒：向量表的列维度由 embedder 的输出维度决定，这里显式设置 dimensions=1024 以便与后续写入的向量对齐；同时注明示例使用单节点便于演示，生产环境建议至少三个节点。

六、集成示例背后的通用运行链路

把上述三类示例放在一起看，会发现 agno Knowledge 集成的代码模式高度统一，这也是它能"无痛"切换数据源与后端的原因。从示例脚本可以归纳出如下调用关系：

1. 构建 Knowledge：传入 vector_db（VectorDB 实现）与可选的 content_sources（远程内容源列表）；
2. 注册检索模型：通过 embedder（如 OpenAIEmbedder）把文本转成向量；Qdrant 等实现还支持 search_type 与 reranker 两个检索增强开关；
3. 写入数据：统一调用 knowledge.insert(...) / knowledge.ainsert(...)（异步），数据入口可以是 path、url、text_content 或 remote_content（来自 S3Config / AzureBlobConfig / GcsConfig / SharePointConfig 的 .file() / .folder()）；
4. 交给 Agent 使用：Agent(knowledge=..., search_knowledge=True) 使模型在回答时自动查询知识库，并把检索结果作为上下文生成引用式回答。

reader 参数的显式传值、SearchType 与 reranker 的配置、以及 embedder 维度的对齐，是三个最容易影响效果的细节：显式 reader 控制解析粒度，混合检索改善专名召回，embedder 维度则必须与向量库 schema 匹配（ScyllaDB 示例即是一例）。

仓库中还提供了若干可以横向扩展阅读的资料：

• cookbook/07_knowledge 总目录下的 Getting Started 与 Building Blocks 等子目录，讲解 Knowledge 与 VectorDB 的基础组装方式；
• 同目录的 rag 子目录收录了 agentic_rag_with_lightrag.py、agentic_rag_infinity_reranker.py、local_rag_langchain_qdrant.py 等端到端 RAG 案例，可视为本目录三类基础集成组合后的进阶应用；
• readers/docling 子目录展示基于 docling 的音频、图片、Markup 等更多输入类型的解析方案。

结语

agno 的 Knowledge 集成层把"读什么"（Readers）、"从哪里读"（Cloud Storage）与"存到哪、怎么搜"（Vector Databases）解耦成三个可自由组合的维度：同一套 Knowledge + Agent 代码，既可以用 LanceDB 在笔记本上五分钟跑通原型，也可以无缝切换到 Qdrant/Pinecone/PgVector/ScyllaDB 承载生产流量。建议你按以下顺序动手验证：先运行 vector_dbs/01_qdrant.py 确认 Qdrant 链路，再用 readers/01_documents.py 体验格式自动检测，最后按自己的数据所在地选择对应的 cloud 示例，即可把整套集成方案落地到真实业务。