这篇能做出什么
跑完这套流程，你会得到一个可查询的长视频知识库，输入一句自然语言问题，输出这样的结果：
```
问题：预算调整是在哪一段讨论的？
预算在 Q3 被下调，主要削减来自市场推广费用。 [00:41:12–00:41:48]
下季度不再新增外包名额，现有外包合同到期不续。 [01:02:30–01:02:55]
该调整需要在下一次经营会上复核。 [01:03:10–01:03:26]
```
每条结论后面都挂着时间区间，点一下就能跳回原片核对。这比"整段视频总结成 200 字"有用得多——因为总结里的每一句话都能被验证，出错的句子能立刻发现并剔除。
整套流程分四件事：把长视频切成带绝对时间戳的片段、对片段做帧采样和索引、按时间戳提问拿到结构化证据、把证据归并成带引用的摘要并落成检索索引。Pegasus 系列是视频理解模型，1.6 具体支持的片段长度上限、鉴权方式、字段名与计费规则以官方文档当前版本为准，本文把所有厂商差异收敛在一个函数里，其余代码可直接照抄。
前置条件清单
- Python 3.10 及以上，能装
requests、numpy - ffmpeg 与 ffprobe 可执行文件在 PATH 中（
ffmpeg -version有输出即可） - 一个视频理解模型的调用凭据：API Key、服务地址。环境变量里放
PEGASUS_API_KEY和PEGASUS_ENDPOINT - 一个用于向量化的 embedding 来源，本地模型或在线接口都行
- 一段待处理的长视频，建议先用 5 分钟片子打通，再上 90 分钟
- 磁盘留出视频体积 1.5 倍以上的空间给切片和抽帧
分步骤
步骤 1：先定产物结构，再动手
不要一上来就调模型。先把输出结构定死，后面每一步都往这个结构里填：
```python
schema.py
Claim = {
"video": str, # 视频标识
"text": str, # 一句结论
"t_start": float, # 绝对起始秒
"t_end": float, # 绝对结束秒
"chunk_id": str, # 来自哪个片段，便于回溯
"confidence": str, # high / medium / low
}
```
没有 t_start/t_end 的结论一律丢弃。这条规则是整个流程可验证性的地基。
步骤 2：规划切片，保留重叠
长视频必须切开，但切点落在句子中间会丢语义，所以相邻片段留一段重叠。
```python
plan.py
def plan_chunks(duration, chunk=180, overlap=15):
step = chunk - overlap
out, start, i = [], 0.0, 0
while start < duration:
end = min(start + chunk, duration)
out.append({"id": f"c{i:04d}", "start": start, "end": end})
if end >= duration:
break
start += step
i += 1
return out
```
先拿到真实时长：
```bash
ffprobe -v error -show_entries format=duration -of csv=p=0 input.mp4
```
步骤 3：用 ffmpeg 切片，保证时间戳准确
关键点是切片要重编码。用 -c copy 会吸附到最近的关键帧，实际起点和规划值可能差好几秒，后续所有时间戳都会偏。
```bash
cut.sh —— 逐片执行，$start / $dur / $id 由 Python 传入
ffmpeg -hide_banner -y \
-ss "$start" -i input.mp4 -t "$dur" \
-c:v libx264 -preset veryfast -crf 20 \
-c:a aac -ar 16000 -ac 1 \
-force_key_frames 0 \
"chunks/${id}.mp4"
```
切完立刻校验实际时长，写回清单：
```python
import subprocess, json
def probe_duration(path):
out = subprocess.check_output([
"ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "csv=p=0", path,
])
return float(out.strip())
```
步骤 4：帧采样与时间戳映射
模型看视频通常是抽帧或抽帧加音轨。固定间隔采样最省事，也最好回溯：
```bash
每 2 秒取一帧，宽度压到 640，便于快速上传和比对
ffmpeg -hide_banner -y -i "chunks/${id}.mp4" \
-vf "fps=0.5,scale=640:-2" -q:v 3 \
"frames/${id}_%05d.jpg"
```
第 n 张帧（从 1 开始计数）在片段内的相对时间是 (n-1) / fps，加上片段起点就是绝对时间。有幻灯片、白板、代码演示的视频，再用一次镜头切换检测补充关键帧：
```bash
ffmpeg -hide_banner -y -i "chunks/${id}.mp4" \
-vf "select='gt(scene,0.35)',showinfo" -vsync vfr \
"frames_scene/${id}_%05d.jpg"
```
把两批帧合并去重后，连同绝对时间戳写进一个 manifest 文件，后面每一步都从 manifest 读时间，不靠推断。
步骤 5：把模型调用收敛成一个函数
不同服务的上传方式、鉴权头、返回字段都不一样。只写一个适配函数，其余代码就不受厂商变动影响：
```python
vlm.py
import os, time, requests
def analyze(clip_path, prompt, retries=4):
"""调用视频理解模型。
请求路径、鉴权头、表单字段名与响应结构，
以你所用服务官方文档当前版本为准，只需改这一个函数。
返回统一结构 {"text": str, "raw": dict}
"""
endpoint = os.environ["PEGASUS_ENDPOINT"]
api_key = os.environ["PEGASUS_API_KEY"]
delay = 2.0
for _ in range(retries):
with open(clip_path, "rb") as f:
resp = requests.post(
endpoint,
headers={"Authorization": f"Bearer {api_key}"},
files={"video": f},
data={"prompt": prompt, "temperature": "0"},
timeout=600,
)
if resp.status_code in (429, 500, 502, 503, 504):
time.sleep(delay)
delay *= 2
continue
resp.raise_for_status()
payload = resp.json()
return {"text": payload.get("text", ""), "raw": payload}
raise RuntimeError("重试次数用尽，检查限流或片段体积")
```
长任务建议改成"提交作业 + 轮询结果"，避免连接被中间层掐断。
步骤 6：按时间戳提问
提问模板里必须写清片段的起点和时长，并且强制模型输出时间戳。否则你拿到的只是漂亮句子，无法回溯。
```python
PROMPT = """你在分析一段视频片段。
片段时长 {dur:.1f} 秒，在完整视频中的起点是 {start:.1f} 秒。
只依据画面与声音中确实出现的信息作答，不要用常识补全。
只输出 JSON，不要输出其它文字：
{{
"answer": "字符串；没有相关信息时填 未提及",
"evidence": [
{{"t_start": 相对片段起点的秒数, "t_end": 秒数, "quote": "画面或台词要点"}}
],
"confidence": "high | medium | low"
}}
问题：{question}
"""
```
拿到结果后立刻把相对时间换成绝对时间，并做基本校验：
```python
def normalize(chunk, result):
claims = []
for ev in result.get("evidence", []):
ts, te = float(ev["t_start"]), float(ev["t_end"])
if not (0 <= ts <= te <= (chunk["end"] - chunk["start"]) + 1):
continue # 时间戳越界，丢弃
claims.append({
"text": ev.get("quote", ""),
"t_start": chunk["start"] + ts,
"t_end": chunk["start"] + te,
"chunk_id": chunk["id"],
"confidence": result.get("confidence", "medium"),
})
return claims
```
步骤 7：生成可引用摘要
分两轮。第一轮逐片段产出要点，第二轮做归并。归并时不要把所有片段原文塞进一次请求，只传"时间区间 + 要点"，上下文压力小很多。
```python
MERGE_PROMPT = """以下是一段长视频各片段的要点，已按时间排序。
请合并成一份结构化摘要，每条结论必须携带原样的时间区间，不要改写时间。
输出 JSON：{"sections":[{"title":"小标题","claims":[{"text":"...","t_start":秒,"t_end":秒}]}]}
要点列表：
{bullets}
"""
```
归并完做一次引用校验：每条结论的时间区间必须落在某个已知片段范围内，否则标记为"未证实"并从正式摘要里剔除。
```python
def verify(claims, clip_ranges, tol=2.0):
kept, dropped = [], []
for c in claims:
ok = any(
r["start"] - tol <= c["t_start"] and c["t_end"] <= r["end"] + tol
for r in clip_ranges
)
(kept if ok else dropped).append(c)
return kept, dropped
```
重叠窗口会让同一件事出现两次。去重规则：同一主题且时间区间重叠超过一半时，保留起点更早的那条。
步骤 8：落成检索索引
用 SQLite 做一层轻量索引：全文检索负责关键词命中，向量负责语义命中，时间字段负责范围过滤。中文全文检索建议用 trigram 分词器，unicode61 对中文几乎不起作用。
```python
import sqlite3, numpy as np
conn = sqlite3.connect("video_index.db")
conn.executescript("""
CREATE TABLE IF NOT EXISTS clips(
video TEXT, chunk_id TEXT, start REAL, end REAL, path TEXT,
summary TEXT, PRIMARY KEY(video, chunk_id));
CREATE VIRTUAL TABLE IF NOT EXISTS clips_fts
USING fts5(chunk_id UNINDEXED, text, tokenize='trigram');
CREATE TABLE IF NOT EXISTS vecs(
chunk_id TEXT PRIMARY KEY, dim INTEGER, vec BLOB);
""")
def to_blob(v):
return np.asarray(v, dtype="float32").tobytes()
def from_blob(b):
return np.frombuffer(b, dtype="float32")
```
检索时三路合并打分：关键词命中、向量AI 词典：余弦相似度">余弦相似度、如果问题里出现"第 40 分钟"这类线索就加时间范围过滤。取前 3 到 5 个片段，再把原始片段喂给模型做一次精读，得到最终带时间戳的回答。
步骤 9：端到端跑一遍
```python
run.py（示意，省略 import）
video = "meeting_2024.mp4"
duration = probe_duration(video)
chunks = plan_chunks(duration, chunk=180, overlap=15)
1) 切片 + 抽帧（调用前面写好的 cut.sh / ffmpeg 命令）
2) 逐片段提问并归一化
for ch in chunks:
res = analyze(f"chunks/{ch['id']}.mp4",
PROMPT.format(dur=ch["end"] - ch["start"],
start=ch["start"],
question="这段里有没有讨论预算调整？"))
ch["claims"] = normalize(ch, parse_json(res["text"]))
3) 归并 + 校验 + 写索引
```
对 90 分钟会议录像，按 180 秒切片、15 秒重叠，得到约 33 个片段。串行跑完大致在一杯咖啡的时间内，并发上限设 2 到 3，别把限流打满。
常见坑与排错
时间戳整体偏移几秒。 九成是切片时用了 -c copy。改成重编码，并且在切片后用 ffprobe 回读实际时长写回清单，不要相信规划值。
模型说"未提及"，被当成否定结论写进摘要。 "未提及"只代表这个片段里没找到，不代表事情没发生。归并阶段要把"未提及"的片段和"有结论"的片段分开处理。
结论重复出现两三次。 重叠窗口导致。按主题加时间重叠度去重，别指望模型自己发现。
返回的 JSON 解析失败。 提示词里明确"只输出 JSON"，解析时先用正则抠出第一个完整花括号块，再做 json.loads；仍然失败就重试一次，别直接抛异常中断整批任务。
429 与 5xx。 指数退避是基础，并发数控制在 2 到 3。批量任务一定要做断点续跑：把每个片段的处理结果落盘，重跑时跳过已完成的。
片段超出模型可处理时长。 把 MAX_CLIP_SECONDS 设得比你以为的安全值再小一点，并且每次调用前用 ffprobe 校验实际时长。超长片段常常表现为静默截断，不报错，但后半段的内容全丢。
声音和画面不同步。 切片时音频单独重编码容易引入偏移，尽量让音视频用同一条命令输出，并检查 -ar、-ac 参数一致。
上传前的合规问题。 会议录像常含个人信息。上传前确认数据保留策略，必要时先做画面遮挡与音频静音处理。
下一步建议
先拿一段 5 分钟的视频把整条链路跑通，确认时间戳能对上原片，再扩到长视频。之后可以往三个方向走：一是做增量索引，新视频进来只处理新增片段；二是把引用做成可点击的时间轴跳转，接进播放器；三是建一个小评测集，人工标注 20 个问题和标准时间区间，每次调整切片长度、帧率或提示词后重跑一遍，用命中率来判断改动是否真的有效。切片粒度和帧采样率这两个参数对结果影响较大，值得单独做几组对照实验。