MinerU部署實(shí)踐：從零開始搭建你的專屬PDF解析服務(wù)

作者：Goldma 2025-04-23 03:00:00

我將詳細(xì)介紹MinerU 的私有化部署流程、PDF 解析服務(wù)開發(fā)，以及如何通過 API 封裝實(shí)現(xiàn)便捷的文檔處理功能。

在多模態(tài)RAG（Retrieval-Augmented Generation）系統(tǒng)中，PDF文件的高效、安全解析與處理是實(shí)現(xiàn)高質(zhì)量知識(shí)檢索和生成的關(guān)鍵環(huán)節(jié)。PDF文件通常包含豐富的文本、圖像和表格信息，這些多模態(tài)數(shù)據(jù)的有效提取和整合對(duì)于提升RAG系統(tǒng)的性能至關(guān)重要。然而，傳統(tǒng)的PDF解析工具往往存在解析精度不足、無法處理復(fù)雜格式（如圖像和表格）等問題，尤其是在涉及私密文檔時(shí)，數(shù)據(jù)安全和隱私保護(hù)也是一大挑戰(zhàn)。

今天，我將詳細(xì)介紹MinerU 的私有化部署流程、PDF 解析服務(wù)開發(fā)，以及如何通過 API 封裝實(shí)現(xiàn)便捷的文檔處理功能。

1、簡(jiǎn)介

MinerU是一款將PDF轉(zhuǎn)化為機(jī)器可讀格式的工具（如markdown、json），可以很方便地抽取為任意格式。主要具有以下功能：

刪除頁(yè)眉、頁(yè)腳、腳注、頁(yè)碼等元素，確保語義連貫
輸出符合人類閱讀順序的文本，適用于單欄、多欄及復(fù)雜排版
保留原文檔的結(jié)構(gòu)，包括標(biāo)題、段落、列表等
提取圖像、圖片描述、表格、表格標(biāo)題及腳注
自動(dòng)識(shí)別并轉(zhuǎn)換文檔中的公式為L(zhǎng)aTeX格式
自動(dòng)識(shí)別并轉(zhuǎn)換文檔中的表格為HTML格式
自動(dòng)檢測(cè)掃描版PDF和亂碼PDF，并啟用OCR功能
OCR支持84種語言的檢測(cè)與識(shí)別
支持多種輸出格式，如多模態(tài)與NLP的Markdown、按閱讀順序排序的JSON、含有豐富信息的中間格式等
支持多種可視化結(jié)果，包括layout可視化、span可視化等，便于高效確認(rèn)輸出效果與質(zhì)檢
支持純CPU環(huán)境運(yùn)行，并支持 GPU(CUDA)/NPU(CANN)/MPS 加速
兼容Windows、Linux和Mac平臺(tái)

項(xiàng)目地址：https://github.com/opendatalab/MinerU

說明文檔：https://mineru.readthedocs.io/en/latest/index.html

2、私有化部署

MinerU官方提供的API，但是其API KEY需要14天要更換一次，并且在數(shù)據(jù)安全和隱私保護(hù)方面也很難控制。下面是對(duì)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

模型權(quán)重下載

方法一：從 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腳本會(huì)自動(dòng)下載模型文件并配置好配置文件中的模型目錄。也可以將MinerU代碼clone到本地，運(yùn)行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到本地，運(yùn)行download_models代碼，可以通過配置一些參數(shù)，將模型下載到制定文件夾。

詳細(xì)參考如何下載模型文件。

修改配置文件以進(jìn)行額外配置

完成下載模型權(quán)重文件步驟后，腳本會(huì)自動(dòng)生成用戶目錄下的magic-pdf.json文件，并自動(dòng)配置默認(rèn)模型路徑。可以在【用戶目錄】下找到magic-pdf.json文件。

windows的用戶目錄為 "C:\Users\用戶名", linux用戶目錄為 "/home/用戶名", macOS用戶目錄為 "/Users/用戶名"

可以修改該文件中的部分配置實(shí)現(xiàn)功能的開關(guān)，如表格識(shí)別功能：

如json內(nèi)沒有如下項(xiàng)目，請(qǐng)手動(dòng)添加需要的項(xiàng)目，并刪除注釋內(nèi)容（標(biāo)準(zhǔn)json不支持注釋）

{
  // other config
  "layout-config": {
    "model": "doclayout_yolo"
  },
  "formula-config": {
    "mfd_model": "yolo_v8_mfd",
    "mfr_model": "unimernet_small",
    "enable": true // 公式識(shí)別功能默認(rèn)是開啟的，如果需要關(guān)閉請(qǐng)修改此處的值為"false"
  },
  "table-config": {
    "model": "rapid_table",
    "sub_model": "slanet_plus",
    "enable": true, // 表格識(shí)別功能默認(rèn)是開啟的，如果需要關(guān)閉請(qǐng)修改此處的值為"false"
    "max_time": 400
  }
}

3、解析代碼

process_pdf是核心解析函數(shù)，主要功能包括：

自動(dòng)識(shí)別PDF類型（普通文本PDF或掃描版PDF）
提取文本內(nèi)容和圖片資源
生成Markdown格式的輸出
可選生成可視化分析結(jié)果

參數(shù)

參數(shù)	類型	默認(rèn)值	描述
pdf_file_name	str	無	要解析的PDF文件路徑
output_dir	str	"output"	輸出文件的主目錄
image_subdir	str	"images"	存放圖片的子目錄名稱
simple_output	bool	True	是否使用簡(jiǎn)單輸出模式（True時(shí)只輸出Markdown和內(nèi)容列表）

代碼

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文件，將其轉(zhuǎn)換為Markdown格式并保存相關(guān)資源
    :param pdf_file_name: PDF文件名
    :param output_dir: 輸出目錄，默認(rèn)為'output'
    :param image_subdir: 圖片子目錄名，默認(rèn)為'images'
    :param simple_output: 是否使用簡(jiǎn)單輸出模式，默認(rèn)為False
    """
    # 獲取不帶后綴的文件名
    name_without_suff = os.path.splitext(os.path.basename(pdf_file_name))[0]
    # 創(chuàng)建輸出子目錄名
    output_subdir = name_without_suff
    # 構(gòu)建圖片目錄和markdown目錄的路徑
    local_image_dir = os.path.join(output_dir, output_subdir, image_subdir)
    local_md_dir = os.path.join(output_dir, output_subdir)
    # 創(chuàng)建必要的目錄
    os.makedirs(local_image_dir, exist_ok=True)
    os.makedirs(local_md_dir, exist_ok=True)
    # 創(chuàng)建文件寫入器
    image_writer, md_writer = FileBasedDataWriter(local_image_dir), FileBasedDataWriter(local_md_dir)
    # 創(chuàng)建文件讀取器并讀取PDF文件
    reader1 = FileBasedDataReader("")
    pdf_bytes = reader1.read(pdf_file_name)
    # 創(chuàng)建數(shù)據(jù)集對(duì)象
    ds = PymuDocDataset(pdf_bytes)
    # 根據(jù)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)
    # 構(gòu)建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:
        # 簡(jiǎn)單輸出模式：只輸出markdown和內(nèi)容列表
        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:
        # 完整輸出模式：輸出所有內(nèi)容
        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)

輸出文件結(jié)構(gòu)

output/
  ├── [PDF文件名]/
  │   ├── images/            # 存放提取的圖片
  │   ├── [PDF文件名].md     # 生成的Markdown文件
  │   ├── [PDF文件名]_content_list.json  # 內(nèi)容列表JSON文件
  │   ├── [PDF文件名]_model.pdf   # 模型可視化結(jié)果（完整模式）
  │   ├── [PDF文件名]_layout.pdf  # 布局可視化結(jié)果（完整模式）
  │   └── [PDF文件名]_spans.pdf   # 文本塊可視化結(jié)果（完整模式）

4、API封裝

API 端點(diǎn)

URL:http://[host]:6601/process_pdf
方法: POST
內(nèi)容類型: multipart/form-data

請(qǐng)求參數(shù)

參數(shù)：pdf_file

類型：文件

描述：要解析的PDF文件

響應(yīng)

成功: 返回包含所有解析結(jié)果的ZIP文件

失敗: 返回JSON格式的錯(cuò)誤信息

代碼

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、調(diào)用示例

下面對(duì)該解析服務(wù)API提供了三種調(diào)用示例，可以根據(jù)需要選擇使用：

代碼

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 文件的基礎(chǔ)名稱（不帶擴(kuò)展名）
    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 相同的基礎(chǔ)文件名
        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 文件的基礎(chǔ)名稱（不帶擴(kuò)展名）
    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:
        # 返回壓縮包內(nèi)容
        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 模塊解壓縮內(nèi)容
    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 模塊解壓縮內(nèi)容
    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"
# 獲取壓縮包內(nèi)容
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"
# 獲取壓縮包內(nèi)容
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)

將解析內(nèi)容保存到本地

pdf_file_path = "/path/to/your.pdf"
output_dir = "/path/to/output/dir"
# 直接調(diào)用API并將結(jié)果保存到指定目錄
parse_pdf_api_to_path(pdf_file_path, output_dir)

責(zé)任編輯：龐桂玉來源：小白學(xué)AI算法