第 23 章 · 扩展：音视频处理与无头渲染管线（FFmpeg/OpenCV/Playwright）

本章目标：

• 用 FFmpeg 对音视频做切分、抽帧、转码与音轨提取，为多模态 Agent 准备数据
• 用 OpenCV 实现「场景切分」与关键帧提取，把长视频压缩成可检索的图像集
• 用 librosa 提取音频特征（能量、静音检测），驱动智能切片与 VAD
• 用 Playwright 无头浏览器构建网页截图/PDF 渲染管线，作为 Agent 的视觉工具
• 将以上能力封装成 Agent 工具链，并做好安全隔离与资源配额

---

上一章解决了「文档」的入口问题，本章处理两类更重的数据：音视频与网页渲染。这两个能力正是 JD 加分项中「音视频/图形处理」与「无头浏览器渲染管线」的落地——它们让 Agent 能看视频、听录音、截网页，是多模态 Agent 的基础设施。

23.1 FFmpeg：音视频处理的瑞士军刀

FFmpeg 是命令行工具，Python 中通过 subprocess 调用。先封装一个安全的执行器：

# agent_prod/media/ffmpeg_tool.py
import subprocess
import json
from pathlib import Path

class FFmpegError(Exception):
    pass

def run_ffmpeg(args: list[str], timeout: int = 300) -> str:
    """统一 FFmpeg 执行入口：超时控制 + 错误捕获"""
    cmd = ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", *args]
    try:
        result = subprocess.run(
            cmd, capture_output=True, text=True, timeout=timeout
        )
    except subprocess.TimeoutExpired:
        raise FFmpegError(f"FFmpeg 超时（>{timeout}s）: {' '.join(args[:4])}...")
    if result.returncode != 0:
        raise FFmpegError(result.stderr[-500:])
    return result.stderr

def probe(path: str) -> dict:
    """ffprobe 读取媒体元信息：时长/分辨率/码率/流信息"""
    result = subprocess.run(
        ["ffprobe", "-v", "quiet", "-print_format", "json",
         "-show_format", "-show_streams", path],
        capture_output=True, text=True,
    )
    return json.loads(result.stdout)

def get_duration(path: str) -> float:
    return float(probe(path)["format"]["duration"])

23.1.1 四个高频操作

def extract_audio(video: str, out: str = "audio.wav") -> str:
    """提取音轨：16kHz 单声道 mono 是 Whisper 的最佳输入格式"""
    run_ffmpeg(["-i", video, "-vn", "-ac", "1", "-ar", "16000", out])
    return out

def extract_frames(video: str, fps: float = 0.5, out_dir: str = "frames/") -> list[str]:
    """按帧率抽帧：fps=0.5 即每 2 秒一帧"""
    Path(out_dir).mkdir(exist_ok=True)
    run_ffmpeg(["-i", video, "-vf", f"fps={fps}", "-q:v", "2", f"{out_dir}/f_%04d.jpg"])
    return sorted(str(p) for p in Path(out_dir).glob("*.jpg"))

def cut_segment(video: str, start: float, end: float, out: str) -> str:
    """精准切片：-ss 放在 -i 前走关键帧快速定位，-accurate_seek 保证精度"""
    run_ffmpeg(["-ss", str(start), "-to", str(end), "-i", video,
                "-c", "copy" if end - start > 30 else "libx264", out])
    return out

def make_thumbnail(video: str, at: float = 1.0, out: str = "thumb.jpg") -> str:
    """截取指定时间点的封面图"""
    run_ffmpeg(["-ss", str(at), "-i", video, "-frames:v", "1", out])
    return out

💡 -c copy 的取舍：直接拷贝流不重编码，速度极快但只能切在关键帧上（可能有几秒偏差）；切片要求精准时必须重编码（libx264）。长视频粗切用 copy，短切片用重编码——这也是一种成本路由。

23.2 OpenCV：场景切分与关键帧提取

固定帧率抽帧会产生大量冗余画面。场景切分只在画面内容显著变化时取帧，是视频理解的正确入口。

pip install opencv-python-headless numpy

# agent_prod/media/scene_detect.py
import cv2
import numpy as np

def detect_scenes(video_path: str, threshold: float = 30.0) -> list[dict]:
    """基于帧差分的场景切分：相邻帧直方图差异超阈值即认为切镜头"""
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    scenes, prev_hist = [], None
    frame_idx = 0
    while True:
        ok, frame = cap.read()
        if not ok:
            break
        # 缩小到 1/4 再算直方图，提速 10 倍以上
        small = cv2.resize(frame, (frame.shape[1] // 4, frame.shape[0] // 4))
        hist = cv2.calcHist([small], [0, 1, 2], None, [8, 8, 8], [0, 256] * 3)
        cv2.normalize(hist, hist)
        if prev_hist is not None:
            diff = cv2.compareHist(prev_hist, hist, cv2.HISTCMP_CHISQR)
            if diff > threshold:
                scenes.append({"time": frame_idx / fps, "frame": frame_idx})
        prev_hist = hist
        frame_idx += 1
    cap.release()
    return scenes

def extract_keyframes(video_path: str, max_frames: int = 20) -> list[str]:
    """每个场景取一帧代表图，输出关键帧路径列表"""
    import subprocess
    from pathlib import Path
    scenes = detect_scenes(video_path)
    out_dir = Path("keyframes"); out_dir.mkdir(exist_ok=True)
    # 场景过多时均匀采样，控制送入多模态模型的图片数量
    step = max(1, len(scenes) // max_frames)
    picked = scenes[::step][:max_frames]
    paths = []
    for i, s in enumerate(picked):
        out = str(out_dir / f"key_{i:03d}_{s['time']:.1f}s.jpg")
        subprocess.run(["ffmpeg", "-y", "-loglevel", "error",
                        "-ss", str(s["time"]), "-i", video_path,
                        "-frames:v", "1", out], check=True)
        paths.append(out)
    return paths

💡 为什么限制 max_frames？ 把关键帧喂给多模态 LLM 时，图片数量直接决定 token 成本。一个 60 分钟视频可能有上百个镜头，均匀采样到 20 帧以内是成本与召回的平衡点（呼应 ch11 的 Token 预算思想）。

23.3 音频特征与智能切片

转录前先做 VAD（语音活动检测），把静音段剔除，能显著提升 Whisper 精度并减少算力：

pip install librosa soundfile

# agent_prod/media/audio_features.py
import librosa
import numpy as np

def detect_speech_segments(path: str, top_db: int = 30) -> list[dict]:
    """基于能量检测语音段：低于最大能量 top_db 的部分视为静音"""
    y, sr = librosa.load(path, sr=16000)
    intervals = librosa.effects.split(y, top_db=top_db)
    return [
        {"start": a / sr, "end": b / sr, "duration": (b - a) / sr}
        for a, b in intervals
    ]

def smart_chunk_audio(path: str, target_len: float = 30.0) -> list[tuple[float, float]]:
    """把音频按语音段聚合成 ~30s 的块，且尽量在静音处切分
    返回 [(start, end), ...]，供 FFmpeg 切片或 Whisper 分段转录
    """
    segments = detect_speech_segments(path)
    chunks, cur_start, cur_end = [], None, None
    for seg in segments:
        if cur_start is None:
            cur_start, cur_end = seg["start"], seg["end"]
        elif seg["end"] - cur_start <= target_len:
            cur_end = seg["end"]          # 合并进当前块
        else:
            chunks.append((cur_start, cur_end))
            cur_start, cur_end = seg["start"], seg["end"]
    if cur_start is not None:
        chunks.append((cur_start, cur_end))
    return chunks

def audio_energy_profile(path: str, frame_sec: float = 1.0) -> np.ndarray:
    """逐秒 RMS 能量曲线：可用于检测「高潮片段」「异常响动」"""
    y, sr = librosa.load(path, sr=16000)
    hop = int(sr * frame_sec)
    rms = librosa.feature.rms(y=y, hop_length=hop)[0]
    return rms / (rms.max() + 1e-9)  # 归一化到 0-1

与上一章的 Whisper 转录串联成完整管线：

# agent_prod/media/video_pipeline.py
from agent_prod.media.ffmpeg_tool import extract_audio, extract_keyframes
from agent_prod.media.audio_features import smart_chunk_audio
from agent_prod.parsers.transcribe import transcribe_to_srt

def process_video(video_path: str) -> dict:
    """视频 → 音轨 → 智能切片 → 转录；同时输出关键帧供多模态分析"""
    audio = extract_audio(video_path, out=f"{video_path}.wav")
    chunks = smart_chunk_audio(audio, target_len=30.0)

    transcript_parts = []
    for start, end in chunks:
        # 逐块转录并保留时间偏移，最终时间轴与原视频对齐
        seg_audio = f"{video_path}.{start:.1f}-{end:.1f}.wav"
        from agent_prod.media.ffmpeg_tool import cut_segment
        cut_segment(audio, start, end, seg_audio)
        result = transcribe_to_srt(seg_audio)
        for s in result["segments"]:
            s["start"] += start
            s["end"] += start
        transcript_parts.extend(result["segments"])

    return {
        "transcript": transcript_parts,
        "keyframes": extract_keyframes(video_path, max_frames=20),
    }

23.4 Playwright 无头渲染管线

有些内容 JS 动态渲染、需要登录态，requests 拿不到。用 Playwright 无头浏览器截图或生成 PDF，作为 Agent 的「眼睛」：

# agent_prod/media/renderer.py
from playwright.sync_api import sync_playwright

def render_page(url: str, *, full_page: bool = True, pdf: bool = False,
                wait_selector: str | None = None, timeout: int = 30_000) -> dict:
    """无头渲染网页：输出截图与可选 PDF，等待关键元素出现后再截"""
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        page = browser.new_page(viewport={"width": 1280, "height": 800})
        page.goto(url, timeout=timeout, wait_until="networkidle")
        if wait_selector:
            page.wait_for_selector(wait_selector, timeout=timeout)
        shot = f"/tmp/render_{abs(hash(url)) % 10**8}.png"
        page.screenshot(path=shot, full_page=full_page)
        result = {"screenshot": shot, "title": page.title()}
        if pdf:
            pdf_path = shot.replace(".png", ".pdf")
            page.pdf(path=pdf_path, format="A4")
            result["pdf"] = pdf_path
        browser.close()
        return result

结合上一章的 OCR，形成「渲染 → 截图 → 文字提取」闭环：

def render_and_extract_text(url: str) -> str:
    """动态页面文字提取：JS 渲染后再抽正文，比纯 HTTP 抓取覆盖面广"""
    result = render_page(url, full_page=True)
    from agent_prod.parsers.ocr import ocr_image
    with open(result["screenshot"], "rb") as f:
        return ocr_image(f.read())

⚠️ 渲染管线三坑：① networkidle 在长轮询页面会永远等不到，务必设 timeout 兜底；② 无头浏览器内存占用高（每实例 200MB+），并发要限流；③ 只渲染可信 URL——恶意页面可能利用浏览器漏洞，生产环境必须跑在沙箱容器里（见 23.5）。

23.5 封装为 Agent 工具链：安全隔离与资源配额

媒体处理是 Agent 工具中最危险的一类：处理用户上传文件、调用外部资源、消耗大量 CPU/磁盘。封装时必须加三层防护：

# agent_prod/media/tools.py
from pathlib import Path
import re
import shutil
from agent_prod.media.video_pipeline import process_video
from agent_prod.media.renderer import render_page

# ── 防护一：路径白名单，防止路径穿越 ──
MEDIA_ROOT = Path("/data/media").resolve()

def safe_path(name: str) -> Path:
    p = (MEDIA_ROOT / name).resolve()
    if not str(p).startswith(str(MEDIA_ROOT)):
        raise ValueError(f"非法路径: {name}")
    return p

# ── 防护二：文件类型与大小白名单 ──
ALLOWED_EXTS = {".mp4", ".mov", ".mp3", ".wav", ".m4a", ".png", ".jpg", ".pdf"}
MAX_SIZE = 500 * 1024 * 1024  # 500MB

def validate_upload(name: str, size: int) -> None:
    if Path(name).suffix.lower() not in ALLOWED_EXTS:
        raise ValueError(f"不支持的类型: {name}")
    if size > MAX_SIZE:
        raise ValueError(f"文件过大: {size} bytes")

# ── 防护三：URL 协议白名单，防 SSRF ──
_URL_RE = re.compile(r"^https://[\w.-]+\.[a-z]{2,}", re.I)

def render_url_tool(url: str) -> dict:
    """Agent 工具：渲染网页截图（仅允许 https 外链）"""
    if not _URL_RE.match(url):
        raise ValueError(f"URL 不合规: {url}")
    return render_page(url, full_page=True)

def analyze_video_tool(file_name: str) -> dict:
    """Agent 工具：视频转录 + 关键帧提取"""
    path = safe_path(file_name)
    if not path.exists():
        raise FileNotFoundError(file_name)
    result = process_video(str(path))
    # 用完即删临时音轨，防止磁盘膨胀
    for tmp in MEDIA_ROOT.glob(f"{file_name}*.wav"):
        tmp.unlink(missing_ok=True)
    return result

# 磁盘配额巡检：超过 80% 触发清理（可挂到 CloudWatch 告警，见 ch18）
def check_disk_quota(max_gb: float = 50) -> bool:
    total = sum(f.stat().st_size for f in MEDIA_ROOT.rglob("*") if f.is_file())
    if total > max_gb * 1024**3 * 0.8:
        shutil.rmtree(MEDIA_ROOT / "tmp", ignore_errors=True)
        return False
    return True

接入 LangGraph 的工具注册（与 ch05 的 ToolNode 模式一致）：

from langchain_core.tools import tool

@tool
def analyze_video(file_name: str) -> str:
    """转录视频并提取关键帧。输入：媒体库中的文件名"""
    result = analyze_video_tool(file_name)
    text = " ".join(s["text"] for s in result["transcript"])
    return f"转录（{len(result['transcript'])} 段）：{text[:2000]}...\n关键帧：{len(result['keyframes'])} 张"

@tool
def render_webpage(url: str) -> str:
    """渲染网页并截图。输入：https:// 开头的 URL"""
    result = render_url_tool(url)
    return f"已截图：{result['screenshot']}（标题：{result['title']}）"

tools = [analyze_video, render_webpage]
# 之后 bind 到 LangGraph agent：llm.bind_tools(tools)

23.6 生产化要点

1. 算力隔离：FFmpeg/OpenCV/Playwright 全部是重资源任务，必须跑在独立 Celery worker 队列（如 media 队列，见 ch02），与 API 进程物理隔离；K8s/ECS 场景给 media worker 单独的实例类型。
2. 超时与重试：每个 FFmpeg 命令设 timeout（本节 run_ffmpeg 已内置），配合 ch01 的指数退避重试；转录失败降级到更小的 Whisper 模型。
3. 临时文件治理：媒体处理产生大量中间文件（音轨、帧图、切片），统一放 /data/media/tmp 并用定时任务 + check_disk_quota 双保险清理。
4. 成本核算：给每个媒体任务打上 request_id，把处理时长、文件大小上报到 CloudWatch（ch18），按用户/租户聚合出媒体处理的单位成本。

本章小结

• FFmpeg：抽音轨（16kHz mono 喂 Whisper）、抽帧、切片（copy 快但粗、重编码准但慢）、封面
• OpenCV：帧直方图差分做场景切分，每场景取关键帧并限制数量控制多模态成本
• 音频特征：能量 VAD 剔除静音，按 ~30s 语音段智能切块，提升转录精度与吞吐
• Playwright：无头渲染动态页面，截图/PDF 双输出，与 OCR 串联提取 JS 渲染后的文字
• 工具化三防护：路径白名单防穿越、类型大小白名单、URL 协议白名单防 SSRF
• 生产四件套：独立 worker 队列、超时重试降级、临时文件治理、成本上报

🛠️ 动手实践

1. 会议视频知识入库：把 process_video 的输出（转录文本 + 关键帧描述）接入第 22 章的 build_ingest_graph，实现「上传会议录像 → 自动进知识库」的完整链路。
2. 媒体处理压测：用 Locust（复用压测课程 ch13 协议扩展思路）对媒体上传接口压测，验证 Celery media 队列在 10 并发下的积压与 Worker 扩容策略。
3. 截图 diff 监控：用 Playwright 每小时渲染同一个仪表盘页面，对截图做像素 diff（OpenCV absdiff），变化超阈值时发 CloudWatch 告警——一个零成本的 UI 巡检机器人。

---

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. FFmpeg 切片时使用 -c copy 与重编码（libx264）的核心取舍是什么？

2. 视频做场景切分后，为什么还要限制送入多模态模型的关键帧数量？

3. 在 Whisper 转录前先做 VAD（语音活动检测）的主要收益是什么？

4. 封装 Playwright 渲染工具时，对 URL 做协议白名单（仅 https）主要是为了防御什么？