全模态模型和"文字模型 + 语音转文字 + 图像描述"的拼接方案，差别在哪里？差在信息在进入模型之前就被翻译成了文字，语气、停顿、画面里的动作全丢了。这篇教程把一个能真正"看"视频、"听"语音的智能体从头搭出来：给它一段录屏或一段口述录音，它能一边流式吐字回答，一边在需要的时候调用工具去查数据。
具体目标是做出一个命令行程序，支持三种输入：本地视频文件、本地音频文件、直接打字。程序把视频拆成画面帧和音轨，按时间轴编排成一次请求发给模型，输出以流式方式实时打印。模型判断需要外部信息时，会发起工具调用，程序执行工具后把结果回填，继续生成。
前置条件清单
- Python 3.10 及以上（以官方文档当前版本要求为准）
- 一个可用的多模态模型 API Key，且模型支持视频/音频输入
- 本机安装
ffmpeg，命令行能直接调用 - 大约 20 分钟，以及一段 30 秒到 2 分钟的自测素材
关于模型标识：不同平台的模型 ID 命名不一样，代码里统一通过环境变量传入。具体填哪个 ID，请看官方文档当前版本的模型列表，选择带 Omni 或明确标注支持音视频输入的那一个。接口地址同理，本文按 OpenAI 兼容协议写，如果你的平台提供兼容端点，直接换 base_url 即可。
```bash
python -m venv .venv
source .venv/bin/activate # Windows 用 .venv\Scripts\activate
pip install openai python-dotenv
ffmpeg -version
```
ffmpeg -version 能打印出版本信息就说明装好了。如果提示找不到命令，先用系统包管理器安装（macOS 用 brew install ffmpeg，Ubuntu/Debian 用 apt install ffmpeg，Windows 可以下载官方构建包并把 bin 目录加进 PATH）。
接着建一个 .env 文件，注意不要把它提交到代码仓库：
```dotenv
OMNI_API_KEY=你的密钥
OMNI_BASE_URL=你的兼容端点地址
OMNI_MODEL=你的多模态模型ID
```
步骤 1：把视频拆成模型能接收的两种模态
全模态模型接收的输入，本质上还是"一段一段的媒体单元"。视频要做的事就是拆开：画面按固定间隔抽成图片，音轨切成若干秒一段的音频片段。
```bash
mkdir -p work/frames work/audio
每 2 秒抽一帧，宽度压到 640
ffmpeg -y -i demo.mp4 -vf "fps=1/2,scale=640:-2" -q:v 3 work/frames/frame_%04d.jpg
抽出 16kHz 单声道 wav，每 30 秒切一段
ffmpeg -y -i demo.mp4 -vn -ac 1 -ar 16000 -f segment -segment_time 30 -c:a pcm_s16le work/audio/chunk_%03d.wav
```
两个参数值得留意。fps=1/2 是"每 2 秒一帧"，2 分钟的视频就是 60 帧；改成 fps=1 是每秒一帧，帧数翻倍，效果更细但请求体积也翻倍。scale=640:-2 里的 -2 表示高度按比例自动算，并且保证是偶数——编码器对奇数高度会报错。
音频统一转成 16kHz 单声道，是因为多数语音前端都按这个采样率工作，采样率对不上容易出现音调异常或者被接口直接拒绝。
步骤 2：按时间轴把帧和音频编排成一次请求
这一步是"统一输入编排"的核心。不要把所有帧和所有音频片段无脑堆进一个数组，而是按时间顺序交错排列，并在每段前面加一行时间戳说明。模型看到"第 0 秒的画面""第 0 到 30 秒的语音"，后面回答"第 12 秒发生了什么"才有依据。
```python
import base64, os, glob, json
from typing import Iterable
def to_data_url(path: str) -> str:
ext = os.path.splitext(path)[1].lower()
mime = {".jpg": "image/jpeg", ".jpeg": "image/jpeg",
".png": "image/png", ".wav": "audio/wav",
".mp3": "audio/mpeg"}.get(ext, "application/octet-stream")
with open(path, "rb") as f:
return f"data:{mime};base64," + base64.b64encode(f.read()).decode()
def build_media_parts(frames_dir: str, audio_dir: str,
frame_every: float = 2.0, audio_chunk: float = 30.0) -> list:
parts = []
frames = sorted(glob.glob(os.path.join(frames_dir, "*.jpg")))
for i, p in enumerate(frames):
t = i * frame_every
parts.append({"type": "text", "text": f"[画面 {t:.0f}s]"})
parts.append({"type": "image_url", "image_url": {"url": to_data_url(p)}})
chunks = sorted(glob.glob(os.path.join(audio_dir, "*.wav")))
for j, p in enumerate(chunks):
t = j * audio_chunk
parts.append({"type": "text", "text": f"[语音 {t:.0f}s-{t + audio_chunk:.0f}s]"})
parts.append({"type": "input_audio",
"input_audio": {"data": to_data_url(p).split(",", 1)[1], "format": "wav"}})
return parts
```
音频字段的写法在不同平台之间有差异，有的用 input_audio 配 data + format，有的用 audio_url 配 data URL。以你所用平台的官方文档为准，改 build_media_parts 里那一行就够了，其余流程不受影响。
如果接口原生支持直接传视频地址或视频文件，可以省掉前面的抽帧步骤，但仍建议保留时间戳提示文本，模型对"第几秒"的定位会稳一些。
步骤 3：先跑通一次非流式请求
别急着上流式，先用最小代码确认模态输入是被接受的。
```python
import os
from dotenv import load_dotenv
from openai import OpenAI
load_dotenv()
client = OpenAI(api_key=os.environ["OMNI_API_KEY"],
base_url=os.environ["OMNI_BASE_URL"])
MODEL = os.environ["OMNI_MODEL"]
parts = build_media_parts("work/frames", "work/audio")
parts.insert(0, {"type": "text", "text": "描述这段录屏里发生了什么，并指出讲解人提到的关键数字。"})
resp = client.chat.completions.create(
model=MODEL,
messages=[{"role": "user", "content": parts}],
)
print(resp.choices[0].message.content)
```
能打印出一段带时间点的描述，说明输入编排这条路是通的。如果报"内容过长"之类的错误，回到步骤 1 把 fps 调小、把 scale 的宽度调小，或者缩短素材。
步骤 4：改成流式输出
流式输出对全模态场景尤其重要，因为一段带画面的回答可能要生成十几秒，用户需要立刻看到反馈。
```python
def stream_once(messages: list):
stream = client.chat.completions.create(
model=MODEL, messages=messages, stream=True)
buf, finish = [], None
for chunk in stream:
if not chunk.choices:
continue
choice = chunk.choices[0]
delta = choice.delta
if delta and delta.content:
print(delta.content, end="", flush=True)
buf.append(delta.content)
if choice.finish_reason:
finish = choice.finish_reason
print()
return "".join(buf), finish
```
几个容易踩的点：chunk.choices 偶尔是空数组，必须判空；delta.content 在只传工具调用时是 None，不能直接字符串相加；finish_reason 只在最后一个分片里出现，中途读到 None 是正常的。
步骤 5：加上工具调用
工具调用在流式模式下有个必须处理的细节：tool_calls 是分片下发的，函数名和参数 JSON 会被拆成好几段。参数必须在全部收完之后再 json.loads，中途解析一定失败。合并时要按 index 字段归位，不能简单按到达顺序追加。
```python
import json
TOOLS = [{
"type": "function",
"function": {
"name": "get_metric",
"description": "查询某个业务指标的当前数值",
"parameters": {
"type": "object",
"properties": {"name": {"type": "string", "description": "指标名"}},
"required": ["name"],
},
},
}]
def merge_tool_calls(acc: dict, deltas: Iterable):
for d in deltas:
idx = d.index if d.index is not None else 0
slot = acc.setdefault(idx, {"id": "", "name": "", "arguments": ""})
if d.id:
slot["id"] = d.id
if d.function and d.function.name:
slot["name"] = d.function.name
if d.function and d.function.arguments:
slot["arguments"] += d.function.arguments
```
工具本体先用一个假实现顶上，接真实系统时替换函数体即可：
```python
def get_metric(name: str) -> dict:
fake = {"日活": 128400, "转化率": "3.7%", "平均响应时长": "420ms"}
return {"name": name, "value": fake.get(name, "暂无数据")}
TOOL_IMPL = {"get_metric": get_metric}
```
步骤 6：串成完整的智能体主循环
把流式、工具调用、消息回填串起来。循环的退出条件是：模型这一轮没有发起工具调用，只有文本输出。
```python
def run_agent(messages: list, max_steps: int = 6):
for _ in range(max_steps):
text_buf, tool_acc, finish = [], {}, None
stream = client.chat.completions.create(
model=MODEL, messages=messages, tools=TOOLS, stream=True)
for chunk in stream:
if not chunk.choices:
continue
choice = chunk.choices[0]
delta = choice.delta
if delta and delta.content:
print(delta.content, end="", flush=True)
text_buf.append(delta.content)
if delta and getattr(delta, "tool_calls", None):
merge_tool_calls(tool_acc, delta.tool_calls)
if choice.finish_reason:
finish = choice.finish_reason
print()
assistant_msg = {"role": "assistant", "content": "".join(text_buf) or None}
if tool_acc:
assistant_msg["tool_calls"] = [
{"id": s["id"], "type": "function",
"function": {"name": s["name"], "arguments": s["arguments"]}}
for _, s in sorted(tool_acc.items())
]
messages.append(assistant_msg)
if not tool_acc:
return messages
for _, s in sorted(tool_acc.items()):
fn = TOOL_IMPL.get(s["name"])
try:
args = json.loads(s["arguments"] or "{}")
result = fn(**args) if fn else {"error": f"未知工具 {s['name']}"}
except Exception as e:
result = {"error": str(e)}
messages.append({"role": "tool", "tool_call_id": s["id"],
"content": json.dumps(result, ensure_ascii=False)})
return messages
```
调用入口：
```python
parts = build_media_parts("work/frames", "work/audio")
parts.insert(0, {"type": "text", "text": (
"这是一段产品演示录屏。请结合画面和讲解语音，说明演示了什么功能，"
"并用 get_metric 查一下讲解人提到的指标现在是多少。")})
msgs = [{"role": "user", "content": parts}]
run_agent(msgs)
```
一个典型的执行过程是：模型先流式输出"这段录屏从第 0 秒开始展示后台界面，讲解人提到日活……"，然后停下发起 get_metric 调用，程序返回真实数值，模型接着把工具结果和画面里的信息缝合起来继续输出。这就是原生全模态的价值——它引用画面、语音和工具结果时用的是同一套上下文。
常见坑与排错
报内容超长。 主要来源是图片和音频的 base64。base64 会让原始字节膨胀约三分之一，60 帧 640 宽的 JPEG 加上两分钟音频，体积很容易上去。优先调小 fps 和 scale 宽度，其次缩短素材。
音频被拒绝或听起来不对。 检查是不是漏了 -ac 1 -ar 16000。另外，视频本身没有音轨时 ffmpeg 的音频命令会报错，抽音轨前可以先用 ffprobe 判断是否存在音频流，或者在脚本里 try/except 掉这一步。
工具参数解析失败。 十有八九是在流式过程中就想解析 JSON。确认参数是拼完整之后才 json.loads，并且用 index 归位。另外要在代码里兜住 json.loads 的异常，把错误信息作为 tool 消息回填，模型通常能自己纠正参数重试。
模型说的时间点和你抽的帧对不上。 时间戳提示里的秒数是按抽帧间隔推算的，不是真实关键帧时间。想更准，可以改成在文件名里带上 ffmpeg 输出的真实时间戳，再把真实时间写进提示文本。
只输出了半句就停了。 检查是否达到了 max_steps 上限，或者单次请求超时。把 max_steps 调到 6 以上，并给客户端配置合理的超时时间。
密钥泄露。 .env 写进 .gitignore，日志里不要打印完整请求体，尤其是带 base64 的那部分。
下一步建议
想要更实用，可以在几个方向上加固。做长视频时，先分段摘要再合并，避免一次塞满上下文；做实时对话时，把麦克风采集接到流式上传，配合语音合成让智能体"说"出来；做生产运行时，把每轮请求的耗时、输入输出 token 数、工具调用次数记下来，这是排查成本与延迟问题的依据。另外，同一段视频的抽帧结果建议做本地缓存，反复调试提示词时不必每次重跑 ffmpeg。
模型 ID、接口字段名与配额限制都可能调整，把这篇里的 OMNI_MODEL、音频字段这类可变量集中在一处，之后按官方文档更新一处就够了。