案例7. PDF中的VQA提取流水线

1432 字约 5 分钟

2025-11-17

1. 概述

PDF VQA 提取流水线 可以自动从教材类 PDF 中提炼高质量问答对，串联了 MinerU 布局解析、面向学科的 LLM 抽取以及结构化后处理。典型场景包括：

构建数学 / 物理 / 化学等学科的题库
生成保留图片引用的问答对Markdown / JSONL 资料

核心阶段：

文档布局解析：通过 MinerU 输出结构化 JSON 与页面切图。
LLM 问答抽取：基于学科定制提示词，解析题目、答案、解答。
合并与过滤：整合问答流、过滤无效条目、导出 JSONL/Markdown 以及图片。

2. 快速开始

步骤 1：安装 Dataflow

安装 Dataflow：

pip install "open-dataflow[pdf2vqa]"

或者从源码安装 Dataflow：

git clone https://github.com/OpenDCAI/DataFlow.git
cd Dataflow
pip install -e ".[pdf2vqa]"

步骤 2：创建工作区

cd /your/working/directory
mkdir run_dataflow
cd run_dataflow

步骤 3：初始化 Dataflow

dataflow init

初始化后即可在 pipelines/ 或任意自定义目录编写脚本。

步骤 4：配置 API 凭证

其中DF_API_KEY用于调用LLM API，MINERU_API_KEY用于调用MinerU进行布局解析。 MINERU_API_KEY可以在 https://mineru.net/apiManage/token 获取， DF_API_KEY可以在你的LLM提供商处获取（例如OpenAI、Google Gemini等）。将它们设置为环境变量：

Linux / macOS:

export DF_API_KEY="sk-xxxxx"
export MINERU_API_KEY="sk2-xxxxx"

Windows PowerShell:

$env:DF_API_KEY = "sk-xxxxx"
$env:MINERU_API_KEY = "sk2-xxxxx"

在脚本中设置接口：

self.llm_serving = APILLMServing_request(
    api_url="https://generativelanguage.googleapis.com/v1beta/openai/chat/completions",
    key_name_of_api_key="DF_API_KEY",
    model_name="gemini-2.5-pro",
    max_workers=100,
)

并设置LLM最大token数量（建议不要设置大于128000，否则LLM因为无法记住细节而效果不好）。

self.vqa_extractor = ChunkedPromptedGenerator(
    llm_serving=self.llm_serving,
    system_prompt = self.vqa_extract_prompt.build_prompt(),
    max_chunk_len=128000,
)

步骤 5：一键运行

python api_pipelines/pdf_vqa_extract_pipeline.py

也可将各算子嵌入其他流程，下文详细介绍数据流。

3. 数据流与流水线逻辑

1. 输入数据

使用 JSONL 描述任务，每行包含 input_pdf_paths 和 name。input_pdf_paths 可以是单个 PDF 或 PDF 列表（问题在前，答案在后）。name 是该任务的标识符。问题和答案可以交错或者分开；它们可以来自同一 PDF 或不同 PDF。

{"input_pdf_paths": "./example_data/PDF2VQAPipeline/questionextract_test.pdf", "name": "math1"}
{"input_pdf_paths": ["./example_data/PDF2VQAPipeline/math_question.pdf", "./example_data/PDF2VQAPipeline/math_answer.pdf"], "name": "math2"}

FileStorage 负责读取与缓存：

self.storage = FileStorage(
            first_entry_file_name="./example_data/PDF2VQAPipeline/vqa_extract_test.jsonl",
            cache_path="./cache",
            file_name_prefix="vqa",
            cache_type="jsonl",
        )

2. 文档布局解析（MinerU）

对每个 PDF（题目、答案或混排）调用 FileOrURLToMarkdownConverterAPI 内部的 _parse_file_with_mineru，MinerU 会产出：

*_content_list.json：结构化布局 token
images/：对应页面切图

Note：如果想要使用本地部署的MinerU模型，可以替换算子为 FileOrURLToMarkdownConverterLocal（opendatalab原版）或 FileOrURLToMarkdownConverterFlash （我们的加速版），并提供相应的模型路径和部署参数。

例如：

self.mineru_executor = FileOrURLToMarkdownConverterAPI(intermediate_dir = "intermediate")

可以等价替换为

self.mineru_executor = FileOrURLToMarkdownConverterLocal(
    intermediate_dir = "intermediate",
    mineru_model_path = "path/to/mineru/model",
)

或者

self.mineru_executor = FileOrURLToMarkdownConverterFlash(
    intermediate_dir = "intermediate",
    mineru_model_path = "path/to/mineru/model",
    batch_size = 4,
    replicas = 1,
    num_gpus_per_replica = 1,
    engine_gpu_util_rate_to_ray_cap = 0.9
)

具体参数和使用方法可以参考 https://github.com/OpenDCAI/DataFlow/blob/main/dataflow/operators/knowledge_cleaning/generate/mineru_operators.py 。

之后会使用MinerU2LLMInputOperator处理成给llm的输入，主要包括展平列表项并重新编号。

3. 问答抽取（VQAExtractor）

ChunkedPromptedGenerator 将布局 JSON 切块以控制 token，利用 QAExtractPrompt 为提示词，并通过 APILLMServing_request 批量调用 LLM。主要特性：

整合、匹配问答对，并将图片插入到正确位置。
同时支持题目答案在不同pdf，以及题目答案混排（question1-answer1-question2-answer2-...）。
将 MinerU 切图复制到 cache_path/name/vqa_images。
解析 <qa_pair>、<question>、<answer>、<solution>、<chapter>、<label> 标签。

4. 后处理与产物

调用QA_Merger进行问答对匹配，这个算子：

对于题目答案混排的，会直接把已经完整的问答对原样写入，等于什么都没做
对于题目答案分离的，会根据章节标题，题目序号进行启发式匹配

这个算子可以设置一个strict_title_match参数，如果设置为True，会对章节标题进行严格匹配，否则会尝试提取标题中的中文/英文序号再匹配。

每个 output_dir （cache_path/name/下面）会得到：

extracted_vqa.jsonl
merged_qa_pairs.jsonl
merged_qa_pairs.md
vqa_images/

此外，cache的主文件最后一个step会包含提取出来的所有问答对，方便后面直接接算子做后处理。

每个qa_item包含：

question：题干文本与图片
answer：答案文本与图片
solution：可选的解题过程
label：题目编号
chapter_title：所在章节/小节标题（如果strict_title_match=False则为标题序号）

示例：

{
  "question": "计算 $x$ 使得 $x^2-1=0$。",
  "answer": "$x = 1$ 或 $x = -1$",
  "solution": "因式分解 $(x-1)(x+1)=0$。",
  "label": 1,
  "chapter_title": "第一章 二次方程"
}

最终调用VQAFormatter算子，将问答对转化为标准ShareGPT格式的数据，方便随后可能进行的微调步骤。

示例：

{
    "messages": [
        {
            "role": "user",
            "content": "<image> $\\\\triangle ABC$ 的内切圆切 $BC$ 于 $D..."
        },
        {
            "role": "assistant",
            "content": "证明 \\n设 $\\\\triangle ABC$ 三对应边为 $a$, $b, c, p = ..."
        }
    ],
    "images": [
        "/path/to/image/.jpg"
    ]
}

5. 流水线示例

from dataflow.operators.knowledge_cleaning import FileOrURLToMarkdownConverterAPI

from dataflow.serving import APILLMServing_request
from dataflow.utils.storage import FileStorage
from dataflow.operators.pdf2vqa import MinerU2LLMInputOperator, LLMOutputParser, QA_Merger, PDF_Merger, VQAFormatter
from dataflow.operators.core_text import ChunkedPromptedGenerator

from dataflow.pipeline import PipelineABC
from dataflow.prompts.pdf2vqa import QAExtractPrompt

from pypdf import PdfWriter
    
class PDF_VQA_extract_optimized_pipeline(PipelineABC):
    def __init__(self):
        super().__init__()
        self.storage = FileStorage(
            first_entry_file_name="./example_data/PDF2VQAPipeline/vqa_extract_test.jsonl",
            cache_path="./cache",
            file_name_prefix="vqa",
            cache_type="jsonl",
        )
        
        self.llm_serving = APILLMServing_request(
            api_url="http://123.129.219.111:3000/v1/chat/completions",
            key_name_of_api_key="DF_API_KEY",
            model_name="gemini-2.5-pro",
            max_workers=100,
        )
        
        self.vqa_extract_prompt = QAExtractPrompt()
        
        self.pdf_merger = PDF_Merger(output_dir="./cache")
        self.mineru_executor = FileOrURLToMarkdownConverterAPI(intermediate_dir = "intermediate")
        self.input_formatter = MinerU2LLMInputOperator()
        self.vqa_extractor = ChunkedPromptedGenerator(
            llm_serving=self.llm_serving,
            system_prompt = self.vqa_extract_prompt.build_prompt(),
            max_chunk_len=128000,
        )
        self.llm_output_parser = LLMOutputParser(output_dir="./cache", intermediate_dir="intermediate")
        self.qa_merger = QA_Merger(output_dir="./cache", strict_title_match=False)
        self.vqa_format_converter = VQAFormatter(output_json_file="./.cache/data/qa.json")
    def forward(self):
        self.pdf_merger.run(
            storage=self.storage.step(),
            input_pdf_list_key="input_pdf_paths",
            input_name_key="name",
            output_pdf_path_key="merged_pdf_path",
        )
        self.mineru_executor.run(
            storage=self.storage.step(),
            input_key="merged_pdf_path",
            output_key="vqa_markdown_path",
        )
        self.input_formatter.run(
            storage=self.storage.step(),
            input_markdown_path_key="vqa_markdown_path",
            output_converted_layout_key="converted_vqa_layout_path",
        )
        self.vqa_extractor.run(
            storage=self.storage.step(),
            input_path_key="converted_vqa_layout_path",
            output_path_key="extracted_llm_vqa_path",
        )
        self.llm_output_parser.run(
            storage=self.storage.step(),
            input_response_path_key="extracted_llm_vqa_path",
            input_converted_layout_path_key="converted_vqa_layout_path",
            input_name_key="name",
            output_qalist_path_key="extracted_vqa_path",
        )
        self.qa_merger.run(
            storage=self.storage.step(),
            input_qalist_path_key="extracted_vqa_path",
            input_name_key="name",
            output_merged_qalist_path_key="output_merged_vqalist_path",
            output_merged_md_path_key="output_merged_md_path",
            output_qa_item_key="vqa_pair",
        )
        self.vqa_format_converter.run(
            storage=self.storage.step(),
            input_qa_item_key="vqa_pair",
            output_messages_key="messages",
            output_images_key="images",
        )


if __name__ == "__main__":
    # jsonl中每一行包含input_pdf_paths, name (math1, math2, physics1, chemistry1, ...)
    pipeline = PDF_VQA_extract_optimized_pipeline()
    pipeline.compile()
    pipeline.forward()

Pipeline 源码：DataFlow/dataflow/statics/pipelines/api_pipelines/pdf_vqa_extract_pipeline.py

利用该流水线可直接从 PDF 教材中沉淀带图引用的结构化问答数据。