长视频理解 Pipeline：Qwen-VL 不支持音频 + 4 小时硬限制的工程化解法

接到一个需求：处理 666 条 2 小时以上的课程视频，做成 AI 教学知识库。

看似简单——调 Qwen-VL 接口就完事了。但真做起来发现：Qwen-VL 单视频 ≤ 2h、文件 ≤ 2GB，而且不支持音频。

这篇文章复盘我们怎么用工程手段补齐模型短板，做出一条”长视频理解 Pipeline”。

一、问题的硬约束

接需求时没仔细看接口文档，等真做起来才发现一堆坑：

限制项	阈值	我们的视频	影响
单视频时长	≤ 2 小时	3-6 小时	超限
单视频文件大小（公网 URL）	≤ 2 GB	3-10 GB	超限
单视频文件大小（本地路径）	≤ 100 MB	3-10 GB	严重超限
单视频文件大小（Base64）	≤ 10 MB	3-10 GB	不可用
音频理解	不支持	含讲师讲解	致命
单次请求视频数	≤ 64 个	1 个就够	不影响

核心矛盾：课程视频常常 3-6 小时、3-10 GB，且含大量语音讲解，仅靠模型原生接口无法直接处理。

二、方案设计原则

我们的核心思路：

不绕开模型，用工程补齐短板 — 模型做它擅长的事（视觉理解），工程做它擅长的事（切片/转写/聚合）
视觉 + 听觉双通道 — 弥补 VL 不支持音频的缺陷
MapReduce 处理范式 — 任意长度视频可水平扩展
异步任务化 — 长耗时操作不阻塞前端

三、整体架构

┌─────────────┐
│ 用户上传视频 │
└──────┬──────┘
       │
       ▼
┌─────────────────────┐
│ ① 上传层（OSS）     │  ── 解决 2GB / 100MB 限制
│ 获得公网 HTTPS URL  │
└──────┬──────────────┘
       │
       ▼
┌─────────────────────┐
│ ② 预处理（ffmpeg）  │  ── 解决 2h 限制
│ - 元信息探测        │
│ - 智能切片          │
│ - 音频分离          │
└──────┬──────────────┘
       │
       ├─────────────────┐
       ▼                 ▼
┌──────────────┐  ┌──────────────────┐
│ ③A 视频切片  │  │ ③B 音频流        │
│ chunk_1..N   │  │ audio.wav        │
└──────┬───────┘  └──────┬───────────┘
       │                  │
       ▼                  ▼
┌──────────────┐  ┌──────────────────┐
│ ④ 任务队列   │  │ ASR 服务         │
│ (RabbitMQ)   │  │ (FunASR)         │
└──────┬───────┘  └──────┬───────────┘
       │                  │
       ▼                  ▼
┌──────────────┐  ┌──────────────────┐
│ Qwen-VL 集群 │  │ 语音文本         │
│ 并发理解     │  │ + 时间戳         │
└──────┬───────┘  └──────┬───────────┘
       │                  │
       └────────┬─────────┘
                ▼
       ┌────────────────────┐
       │ ⑤ 聚合层（LLM）    │
       │ 视觉摘要 + 语音    │
       │ → 完整课程笔记    │
       └────────┬───────────┘
                ▼
       ┌────────────────────┐
       │ ⑥ 结构化输出       │
       │ - 课程讲义         │
       │ - 知识点切片       │
       │ - 问答对           │
       └────────────────────┘

四、关键工程实现

4.1 解决 2GB / 100MB 限制：OSS 中转

禁止用本地路径或 Base64（限额太小）。强制走公网 URL 通道。

import oss2
from dataclasses import dataclass

@dataclass
class UploadResult:
    url: str          # 签名 URL，Qwen-VL 直接消费
    object_key: str   # OSS 内部 key
    expire_at: int    # 过期时间戳


class VideoUploader:
    def __init__(self, access_key, secret_key, endpoint, bucket_name):
        auth = oss2.Auth(access_key, secret_key)
        self.bucket = oss2.Bucket(auth, endpoint, bucket_name)

    def upload(self, local_path: str, ttl: int = 3600) -> UploadResult:
        object_key = f"videos/{self._hash(local_path)}{Path(local_path).suffix}"
        # 简单上传（适合 <5GB）
        self.bucket.put_object_from_file(object_key, local_path)
        # 签名 URL（Qwen-VL 用）
        url = self.bucket.sign_url('GET', object_key, ttl)
        return UploadResult(
            url=url,
            object_key=object_key,
            expire_at=int(time.time()) + ttl,
        )

关键细节：签名 URL 的 TTL 要大于 Qwen-VL 处理时间。我们设 3600 秒（1 hour），足够。

4.2 解决 2h 限制：智能切片

切片不是简单按时间切——要找语义断点才切，否则会切断讲解。

我们设计了三级切片策略：

class SmartSplitter:
    MAX_DURATION = 6840  # 2 小时硬上限，留 5% buffer = 6840 秒

    def split(self, video_path: str) -> List[VideoChunk]:
        meta = self.get_metadata(video_path)

        # 1️⃣ 优先：按 MP4 章节切
        if meta.get('chapters'):
            return self.split_by_chapters(video_path)

        # 2️⃣ 次选：按静音段切
        if self._has_dense_silence(video_path):
            return self.split_by_silence(video_path)

        # 3️⃣ 兜底：按固定时长切
        return self.split_by_duration(video_path, chunk_seconds=6300)

切片策略对比

策略	切点位置	优点	缺点	适用
章节优先	MP4 内嵌章节	语义完整	依赖源文件	标准课程
静音检测	静音段	不切断讲解	切点稀疏不均	讲解型
固定时长	每 105 分钟	简单稳定	可能切断讲解	监控录像

ffmpeg 切片命令

# -c copy 避免重新编码，秒级完成（vs 重新编码 30 分钟）
ffmpeg -y -ss 0 -to 6300 \
  -i input.mp4 \
  -c copy \
  -avoid_negative_ts make_zero \
  output/chunk_000.mp4

-c copy 是性能关键：4 hour 视频切片 < 30 秒，vs 重新编码 30 分钟，提速 60 倍。

4.3 解决音频丢失：ASR 独立通道

这是最关键的发现：Qwen-VL 只看画面，讲师的口头讲解、黑板书写时说的话全部丢失。

我们必须独立处理音频：

class ASRService:
    def __init__(self):
        self.model = AutoModel(
            model="paraformer-zh",       # 中文 SOTA
            vad_model="fa-zh",           # 语音活动检测
            punc_model="ct-punc",        # 加标点
            spk_model="cam++",           # 说话人分离（多人课堂）
        )

    def transcribe(self, video_path: str) -> List[dict]:
        """返回带时间戳的转写结果"""
        audio_path = self.extract_audio(video_path)
        result = self.model.generate(
            audio_path,
            batch_size_s=300,    # 5 分钟一批
            return_timestamp=True,
        )
        return self._format_result(result)

    def extract_audio(self, video_path: str) -> str:
        """从视频中提取 16kHz 单声道 wav"""
        audio_path = str(Path(video_path).with_suffix('.wav'))
        subprocess.run([
            'ffmpeg', '-y', '-i', video_path,
            '-vn', '-acodec', 'pcm_s16le',
            '-ac', '1', '-ar', '16000',
            audio_path,
        ], check=True)
        return audio_path

为什么选 FunASR 而不是 Whisper

指标	FunASR (paraformer-zh)	Whisper Large-v3
中文 WER	3-5%	8-12%
说话人分离	✅ cam++	❌ 无
4h 音频成本	¥0.5（GPU 推理）	¥0.7（GPU 推理）
数据合规	✅ 私有化	✅ 私有化

核心优势：中文 WER 低 2-3 倍 + 说话人分离（多人课堂必需）。

4.4 MapReduce 聚合处理

class VideoUnderstandingPipeline:
    def __init__(self, qwen_client, asr_service, llm_client, max_concurrent=5):
        self.qwen = qwen_client
        self.asr = asr_service
        self.llm = llm_client
        self.semaphore = asyncio.Semaphore(max_concurrent)

    async def process(self, video_path: str, prompt: str) -> dict:
        # 1. 上传 OSS
        upload = self.uploader.upload(video_path)

        # 2. 切片
        chunks = self.splitter.smart_split(video_path)

        # 3. 并发处理（视觉 + 听觉并行）
        visual_tasks = [self._process_visual(c, prompt) for c in chunks]
        asr_result = self.asr.transcribe(video_path)  # 一次性转写

        visual_results = await asyncio.gather(*visual_tasks)

        # 4. 聚合（视觉摘要 + 语音文本 → 完整课程笔记）
        final = await self._aggregate(visual_results, asr_result, prompt)
        return final

    async def _aggregate(self, visual_results, asr_segments, prompt):
        """LLM 二次聚合：把分段摘要合并为完整理解"""
        sorted_visual = sorted(visual_results, key=lambda r: r.chunk_index)

        context_parts = []
        for vr in sorted_visual:
            ts_start = self._format_ts(vr.start_time)
            ts_end = self._format_ts(vr.end_time)
            context_parts.append(f"""
## 时间段 {ts_start} - {ts_end}

### 画面内容
{vr.visual_summary}

### 讲师讲解（ASR 转写）
{self._get_asr_for_range(asr_segments, vr.start_time, vr.end_time)}
""")

        context = "\n".join(context_parts)

        aggregate_prompt = f"""以下是按时间顺序排列的课程分段理解结果。
请合并为一份完整的、结构化的课程讲义，包含：
1. 课程主题与目标
2. 核心知识点（按时间顺序）
3. 关键概念定义
4. 课堂示例与演示
5. 课后要点总结

{prompt}

分段内容：
{context}
"""
        return await self.llm.completion(aggregate_prompt)

关键点：聚合时按时间戳对齐视觉和听觉——讲师讲解的图片刚好对应讲解内容。

4.5 异步任务化

长视频处理耗时 5-30 分钟，必须异步化。

from celery import Celery

app = Celery('video_tasks', broker='redis://localhost:6379/0')

@app.task(bind=True, max_retries=3)
def process_video_task(self, video_id: str, prompt: str):
    try:
        pipeline = VideoUnderstandingPipeline(...)
        result = asyncio.run(pipeline.process(...))
        VideoResult.objects.filter(id=video_id).update(
            status='completed', result=result,
        )
        return result
    except Exception as exc:
        self.retry(exc=exc, countdown=60)

前端通过 SSE / 轮询查询进度。

五、成本与性能估算

以 4 小时 / 5 GB / 1080p 课程视频 为基准：

阶段	耗时	成本	备注
OSS 上传	2-5 min	¥0.5 (存储)	一次性
ffmpeg 切片	< 30 s	¥0	-c copy 几乎免费
Qwen-VL 调用 × 3 段	60-90 s	¥3-5	并发
ASR 转写	90-120 s	¥2-3	4h 音频
LLM 聚合	10-20 s	¥0.5	上下文较长
总计	~5 min	¥6-9	端到端

优化后（开启缓存、批处理、模型选择优化）：可降至 ¥4-6 / 视频。

六、踩过的坑（10 个真实教训）

坑 1：不要传 Base64 给 Qwen-VL

最初图省事，直接把视频读成 Base64 传。结果返回错误：

Error: Base64 payload exceeds 10MB limit

教训：永远走 OSS 中转，不要尝试其他方式。

坑 2：切片后丢失跨段上下文

第一版只切视频，每个 chunk 独立给 Qwen-VL 处理。结果：

章节”第一章入门”在 chunk_001 末尾讲，”第二章进阶”在 chunk_002 开头
切在中间，Qwen-VL 看不到完整章节

解法：聚合阶段 LLM 看完整 chunks 列表，自动判断章节边界。

坑 3：ASR 时间戳与视频帧不对齐

最初直接用 ffmpeg 提取音频的 duration，与视频 chunks 时间戳对齐。结果发现：

视频是 25 fps，音频是 16kHz，时间戳粒度不同
ASR 输出的 timestamp 精度是 100ms，与视频的 40ms 不对齐

解法：在 ASR 输出时按帧对齐，统一以 ms 为单位。

坑 4：Qwen-VL 限流

高峰期（每天处理 100 条视频）触发 Qwen-VL 限流：

Error: rate limit exceeded (60 req/min)

解法：加任务队列背压 + 指数退避重试。

坑 5：FFmpeg 中文路径乱码

最初用 Python 调 ffmpeg 处理中文路径视频：

subprocess.run(['ffmpeg', '-i', '/data/课程视频/语文.mp4', ...])
# ffmpeg 报：No such file or directory

解法：用 subprocess 时把 cwd 切到视频所在目录，或用 os.fsencode。

坑 6：切片时 metadata 丢失

-c copy 切片后，部分视频的 metadata（如字幕轨）丢失。

解法：用 -map 0 复制所有流：

ffmpeg -i input.mp4 -map 0 -c copy -segment_time 6300 chunk_%03d.mp4

坑 7：ASR 说话人标签跳变

FunASR 的 cam++ 说话人分离偶尔会把同一个人识别成不同 speaker。

解法：聚类合并相似 embedding 的 speaker 标签。

坑 8：聚合 LLM 上下文超限

666 条视频聚合到 LLM 时，每条 3 chunks × 500 tokens = 1500 tokens。聚合时还要把所有 chunks 拼起来，远超 32K 上下文。

解法：分层聚合（每个 chunk 单独聚合 → 中间结果再聚合）。

坑 9：视频损坏无法切片

~2% 的视频 ffprobe 失败（编码异常、文件截断）。

解法：标记为”损坏”，转码后重试。

坑 10：OSS 签名 URL 过期

我们最初设签名 URL TTL 1 小时。结果某次 Qwen-VL 处理时间超 1 小时（5 段聚合），最后一段 URL 过期，返回 403。

解法：动态续签 URL，或用 STS Token。

七、兜底方案：抽帧 + 图像理解

当视频无法切片（如加密流、直播回放），或切片成本过高时使用：

def extract_keyframes(video_path: str, interval_seconds=30, max_frames=8000) -> List[dict]:
    """每 30 秒抽 1 帧 + 场景变化帧"""
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
    duration = total_frames / fps

    frames = []
    last_frame = None
    for sec in range(0, int(duration), int(interval_seconds)):
        cap.set(cv2.CAP_PROP_POS_MSEC, sec * 1000)
        ret, frame = cap.read()
        if not ret:
            continue
        if last_frame is not None:
            diff = cv2.absdiff(frame, last_frame).mean()
            if diff < 5.0:  # 几乎无变化，跳过
                continue
        frames.append({
            "timestamp": sec,
            "image_b64": base64.b64encode(cv2.imencode('.jpg', frame)[1]).decode(),
        })
        last_frame = frame
        if len(frames) >= max_frames:
            break
    return frames

然后用 Qwen3-VL 图像理解接口一次性传入（支持 8000 张图）。

八、给类似场景的建议

如果你也要做长视频 / 长音频理解：

永远走 OSS 中转——不要尝试 Base64 / 本地路径
智能切片优于固定切片——找语义断点（章节 > 静音 > 兜底）
-c copy 是性能关键——提速 30-60 倍
视觉 + 听觉双通道——VL 不支持音频，必须独立 ASR
MapReduce 聚合——分层聚合避免上下文超限
异步任务化——5-30 分钟处理，必须异步 + 进度查询
限流 + 重试 + 降级——高峰期必备

九、总结

长视频理解不是”调一个 API”那么简单——是工程问题，不是 AI 问题。

核心原则：

MapReduce 处理范式——任意长度可扩展
双通道融合——视觉 + 听觉缺一不可
智能切片策略——找语义断点，不要机械切
异步任务化——长耗时必须后台跑
限流 + 重试 + 降级——稳定性保障

最后一句话：模型是发动机，工程是变速箱。两者配合才能跑得远、跑得稳。

十、参考资料

作者：魏远标，贝斯平 AI 架构师。技术博客：javai.tech

你做过最长的视频 AI 处理是多久？留言聊聊～