视频问答生成流水线

1303 字约 4 分钟

2025-07-16

1. 概述

视频问答生成流水线从视频内容自动生成高质量的问答对，通过先生成视频描述，再基于描述生成问答，适用于视频问答数据集构建、视频理解评测和多模态训练数据生成。

我们支持以下应用场景：

视频问答数据集自动构建
视频理解能力评测数据生成
多模态对话训练数据合成
视频内容理解与问答

流水线的主要流程包括：

视频描述生成：利用 VLM 模型分析视频内容并生成详细描述。
问答对生成：基于视频描述（可选结合视频）生成问题和答案。

2. 快速开始

第一步：创建新的 DataFlow 工作文件夹

mkdir run_dataflow_mm
cd run_dataflow_mm

第二步：初始化 DataFlow-MM

dataflowmm init

这时你会看到：

run_dataflow_mm/playground/video_qa_pipeline.py

第三步：配置模型路径

在 video_qa_pipeline.py 中配置 VLM 模型路径：

self.vlm_serving = LocalModelVLMServing_vllm(
    hf_model_name_or_path="Qwen/Qwen2.5-VL-7B-Instruct",  # 修改为你的模型路径
    hf_cache_dir="./dataflow_cache",
    vllm_tensor_parallel_size=1,
    vllm_temperature=0.7,
    vllm_top_p=0.9,
    vllm_max_tokens=2048,
    vllm_max_model_len=51200,
    vllm_gpu_memory_utilization=0.9
)

第四步：一键运行

python playground/video_qa_pipeline.py

API 版本

如果你希望使用 API 服务而非本地模型，可以使用 API 版本的流水线：

python api_pipelines/video_qa_api_pipeline.py

API 版本的使用方式与本地版本类似，只需配置 API Key 和服务地址即可。详情参见 api_pipelines/video_qa_api_pipeline.py 中的配置说明。

此外，你可以根据自己的需求调整生成策略（是否在生成 QA 时使用视频输入）。接下来，我们会详细介绍流水线中的各个步骤和参数配置。

3. 数据流与流水线逻辑

1. 输入数据

该流程的输入数据包括以下字段：

video：视频文件路径列表，如 ["path/to/video.mp4"]
conversation：对话格式数据，如 [{"from": "human", "value": ""}]
image（可选）：图像文件路径列表，用于同时处理图像

这些输入数据可以存储在指定的文件中（如 json 或 jsonl），并通过 FileStorage 对象进行管理和读取：

self.storage = FileStorage(
    first_entry_file_name="./dataflow/example/video_caption/sample_data.json",
    cache_path="./cache",
    file_name_prefix="video_vqa",
    cache_type="json",
)

输入数据示例：

[
    {
        "video": ["./videos/sample1.mp4"],
        "conversation": [{"from": "human", "value": ""}]
    },
    {
        "video": ["./videos/sample2.mp4"],
        "conversation": [{"from": "human", "value": ""}]
    }
]

2. 视频描述生成（PromptedVQAGenerator）

流程的第一步是使用通用 VQA 生成器（PromptedVQAGenerator）结合 VideoCaptionGeneratorPrompt 为视频生成详细的文本描述。

功能：

利用 VLM 模型分析视频内容并生成描述文本
为后续的问答生成提供内容基础
使用提示模板（Prompt Template）配置生成内容的格式和风格

输入：视频文件路径和对话格式数据
输出：生成的视频描述文本（caption 字段）

算子初始化：

from dataflow.prompts.video import VideoCaptionGeneratorPrompt

self.prompt_template = VideoCaptionGeneratorPrompt()

self.prompted_vqa_generator = PromptedVQAGenerator(
    serving=self.vlm_serving,
    system_prompt="You are a helpful assistant.",
    prompt_template=self.prompt_template
)

算子运行：

self.prompted_vqa_generator.run(
    storage=self.storage.step(),
    input_image_key="image",              # 输入图像字段（可选）
    input_video_key="video",              # 输入视频字段
    input_conversation_key="conversation", # 输入对话字段
    output_answer_key="caption",          # 输出描述字段
)

3. 问答对生成（VideoCaptionToQAGenerator）

流程的第二步是使用问答生成器（VideoCaptionToQAGenerator）基于视频描述生成问答对。

功能：

基于视频描述生成相关的问题和答案
支持两种模式：
- 使用视频输入：同时基于描述和视频内容生成 QA（更准确）
- 仅使用描述：仅基于文本描述生成 QA（更快速）

输入：视频描述、视频文件（可选）、对话数据
输出：生成的问答对（qa 字段）

算子初始化：

self.videocaption_to_qa_generator = VideoCaptionToQAGenerator(
    vlm_serving=self.vlm_serving,
    use_video_input=True,  # 控制是否使用视频输入
)

参数说明：

use_video_input=True: 同时使用 caption 和视频生成问题
use_video_input=False: 仅使用 caption 生成问题

算子运行：

self.videocaption_to_qa_generator.run(
    storage=self.storage.step(),
    input_image_key="image",              # 输入图像字段（可选）
    input_video_key="video",              # 输入视频字段
    input_conversation_key="conversation", # 输入对话字段
    output_key="qa",                      # 输出问答字段
)

4. 输出数据

最终，流水线生成的输出数据将包含以下内容：

video：原始视频路径
conversation：更新后的对话数据
caption：生成的视频描述文本
qa：生成的问答对（包含问题和答案）

输出数据示例：

{
    "video": ["./videos/sample1.mp4"],
    "conversation": [{"from": "human", "value": "Please describe the video in detail."}],
    "caption": "这段视频展示了一个人在公园中散步的场景。天气晴朗，阳光明媚，背景中可以看到绿树和长椅。",
    "qa": {
        "question": "视频中的人在做什么？",
        "answer": "视频中的人正在公园中散步，享受晴朗的天气。"
    }
}

4. 流水线示例

以下给出示例流水线，展示如何使用 VideoVQAGenerator 进行视频问答生成。

from dataflow.operators.core_vision import PromptedVQAGenerator, VideoCaptionToQAGenerator
from dataflow.serving import LocalModelVLMServing_vllm
from dataflow.utils.storage import FileStorage
from dataflow.prompts.video import VideoCaptionGeneratorPrompt

class VideoVQAGenerator():
    def __init__(self):
        """
        Initialize VideoVQAGenerator with default parameters.
        """
        self.storage = FileStorage(
            first_entry_file_name="./dataflow/example/video_caption/sample_data.json",
            cache_path="./cache",
            file_name_prefix="video_vqa",
            cache_type="json",
        )

        self.vlm_serving = LocalModelVLMServing_vllm(
            hf_model_name_or_path="Qwen/Qwen2.5-VL-7B-Instruct",
            hf_cache_dir="./dataflow_cache",
            vllm_tensor_parallel_size=1,
            vllm_temperature=0.7,
            vllm_top_p=0.9, 
            vllm_max_tokens=2048,
            vllm_max_model_len=51200,  
            vllm_gpu_memory_utilization=0.9
        )

        self.prompt_template = VideoCaptionGeneratorPrompt()
        
        self.prompted_vqa_generator = PromptedVQAGenerator(
            serving=self.vlm_serving,
            system_prompt="You are a helpful assistant.",
            prompt_template=self.prompt_template
        )
        
        self.videocaption_to_qa_generator = VideoCaptionToQAGenerator(
            vlm_serving=self.vlm_serving,
            use_video_input=True,  # 控制是否使用视频输入
        )

    def forward(self):
        # Step 1: Generate video captions using PromptedVQAGenerator
        self.prompted_vqa_generator.run(
            storage=self.storage.step(),
            input_image_key="image",
            input_video_key="video",
            input_conversation_key="conversation",
            output_answer_key="caption",
        )
        
        # Step 2: Generate QA from captions
        self.videocaption_to_qa_generator.run(
            storage = self.storage.step(),
            input_image_key="image",
            input_video_key="video",
            input_conversation_key="conversation",
            output_key="qa",
        )

if __name__ == "__main__":
    model = VideoVQAGenerator()
    model.forward()

图像生成

图像编辑