【项目】多模态RAG—本地部署MinerU实现多类文档解析

（一）MinerU基本介绍

“你是否遇到过Python解析PDF时格式错乱？或从Word表格提取数据丢失样式？传统库（如PyPDF2、python-docx）对复杂格式支持有限，而商业API又贵又慢。今天介绍一款开箱即用的开源工具——MinerU，支持本地部署，API调用，完美保留文档原始格式。”

MinerU是一款由上海人工智能实验室 OpenDataLab 团队开源的一款高质量数据提取工具，将PDF转换成Markdown和JSON格式。可用于文档数字化、RAG、LLM语料等场景。

具备功能如下：

• 删除页眉、页脚、脚注、页码等元素，确保语义连贯

• 输出符合人类阅读顺序的文本，适用于单栏、多栏及复杂排版

• 保留原文档的结构，包括标题、段落、列表等

• 提取图像、图片描述、表格、表格标题及脚注

• 自动识别并转换文档中的公式为LaTeX格式

• 自动识别并转换文档中的表格为HTML格式

• 自动检测扫描版PDF和乱码PDF，并启用OCR功能

• OCR支持84种语言的检测与识别

• 支持多种输出格式，如多模态与NLP的Markdown、按阅读顺序排序的JSON、含有丰富信息的中间格式等

• 支持多种可视化结果，包括layout可视化、span可视化等，便于高效确认输出效果与质检

• 支持纯CPU环境运行，并支持 GPU(CUDA)/NPU(CANN)/MPS 加速

• 兼容Windows、Linux和Mac平台

从项目提交记录上看，MinerU一直非常活跃，是一个值得持续跟踪的文档解析工具。 相关链接如下：

• 论文链接： https://arxiv.org/abs/2409.18839

• 代码链接： https://github.com/opendatalab/MinerU

• 官网试用链接： https://mineru.net/OpenSourceTools/Extractor

（二）实际需求背景

实际需求：

1. 目前有大量的政策数据，为了实现基于知识库的AI政务问答系统，原文中大量附件不能被大模型很好的识别，直接读取文本，又会丢失格式，所以目前最好的解决方案就是把附件都转化成markdown格式（AI识别效果好）
2. 给客户展示数据的时候，又需要能在网页中直接看到附件内容，所以还需要将markdown格式转化成html标签格式。
3. 需要解析的格式：“pdf”, “doc”, “docx”, “xls”, “xlsx”, “jpg”

（三）效果图

原始页面：

解析附件后拼接效果：

带表格附件解析效果：

（四）MinerU安装

Linux系统安装环境，本地部署MinerU项目，编写服务端程序提供API接口，方便其他项目调用

支持解析的文档格式：“pdf”, “doc”, “docx”, “xls”, “xlsx”, “jpg”

详细使用：

拉取项目、进入项目、从源码安装环境

git clone https://gitee.com/myhloli/MinerU.git
cd MinerU
uv pip install -e .[core] -i https://mirrors.aliyun.com/pypi/simple

下载模型到本地

mineru-models-download
根据提示选择下一步
下载方式选择modelscope
回车
如图所示，即成功开始下载

下载完成后就可以使用命令行调用了

mineru -p <input_path> -o <output_path> --source local

<input_path>：需要解析的文件地址

<output_path> ：测解析后输出文件的地址

–source local： 使用本地模型解析

（五）服务端

我们使用python写一个服务端程序，方便项目可以使用API调用

脚本放在MinerU项目的根目录即可

mineru_service.py 脚本代码如下：

# -*- coding: utf-8 -*-
import os
import uuid
import shutil
import time
import json
import asyncio
import logging
from pathlib import Path
from typing import List, Optional
from fastapi import FastAPI, UploadFile, File, Form
from pydantic import BaseModel
from fastapi.responses import FileResponse, JSONResponse
import markdown
import uvicorn
import mimetypes
import subprocess
from tempfile import NamedTemporaryFile

# 设置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mineru")

# 创建 FastAPI 应用
app = FastAPI()

# 临时输出目录
TEMP_BASE_DIR = Path("./temp_output")
TEMP_BASE_DIR.mkdir(exist_ok=True)

# 启动清理任务（清除超过 1 小时的目录）
async def clean_temp_dirs():
    while True:
        try:
            for dir in TEMP_BASE_DIR.iterdir():
                if dir.is_dir() and (time.time() - dir.stat().st_mtime) > 3600:
                    shutil.rmtree(dir, ignore_errors=True)
                    logger.info(f"清理目录: {dir}")
        except Exception as e:
            logger.exception(f"清理任务异常: {e}")
        await asyncio.sleep(3600)

@app.on_event("startup")
async def startup_event():
    asyncio.create_task(clean_temp_dirs())

# 请求体模型
class ParseRequest(BaseModel):
    file_paths: Optional[List[str]] = None
    lang: Optional[str] = "ch"
    backend: Optional[str] = "pipeline"
    method: Optional[str] = "auto"
    output_format: Optional[str] = "md"

# 转换常见办公文档为 PDF
def convert_to_pdf(input_path: Path) -> Path:
    suffix = input_path.suffix.lower()
    if suffix in ['.doc', '.docx', '.xls', '.xlsx']:
        output_path = input_path.with_suffix('.pdf')
        cmd = [
            "libreoffice",
            "--headless",
            "--convert-to", "pdf",
            "--outdir", str(input_path.parent),
            str(input_path)
        ]
        result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)

        if result.returncode != 0 or not output_path.exists():
            logger.error(
                f"❌ 文档转 PDF 失败: {input_path}\n"
                f"➡️ 命令: {' '.join(cmd)}\n"
                f"  stdout: {result.stdout}\n"
                f"  stderr: {result.stderr}"
            )
            # 可以选择返回原文件而不是报错中断（根据业务需要）
            # return input_path
            raise RuntimeError(f"文档转 PDF 失败: {input_path}")
        else:
            logger.info(f"✅ 转换成功: {input_path} -> {output_path}")
            return output_path
    else:
        return input_path
    
# 调用 mineru 命令行接口
def run_mineru_cli(input_path: str, output_path: str, lang: str = "ch", backend: str = "pipeline", method: str = "auto"):
    os.environ['MINERU_MODEL_SOURCE'] = "modelscope"
    cmd = f"mineru -p {input_path} -o {output_path} --source local"
    logger.info(f"运行命令: {cmd}")
    code = os.system(cmd)
    if code != 0:
        raise RuntimeError(f"mineru 命令执行失败，退出码 {code}")
@app.post("/parse")
async def parse_from_path(req: ParseRequest):
    result = {}
    for path in req.file_paths:
        name = Path(path).stem
        task_id = uuid.uuid4().hex[:8]
        task_output_dir = TEMP_BASE_DIR / task_id
        task_output_dir.mkdir(exist_ok=True)

        input_path = convert_to_pdf(Path(path)) 
        run_mineru_cli(input_path, str(task_output_dir), lang=req.lang, backend=req.backend, method=req.method)

        output_md_path = task_output_dir / name / f"auto"
        output_md_path = output_md_path / f"{name}.md"
        if req.output_format == "html":
            md_content = output_md_path.read_text(encoding="utf-8")
            result[name] = convert_md_to_html(md_content)
        else:
            result[name] = output_md_path.read_text(encoding="utf-8")
    return JSONResponse(content=result)
@app.post("/upload")
async def parse_from_upload(
    file: UploadFile = File(...),
    lang: str = Form("ch"),
    backend: str = Form("pipeline"),
    method: str = Form("auto"),
    output_format: str = Form("md")
):
    content = await file.read()
    name = Path(file.filename).stem
    task_id = uuid.uuid4().hex[:8]
    task_dir = TEMP_BASE_DIR / task_id
    task_dir.mkdir(exist_ok=True)

    input_path = task_dir / file.filename

    with open(input_path, "wb") as f:
        f.write(content)

    input_path = convert_to_pdf(input_path)
    run_mineru_cli(str(input_path), str(task_dir), lang=lang, backend=backend, method=method)

    output_md_path = task_dir / name / f"auto"
    output_md_path = output_md_path / f"{name}.md"
    if output_format == "html":
        md_content = output_md_path.read_text(encoding="utf-8")
        html_content = convert_md_to_html(md_content)
        return html_content
    else:
        return output_md_path.read_text(encoding="utf-8")

def convert_md_to_html(md_text: str) -> str:
    html = markdown.markdown(md_text, extensions=['extra', 'tables', 'nl2br'])
    html = html.replace("\n", "")  # 可选：去除残余换行符
    return html

# 启动服务
if __name__ == '__main__':
    uvicorn.run("__main__:app", host="0.0.0.0", port=5002, reload=False)

需要用到的依赖：

fastapi
uvicorn[standard]
markdown
pydantic
python-multipart

可以创建一个requirements.txt，把上面依赖粘贴进去

然后执行

pip install -r requirements.txt

安装完毕后，运行脚本

python mineru_service.py

如上图所示，即运行成功

需要把服务器防火墙的5002端口开放（不同系统使用的防火墙不同，根据自己的系统来）

sudo ufw allow 5002/tcp
或者
sudo firewall-cmd --zone=public --add-port=5002/tcp --permanent
sudo firewall-cmd --reload

由于解析的文件太多，所以在代码里加了一个每隔1小时，清除所有的已解析文件方法，如果修改了导出文件的目录，记得一并修改

（六）接口文档

（1）接口1：批量解析文档路径

请求方式：POST /parse

请求体 JSON（application/json）

{
  "file_paths": ["/path/to/doc1.docx", "/path/to/doc2.xlsx"],
  "lang": "ch",
  "backend": "pipeline",
  "method": "auto",
  "output_format": "md"
}

✅ 响应示例（返回 JSON）

{
  "doc1": "# 一级标题\n- 抽取内容段落1\n- 抽取内容段落2",
  "doc2": "<h1>标题</h1><p>段落内容</p>"
}

键为文件名（无后缀），值为 markdown 或 html 格式文本

（2）接口2：上传单个文件进行解析

请求方式：POST /upload

请求类型：multipart/form-data

表单字段

✅ 响应示例：

• Markdown 响应（output_format=md）：

# 一级标题
- 抽取内容段落
- 抽取内容段落

• HTML 响应（output_format=html）：

<h1>一级标题</h1><ul><li>抽取内容段落</li></ul>

错误情况说明

（七）测试

上传需要解析的文件到服务器

使用postman工具，尝试调用（大部分参数有默认值，所以我只填必传参数了）

（八）总结

实际使用中，解析受显卡GPU影响，如果配置不够高建议单线程慢慢解析，所有的数据附件都存储在文件服务器中（nfs共享服务器），可以直接把附件服务器挂载到安装MinerU的服务器上，方便获取文件，只需要根据文件的名称拼接路径即可读取数据