video-edit¶
第5章 · Coding Agent 与代码生成 · 配套项目
chapter5/video-edit
项目说明¶
实验 5-6：基于 API 的智能视频剪辑¶
《深入理解 AI Agent》配套实验。用户给一段含多个场景的视频 + 一句自然语言需求 （如"把冲浪部分剪出来"），Agent 自动定位目标场景、生成 Blender Python API 脚本 剪出片段并自我审查。
目的¶
验证三个核心机制在多媒体处理中的作用：
- 两步 Vision 定位：Proposer 无法直接"看懂"视频，于是委托一个视频分析子 Agent， 用 ffmpeg 抽帧 + Vision LLM 读图来定位目标场景的时间边界。
- 代码生成（Blender Python API）：Proposer 把剪辑计划翻译成一段调用 Blender
Python API（bpy） 的脚本——导入 / 裁剪 / 字幕 / 变速 / 渲染各对应一个 API 调用，
用
blender --background --python edit.py无头执行。这正是书中"把视频编辑重构为 API 调用和代码生成问题"的落地。未装 Blender 时脚本照常生成（代码生成产物）， 实际渲染回退到 ffmpeg（见下文"剪辑后端"）。 - 提议者-审核者（Proposer / Reviewer）：Proposer 剪辑后无法自证效果， 由 Reviewer 抽取成片关键帧、用 Vision LLM 检查是否剪对，不合格则反馈、迭代。
两步定位原理¶
Vision LLM 逐帧扫全片既慢又贵，因此采用"先粗后细"：
- 第一步（粗粒度）：每 10 秒抽一帧，把全片的稀疏截图连同"要找哪个场景"一起 交给 Vision，得到大致区间（如"冲浪在 20–30s"）。
- 第二步（细粒度）：在粗区间上下各外扩一个粗间隔，每 1 秒抽一帧， 再问 Vision 精确边界（如"15–29s"）。
把这套抽帧-读图封装成独立子 Agent：几十张截图只进入子 Agent 的一次性上下文， 不会污染主 Agent（Proposer/Reviewer）的对话历史。demo 末尾会打印两者的 token 对比。
提议者-审核者¶
NL 需求 ──► Proposer 解析意图（目标场景 + 特效）
│
▼
视频分析子 Agent 两步定位 ──► [start, end]
│
▼
Proposer 生成 Blender bpy 脚本（edit.py）──► 渲染剪辑（可加字幕/慢动作）
│ 装了 Blender 用 bpy，否则回退 ffmpeg
▼
Reviewer 抽首/中/尾关键帧 ──► Vision 检查 pass/fail + 反馈
│pass? 否 → Proposer 据反馈修正边界，重剪（最多 3 轮）
▼是
输出成片 final.mp4
运行¶
pip install -r requirements.txt
cp env.example .env # 填入 OPENAI_API_KEY（未配置时设 OPENROUTER_API_KEY 自动改走 OpenRouter）
python demo.py # 默认需求"把冲浪的部分剪出来"（完整流程）
python demo.py "把滑雪部分剪出来，并加上字幕 Winter" # 自定义需求
python demo.py -i my.mp4 -o out.mp4 "把演讲开场剪出来" # 用自己的视频 + 自定义输出
python demo.py --backend blender # 强制用 Blender Python API 无头渲染（需装 Blender）
python demo.py --vision-model gpt-5.6-luna # 覆盖模型（也可用 --text-model）
python demo.py --quick # 快速模式：粗采样 + 单轮审查，Vision 调用最少（省时省钱）
python demo.py --smoke # 冒烟自检：仅剪辑链路 + 生成 bpy 脚本，不调用任何 API
python demo.py --help # 查看全部参数
常用参数（完整见 --help）：--input/-i 输入视频、--output/-o 成片路径、
--backend {auto,blender,ffmpeg} 剪辑后端、--text-model/--vision-model 覆盖模型。
一条命令即可跑通：生成/读取视频 → 两步定位 → 生成 bpy 脚本剪辑 → 审查 → 输出成片。
每次运行都会清空 output/，从干净状态开始（幂等可重复）。完整流程会多次调用
Vision 模型（较慢/耗费额度）；只想验证链路时先跑 --smoke（零 API），或用 --quick。
预期输出示例¶
--smoke（零 API，可复现）¶
以下为 python demo.py --smoke 的真实输出（无需 OpenAI Key，仅需 ffmpeg）：
==========================================================================
冒烟自检 | 剪辑链路 + bpy 脚本生成，不调用任何 API
==========================================================================
[1/3] 生成测试视频 OK：output/source.mp4（场景真值={'hiking': (0, 15), 'surfing': (15, 30), 'skiing': (30, 42), 'cycling': (42, 54)}）
[2/3] 抽帧 OK：output/frames/smoke.png
[3/3] 剪辑+字幕 OK（后端=ffmpeg（未装 Blender，回退））：
文件: smoke_cut.mp4
时长: 5.03s
容器: mov,mp4,m4a,3gp,3g2,mj2
大小: 121.4 KB
视频流: h264 1280x720 @ 30/1 fps
音频流: aac 44100Hz 1ch
已生成 Proposer 的 Blender 脚本：output/edit.py
（这正是书中'生成 Blender Python API 代码'的产物；装好 Blender 后可直接
`blender --background --python output/edit.py` 无头渲染。）
✓ 冒烟自检通过：剪辑链路正常 + bpy 脚本已生成（未调用 OpenAI）。
生成的 output/edit.py 是一段可执行的 Blender bpy 脚本（new_movie 导入、
frame_offset_start/frame_final_duration 裁剪、new_effect(type='TEXT') 字幕、
bpy.ops.render.render 渲染），本机对其做过 py_compile 语法校验。
--quick（完整链路，需 API）¶
以下为 python demo.py --quick（默认需求"把冲浪的部分剪出来"）的真实节选（定位/误差/
token 部分与剪辑后端无关，故不受 bpy/ffmpeg 后端切换影响）：
步骤 1 | Proposer 解析自然语言需求
解析结果：目标场景='surfing scene' 特效=[]
步骤 2 | 视频分析子 Agent：两步 Vision 定位（--quick 快速采样）
[粗粒度] 每 15s 采样 5 帧 → Vision 得区间 [15, 30]s（依据：The word 'SURFING' appears at t=15s and changes at t=30s.）
[细粒度] 窗口 [0.0, 45.0] 内每 2s 采样 23 帧 → 精确边界 [16.0, 28.0]s
>>> 最终定位：起 16.0s 止 28.0s
真值 [15, 30]s → 起点误差 1.0s，终点误差 2.0s（验收要求 ≤ 3s）
步骤 3-4 | Proposer 剪辑 + Reviewer 审查（迭代）
Proposer 剪出片段 [16.0, 28.0]s，成片时长 12.0s
Reviewer：pass=... score=... 检查帧=['0.5', '6.0', '11.5']
Token 统计（子 Agent 隔离截图，主上下文不被污染）
主 Agent（Proposer+Reviewer）：573 tokens
子 Agent（两步定位截图） ：2934 tokens
产物（output/ 目录，真实文件）：
| 文件 | 时长 | 说明 |
|---|---|---|
source.mp4 |
54.0s | 程序化生成的 4 场景测试原片 |
edit_round1.py |
— | Proposer 生成的 Blender bpy 脚本（代码生成产物，可换机执行） |
cut_round1.mp4 |
12.0s | 第 1 轮剪出的候选片段 |
final.mp4 |
12.0s | 采用的成片（H.264 + AAC，1280x720@30fps） |
Token 统计印证了核心结论：几十张截图（2934 tokens）只进入子 Agent的一次性 上下文，主 Agent 的对话历史（573 tokens）几乎不受截图污染。 （注：合成测试片仅显示"SURFING"字样而非真实冲浪画面，Reviewer 有时会据此判为 不通过——这正是审核者按画面内容如实反馈的体现；换真实视频即无此现象。）
依赖¶
- ffmpeg / ffprobe：本机实际剪辑与抽帧。
brew install ffmpeg（macOS）/apt install ffmpeg（Ubuntu）。本项目在 ffmpeg 8.0 上验证通过。 - OPENAI_API_KEY：用
gpt-5.6-luna做视觉定位/审查与文本规划（视觉模型须支持图像输入）；未配置时用OPENROUTER_API_KEY兜底，自动改走 OpenRouter。
如何适配 / 扩展¶
换模型 / 供应商¶
模型与端点全部通过环境变量注入（见 env.example），无需改代码：
TEXT_MODEL：规划/边界修正的文本模型（默认gpt-5.6-luna）。VISION_MODEL：定位/审查的视觉模型，必须支持图像输入（默认gpt-5.6-luna）。OPENAI_BASE_URL：换成任何兼容 OpenAI 协议的端点（自建代理、Azure OpenAI、 或其他厂商网关），配合对应的OPENAI_API_KEY即可。
export OPENAI_BASE_URL=https://your-gateway.example.com/v1
export VISION_MODEL=gpt-5.6-luna # 例：用当前廉价旗舰视觉模型
export TEXT_MODEL=gpt-5.6-luna
agents.py 里的 OpenAI() 客户端会自动读取上述变量（client() 惰性初始化）。
换输入视频¶
make_test_video.py 用 ffmpeg 程序化生成一段 54s 的视频，含 4 个明显不同的场景
（HIKING 绿 / SURFING 蓝 / SKIING 白 / CYCLING 橙），每段都叠加大号场景名与时间码水印，
让 Vision 仅凭画面就能准确定位——便于复现验收。
换成你自己的真实视频：直接 python demo.py -i 你的.mp4 -o 输出.mp4 "剪辑需求"
即可（无需改代码）。此时跳过测试片生成，也不再打印定位误差（外部视频无真值）。
Blender vs. ffmpeg（剪辑后端）¶
书中原方案用 Blender Python API（bpy） 驱动视频序列编辑器（VSE）完成剪辑。
本项目把它实现为一等后端：blender_editor.generate_bpy_script() 把剪辑计划翻译成
一段真实可执行的 bpy 脚本（new_movie 导入、frame_offset_start/frame_final_duration
裁剪、new_effect(type='TEXT'/'SPEED') 字幕/变速、bpy.ops.render.render 渲染），
render_with_blender() 再用 blender --background --python edit.py 无头执行。
--backend blender：强制走 Blender（需blender --version可用）；--backend ffmpeg：强制走 ffmpeg；--backend auto（默认）：装了 Blender 用 bpy，否则回退 ffmpeg。
关键点：无论哪个后端，Proposer 生成的 bpy 脚本都会落盘到 output/edit_round*.py
（--smoke 下为 output/edit.py）——即"生成 Blender Python API 代码"这一核心产物，
可人工核对、也可拷到装了 Blender 的机器上执行。本机未安装 Blender，故本仓库的实际
渲染由 ffmpeg 完成并验证；bpy 脚本已通过 py_compile 语法校验，但未在真实 Blender
上跑过渲染（装好 Blender 后即可用 --backend blender 端到端执行）。两种后端的取舍：
| ffmpeg | Blender（bpy） | |
|---|---|---|
| 定位 | 裁剪/拼接/字幕/变速等 2D 流水线 | 3D 场景、合成、关键帧动画、粒子/摄像机 |
| 上手 | 单二进制、无 GUI、CI 友好 | 需装完整 Blender，体积大、渲染慢 |
| 适用 | 绝大多数"剪一段 + 简单特效"需求 | 需要 3D 合成/复杂转场/图层混合时才值得 |
核心的"两步 Vision 定位 + 提议者-审核者"与执行层解耦，两个后端共用同一份剪辑计划，
agents.py/demo.py 无需为切换后端改动逻辑。
文件¶
| 文件 | 作用 |
|---|---|
demo.py |
一条命令跑通的编排入口（CLI、启动自检、迭代循环、token 统计） |
agents.py |
VideoAnalyzerAgent（两步定位）/ ProposerAgent / ReviewerAgent |
blender_editor.py |
Blender bpy 脚本生成 + 无头渲染（书中原方案，核心实验点） |
video_editor.py |
剪辑执行层：apply_edit() 统一入口，调度 Blender/ffmpeg 双后端 |
make_test_video.py |
程序化生成含 4 个场景的测试视频 |
ffmpeg_utils.py |
ffmpeg/ffprobe 薄封装（统一错误检查、抽帧、探测时长/流） |
output/（生成的视频、截图、成片）已被 .gitignore 忽略，避免仓库膨胀。
局限¶
- 定位精度取决于场景在画面上的可辨识度；真实视频若场景过渡渐变，边界误差会大于纯色测试片。
- 细粒度步长固定 1s，边界精度上限即 ±1s 量级（满足书中 ±3s 验收）。
- 慢动作音频用
atempo变速，倍率过大时音质下降；转场/多轨混音等复杂特效未覆盖。 - Reviewer 仅抽首/中/尾三帧，长片段中段的偶发错误可能漏检（可调高抽帧密度）。
源代码¶
agents.py¶
"""
实验 5-6 的三个 Agent：
VideoAnalyzerAgent —— 视频分析子 Agent，用"两步 Vision 定位"找目标场景边界。
ProposerAgent —— 把自然语言需求解析成剪辑计划，调用子 Agent 定位并执行剪辑。
ReviewerAgent —— 抽取成片关键帧，用 Vision 检查是否剪对，给出结构化反馈。
把视频分析封装为独立子 Agent 的意义：大量截图只进入子 Agent 的一次性上下文，
不会污染主 Agent（Proposer/Reviewer）的对话历史——见 demo.py 打印的 token 统计。
"""
import base64
import json
import os
import re
from openai import OpenAI
from ffmpeg_utils import extract_frame, probe_duration
TEXT_MODEL = os.getenv("TEXT_MODEL", "gpt-5.6-luna")
VISION_MODEL = os.getenv("VISION_MODEL", "gpt-5.6-luna") # 必须支持图像输入
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
_client = None
def map_model_to_openrouter(model: str) -> str:
"""把直连模型名映射为 OpenRouter 上的 id（非可映射 id 统一兜底到当前廉价旗舰）。"""
if not model or "/" in model:
return model or "openai/gpt-5.6-luna"
m = model.lower()
if m.startswith(("gpt-", "o1", "o3", "o4")):
return "openai/" + model
if m.startswith("claude"):
if "haiku" in m:
return "anthropic/claude-haiku-4.5"
if "sonnet" in m:
return "anthropic/claude-sonnet-4.6"
return "anthropic/claude-opus-4.8"
if m.startswith("gemini"):
return "google/" + model
return "openai/gpt-5.6-luna"
def _temp_for(model):
"""推理模型（gpt-5 / o 系列等）不接受 temperature=0。"""
return (1 if any(k in (model or "").lower()
for k in ("gpt-5", "o1", "o3", "o4", "thinking", "reasoner", "kimi-k3"))
else 0)
def client() -> OpenAI:
"""构造（并缓存）OpenAI 客户端，含通用 OpenRouter 兜底。
- 有 OPENAI_API_KEY：直连；但默认模型 gpt-5.x（直连需组织实名认证）且设置了
OPENROUTER_API_KEY 时优先走 OpenRouter。
- 无 OPENAI_API_KEY 但有 OPENROUTER_API_KEY：改走 OpenRouter（模型名自动映射）。
"""
global _client, TEXT_MODEL, VISION_MODEL
if _client is None:
api_key = os.getenv("OPENAI_API_KEY")
base_url = os.getenv("OPENAI_BASE_URL")
orkey = os.getenv("OPENROUTER_API_KEY")
prefer_or = bool(orkey) and (
(TEXT_MODEL or "").lower().startswith("gpt-5") or (VISION_MODEL or "").lower().startswith("gpt-5")
)
if prefer_or or (not api_key and orkey):
api_key, base_url = orkey, OPENROUTER_BASE_URL
TEXT_MODEL = map_model_to_openrouter(TEXT_MODEL)
VISION_MODEL = map_model_to_openrouter(VISION_MODEL)
kw = {}
if api_key:
kw["api_key"] = api_key
if base_url:
kw["base_url"] = base_url
_client = OpenAI(**kw)
return _client
def _img_part(path: str) -> dict:
with open(path, "rb") as f:
b64 = base64.b64encode(f.read()).decode()
return {"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{b64}", "detail": "low"}}
def _extract_json(text: str) -> dict:
"""从 LLM 回复里稳健地抠出第一个 JSON 对象。"""
m = re.search(r"\{.*\}", text, re.DOTALL)
if not m:
raise ValueError(f"未能从回复中解析 JSON：{text[:200]}")
return json.loads(m.group(0))
def _num(value, default: float) -> float:
"""把 LLM 返回的数值字段转成 float；字段缺失、为 null 或非法时回退 default。"""
try:
return float(value)
except (TypeError, ValueError):
return default
class TokenMeter:
"""累计 token，用于对比'子 Agent 隔离截图'带来的主上下文节省。"""
def __init__(self):
self.prompt = 0
self.completion = 0
def add(self, resp):
u = getattr(resp, "usage", None)
if u:
self.prompt += u.prompt_tokens
self.completion += u.completion_tokens
def total(self):
return self.prompt + self.completion
# --------------------------------------------------------------------------- #
# 视频分析子 Agent：两步 Vision 定位
# --------------------------------------------------------------------------- #
class VideoAnalyzerAgent:
def __init__(self, meter: TokenMeter = None):
self.meter = meter or TokenMeter()
def _vision_locate(self, video, timestamps, question, frame_dir):
"""抽取给定时间点的帧，连同问题交给 Vision LLM，返回 {start,end}。"""
content = [{
"type": "text",
"text": (
f"下面是同一段视频在不同时间点的截图（每张图前标注了该帧的时间，单位秒）。\n"
f"目标问题：{question}\n"
f"请判断'目标场景'在视频中出现的时间区间。只依据画面内容判断。\n"
f"严格输出 JSON：{{\"start\": <起点秒>, \"end\": <终点秒>, "
f"\"reason\": \"<简要依据>\"}}。若所有截图都看不到目标场景，"
f"令 start=end=-1。"
),
}]
for t in timestamps:
png = os.path.join(frame_dir, f"f_{t:.1f}.png")
extract_frame(video, t, png)
content.append({"type": "text", "text": f"[时间 t={t:.1f}s]"})
content.append(_img_part(png))
resp = client().chat.completions.create(
model=VISION_MODEL,
messages=[{"role": "user", "content": content}],
temperature=_temp_for(VISION_MODEL),
max_tokens=300,
)
self.meter.add(resp)
data = _extract_json(resp.choices[0].message.content)
# 模型可能省略 start/end 或返回 null——按约定的 -1 哨兵处理，走兜底逻辑。
return _num(data.get("start"), -1.0), _num(data.get("end"), -1.0), data.get("reason", "")
def locate(self, video, question, coarse_interval=10.0, fine_interval=1.0,
frame_dir="output/frames"):
"""
两步定位：
第一步（粗）：每 coarse_interval 秒一帧，Vision 给出大致场景区间。
第二步（细）：在粗区间上下各扩一个粗间隔，每 fine_interval 秒一帧，
Vision 精确定位边界。
返回 (start, end, trace)。
"""
os.makedirs(frame_dir, exist_ok=True)
duration = probe_duration(video)
trace = {}
# ---- 第一步：粗粒度 ----
coarse_ts = [t for t in _frange(0, duration, coarse_interval)]
cs, ce, creason = self._vision_locate(video, coarse_ts, question, frame_dir)
trace["coarse"] = {"timestamps": coarse_ts, "start": cs, "end": ce,
"reason": creason}
if cs < 0 or ce < 0:
# 兜底：粗定位失败——退化为全视频精扫（步长放大以控制成本）。
trace["coarse_fallback"] = True
step = max(fine_interval, duration / 20.0)
scan_ts = list(_frange(0, duration, step))
cs, ce, creason = self._vision_locate(video, scan_ts, question, frame_dir)
trace["coarse"]["fallback_scan"] = {"start": cs, "end": ce}
if cs < 0:
raise RuntimeError(
"Vision 定位失败：在整段视频里都没找到匹配'{}'的场景。\n"
"请检查需求描述是否与视频内容相符，或更换视频。".format(question)
)
# ---- 第二步：细粒度（在粗区间外扩一个粗间隔）----
lo = max(0.0, cs - coarse_interval)
hi = min(duration, ce + coarse_interval)
fine_ts = list(_frange(lo, hi, fine_interval))
fs, fe, freason = self._vision_locate(video, fine_ts, question, frame_dir)
trace["fine"] = {"window": [lo, hi], "timestamps_count": len(fine_ts),
"start": fs, "end": fe, "reason": freason}
if fs < 0 or fe < 0 or fe <= fs:
# 兜底：细定位失败——采用粗定位结果，保证流程可继续。
trace["fine_fallback"] = True
fs, fe = cs, ce
# 收敛到视频范围内。
fs = max(0.0, fs)
fe = min(duration, fe)
return fs, fe, trace
def _frange(start, stop, step):
"""浮点 range（含首、含接近末尾的采样点）。"""
out = []
t = start
while t < stop - 1e-6:
out.append(round(t, 3))
t += step
# 补一个接近末尾的采样点，确保末段场景被覆盖。
last = round(max(start, stop - 0.5), 3)
if not out or abs(out[-1] - last) > step / 2:
out.append(last)
return out
# --------------------------------------------------------------------------- #
# Proposer Agent
# --------------------------------------------------------------------------- #
class ProposerAgent:
def __init__(self, meter: TokenMeter = None):
self.meter = meter or TokenMeter()
def parse_request(self, nl_request: str) -> dict:
"""把自然语言需求解析成结构化意图：目标场景描述 + 特效列表。"""
resp = client().chat.completions.create(
model=TEXT_MODEL,
temperature=_temp_for(TEXT_MODEL),
max_tokens=400,
messages=[{
"role": "user",
"content": (
"你是视频剪辑规划器。把用户的中文剪辑需求解析成 JSON。\n"
"字段：\n"
" target_query: 用于视觉定位的一句话描述（英文更利于匹配画面文字），"
"说明要剪出哪个场景；\n"
" effects: 特效数组，元素形如 "
"{\"type\":\"subtitle\",\"text\":\"...\"} 或 "
"{\"type\":\"slowmo\",\"factor\":2.0}，无特效则为 []。\n"
f"用户需求：{nl_request}\n"
"只输出 JSON。"
),
}],
)
self.meter.add(resp)
return _extract_json(resp.choices[0].message.content)
def revise_bounds(self, start, end, feedback, duration):
"""根据 Reviewer 反馈微调边界（保守外扩/内收）。"""
resp = client().chat.completions.create(
model=TEXT_MODEL,
temperature=_temp_for(TEXT_MODEL),
max_tokens=200,
messages=[{
"role": "user",
"content": (
f"当前剪辑区间 start={start:.1f}s end={end:.1f}s，视频总长 {duration:.1f}s。\n"
f"审核反馈：{feedback}\n"
"请给出修正后的区间，输出 JSON {\"start\":..,\"end\":..}。"
"若反馈指出包含了无关片段则内收，若指出遗漏内容则外扩，幅度 1~5 秒。"
),
}],
)
self.meter.add(resp)
d = _extract_json(resp.choices[0].message.content)
# 模型可能省略 start/end 或返回 null——缺失时维持当前区间不变。
return max(0.0, _num(d.get("start"), start)), min(duration, _num(d.get("end"), end))
# --------------------------------------------------------------------------- #
# Reviewer Agent
# --------------------------------------------------------------------------- #
class ReviewerAgent:
def __init__(self, meter: TokenMeter = None):
self.meter = meter or TokenMeter()
def review(self, clip_path, target_query, frame_dir="output/review_frames"):
"""
抽取成片的首/中/尾关键帧，用 Vision 检查：
- 是否完整包含目标场景（无遗漏）；
- 是否夹带了无关场景（无多余）。
返回结构化结果 {pass, score, feedback, frames_checked}。
"""
os.makedirs(frame_dir, exist_ok=True)
dur = probe_duration(clip_path)
# 取首/中/尾，并在首尾稍微内缩避开黑帧。
keyts = [min(0.5, dur * 0.1), dur / 2.0, max(0.0, dur - 0.5)]
content = [{
"type": "text",
"text": (
f"这是剪辑成片的几个关键帧（首/中/尾）。剪辑目标是：{target_query}。\n"
"请检查：(1) 成片是否完整呈现了目标场景；(2) 是否夹带了不该出现的其他场景。\n"
"严格输出 JSON：{\"pass\": true/false, \"score\": 0-10, "
"\"feedback\": \"<发现的问题或确认无误>\"}。"
),
}]
for t in keyts:
png = os.path.join(frame_dir, f"r_{t:.1f}.png")
extract_frame(clip_path, t, png)
content.append({"type": "text", "text": f"[成片内 t={t:.1f}s]"})
content.append(_img_part(png))
resp = client().chat.completions.create(
model=VISION_MODEL,
temperature=_temp_for(VISION_MODEL),
max_tokens=300,
messages=[{"role": "user", "content": content}],
)
self.meter.add(resp)
data = _extract_json(resp.choices[0].message.content)
data["frames_checked"] = keyts
return data
blender_editor.py¶
"""
Blender Python API（bpy）剪辑执行层 —— 实验 5-6 的核心。
书中方案强调"代码生成"：Proposer Agent 不去点 GUI，而是**生成一段调用
Blender Python API 的脚本**，每个编辑操作（导入 / 裁剪 / 字幕 / 变速 / 渲染）
对应一个清晰的函数调用，再用 `blender --background --python edit.py` 无头执行。
本模块两个出口：
generate_bpy_script(source, plan, out_video) -> str
纯字符串生成，**不依赖 bpy**，任何机器都能产出这段脚本（体现代码生成能力，
可人工核对，或拷到装了 Blender 的机器上执行）。
render_with_blender(source, plan, out_video, script_path) -> str
若本机 `blender` 可执行，写出脚本并无头渲染产出成片；否则抛错由调用方回退。
裁剪基于 Blender 视频序列编辑器（VSE）：new_movie / new_sound 导入素材，
frame_offset_start + frame_final_duration 完成裁剪，TEXT / SPEED 特效条叠加，
FFMPEG(H.264+AAC) 容器渲染。API 面向 Blender 3.x / 4.x。
"""
import os
import shutil
import subprocess
def blender_available() -> bool:
"""本机是否有 blender 可执行文件（决定 backend=auto 时走 Blender 还是 ffmpeg）。"""
return shutil.which("blender") is not None
# 生成的 bpy 脚本模板。占位符全部通过 repr() 注入，保证是合法的 Python 字面量。
_BPY_TEMPLATE = '''"""
本文件由 blender_editor.generate_bpy_script() 自动生成（实验 5-6）。
执行：blender --background --python edit.py
它把一条剪辑计划翻译成 Blender 视频序列编辑器（VSE）的 API 调用序列。
"""
import os
import bpy
SRC = {src}
OUT = {out}
FPS = {fps}
START = {start} # 目标片段起点（秒）
END = {end} # 目标片段终点（秒）
SUBTITLE = {subtitle} # None 或字幕文本
SLOWMO = {slowmo} # None 或放慢倍率（factor>1 表示放慢 factor 倍）
scene = bpy.context.scene
scene.render.fps = FPS
scene.render.fps_base = 1.0
# 清掉可能存在的旧序列，保证幂等
if scene.sequence_editor:
bpy.ops.sequencer.select_all(action='SELECT')
bpy.ops.sequencer.delete()
se = scene.sequence_editor_create()
start_frame = int(round(START * FPS))
dur_frames = max(1, int(round((END - START) * FPS)))
# 1) 导入影片 + 音轨（new_sound 在无音轨素材上会抛 RuntimeError，忽略即可）
movie = se.sequences.new_movie(name="clip", filepath=SRC, channel=1, frame_start=1)
try:
sound = se.sequences.new_sound(name="audio", filepath=SRC, channel=2, frame_start=1)
except RuntimeError:
sound = None
# 2) 裁剪 [START, END]：偏移掉片头，再固定成片时长
for strip in (movie, sound):
if strip is None:
continue
strip.frame_offset_start = start_frame
strip.frame_final_duration = dur_frames
top_channel = 3
# 3) 慢动作：SPEED 特效条（MULTIPLY 模式，speed_factor = 1/倍率）
if SLOWMO:
speed = se.sequences.new_effect(
name="slowmo", type='SPEED', channel=top_channel,
frame_start=1, frame_end=1 + dur_frames, seq1=movie,
)
speed.use_default_fade = False
speed.speed_control = 'MULTIPLY'
speed.speed_factor = 1.0 / SLOWMO
top_channel += 1
# 放慢后成片总帧数按倍率拉长
render_dur = int(round(dur_frames * SLOWMO))
movie.frame_final_duration = render_dur
else:
render_dur = dur_frames
# 4) 字幕：TEXT 特效条，底部居中带半透明底框
if SUBTITLE:
txt = se.sequences.new_effect(
name="subtitle", type='TEXT', channel=top_channel,
frame_start=1, frame_end=1 + render_dur,
)
txt.text = SUBTITLE
txt.font_size = 100
txt.location = (0.5, 0.12)
txt.align_x = 'CENTER'
txt.align_y = 'BOTTOM'
txt.use_box = True
txt.box_color = (0.0, 0.0, 0.0, 0.6)
# 5) 渲染范围 + 输出为 mp4(H.264+AAC)
scene.frame_start = 1
scene.frame_end = render_dur
r = scene.render
r.image_settings.file_format = 'FFMPEG'
r.ffmpeg.format = 'MPEG4'
r.ffmpeg.codec = 'H264'
r.ffmpeg.audio_codec = 'AAC'
r.filepath = OUT
os.makedirs(os.path.dirname(OUT) or ".", exist_ok=True)
bpy.ops.render.render(animation=True)
print("BLENDER_RENDER_DONE", OUT)
'''
def _plan_fields(plan: dict):
"""从剪辑计划里抽出 bpy 脚本需要的字段。"""
start, end = float(plan["start"]), float(plan["end"])
if end <= start:
raise ValueError(f"剪辑区间非法：start={start} >= end={end}")
effects = plan.get("effects", []) or []
subtitle = None
slowmo = None
for eff in effects:
etype = eff.get("type")
if etype == "subtitle":
subtitle = eff.get("text", "")
elif etype == "slowmo":
slowmo = float(eff.get("factor", 2.0))
return start, end, subtitle, slowmo
def generate_bpy_script(source: str, plan: dict, out_video: str, fps: int = 30) -> str:
"""把剪辑计划渲染成一段可执行的 Blender Python(bpy) 脚本文本（不依赖 bpy）。"""
start, end, subtitle, slowmo = _plan_fields(plan)
return _BPY_TEMPLATE.format(
src=repr(os.path.abspath(source)),
out=repr(os.path.abspath(out_video)),
fps=int(fps),
start=repr(start),
end=repr(end),
subtitle=repr(subtitle),
slowmo=repr(slowmo),
)
def write_bpy_script(source: str, plan: dict, out_video: str,
script_path: str, fps: int = 30) -> str:
"""生成 bpy 脚本并落盘（无论用哪个后端都会产出，作为代码生成产物）。"""
script = generate_bpy_script(source, plan, out_video, fps=fps)
os.makedirs(os.path.dirname(script_path) or ".", exist_ok=True)
with open(script_path, "w") as f:
f.write(script)
return script_path
def render_with_blender(source: str, plan: dict, out_video: str,
script_path: str, fps: int = 30) -> str:
"""写出 bpy 脚本并用 `blender --background --python` 无头执行，产出成片。"""
if not blender_available():
raise RuntimeError(
"指定用 Blender 后端，但未找到 blender 可执行文件。\n"
" 安装后确保 `blender --version` 可用：https://www.blender.org/download/\n"
" 或改用 --backend ffmpeg。"
)
write_bpy_script(source, plan, out_video, script_path, fps=fps)
os.makedirs(os.path.dirname(out_video) or ".", exist_ok=True)
proc = subprocess.run(
["blender", "--background", "--python", script_path],
capture_output=True, text=True,
)
if proc.returncode != 0 or not os.path.exists(out_video):
tail = "\n".join(proc.stderr.strip().splitlines()[-12:])
raise RuntimeError(
f"Blender 渲染失败（exit={proc.returncode}）：\n{tail}\n"
f" 可人工检查生成的脚本：{script_path}"
)
return out_video
if __name__ == "__main__":
# 零依赖自检：打印一段"含裁剪 + 字幕"的示例 bpy 脚本（可 py_compile 校验其语法）。
demo_plan = {
"start": 16.0,
"end": 28.0,
"effects": [{"type": "subtitle", "text": "SURFING"}],
}
print("# blender available:", blender_available())
print("# ---- generated edit.py ----")
print(generate_bpy_script("output/source.mp4", demo_plan, "output/final.mp4"))
demo.py¶
"""
实验 5-6：基于 API 的智能视频剪辑（两步 Vision 定位 + 提议者-审核者）
一条命令跑通：
python demo.py # 默认需求"把冲浪的部分剪出来"
python demo.py "把滑雪部分剪出来，并加上字幕 Winter" # 自定义需求
流程：
1. 程序化生成含 4 个明显不同场景的测试视频（HIKING/SURFING/SKIING/CYCLING）；
2. Proposer 解析自然语言需求 → 目标场景 + 特效；
3. 视频分析子 Agent 两步定位（粗粒度每 10s → 细粒度每 1s）找到精确边界；
4. Proposer 生成 Blender Python API（bpy）脚本剪出片段（可含字幕/慢动作）；
装了 Blender 则无头渲染，否则回退 ffmpeg——但 bpy 脚本始终生成（代码生成产物）；
5. Reviewer 检查成片关键帧，给出反馈；不合格则 Proposer 修正边界重剪，迭代。
依赖：ffmpeg/ffprobe（回退后端 + 抽帧）、OPENAI_API_KEY（gpt-5.6-luna 视觉 + 文本；未配置时可用 OPENROUTER_API_KEY 兜底）；
可选 Blender（书中原方案，`--backend blender`）。
常用命令（完整用法见 `python demo.py --help`）：
python demo.py # 默认需求，完整流程
python demo.py --quick # 快速模式：粗采样 + 单轮审查，省时省钱
python demo.py --smoke # 冒烟自检：仅剪辑链路 + 生成 bpy 脚本，不调用任何 API
"""
import argparse
import os
import shutil
import sys
from dotenv import load_dotenv
load_dotenv()
HERE = os.path.dirname(os.path.abspath(__file__))
OUT_DIR = os.path.join(HERE, "output")
SOURCE_VIDEO = os.path.join(OUT_DIR, "source.mp4") # 测试片输出位置
FINAL_VIDEO = os.path.join(OUT_DIR, "final.mp4")
MAX_ROUNDS = 3 # Reviewer 反馈后最多重剪次数（默认，可用 --max-rounds 覆盖）
DEFAULT_REQUEST = "把冲浪的部分剪出来"
def banner(title):
print("\n" + "=" * 74)
print(f" {title}")
print("=" * 74)
def build_arg_parser() -> argparse.ArgumentParser:
"""命令行参数：位置参数为中文剪辑需求，另有输入/输出/后端/模型/快速等开关。"""
p = argparse.ArgumentParser(
prog="demo.py",
description="实验 5-6：基于 API 的智能视频剪辑（两步 Vision 定位 + 提议者-审核者）",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"示例：\n"
" python demo.py\n"
" python demo.py \"把滑雪部分剪出来，并加上字幕 Winter\"\n"
" python demo.py -i my.mp4 -o out.mp4 \"把演讲开场剪出来\"\n"
" python demo.py --backend blender # 强制用 Blender Python API 渲染\n"
" python demo.py --quick # 更少 Vision 调用，快速验证链路\n"
" python demo.py --smoke # 只跑剪辑链路 + 生成 bpy 脚本，不调用任何 API\n"
),
)
p.add_argument("request", nargs="?", default=DEFAULT_REQUEST,
help="中文剪辑需求（默认：%(default)s）")
p.add_argument("--input", "-i", metavar="VIDEO", default=None,
help="输入视频路径（不指定则程序化生成 4 场景测试片）")
p.add_argument("--output", "-o", metavar="VIDEO", default=FINAL_VIDEO,
help="成片输出路径（默认 output/final.mp4）")
p.add_argument("--backend", choices=["auto", "blender", "ffmpeg"], default="auto",
help="剪辑后端：auto=装了 Blender 用 bpy 否则 ffmpeg；"
"blender=强制 Blender Python API；ffmpeg=强制 ffmpeg（默认 auto）")
p.add_argument("--text-model", metavar="NAME", default=None,
help="覆盖文本模型（否则用 $TEXT_MODEL，默认 gpt-5.6-luna）")
p.add_argument("--vision-model", metavar="NAME", default=None,
help="覆盖视觉模型，须支持图像输入（否则用 $VISION_MODEL，默认 gpt-5.6-luna）")
p.add_argument("--quick", action="store_true",
help="快速模式：粗采样（15s/2s）+ 单轮审查，减少 Vision API 调用")
p.add_argument("--max-rounds", type=int, default=MAX_ROUNDS, metavar="N",
help="Reviewer 反馈后最多重剪轮数（默认 %(default)s；--quick 时强制为 1）")
p.add_argument("--smoke", action="store_true",
help="冒烟自检：仅剪辑链路 + 生成 bpy 脚本，不调用任何 API")
return p
def smoke_check():
"""冒烟自检：不触碰 OpenAI，验证剪辑链路可用并生成 Proposer 的 bpy 脚本。"""
from blender_editor import blender_available
from ffmpeg_utils import ensure_ffmpeg, extract_frame, format_probe
from make_test_video import GROUND_TRUTH, make as make_test_video
from video_editor import apply_edit
banner("冒烟自检 | 剪辑链路 + bpy 脚本生成，不调用任何 API")
try:
ensure_ffmpeg()
except RuntimeError as e:
print(f"\n[错误] {e}")
sys.exit(1)
if os.path.isdir(OUT_DIR):
shutil.rmtree(OUT_DIR)
os.makedirs(OUT_DIR, exist_ok=True)
make_test_video(SOURCE_VIDEO)
print(f"[1/3] 生成测试视频 OK：{SOURCE_VIDEO}（场景真值={GROUND_TRUTH}）")
frame_dir = os.path.join(OUT_DIR, "frames")
os.makedirs(frame_dir, exist_ok=True) # extract_frame 要求目录已存在
frame = extract_frame(SOURCE_VIDEO, 20.0, os.path.join(frame_dir, "smoke.png"))
print(f"[2/3] 抽帧 OK：{frame}")
clip = os.path.join(OUT_DIR, "smoke_cut.mp4")
script_path = os.path.join(OUT_DIR, "edit.py")
# backend="auto"：未装 Blender 则用 ffmpeg 实际渲染，但仍生成 bpy 脚本（代码生成产物）。
apply_edit(SOURCE_VIDEO, {"start": 15.0, "end": 20.0,
"effects": [{"type": "subtitle", "text": "SMOKE"}]},
clip, backend="auto", script_path=script_path)
used = "Blender bpy" if blender_available() else "ffmpeg（未装 Blender，回退）"
print(f"[3/3] 剪辑+字幕 OK（后端={used}）：\n{format_probe(clip)}")
print(f"\n已生成 Proposer 的 Blender 脚本：{script_path}")
print("（这正是书中'生成 Blender Python API 代码'的产物；装好 Blender 后可直接")
print(f" `blender --background --python {script_path}` 无头渲染。）")
print("\n✓ 冒烟自检通过：剪辑链路正常 + bpy 脚本已生成（未调用 OpenAI）。")
def preflight():
"""启动自检：给出清晰中文报错，而非 traceback。"""
from ffmpeg_utils import ensure_ffmpeg
if not (os.getenv("OPENAI_API_KEY") or os.getenv("OPENROUTER_API_KEY")):
print("\n[错误] 未检测到 OPENAI_API_KEY（或 OPENROUTER_API_KEY 兜底）。\n"
" 请复制 env.example 为 .env 并填入有效的 OpenAI Key，或执行：\n"
" export OPENAI_API_KEY=sk-... # 或 export OPENROUTER_API_KEY=sk-or-...\n"
" 本实验用 gpt-5.6-luna 做视觉定位与审查，必须提供有效 Key。")
sys.exit(1)
try:
ensure_ffmpeg()
except RuntimeError as e:
print(f"\n[错误] {e}")
sys.exit(1)
def main():
args = build_arg_parser().parse_args()
if args.smoke: # 仅剪辑链路，不需要 API Key，提前返回。
smoke_check()
return
nl_request = args.request
# --quick：粗化采样步长并只审查一轮，把 Vision 调用降到最少（用于快速验证链路）。
coarse_interval = 15.0 if args.quick else 10.0
fine_interval = 2.0 if args.quick else 1.0
max_rounds = 1 if args.quick else max(1, args.max_rounds)
# 模型覆盖：写回环境变量，供 agents 模块（惰性初始化）读取。须在导入 agents 前设置。
if args.text_model:
os.environ["TEXT_MODEL"] = args.text_model
if args.vision_model:
os.environ["VISION_MODEL"] = args.vision_model
preflight()
# 延迟导入：确保 preflight 的报错优先于任何 SDK 初始化。
from agents import (ProposerAgent, ReviewerAgent, VideoAnalyzerAgent,
TokenMeter, TEXT_MODEL, VISION_MODEL)
from blender_editor import blender_available
from ffmpeg_utils import format_probe, probe_duration
from make_test_video import make as make_test_video, GROUND_TRUTH
from video_editor import apply_edit
# 幂等：每次从干净的 output/ 开始。
if os.path.isdir(OUT_DIR):
shutil.rmtree(OUT_DIR)
os.makedirs(OUT_DIR, exist_ok=True)
ground_truth = None
if args.input:
banner("步骤 0 | 使用外部输入视频")
source_video = os.path.abspath(args.input)
if not os.path.isfile(source_video):
print(f"\n[错误] 输入视频不存在：{source_video}")
sys.exit(1)
print(f"输入视频：{source_video}")
else:
banner("步骤 0 | 生成测试视频（4 个明显不同的场景）")
source_video = SOURCE_VIDEO
make_test_video(source_video)
ground_truth = GROUND_TRUTH
print(f"已生成 {source_video}")
print(f"场景真值（用于核对定位误差）：{ground_truth}")
total_dur = probe_duration(source_video)
print(f"时长 {total_dur:.1f}s")
print(f"文本模型={TEXT_MODEL} 视觉模型={VISION_MODEL} 剪辑后端={args.backend}")
# 分离的 token 计量：主 Agent（Proposer+Reviewer）vs 子 Agent（截图定位）。
main_meter = TokenMeter()
sub_meter = TokenMeter()
proposer = ProposerAgent(main_meter)
reviewer = ReviewerAgent(main_meter)
analyzer = VideoAnalyzerAgent(sub_meter)
banner("步骤 1 | Proposer 解析自然语言需求")
print(f"用户需求：{nl_request}")
intent = proposer.parse_request(nl_request)
# 模型可能省略 target_query 或返回 null——退化为用原始需求文本做视觉定位。
target_query = intent.get("target_query") or nl_request
effects = intent.get("effects", [])
print(f"解析结果：目标场景='{target_query}' 特效={effects}")
banner("步骤 2 | 视频分析子 Agent：两步 Vision 定位"
+ ("（--quick 快速采样）" if args.quick else ""))
start, end, trace = analyzer.locate(
source_video, target_query,
coarse_interval=coarse_interval, fine_interval=fine_interval,
frame_dir=os.path.join(OUT_DIR, "frames"),
)
c = trace["coarse"]
print(f" [粗粒度] 每 {coarse_interval:.0f}s 采样 {len(c['timestamps'])} 帧 → Vision 得区间 "
f"[{c['start']:.0f}, {c['end']:.0f}]s（依据：{c['reason']}）")
f = trace["fine"]
print(f" [细粒度] 窗口 {f['window']} 内每 {fine_interval:.0f}s 采样 {f['timestamps_count']} 帧 → "
f"精确边界 [{f['start']:.1f}, {f['end']:.1f}]s（依据：{f['reason']}）")
print(f" >>> 最终定位：起 {start:.1f}s 止 {end:.1f}s")
# 与真值对比，打印定位误差（验收：误差 ≤ ±3s）。仅测试片有真值。
key = _match_ground_truth(target_query, ground_truth) if ground_truth else None
if key:
gs, ge = ground_truth[key]
print(f" 真值 [{gs}, {ge}]s → 起点误差 {abs(start - gs):.1f}s，"
f"终点误差 {abs(end - ge):.1f}s（验收要求 ≤ 3s）")
banner("步骤 3-4 | Proposer 生成 bpy 脚本剪辑 + Reviewer 审查（迭代）")
plan = {"start": start, "end": end, "effects": effects}
final_path = None
for rnd in range(1, max_rounds + 1):
print(f"\n--- 第 {rnd} 轮 ---")
clip = os.path.join(OUT_DIR, f"cut_round{rnd}.mp4")
script_path = os.path.join(OUT_DIR, f"edit_round{rnd}.py")
apply_edit(source_video, plan, clip, backend=args.backend,
script_path=script_path)
cdur = probe_duration(clip)
used = "Blender bpy" if (args.backend == "blender" or
(args.backend == "auto" and blender_available())) else "ffmpeg"
print(f" Proposer 生成 Blender 脚本 → {script_path}")
print(f" 剪出片段 [{plan['start']:.1f}, {plan['end']:.1f}]s（后端={used}），"
f"成片时长 {cdur:.1f}s")
review = reviewer.review(clip, target_query,
frame_dir=os.path.join(OUT_DIR, "review_frames"))
print(f" Reviewer：pass={review.get('pass')} score={review.get('score')} "
f"检查帧={['%.1f' % t for t in review.get('frames_checked', [])]}")
print(f" Reviewer 反馈：{review.get('feedback', '（无）')}")
if review.get("pass"):
final_path = clip
print(" ✓ 审核通过。")
break
if rnd == max_rounds:
final_path = clip
print(" 达到最大轮数，采用当前成片。")
break
# 未通过：Proposer 据反馈修正边界后重剪。
ns, ne = proposer.revise_bounds(plan["start"], plan["end"],
review.get("feedback", ""), total_dur)
print(f" Proposer 据反馈修正边界：[{ns:.1f}, {ne:.1f}]s")
plan["start"], plan["end"] = ns, ne
output_path = os.path.abspath(args.output)
os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)
shutil.copy(final_path, output_path)
banner("步骤 5 | 成片信息（ffprobe）")
print(format_probe(output_path))
banner("Token 统计（子 Agent 隔离截图，主上下文不被污染）")
print(f" 主 Agent（Proposer+Reviewer）：{main_meter.total()} tokens "
f"(prompt={main_meter.prompt}, completion={main_meter.completion})")
print(f" 子 Agent（两步定位截图） ：{sub_meter.total()} tokens "
f"(prompt={sub_meter.prompt}, completion={sub_meter.completion})")
print(f"\n完成：{output_path}")
def _match_ground_truth(query, gt):
q = query.lower()
for key in gt:
if key in q:
return key
# 中文关键词兜底映射。
zh = {"冲浪": "surfing", "徒步": "hiking", "滑雪": "skiing", "骑": "cycling",
"hik": "hiking", "surf": "surfing", "ski": "skiing", "cycl": "cycling"}
for k, v in zh.items():
if k in q:
return v
return None
if __name__ == "__main__":
main()
ffmpeg_utils.py¶
"""
ffmpeg / ffprobe 薄封装：所有对外部进程的调用都集中在这里，统一做错误检查。
设计要点：
- run() 捕获非零退出码并抛出带 stderr 的清晰异常（而非让 traceback 泄漏）；
- 提供 probe_duration / probe_streams，供 Reviewer 与验证环节读取成片信息；
- extract_frame 把某一时间点抽成一张 PNG（缩放到 512 宽以节省 Vision token）。
"""
import json
import os
import shutil
import subprocess
# macOS 自带字体；换平台时改这里即可（Linux 常见 DejaVuSans.ttf）。
FONT_CANDIDATES = [
"/System/Library/Fonts/Supplemental/Arial.ttf",
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
"/Library/Fonts/Arial.ttf",
]
def find_font() -> str:
for p in FONT_CANDIDATES:
if os.path.exists(p):
return p
return "" # drawtext 会退化为默认字体
def ensure_ffmpeg():
"""启动前自检：ffmpeg / ffprobe 是否可用，给出清晰中文报错。"""
for tool in ("ffmpeg", "ffprobe"):
if shutil.which(tool) is None:
raise RuntimeError(
f"未找到 {tool}，本项目用 ffmpeg 完成实际剪辑。\n"
f" macOS: brew install ffmpeg\n"
f" Ubuntu: sudo apt install ffmpeg"
)
def run(cmd, desc="ffmpeg 命令"):
"""执行命令，失败时抛出带 stderr 尾部的异常。"""
proc = subprocess.run(cmd, capture_output=True, text=True)
if proc.returncode != 0:
tail = "\n".join(proc.stderr.strip().splitlines()[-8:])
raise RuntimeError(f"{desc} 执行失败（exit={proc.returncode}）：\n{tail}")
return proc
def probe_duration(path: str) -> float:
"""返回视频时长（秒）。文件缺少时长元数据时 ffprobe 输出 N/A，给出清晰报错。"""
proc = run(
["ffprobe", "-v", "error", "-show_entries", "format=duration",
"-of", "default=noprint_wrappers=1:nokey=1", path],
desc="ffprobe 读取时长",
)
out = proc.stdout.strip()
if not out or out == "N/A":
raise RuntimeError(f"ffprobe 无法读取时长（文件缺少时长元数据或不是音视频文件）：{path}")
return float(out)
def probe_streams(path: str) -> dict:
"""返回 ffprobe 的 JSON（format + streams），用于打印成片信息。"""
proc = run(
["ffprobe", "-v", "error", "-show_format", "-show_streams",
"-of", "json", path],
desc="ffprobe 读取流信息",
)
return json.loads(proc.stdout)
def format_probe(path: str) -> str:
"""把成片信息格式化成一行行的人类可读文本（用于验证输出）。"""
info = probe_streams(path)
fmt = info.get("format", {})
lines = [
f" 文件: {os.path.basename(path)}",
f" 时长: {float(fmt.get('duration', 0)):.2f}s",
f" 容器: {fmt.get('format_name', '?')}",
f" 大小: {int(fmt.get('size', 0)) / 1024:.1f} KB",
]
for s in info.get("streams", []):
if s.get("codec_type") == "video":
lines.append(
f" 视频流: {s.get('codec_name')} {s.get('width')}x{s.get('height')} "
f"@ {s.get('r_frame_rate')} fps"
)
elif s.get("codec_type") == "audio":
lines.append(
f" 音频流: {s.get('codec_name')} {s.get('sample_rate')}Hz "
f"{s.get('channels')}ch"
)
return "\n".join(lines)
def extract_frame(video: str, t: float, out_png: str, width: int = 512):
"""抽取 t 秒处的一帧，缩放到 width 宽存为 PNG。"""
run(
["ffmpeg", "-y", "-ss", f"{t:.3f}", "-i", video,
"-frames:v", "1", "-vf", f"scale={width}:-1", out_png],
desc=f"抽帧 t={t:.1f}s",
)
return out_png
make_test_video.py¶
"""
程序化生成一段"含多个明显不同场景"的测试视频（无需任何素材文件）。
每个场景 = 一种纯色背景 + 一个大号运动标题（场景英文名）+ 时间码水印。
标题让 Vision LLM 能仅凭画面就准确判断"这是哪个场景"，从而验证两步定位。
换成真实视频时：把 demo.py 里的 SOURCE_VIDEO 指向你自己的 mp4 即可（见 README）。
"""
import os
from ffmpeg_utils import find_font, run
# 每个场景：(名称, 背景色, 起始秒, 时长秒)。刻意让每段 > 10s，
# 使"每 10s 一张"的粗粒度采样必然命中每个场景。
SCENES = [
("HIKING", "0x1E6B3A", 0, 15), # 森林绿
("SURFING", "0x1565C0", 15, 15), # 海洋蓝
("SKIING", "0xE0E0E0", 30, 12), # 雪地白
("CYCLING", "0xE65100", 42, 12), # 落日橙
]
TOTAL = SCENES[-1][2] + SCENES[-1][3] # 54s
W, H, FPS = 1280, 720, 30
def _drawtext(text, size, y_expr, color="white", box=False):
font = find_font()
parts = [f"text='{text}'", f"fontsize={size}", f"fontcolor={color}",
"x=(w-text_w)/2", f"y={y_expr}"]
if font:
parts.insert(0, f"fontfile={font}")
if box:
parts += ["box=1", "boxcolor=black@0.4", "boxborderw=20"]
return "drawtext=" + ":".join(parts)
def make(out_path: str) -> str:
"""生成测试视频，返回路径。幂等：每次覆盖，保证从干净状态开始。"""
out_dir = os.path.dirname(out_path)
if out_dir:
os.makedirs(out_dir, exist_ok=True)
clip_paths = []
tmp_dir = out_dir or "."
for i, (name, color, start, dur) in enumerate(SCENES):
clip = os.path.join(tmp_dir, f"_scene_{i}.mp4")
# 让标题上下缓慢漂移，制造真实"运动画面"，避免纯静止帧。
title = _drawtext(name, 140, "(h-text_h)/2 + 60*sin(t)", box=True)
# 左上角时间码：t 为片段内相对时间，加 start 得到全局时间。
# drawtext 里表达式含冒号，必须转义为 \: 否则被当成选项分隔符。
clock = _drawtext(rf"t=%{{eif\:t+{start}\:d}}s", 48,
"40", color="yellow")
vf = f"{title},{clock}"
run(
["ffmpeg", "-y",
"-f", "lavfi", "-i", f"color=c={color}:s={W}x{H}:d={dur}:r={FPS}",
"-f", "lavfi", "-i", f"sine=frequency={220 + i * 110}:duration={dur}",
"-vf", vf, "-pix_fmt", "yuv420p",
"-c:v", "libx264", "-c:a", "aac", "-shortest", clip],
desc=f"生成场景 {name}",
)
clip_paths.append(clip)
# 用 concat demuxer 无缝拼接成完整原始素材。
list_file = os.path.join(tmp_dir, "_concat_list.txt")
with open(list_file, "w") as f:
for c in clip_paths:
f.write(f"file '{os.path.abspath(c)}'\n")
run(
["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", list_file,
"-c", "copy", out_path],
desc="拼接测试视频",
)
# 清理中间片段。
for c in clip_paths:
os.remove(c)
os.remove(list_file)
return out_path
# 供 demo / README 引用：场景真值表，用于验证定位误差。
GROUND_TRUTH = {name.lower(): (start, start + dur) for name, _, start, dur in SCENES}
test_agents_llm_json.py¶
"""LLM 返回的 JSON 缺字段/为 null 时，Agent 解析应按约定哨兵处理，不应崩溃。"""
import types
import pytest
import agents
import ffmpeg_utils
def _fake_client(content):
resp = types.SimpleNamespace(
choices=[types.SimpleNamespace(
message=types.SimpleNamespace(content=content))],
usage=None)
completions = types.SimpleNamespace(create=lambda **kw: resp)
return types.SimpleNamespace(chat=types.SimpleNamespace(completions=completions))
def _stub_io(monkeypatch, content):
"""替换掉网络与帧抽取 IO，让 Agent 直接吃到给定的 LLM 回复文本。"""
monkeypatch.setattr(agents, "client", lambda: _fake_client(content))
monkeypatch.setattr(agents, "extract_frame", lambda *a, **k: None)
monkeypatch.setattr(agents, "_img_part",
lambda p: {"type": "image_url", "image_url": {"url": "data:,"}})
def test_vision_locate_missing_keys(monkeypatch):
"""模型省略 start/end → 按 -1 哨兵返回（走兜底逻辑），不抛 KeyError。"""
_stub_io(monkeypatch, '{"reason": "画面里看不到目标场景"}')
start, end, reason = agents.VideoAnalyzerAgent()._vision_locate(
"fake.mp4", [0.0], "目标", "frames")
assert (start, end) == (-1.0, -1.0)
assert reason == "画面里看不到目标场景"
def test_vision_locate_null_fields(monkeypatch):
"""模型返回显式 null → 同样按 -1 哨兵返回，不抛 TypeError。"""
_stub_io(monkeypatch, '{"start": null, "end": null, "reason": "not visible"}')
start, end, _ = agents.VideoAnalyzerAgent()._vision_locate(
"fake.mp4", [0.0], "目标", "frames")
assert (start, end) == (-1.0, -1.0)
def test_revise_bounds_null_start_keeps_current(monkeypatch):
"""修正区间为 null/缺失时维持当前值，正常数值仍生效。"""
_stub_io(monkeypatch, '{"start": null, "end": 5}')
ns, ne = agents.ProposerAgent().revise_bounds(1.0, 3.0, "反馈", 10.0)
assert ns == 1.0
assert ne == 5.0
def test_probe_duration_na(monkeypatch):
"""ffprobe 输出 N/A（无时长元数据）→ 清晰的 RuntimeError，而非 ValueError。"""
fake_proc = types.SimpleNamespace(stdout="N/A\n")
monkeypatch.setattr(ffmpeg_utils, "run", lambda *a, **k: fake_proc)
with pytest.raises(RuntimeError, match="时长"):
ffmpeg_utils.probe_duration("no_duration.bin")
def test_probe_duration_normal(monkeypatch):
fake_proc = types.SimpleNamespace(stdout="12.5\n")
monkeypatch.setattr(ffmpeg_utils, "run", lambda *a, **k: fake_proc)
assert ffmpeg_utils.probe_duration("a.mp4") == 12.5
test_bare_output_filename.py¶
import os
import make_test_video
from video_editor import apply_edit
def test_make_test_video_bare_filename():
out_name = "test_temp_bare_video.mp4"
if os.path.exists(out_name):
os.remove(out_name)
try:
path = make_test_video.make(out_name)
assert os.path.exists(path)
assert path == out_name
finally:
if os.path.exists(out_name):
os.remove(out_name)
for i in range(len(make_test_video.SCENES)):
scene_file = f"_scene_{i}.mp4"
if os.path.exists(scene_file):
os.remove(scene_file)
def test_apply_edit_bare_filename():
source = "test_source.mp4"
make_test_video.make(source)
out_name = "test_output_bare_video.mp4"
if os.path.exists(out_name):
os.remove(out_name)
plan = {
"start": 0.0,
"end": 2.0,
"effects": [{"type": "subtitle", "text": "hello"}],
}
try:
path = apply_edit(source, plan, out_name, backend="ffmpeg")
assert os.path.exists(path)
assert path == out_name
finally:
for name in (out_name, source, "edit.py"):
if os.path.exists(name):
os.remove(name)
for i in range(len(make_test_video.SCENES)):
scene_file = f"_scene_{i}.mp4"
if os.path.exists(scene_file):
os.remove(scene_file)
video_editor.py¶
"""
视频剪辑执行层（双后端）。
书中实验 5-6 的核心是"代码生成"：Proposer Agent 生成一段调用 Blender Python API
（bpy）的脚本来完成剪辑。因此本层有两个后端：
- blender：把剪辑计划翻译成 bpy 脚本，用 `blender --background --python` 无头渲染
（见 blender_editor.py）——书中原方案，需安装 Blender；
- ffmpeg ：用 ffmpeg 完成等价的裁剪/字幕/慢动作，单二进制、CI 友好，本机已验证。
apply_edit(..., backend=) 统一入口：backend="auto" 时装了 Blender 走 bpy，否则回退
ffmpeg。**无论走哪个后端，都会把 Proposer 生成的 bpy 脚本落盘（output/edit.py）**，
体现"生成 Blender Python API 代码"这一核心，便于人工核对或换机执行。
支持的操作：
- trim 裁剪 [start, end] 片段
- subtitle 在片段上叠加字幕
- slowmo 慢动作（放慢到 factor 倍时长）
所有操作最终产出一个标准 mp4（H.264 + AAC）。
"""
import os
from blender_editor import blender_available, render_with_blender, write_bpy_script
from ffmpeg_utils import find_font, run
def _esc(text: str) -> str:
"""转义 drawtext/subtitles 文本中的特殊字符。"""
return text.replace("\\", "\\\\").replace(":", r"\:").replace("'", r"\'")
def apply_edit(source: str, plan: dict, out_path: str,
backend: str = "auto", script_path: str = None) -> str:
"""
按剪辑计划 plan 生成成片。
plan 结构（由 Proposer Agent 产出）:
{
"start": float, # 目标片段起点（秒）
"end": float, # 目标片段终点（秒）
"effects": [ # 可选特效列表
{"type": "subtitle", "text": "..."},
{"type": "slowmo", "factor": 2.0}
]
}
backend:
"auto" 装了 blender 用 bpy，否则回退 ffmpeg（默认）。
"blender" 强制用 Blender bpy 无头渲染（未装 blender 则报错）。
"ffmpeg" 强制用 ffmpeg 剪辑。
无论哪个后端，都会把 Proposer 生成的 bpy 脚本写到 script_path
（默认 out 同目录 edit.py），作为"生成 Blender API 代码"的产物。
返回成片路径。
"""
os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True)
if script_path is None:
script_path = os.path.join(os.path.dirname(out_path) or ".", "edit.py")
# 始终产出 bpy 脚本：这正是书中"Proposer 生成 Blender 脚本"的落地产物。
write_bpy_script(source, plan, out_path, script_path)
use_blender = backend == "blender" or (backend == "auto" and blender_available())
if use_blender:
return render_with_blender(source, plan, out_path, script_path)
if backend == "blender": # 显式要求 Blender 却没装
return render_with_blender(source, plan, out_path, script_path) # 抛清晰错误
return _apply_edit_ffmpeg(source, plan, out_path)
def _apply_edit_ffmpeg(source: str, plan: dict, out_path: str) -> str:
"""ffmpeg 后端：与 bpy 脚本等价的裁剪/字幕/慢动作，产出 H.264+AAC 的 mp4。"""
start, end = float(plan["start"]), float(plan["end"])
if end <= start:
raise ValueError(f"剪辑区间非法：start={start} >= end={end}")
effects = plan.get("effects", []) or []
vf_chain = [] # 视频滤镜链
af_chain = [] # 音频滤镜链
font = find_font()
for eff in effects:
etype = eff.get("type")
if etype == "subtitle":
txt = _esc(eff.get("text", ""))
opts = [f"text='{txt}'", "fontsize=52", "fontcolor=white",
"x=(w-text_w)/2", "y=h-text_h-50",
"box=1", "boxcolor=black@0.6", "boxborderw=16"]
if font:
opts.insert(0, f"fontfile={font}")
vf_chain.append("drawtext=" + ":".join(opts))
elif etype == "slowmo":
factor = float(eff.get("factor", 2.0))
vf_chain.append(f"setpts={factor}*PTS")
# atempo 只支持 0.5~2.0，用 1/factor 放慢音频。
af_chain.append(f"atempo={max(0.5, min(2.0, 1.0 / factor))}")
cmd = ["ffmpeg", "-y", "-ss", f"{start:.3f}", "-to", f"{end:.3f}", "-i", source]
if vf_chain:
cmd += ["-vf", ",".join(vf_chain)]
if af_chain:
cmd += ["-af", ",".join(af_chain)]
cmd += ["-c:v", "libx264", "-c:a", "aac", "-pix_fmt", "yuv420p", out_path]
out_dir = os.path.dirname(out_path)
if out_dir:
os.makedirs(out_dir, exist_ok=True)
run(cmd, desc="ffmpeg 剪辑")
return out_path
# --------------------------------------------------------------------------- #
# Blender 后端说明
# --------------------------------------------------------------------------- #
# 真实的 Blender bpy 脚本生成与无头渲染已实现在 blender_editor.py：
# - generate_bpy_script()：把剪辑计划翻译成一段 bpy 脚本（new_movie / 裁剪 /
# TEXT / SPEED / render）——即书中"Proposer 生成 Blender 脚本"；
# - render_with_blender()：用 `blender --background --python edit.py` 无头执行。
# apply_edit(backend="blender") 即走这条路径；backend="auto"（默认）在未装 Blender 时
# 回退到本文件的 ffmpeg 后端。核心的"两步 Vision 定位 + 提议者-审核者"与执行层解耦，
# 两个后端共用同一份剪辑计划（plan），agents.py / demo.py 无需为切换后端改动逻辑。