ag2-multimodal-input

用途

当需要让 Agent 处理非文本输入时使用该能力：描述图片、转写音频、总结视频，或从 PDF / 文档中提取内容。同一套工厂模式可跨供应商使用，但各供应商支持情况有差异。

主要能力

输入工厂

工厂	支持格式
ImageInput(...)	JPEG、PNG、GIF、WebP
AudioInput(...)	WAV、MP3、OGG、FLAC、AAC
VideoInput(...)	MP4、WebM、MOV、MKV、MPEG
DocumentInput(...)	PDF、TXT、HTML、Markdown、CSV、JSON、Office 格式

每个工厂都接受相同的四种数据来源：

from ag2.events import ImageInput

ImageInput("https://example.com/photo.jpg")     # URL
ImageInput(path="photo.jpg")                    # local file
ImageInput(data=raw_bytes, media_type="image/png")  # bytes
ImageInput(file_id="file-abc123")               # provider-uploaded

供应商支持矩阵

输入类型	OpenAI	OpenAI Responses	Gemini	Anthropic
Text	✓	✓	✓	✓
Image (URL)	✓	✓	✓	✓
Image (binary)	✓	✓	✓	✓
Audio (URL)	–	–	✓	–
Audio (binary)	✓	–	✓	–
Video (URL)	–	–	✓	–
Video (binary)	–	–	✓	–
Document (URL)	–	✓	✓	✓
Document (binary)	✓	✓	✓	✓
File ID	✓	✓	✓	✓

不支持的组合会抛出 UnsupportedInputError 并附带清晰提示。Gemini 的多模态支持最广；如果不确定该为多模态任务选哪个供应商，可以先用它。

使用方式

60 秒示例

from ag2 import Agent
from ag2.config import GeminiConfig
from ag2.events import ImageInput

agent = Agent(
    "vision",
    "You describe images.",
    config=GeminiConfig(model="gemini-3-flash-preview"),
)

image = ImageInput("https://example.com/photo.jpg")
reply = await agent.ask("Describe this image in detail.", image)
print(reply.body)

一次 ask 也可以传入多个输入：

reply = await agent.ask(
    "Compare these two images.",
    ImageInput("https://example.com/before.jpg"),
    ImageInput("https://example.com/after.jpg"),
)

供应商专属特性

Gemini —— 可直接使用 YouTube URL：

from ag2.events import VideoInput

video = VideoInput("https://www.youtube.com/watch?v=dQw4w9WgXcQ")
reply = await agent.ask("Summarize this video.", video)

Gemini —— 通过 Google Files API 处理大于 20MB 的大文件：

from google import genai
from ag2.events import VideoInput
import time

client = genai.Client()
uploaded = client.files.upload(file="large_video.mp4")
while uploaded.state.name == "PROCESSING":
    time.sleep(2)
    uploaded = client.files.get(name=uploaded.name)

video = VideoInput(uploaded.uri)

Gemini —— vendor_metadata：

键	用途
media_resolution	MEDIA_RESOLUTION_LOW/MEDIUM/HIGH/ULTRA_HIGH —— 影响 token 与成本
video_metadata	裁剪（start_offset、end_offset）与 fps
display_name	文件显示名称

ImageInput(data=raw, media_type="image/jpeg", vendor_metadata={"media_resolution": "MEDIA_RESOLUTION_LOW"})

VideoInput(path="lecture.mp4", vendor_metadata={
    "video_metadata": {"start_offset": "60s", "end_offset": "120s", "fps": 0.5},
})

OpenAI —— 图片 detail：

ImageInput(data=raw, media_type="image/png", vendor_metadata={"detail": "low"})  # "low" | "high" | "auto"

Anthropic —— File ID + prompt 缓存：

import anthropic
from ag2.events import ImageInput, DocumentInput

client = anthropic.Anthropic()
uploaded = client.beta.files.upload(file=("photo.jpg", open("photo.jpg", "rb"), "image/jpeg"))

# filename determines block type (image vs document)
image = ImageInput(file_id=uploaded.id, filename="photo.jpg")

# Cache an attachment so subsequent turns skip re-uploading
doc = DocumentInput(path="report.pdf", vendor_metadata={"cache_control": {"type": "ephemeral"}})

FilesAPI —— 与供应商无关的上传生命周期

对任何提供文件 API 的供应商（OpenAIConfig、OpenAIResponsesConfig、AnthropicConfig、GeminiConfig）：

from ag2 import FilesAPI
from ag2.config import OpenAIResponsesConfig

files = FilesAPI(OpenAIResponsesConfig(model="gpt-5-mini"))

uploaded = await files.upload(path="report.pdf", purpose="assistants")
print(uploaded.file_id)

# Or from bytes (filename required)
uploaded = await files.upload(data=b"...", filename="hello.txt", purpose="assistants")

# List, read, delete
all_files = await files.list()
data = await files.read(uploaded.file_id)        # NotImplementedError on Gemini
await files.delete(uploaded.file_id)

将 file_id 传给 DocumentInput、ImageInput 等：

from ag2.events import DocumentInput

doc = DocumentInput(file_id=uploaded.file_id)
reply = await agent.ask("Summarize this report.", doc)

适用场景

图片描述、图片对比
音频转写
视频总结（含 YouTube 链接、视频裁剪与抽帧）
PDF / 文档内容提取
需要跨供应商复用同一套输入工厂的多模态任务

限制与注意事项

供应商支持差异：选到不支持该输入类型的供应商会抛出 UnsupportedInputError，需先查支持矩阵；Gemini 覆盖最广。
FilesAPI.read() 在 Gemini 上会抛出 NotImplementedError，Gemini 不提供下载。
调用 files.upload(data=...) 时必须提供 filename=，否则抛出 ValueError。
给一个工厂提供多个来源不会报错：解析优先级为 url > file_id > path > data，多余来源会被静默忽略；只提供一个来源才能得到预期结果。提供零个来源会抛出 ValueError。
Anthropic 的 ImageInput(file_id=...) 必须带 filename=，因为 Anthropic 依靠文件扩展名判断块类型（image 还是 document）。
Gemini 的 vendor_metadata 键是嵌套的——video_metadata 本身接收一个字典，需参照文档表格确认结构。
使用 Gemini 大文件上传时需等待 PROCESSING 状态完成，轮询 client.files.get(name=...) 直到就绪后再引用 URI。

技能介绍

用途