一. 前言：影片不是「很多張圖片」這麼簡單 #
單張圖片問答跑通之後，很容易冒出一個危險念頭：
把影片每隔幾秒截一張圖，再全部丟給模型，不就好了？ 能跑，但不一定能用。 影片分析至少多了三份帳：
- 時間：每張圖到底對應第幾秒？
- 覆蓋：重要事件有沒有剛好落在抽樣空隙？
- 合併：連續五張都看到同一件事，不能算成五個事件。 今天拍拍君要做的不是一次性的「幫我看這段影片」，而是一條可檢查的本機管線：
- 用
ffprobe讀取影片資訊； - 產生明確的抽樣時間戳；
- 用 FFmpeg 擷取對應影格；
- 讓 MLX-VLM 回傳逐幀結構化觀察；
- 合併成事件時間軸；
- 建立可搜尋的 JSONL 索引；
- 驗證排序、覆蓋率與證據。 如果你只需要單張圖片問答，先看 MLX-VLM 本地圖片問答； 若你處理的是彼此獨立的大量圖片，則參考 MLX-VLM 批次結構化擷取。 這篇只處理它們沒有處理的東西：時間關係。
二. 先畫資料流：時間戳是第一級資料 #
我們的輸出目錄長這樣：
video-lab/
├── input/
│ └── demo.mp4
├── frames/
│ ├── frame_000000.jpg
│ ├── frame_000005.jpg
│ └── frame_000010.jpg
├── observations.jsonl
├── timeline.json
└── video_pipeline.py
檔名裡的數字是毫秒或 frame number 都可以，但要先定義清楚。 這裡採用整數秒，並在 JSON 裡保留浮點秒數：
{
"timestamp_s": 5.0,
"frame_path": "frames/frame_000005.jpg",
"label": "person_enters",
"summary": "一人從畫面左側進入",
"confidence": 0.82
}
三. 安裝：把影片處理與模型服務分開 #
先確認系統有 FFmpeg：
brew install ffmpeg
ffmpeg -version
ffprobe -version
再建立 Python 環境：
uv init video-lab
cd video-lab
uv add openai pydantic pillow
MLX-VLM 可以使用 CLI、Python API 或本機 server。 為了讓抽幀、重試與模型推論解耦，這篇沿用 OpenAI-compatible server：
uv tool install mlx-vlm
mlx_vlm.server \
--model mlx-community/Qwen2.5-VL-7B-Instruct-4bit \
--host 127.0.0.1 \
--port 8080
這篇刻意不直接傳影片給模型。 MLX-VLM 的原生影片能力會依模型與 processor 而異；顯式抽幀比較慢，卻能把時間戳、抽樣率與失敗影格完整留下。
四. 用 ffprobe 讀取 duration，不要猜 #
先寫一個小資料模型：
from __future__ import annotations
import json
import subprocess
from dataclasses import dataclass
from pathlib import Path
@dataclass(frozen=True)
class VideoInfo:
path: Path
duration_s: float
接著呼叫 ffprobe：
def probe_video(path: Path) -> VideoInfo:
completed = subprocess.run([
"ffprobe",
"-v", "error",
"-select_streams", "v:0",
"-show_entries", "format=duration",
"-of", "json",
str(path),
], check=True, capture_output=True, text=True)
payload = json.loads(completed.stdout)
duration = float(payload["format"]["duration"])
if duration <= 0:
raise ValueError("影片 duration 必須大於 0")
return VideoInfo(path=path, duration_s=duration)
五. 產生抽樣時間：先固定規則，再談聰明抽樣 #
最容易重現的策略是固定間隔：
def sample_timestamps(duration_s: float, every_s: float) -> list[float]:
if every_s <= 0:
raise ValueError("every_s 必須大於 0")
timestamps: list[float] = []
current = 0.0
while current < duration_s:
timestamps.append(round(current, 3))
current += every_s
tail = max(0.0, duration_s - 0.05)
if not timestamps or tail > timestamps[-1] + 0.5:
timestamps.append(round(tail, 3))
return timestamps
最後補一張靠近片尾的影格，可以避免 62 秒影片以 10 秒抽樣時只看到第 0 到 60 秒。 但不要誤會：固定 5 秒抽樣不代表能看見 5 秒內發生的所有事。 它只代表「每 5 秒取一個觀察點」。
六. 精準擷取影格：一張圖對一個 timestamp #
我們用 -ss 指定時間，輸出一張 JPEG：
def extract_frame(
video_path: Path,
timestamp_s: float,
output_path: Path,
) -> None:
output_path.parent.mkdir(parents=True, exist_ok=True)
command = [
"ffmpeg",
"-hide_banner",
"-loglevel", "error",
"-ss", f"{timestamp_s:.3f}",
"-i", str(video_path),
"-frames:v", "1",
"-vf", "scale='min(1280,iw)':-2",
"-q:v", "2",
"-y",
str(output_path),
]
subprocess.run(command, check=True)
再批次建立 manifest：
@dataclass(frozen=True)
class FrameSample:
timestamp_s: float
path: Path
def extract_samples(info: VideoInfo, every_s: float) -> list[FrameSample]:
samples: list[FrameSample] = []
for timestamp in sample_timestamps(info.duration_s, every_s):
name = f"frame_{int(timestamp):06d}.jpg"
output_path = Path("frames") / name
extract_frame(info.path, timestamp, output_path)
samples.append(FrameSample(timestamp, output_path))
return samples
若抽樣間隔小於一秒，檔名不能再用 int(timestamp)，否則會互相覆寫。
可以改用毫秒：round(timestamp * 1000)。
七. 定義逐幀輸出契約 #
先讓每張影格只回答可觀察的事，不要叫它直接編完整故事：
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field
class FrameObservation(BaseModel):
model_config = ConfigDict(extra="forbid")
label: Literal[
"empty",
"person_enters",
"person_present",
"person_leaves",
"object_change",
"unknown",
]
summary: str = Field(min_length=1, max_length=160)
evidence: list[str] = Field(max_length=4)
confidence: float = Field(ge=0.0, le=1.0)
class TimestampedObservation(FrameObservation):
timestamp_s: float = Field(ge=0.0)
frame_path: str
evidence 應描述畫面中能指認的線索，例如「左側門口出現人影」。
不要放「大概」「看起來像」這種只有語氣、沒有位置或物件的句子。
八. 把影格送進 MLX-VLM #
先把圖片轉成 data URL：
import base64
import mimetypes
import os
from openai import OpenAI
def image_data_url(path: Path) -> str:
mime = mimetypes.guess_type(path.name)[0] or "image/jpeg"
encoded = base64.b64encode(path.read_bytes()).decode("ascii")
return f"data:{mime};base64,{encoded}"
client = OpenAI(base_url=os.getenv(
"MLX_VLM_BASE_URL", "http://127.0.0.1:8080/v1"
), api_key="not-needed")
MODEL = os.environ["MLX_VLM_MODEL"]
請求時把 timestamp 一起寫進 prompt，但不要讓模型自己推算時間：
def observe_frame(sample: FrameSample) -> TimestampedObservation:
response = client.chat.completions.create(
model=MODEL,
messages=[{
"role": "user",
"content": [
{
"type": "text",
"text": f"這是第 {sample.timestamp_s:.3f} 秒。只描述畫面可支持的事件；不確定就選 unknown。",
},
{
"type": "image_url",
"image_url": {"url": image_data_url(sample.path)},
},
],
}],
response_format={
"type": "json_schema",
"json_schema": {
"name": "FrameObservation",
"strict": True,
"schema": FrameObservation.model_json_schema(),
},
},
temperature=0,
max_tokens=256,
)
raw = response.choices[0].message.content or ""
parsed = FrameObservation.model_validate_json(raw)
return TimestampedObservation(
**parsed.model_dump(),
timestamp_s=sample.timestamp_s,
frame_path=str(sample.path),
)
Structured Output 保證格式，不保證內容正確；confidence=0.99 也不能取代抽樣檢查。
九. 逐筆落盤：中斷後不要重跑全部 #
影片可能有幾百張抽樣影格。 每完成一張就追加 JSONL：
def append_jsonl(path: Path, item: TimestampedObservation) -> None:
with path.open("a", encoding="utf-8") as handle:
handle.write(item.model_dump_json() + "\n")
def completed_timestamps(path: Path) -> set[float]:
if not path.exists():
return set()
values: set[float] = set()
for line in path.read_text(encoding="utf-8").splitlines():
if line.strip():
values.add(json.loads(line)["timestamp_s"])
return values
浮點數直接當 key 有風險，因此前面統一 round 到小數三位。
更嚴格的版本可以使用整數毫秒 timestamp_ms。
十. 合併連續觀察，建立事件時間軸 #
現在每五秒一筆觀察，但使用者想看的是事件：
@dataclass
class TimelineEvent:
start_s: float
end_s: float
label: str
summaries: list[str]
evidence_frames: list[str]
def merge_timeline(
observations: list[TimestampedObservation],
max_gap_s: float,
) -> list[TimelineEvent]:
ordered = sorted(observations, key=lambda item: item.timestamp_s)
events: list[TimelineEvent] = []
for item in ordered:
if item.label in {"empty", "unknown"}:
continue
previous = events[-1] if events else None
same_event = (
previous is not None
and previous.label == item.label
and item.timestamp_s - previous.end_s <= max_gap_s
)
if same_event:
previous.end_s = item.timestamp_s
previous.summaries.append(item.summary)
previous.evidence_frames.append(item.frame_path)
else:
events.append(TimelineEvent(
start_s=item.timestamp_s,
end_s=item.timestamp_s,
label=item.label,
summaries=[item.summary],
evidence_frames=[item.frame_path],
))
return events
這是保守的 baseline：只合併相同 label，而且時間差不能超過門檻。 不要一開始就讓第二個 LLM 自由改寫整條時間軸，否則它可能把兩段遠離的事件接成一段。
十一. 事件索引與品質檢查 #
時間軸適合人讀，事件索引適合程式查。每筆索引至少保留 event_id、start_s、end_s、label、摘要與證據影格：
def event_record(event: TimelineEvent, event_id: int) -> dict[str, object]:
return {
"event_id": f"evt-{event_id:04d}",
"start_s": event.start_s,
"end_s": event.end_s,
"label": event.label,
"summary": "；".join(dict.fromkeys(event.summaries)),
"evidence_frames": event.evidence_frames,
}
寫檔前先做 deterministic QA：
def validate_results(
observations: list[TimestampedObservation],
events: list[TimelineEvent],
duration_s: float,
) -> list[str]:
issues: list[str] = []
times = [item.timestamp_s for item in observations]
if times != sorted(times):
issues.append("observations 未依時間排序")
if len(times) != len(set(times)):
issues.append("存在重複 timestamp")
if any(value < 0 or value > duration_s for value in times):
issues.append("timestamp 超出影片範圍")
for index, event in enumerate(events, 1):
if event.start_s > event.end_s or not event.evidence_frames:
issues.append(f"event {index} 邊界或證據無效")
if any(not Path(frame).is_file() for frame in event.evidence_frames):
issues.append(f"event {index} 的證據影格不存在")
return issues
另外計算相鄰 timestamp 的最大空隙。預期每 5 秒一張卻出現 25 秒空洞，代表抽幀、續跑或寫檔可能失敗。 「抽樣完成率」不是「內容理解率」；名稱要誠實，報表才不會自己騙自己。
十二. 抽樣策略、回歸測試與保留政策 #
| 策略 | 優點 | 風險 |
|---|---|---|
| 固定間隔 | 可重現、容易估成本 | 可能漏掉短事件 |
| 場景切換 | 適合剪輯明顯的影片 | 鏡頭內事件可能漏掉 |
| 混合策略 | 覆蓋與細節較平衡 | 規則與 QA 較複雜 |
| 先做固定間隔 baseline；看過漏報案例後，再加入 scene detection 或 motion-based 補幀。 | ||
回歸測試至少固定抽樣設定、模型、prompt 與一段已標記事件的短片，並保存誤報、漏報與 unknown 比例。 deterministic 部分則直接測： |
def test_sample_timestamps_adds_tail() -> None:
assert sample_timestamps(12.2, 5.0) == [0.0, 5.0, 10.0, 12.15]
上線前還要決定原始影片與抽樣影格保留多久、是否含人臉或個資、哪些錯誤可重試，以及是否用內容雜湊避免重跑。 本機推論減少第三方傳輸，不代表資料自動安全；frames、shell history、backup 與共享磁碟都可能留下副本。 最後量測 FFmpeg 抽幀、VLM prefill/decode，以及每分鐘影片產生的影格與 token，才能合理選 1、5 或 30 秒間隔。
十三. 常見踩雷 #
1. 用 frame index 當時間 #
可變幀率影片裡，frame number 不一定能直接除以一個固定 FPS。 保留 timestamp 才是穩定契約。
2. 把每張影格當獨立真相 #
單張可能是轉場、模糊或遮擋。 事件應由鄰近影格交叉確認。
3. 相同 label 就全部合併 #
第 10 秒與第 10 分鐘的 person_present 不一定是同一事件。
合併一定要有時間 gap 上限。
4. JSON 合法就直接發布 #
Schema 只能約束形狀。 時間邊界、證據影格與人工抽查仍不可少。
5. 重試時產生重複紀錄 #
以影片 ID 加 timestamp 建立唯一鍵。 續跑前先讀已完成集合，寫入後再更新狀態。
結語：先讓每個結論都能回到某一秒 #
影片分析最有價值的輸出，不是最華麗的摘要，而是可追溯的摘要。 今天這條管線把工作拆成幾個清楚邊界：
- FFmpeg 負責可重現地抽出影格；
- timestamp 負責時間契約；
- MLX-VLM 負責逐幀觀察；
- deterministic code 負責排序、合併與 QA；
- evidence frame 負責讓人回頭驗證。 等 baseline 穩定，再加入場景切換、語音轉錄、向量搜尋或更進階的原生影片模型。 先把「第幾秒、看見什麼、證據在哪」說清楚，系統才真的值得信任。🎬