Agno 图像提取到向量数据库：用 Agent 构建可搜索媒体库的完整流水线

在 agno 项目中，cookbook/data_labeling/_09_image_extraction_to_vectordb/ 演示了一条经典的数据标注（labeling）工作流：先用多模态 Agent 从图片中抽取结构化的文字描述，再把描述文本嵌入向量并存入向量数据库，最终通过自然语言查询实现相似图片检索——这正是"构建可搜索媒体库"的标准范式。读完本文，你将掌握如何用 agno 的 Agent、GeminiEmbedder 与 LanceDb 在一个文件内搭起"提取 → 嵌入 → 存储 → 检索"的端到端流水线，并理解其底层实现细节与适用场景。

流水线总览：四条核心步骤

该示例的完整入口是 basic.py，文件虽只有一个，但价值在于它串起了整条链路，共四个环节：

1. 结构化提取：针对每张图片的 URL，一个多模态 Agent 调用视觉模型，抽取一个结构化的 ImageDescription（主体 subject、场景 setting、氛围 mood、关键物体 key_objects）；
2. 文本扁平化：将结构化字段拼装成一段可检索的纯文本；
3. 嵌入与入库：用 GeminiEmbedder 将文本嵌入为向量，存储到 LanceDb 表 data_labeling_images；
4. 相似检索：用自然语言查询向量索引，返回最相似的图片文档。

关键设计点：本流水线检索的是"图片的文本描述向量"，而非图片本身的视觉嵌入。两张图片是否匹配，取决于它们的文字描述是否相近。这一点会在"适用场景"一节中展开讨论。

第一步：定义结构化输出 Schema

提取阶段的核心是让 Agent 输出可预测、可入库的结构化结果。示例使用 Pydantic 模型定义输出契约：

from typing import List
from pydantic import BaseModel, Field

class ImageDescription(BaseModel):
    subject: str = Field(..., description="The main subject of the image")
    setting: str = Field(..., description="Where the image takes place")
    mood: str = Field(..., description="Overall mood or tone")
    key_objects: List[str] = Field(
        default_factory=list, description="Up to five notable objects"
    )

四个字段分工明确：

字段	类型	语义
subject	str（必填）	图片主体，如"被纺织会馆拱门框住的圣玛丽教堂"
setting	str（必填）	场景地点，如图片拍摄发生的环境
mood	str（必填）	整体氛围或基调，如"魔幻而宁静"
key_objects	List[str]（默认空列表）	最多五个值得注意的物体

使用 output_schema=ImageDescription 后，Agent 的输出会被强制校验并反序列化为该模型实例，后续代码可以直接以属性方式访问字段，这是保证下游入库稳定的关键。

第二步：创建提取 Agent

from agno.agent import Agent

extractor = Agent(
    model="google:gemini-3.5-flash",
    instructions="You describe images as structured, search-friendly metadata.",
    output_schema=ImageDescription,
)

• model="google:gemini-3.5-flash"：多模态视觉模型，用于理解图片内容。结合本仓库的 TEST_LOG.md 可知，该示例在 agno 2.7.4 下针对 gemini-3.5-flash 实测通过。
• instructions：提示词明确要求"将图片描述为结构化、利于搜索的元数据"，这比笼统的"描述图片"更能引导模型产出适合检索的文本。
• output_schema=ImageDescription：绑定上面的 Pydantic 模型。

图片通过 agno.media.Image 传入。从源码 media.py 可见，Image 是统一的多模态输入类，url、filepath、content（原始字节）三选一且只能提供一个；本示例使用 Image(url=url) 传入远程图片地址。

第三步：配置向量数据库

from agno.vectordb.lancedb import LanceDb, SearchType

vector_db = LanceDb(
    uri="tmp/lancedb",
    table_name="data_labeling_images",
    search_type=SearchType.vector,
    embedder=GeminiEmbedder(id="gemini-embedding-001"),
)

LanceDb 关键参数

对照源码 lance_db.py 的构造函数，可以梳理出以下核心配置项：

• uri：LanceDB 数据库地址。本地路径即持久化目录，示例写入仓库根目录下的 tmp/lancedb/；也可以传 LanceDB Cloud 的 db://... 地址（此时必须提供 api_key 或设置环境变量 LANCEDB_API_KEY，否则构造函数会直接抛出 ValueError）。
• table_name：表名。若本地库中已存在同名表会直接打开复用；不存在则在首次 create() 或插入时创建。
• search_type：检索类型。由 search.py 中的枚举定义，共三种：SearchType.vector（向量检索）、SearchType.keyword（全文检索，需要 FTS 索引）、SearchType.hybrid（混合检索，会创建全文索引并合并两路结果）。示例采用纯向量检索。
• embedder：负责文本↔向量转换的嵌入器。若不传，默认回退为 OpenAIEmbedder（源码中会打印 "Embedder not provided, using OpenAIEmbedder as default." 日志）。
• distance：距离度量，默认 Distance.cosine（余弦相似度）。
• nprobes：ANN 检索时的探针数，用于权衡召回与速度。
• reranker：可选的检索后重排器。
• on_bad_vectors：遇到异常向量时的处理策略（"error" / "drop" / "fill" / "null"），配合 fill_value 使用。

另外注意：LanceDb 要求 embedder.dimensions 必须已设置，否则初始化会抛出 ValueError("Embedder.dimensions must be set.")。

GeminiEmbedder 细节

GeminiEmbedder 的默认值为：

• id="gemini-embedding-001"（示例显式指定，保持一致）；
• task_type="RETRIEVAL_QUERY"，用于告诉 Google 嵌入 API 该向量的用途（查询端）；
• dimensions=1536，即输出向量维度，同时决定了 LanceDb 表的向量列 schema。

嵌入时若设置了 dimensions，会通过 output_dimensionality 参数传给 Google API；认证方面，非 Vertex 模式下读取环境变量 GOOGLE_API_KEY，因此运行前必须配置该密钥。若想改用 Vertex AI，可设置环境变量 GOOGLE_GENAI_USE_VERTEXAI=true 并配合 GOOGLE_CLOUD_PROJECT / GOOGLE_CLOUD_LOCATION。

第四步：编写流水线核心函数

单图提取：describe

def describe(url: str) -> ImageDescription:
    return extractor.run("Describe this image.", images=[Image(url=url)]).content

Agent.run() 接收文本提示词与图片列表，.content 因绑定了 output_schema 而直接是 ImageDescription 实例。

文本扁平化：to_searchable_text

def to_searchable_text(d: ImageDescription) -> str:
    return (
        f"Subject: {d.subject}. Setting: {d.setting}. Mood: {d.mood}. "
        f"Objects: {', '.join(d.key_objects)}."
    )

这一步把结构化字段拼成一段紧凑的、语义完整的自然语言文本。以 TEST_LOG 中记录的 Krakow 提取结果为例，拼装后的文本形如：

Subject: St. Mary's Basilica framed by the arches of the Cloth Hall. Setting: ... Mood: Magical and serene. Objects: ...

扁平化的意义在于：向量模型对完整句子的语义编码效果通常优于对分散字段的分别编码，且检索时用户的自然语言查询可以与该文本直接对齐。

批量入库：index_images

def index_images(urls: List[str]) -> None:
    vector_db.create()
    docs: List[Document] = []
    for url in urls:
        description = describe(url)
        docs.append(
            Document(
                name=url,
                content=to_searchable_text(description),
                meta_data={"url": url, "subject": description.subject},
            )
        )
    vector_db.insert(content_hash="image_batch_1", documents=docs)

要点拆解：

• vector_db.create()：建表（幂等，已存在则跳过）。从源码看，LanceDb 的表 schema 为 [vector(定长 float32 列表), id(string), payload(string), user_id(string 可空)]，其中 user_id 列用于多用户数据隔离，None 表示共享内容。
• Document：agno 的文档抽象，name、content、meta_data 分别记录名称、正文与可检索的元数据。此处把图片 URL 存进 name 和 meta_data["url"]，把抽取出的 subject 也放入元数据，方便检索后溯源。
• vector_db.insert(content_hash=..., documents=...)：content_hash 是批次标识，会参与文档 ID 的确定性计算（源码中基于 md5(base_id + content_hash) 生成），因此不同批次、不同批次内同名文档都不会互相覆盖；插入时若文档尚无 embedding，会自动调用 document.embed(embedder=self.embedder) 完成嵌入。底层每一行会把 name / meta_data / content / content_hash 等序列化为 JSON 存入 payload 字段。

自然语言检索

for query in ["a historic city at night", "wildlife on the savanna"]:
    results = vector_db.search(query, limit=2)
    pprint({"query": query, "hits": [r.meta_data for r in results]})

vector_db.search(query, limit=2) 会先用同一个 GeminiEmbedder 对查询文本求嵌入（注意查询端使用 task_type="RETRIEVAL_QUERY"），再在表中做 ANN 检索并返回 Top-K 个 Document。从源码可见 search() 会根据 search_type 分流到 vector_search / keyword_search / hybrid_search；返回结果的 meta_data 即插入时写入的 {"url": ..., "subject": ...}，可直接用来展示命中图片。

运行方式与前置条件

按 README.md 与脚本头部注释，运行步骤如下：

pip install lancedb
python cookbook/data_labeling/_09_image_extraction_to_vectordb/basic.py

前置条件与说明：

1. 必须配置 GOOGLE_API_KEY 环境变量，Agent 的视觉模型与 GeminiEmbedder 都依赖 Google API；
2. 必须先安装 lancedb（以及其依赖的 pyarrow），否则导入 LanceDb 时会抛出 ImportError；
3. 数据写入仓库根目录下的 tmp/lancedb/（uri="tmp/lancedb" 是相对路径）；
4. 示例内置了 3 张公开示例图（Krakow 教堂街景、瀑布风景、非洲草原野生动物），无需自行准备图片；
5. 纯向量检索不依赖 tantivy（全文检索引擎），按 TEST_LOG 所述，未安装 tantivy 也能正常运行。

实测验证记录

仓库自带的 TEST_LOG.md 记录了 2026-07-18 在 agno 2.7.4 下从干净环境（删除 tmp/lancedb/）实测 basic.py 的结果：

• 3 张图片全部成功完成结构化提取，例如 Krakow 教堂图得到 subject="St. Mary's Basilica framed by the arches of the Cloth Hall"、mood="Magical and serene"；草原图得到 subject="A diverse group of African wildlife, including elephants, giraffes, zebras, and a crocodile, gathered at a watering hole"；
• 共插入 3 条文档；
• 查询 "a historic city at night" 命中 Krakow 教堂图居首，查询 "wildlife on the savanna" 命中草原野生动物图居首；
• 单张图片的提取耗时约 2.7～4.8 秒。

这组记录直接验证了"结构化提取 → 嵌入 → 检索"整条链路在真实模型下的可用性。

适用场景与边界

README 明确了三类典型使用场景：

• 用自然语言检索图片库：例如在正版图库中搜索"夜晚的历史城市"、"草原上的野生动物"；
• 按描述相似度给商品目录去重：注意检索基于抽取出的文字描述而非图片嵌入，因此"两张图描述一致"即视为重复，对"同一物体的不同拍摄角度/光线"这类描述趋同的图会判定为相似；
• 构建"找相似"（Find more like this）功能：对用户上传的媒体，先抽取描述入库，再以描述向量进行相似推荐。

边界在于：由于索引的是文本描述而非视觉特征，模型的描述质量直接决定检索效果。描述字段（如 mood、setting）设计得越贴近用户可能的查询用语，检索命中率越高；这也正是 instructions 强调"search-friendly metadata"的原因。

延伸：如何改造成生产级流水线

basic.py 是单文件最小实现，从 agno 的库结构看，可以沿着以下方向扩展（均可在本仓库找到对应模块）：

• 接入知识库 / AgentOS 的 knowledge 机制：LanceDb 实现了 agno 的 VectorDb 抽象，可以配合 knowledge 目录下的各种知识库示例，让 Agent 在问答时自动检索已入库的图片描述；
• 切换嵌入器：agno 在 knowledge/embedder/ 下提供了 OpenAI、Cohere、Mistral、Ollama、SentenceTransformer、VoyageAI 等多种嵌入器，embedder 参数直接替换即可；
• 切换存储后端：仓库的 06_storage 覆盖 Postgres（pgvector）、SQLite、Redis、Mongo、MySQL 等向量后端；
• 切换检索模式：把 search_type 改为 SearchType.hybrid 可融合全文与向量检索，LanceDb 会为 payload 列自动创建 FTS 索引并合并两路结果、按内容去重；
• 多用户隔离：insert / search 均支持 user_id 参数，底层利用表的 user_id 列做预过滤（prefilter=True），适合做成多租户媒体库；
• 引入重排：传入 reranker 可在召回后精排，提升检索质量。

小结

_09_image_extraction_to_vectordb 用约百行代码展示了 agno 最具代表性的多模态数据标注流水线：Pydantic 输出契约保证提取结构化，GeminiEmbedder 完成文本向量化，LanceDb 提供建表、去重、持久化与多模式检索，而 TEST_LOG.md 中的实测记录则证明了这条链路可直接运行。无论你要做图库语义搜索、商品去重，还是"以图搜图"的相似推荐，都可以以 basic.py 为起点快速落地。