MinerU部署实践：从零开始搭建你的专属PDF解析服务

在多模态RAG（Retrieval-Augmented Generation）系统中，PDF文件的高效、安全解析与处理是实现高质量知识检索和生成的关键环节。PDF文件通常包含丰富的文本、图像和表格信息，这些多模态数据的有效提取和整合对于提升RAG系统的性能至关重要。然而，传统的PDF解析工具往往存在解析精度不足、无法处理复杂格式（如图像和表格）等问题，尤其是在涉及私密文档时，数据安全和隐私保护也是一大挑战。

今天，我将详细介绍MinerU 的私有化部署流程、PDF 解析服务开发，以及如何通过 API 封装实现便捷的文档处理功能。

1、简介

MinerU是一款将PDF转化为机器可读格式的工具（如markdown、json），可以很方便地抽取为任意格式。 主要具有以下功能：

• 删除页眉、页脚、脚注、页码等元素，确保语义连贯
• 输出符合人类阅读顺序的文本，适用于单栏、多栏及复杂排版
• 保留原文档的结构，包括标题、段落、列表等
• 提取图像、图片描述、表格、表格标题及脚注
• 自动识别并转换文档中的公式为LaTeX格式
• 自动识别并转换文档中的表格为HTML格式
• 自动检测扫描版PDF和乱码PDF，并启用OCR功能
• OCR支持84种语言的检测与识别
• 支持多种输出格式，如多模态与NLP的Markdown、按阅读顺序排序的JSON、含有丰富信息的中间格式等
• 支持多种可视化结果，包括layout可视化、span可视化等，便于高效确认输出效果与质检
• 支持纯CPU环境运行，并支持 GPU(CUDA)/NPU(CANN)/MPS 加速
• 兼容Windows、Linux和Mac平台

项目地址：https://github.com/opendatalab/MinerU

说明文档：https://mineru.readthedocs.io/en/latest/index.html

2、私有化部署

MinerU官方提供的API，但是其API KEY需要14天要更换一次，并且在数据安全和隐私保护方面也很难控制。下面是对MinerU的私有化部署介绍：

安装magic-pdf

conda create -n mineru pythnotallow=3.10 
conda activate mineru
pip install -U"magic-pdf[full]"-i https://mirrors.aliyun.com/pypi/simple

模型权重下载

方法一：从 Hugging Face 下载模型

使用python脚本 从Hugging Face下载模型文件

pip install huggingface_hub
wget https://gcore.jsdelivr.net/gh/opendatalab/MinerU@master/scripts/download_models_hf.py -O download_models_hf.py
python download_models_hf.py

python脚本会自动下载模型文件并配置好配置文件中的模型目录。也可以将MinerU代码clone到本地，运行download_models_hf代码

方法二：从 ModelScope 下载模型

pip install modelscope
wget https://gcore.jsdelivr.net/gh/opendatalab/MinerU@master/scripts/download_models.py -O download_models.py
python download_models.py

也可以将MinerU代码clone到本地，运行download_models代码，可以通过配置一些参数，将模型下载到制定文件夹。

详细参考如何下载模型文件。

修改配置文件以进行额外配置

完成下载模型权重文件步骤后，脚本会自动生成用户目录下的magic-pdf.json文件，并自动配置默认模型路径。 可以在【用户目录】下找到magic-pdf.json文件。

windows的用户目录为 "C:\Users\用户名", linux用户目录为 "/home/用户名", macOS用户目录为 "/Users/用户名"

可以修改该文件中的部分配置实现功能的开关，如表格识别功能：

如json内没有如下项目，请手动添加需要的项目，并删除注释内容（标准json不支持注释）

{
  // other config
  "layout-config": {
    "model": "doclayout_yolo"
  },
  "formula-config": {
    "mfd_model": "yolo_v8_mfd",
    "mfr_model": "unimernet_small",
    "enable": true // 公式识别功能默认是开启的，如果需要关闭请修改此处的值为"false"
  },
  "table-config": {
    "model": "rapid_table",
    "sub_model": "slanet_plus",
    "enable": true, // 表格识别功能默认是开启的，如果需要关闭请修改此处的值为"false"
    "max_time": 400
  }
}

3、解析代码

process_pdf是核心解析函数，主要功能包括：

• 自动识别PDF类型（普通文本PDF或扫描版PDF）
• 提取文本内容和图片资源
• 生成Markdown格式的输出
• 可选生成可视化分析结果

参数

代码

import os
from magic_pdf.data.data_reader_writer import FileBasedDataWriter, FileBasedDataReader
from magic_pdf.data.dataset import PymuDocDataset
from magic_pdf.model.doc_analyze_by_custom_model import doc_analyze
from magic_pdf.config.enums import SupportedPdfParseMethod




def process_pdf(pdf_file_name, output_dir="output", image_subdir="images", simple_output=True):
    """
    处理PDF文件，将其转换为Markdown格式并保存相关资源
    :param pdf_file_name: PDF文件名
    :param output_dir: 输出目录，默认为'output'
    :param image_subdir: 图片子目录名，默认为'images'
    :param simple_output: 是否使用简单输出模式，默认为False
    """
    # 获取不带后缀的文件名
    name_without_suff = os.path.splitext(os.path.basename(pdf_file_name))[0]
    # 创建输出子目录名
    output_subdir = name_without_suff
    # 构建图片目录和markdown目录的路径
    local_image_dir = os.path.join(output_dir, output_subdir, image_subdir)
    local_md_dir = os.path.join(output_dir, output_subdir)
    # 创建必要的目录
    os.makedirs(local_image_dir, exist_ok=True)
    os.makedirs(local_md_dir, exist_ok=True)
    # 创建文件写入器
    image_writer, md_writer = FileBasedDataWriter(local_image_dir), FileBasedDataWriter(local_md_dir)
    # 创建文件读取器并读取PDF文件
    reader1 = FileBasedDataReader("")
    pdf_bytes = reader1.read(pdf_file_name)
    # 创建数据集对象
    ds = PymuDocDataset(pdf_bytes)
    # 根据PDF类型选择处理方式
    if ds.classify() == SupportedPdfParseMethod.OCR:
        # 使用OCR模式处理
        infer_result = ds.apply(doc_analyze, ocr=True)
        pipe_result = infer_result.pipe_ocr_mode(image_writer)
    else:
        # 使用文本模式处理
        infer_result = ds.apply(doc_analyze, ocr=False)
        pipe_result = infer_result.pipe_txt_mode(image_writer)
    # 构建markdown文件的完整路径
    md_file_path = os.path.join(os.getcwd(), local_md_dir, f"{name_without_suff}.md")
    abs_md_file_path = os.path.abspath(md_file_path)
    if simple_output:
        # 简单输出模式：只输出markdown和内容列表
        pipe_result.dump_md(md_writer, f"{name_without_suff}.md", os.path.basename(local_image_dir))
        pipe_result.dump_content_list(md_writer, f"{name_without_suff}_content_list.json",
                                      os.path.basename(local_image_dir))
        return abs_md_file_path
    else:
        # 完整输出模式：输出所有内容
        pipe_result.dump_md(md_writer, f"{name_without_suff}.md", os.path.basename(local_image_dir))
        pipe_result.dump_content_list(md_writer, f"{name_without_suff}_content_list.json",
                                      os.path.basename(local_image_dir))
    # 生成可视化文件
    infer_result.draw_model(os.path.join(local_md_dir, f"{name_without_suff}_model.pdf"))
    pipe_result.draw_layout(os.path.join(local_md_dir, f"{name_without_suff}_layout.pdf"))
    pipe_result.draw_span(os.path.join(local_md_dir, f"{name_without_suff}_spans.pdf"))
    return abs_md_file_path


if __name__ == "__main__":
    # 指定要处理的PDF文件名
    pdf_file_name = "/path/to/demo1.pdf"
    # 处理PDF文件并获取生成的markdown文件路径
    md_file_path = process_pdf(pdf_file_name, output_dir="/path/to/output", simple_output=False)
    # 打印生成的markdown文件路径
    print(md_file_path)

输出文件结构

output/
  ├── [PDF文件名]/
  │   ├── images/            # 存放提取的图片
  │   ├── [PDF文件名].md     # 生成的Markdown文件
  │   ├── [PDF文件名]_content_list.json  # 内容列表JSON文件
  │   ├── [PDF文件名]_model.pdf   # 模型可视化结果（完整模式）
  │   ├── [PDF文件名]_layout.pdf  # 布局可视化结果（完整模式）
  │   └── [PDF文件名]_spans.pdf   # 文本块可视化结果（完整模式）

4、API封装

API 端点

• URL:http://[host]:6601/process_pdf
• 方法: POST
• 内容类型: multipart/form-data

请求参数

参数：pdf_file

类型：文件

描述：要解析的PDF文件

响应

成功: 返回包含所有解析结果的ZIP文件

失败: 返回JSON格式的错误信息

代码

from flask import Flask, request, send_file, jsonify
import os
import shutil
import zipfile
from scripts.mineru_process_pdf import process_pdf
app = Flask(__name__)
def create_zip_from_directory(directory_path, zip_file_path):
    with zipfile.ZipFile(zip_file_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
        for root, dirs, files in os.walk(directory_path):
            for file in files:
                file_path = os.path.join(root, file)
                arcname = os.path.relpath(file_path, directory_path)
                zipf.write(file_path, arcname)
@app.route('/process_pdf', methods=['POST'])
def process_pdf_api():
    if 'pdf_file' not in request.files:
        return jsonify({'error': 'No file part'}), 400
    file = request.files['pdf_file']
    if file.filename == '':
        return jsonify({'error': 'No selected file'}), 400
    # Save the uploaded file to a temporary location
    input_pdf_path = os.path.join('temp', file.filename)
    os.makedirs('temp', exist_ok=True)
    file.save(input_pdf_path)
    try:
        # Process the PDF file
        output_dir = '/path/to/output'
        markdown_file_path = process_pdf(input_pdf_path, output_dir=output_dir, simple_output=False)
        # Create a zip file from the output directory
        temp_path = '/path/to/temp'
        os.makedirs(temp_path, exist_ok=True)
        zip_file_path = os.path.join(temp_path, f"{os.path.splitext(file.filename)[0]}.zip")
        create_zip_from_directory(os.path.join(output_dir, os.path.splitext(file.filename)[0]), zip_file_path)
        # Send the zip file as a response
        return send_file(zip_file_path, as_attachment=True)
    except Exception as e:
        return jsonify({'error': str(e)}), 500
    finally:
        # Clean up temporary files
        if os.path.exists(input_pdf_path):
            os.remove(input_pdf_path)
        if os.path.exists(zip_file_path):
            os.remove(zip_file_path)
        if os.path.exists(output_dir):
            shutil.rmtree(output_dir)
if __name__ == '__main__':
    app.run(debug=True, host='0.0.0.0', port=6601)

5、调用示例

下面对该解析服务API提供了三种调用示例，可以根据需要选择使用：

代码

import requests
import os
import zipfile
import io


def parse_pdf_api_to_path(pdf_file_path, output_dir):
    url = "http://localhost:6601/process_pdf"
    # 确保输出目录存在
    os.makedirs(output_dir, exist_ok=True)
    # 获取 PDF 文件的基础名称（不带扩展名）
    base_filename = os.path.splitext(os.path.basename(pdf_file_path))[0]
    with open(pdf_file_path, 'rb') as pdf_file:
        files = {'pdf_file': pdf_file}
        response = requests.post(url, files=files)
    if response.status_code == 200:
        # 保存返回的 zip 文件到指定目录，使用与 PDF 相同的基础文件名
        output_zip_path = os.path.join(output_dir, f'{base_filename}.zip')
        with open(output_zip_path, 'wb') as f:
            f.write(response.content)
        print(f"Test passed: Received zip file and saved to {output_zip_path}.")
    else:
        print(f"Test failed: {response.status_code} - {response.json()}")


def parse_pdf_api_to_content(pdf_file_path):
    url = "http://localhost:6601/process_pdf"
    # 获取 PDF 文件的基础名称（不带扩展名）
    base_filename = os.path.splitext(os.path.basename(pdf_file_path))[0]
    with open(pdf_file_path, 'rb') as pdf_file:
        files = {'pdf_file': pdf_file}
        response = requests.post(url, files=files)
    if response.status_code == 200:
        # 返回压缩包内容
        print(f"Request successful: Received zip file for {base_filename}.")
        return response.content
    else:
        error_message = f"Request failed: {response.status_code} - {response.json()}"
        print(error_message)
        raise Exception(error_message)


def save_zip_content_to_directory(zip_content, output_dir):
    # 确保输出目录存在
    os.makedirs(output_dir, exist_ok=True)
    # 使用 zipfile 模块解压缩内容
    with zipfile.ZipFile(io.BytesIO(zip_content)) as z:
        z.extractall(output_dir)
    print(f"Files extracted to {output_dir}")


def save_zip_and_content_to_directory(zip_content, output_dir, zip_filename):
    # 确保输出目录存在
    os.makedirs(output_dir, exist_ok=True)
    # 保存压缩包到指定目录
    zip_path = os.path.join(output_dir, zip_filename)
    with open(zip_path, 'wb') as f:
        f.write(zip_content)
    print(f"Zip file saved to {zip_path}")
    # 使用 zipfile 模块解压缩内容
    with zipfile.ZipFile(io.BytesIO(zip_content)) as z:
        z.extractall(output_dir)
    print(f"Files extracted to {output_dir}")

直接解压并保存到指定目录

pdf_file_path = "/path/to/your.pdf"
output_unzip_dir = "/path/to/output/dir"
# 获取压缩包内容
zip_content = parse_pdf_api_to_content(pdf_file_path)
# 解压并保存到指定目录
save_zip_content_to_directory(zip_content, output_unzip_dir)

保存压缩包到指定目录并解压

pdf_file_path = "/path/to/your.pdf"
output_unzip_dir = "/path/to/output/dir"
# 获取压缩包内容
zip_content = parse_pdf_api_to_content(pdf_file_path)
# 定义压缩包文件名
zip_filename = os.path.splitext(os.path.basename(pdf_file_path))[0] + ".zip"
# 保存压缩包并解压
save_zip_and_content_to_directory(zip_content, output_unzip_dir, zip_filename)

将解析内容保存到本地

pdf_file_path = "/path/to/your.pdf"
output_dir = "/path/to/output/dir"
# 直接调用API并将结果保存到指定目录
parse_pdf_api_to_path(pdf_file_path, output_dir)