OCRVLM

专为 LLM · RAG · Agent 场景构建的高精度文档解析引擎

安装 MinerU

使用pip或uv安装MinerU

• 创建虚拟环境

mkdir -p ~/mineru
python3 -m venv ~/mineru/venv
source ~/mineru/venv/bin/activate

• 安装 MinerU

pip install --upgrade pip -i https://mirrors.aliyun.com/pypi/simple
pip install uv -i https://mirrors.aliyun.com/pypi/simple
uv pip install -U "mineru[all]" -i https://mirrors.aliyun.com/pypi/simple 
uv pip install -U "mineru[core]" -i https://mirrors.aliyun.com/pypi/simple 

带显卡用[all]、纯CPU环境用[core]

• 下载模型MinerU2.5pro-1.2b

export MINERU_MODEL_SOURCE=modelscope
mineru-models-download

• 运行测试

# hybrid
mineru -p 输入.pdf -o 输出目录

# 纯CPU
mineru -p 输入.pdf -o 输出目录 -b pipeline

• 启动mineru-api

mineru-api

使用docker部署Mineru

• 使用 Dockerfile 构建镜像

wget https://gcore.jsdelivr.net/gh/opendatalab/MinerU@master/docker/china/Dockerfile

docker build -t mineru:latest -f Dockerfile .

• 通过 Docker Compose

wget https://gcore.jsdelivr.net/gh/opendatalab/MinerU@master/docker/compose.yaml

• 启动 openai兼容接口 服务

docker compose -f compose.yaml --profile openai-server up -d

在另一个终端中通过http client连接openai server（只需cpu与网络，不需要vllm环境）

mineru -p <input_path> -o <output_path> -b vlm-http-client -u http://<server_ip>:30000

• 启动 Web API 服务

docker compose -f compose.yaml --profile api up -d

在浏览器中访问 http://<server_ip>:8000/docs 查看API文档。

• 启动 MinerU Router 服务

docker compose -f compose.yaml --profile router up -d

在浏览器中访问 http://<server_ip>:7860 使用 Gradio WebUI。

测速程序

安装依赖pip install requests

python mineru_speed_test.py -i /root/test/test.pdf -o /root/test/output3

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import argparse
import os
import time
import requests
import sys


def parse_pdf(api_url, input_file, output_dir):

    if not os.path.exists(input_file):
        print(f"输入文件不存在: {input_file}")
        sys.exit(1)

    os.makedirs(output_dir, exist_ok=True)

    filename = os.path.basename(input_file)

    print("=" * 70)
    print("MinerU API 性能测试")
    print(f"输入文件 : {input_file}")
    print(f"输出目录 : {output_dir}")
    print(f"API地址  : {api_url}")
    print("=" * 70)


    try:
        with open(input_file, "rb") as f:

            files = {
                "files": (
                    filename,
                    f,
                    "application/pdf"
                )
            }

            data = {
                "output_dir": output_dir,
                "output_dir_mode": "custom",
                "parse_method": "auto"
            }


            # 开始计时
            start = time.perf_counter()


            response = requests.post(
                api_url,
                files=files,
                data=data,
                timeout=3600
            )


            end = time.perf_counter()


        cost = end - start


    except Exception as e:
        print("\n请求异常:")
        print(e)
        sys.exit(1)


    print("\n结果:")
    print("-" * 70)

    print(f"HTTP状态码 : {response.status_code}")
    print(f"接口耗时   : {cost:.3f} 秒")


    if response.status_code == 200:

        print("\n解析成功")


        try:
            result = response.json()

            print("\n返回结果:")
            print(result)


        except Exception:

            print(response.text)


    else:

        print("\n解析失败:")
        print(response.text)



if __name__ == "__main__":


    parser = argparse.ArgumentParser(
        description="MinerU 2.5-Pro API速度测试"
    )


    parser.add_argument(
        "-i",
        "--input",
        required=True,
        help="PDF文件路径"
    )


    parser.add_argument(
        "-o",
        "--output",
        required=True,
        help="输出目录"
    )


    parser.add_argument(
        "--url",
        default="http://127.0.0.1:8000/file_parse",
        help="MinerU API地址"
    )


    args = parser.parse_args()


    parse_pdf(
        args.url,
        args.input,
        args.output
    )

• 多线程测试

python mineru_batch_speed_test.py -i 输入文件 -o 输出目录 -n 任务数 -c 并发数 --url api地址

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import requests
import time
import os
import argparse
from concurrent.futures import ThreadPoolExecutor, as_completed


def run_task(task_id, api_url, pdf_file, output_root):

    output_dir = os.path.join(
        output_root,
        f"task_{task_id}"
    )

    os.makedirs(output_dir, exist_ok=True)


    start = time.perf_counter()

    try:

        with open(pdf_file, "rb") as f:

            files = {
                "files": (
                    os.path.basename(pdf_file),
                    f,
                    "application/pdf"
                )
            }


            data = {
                "output_dir": output_dir,
                "output_dir_mode": "custom",
                "parse_method": "auto"
            }


            r = requests.post(
                api_url,
                files=files,
                data=data,
                timeout=3600
            )


        cost = time.perf_counter() - start


        if r.status_code == 200:

            return {
                "task": task_id,
                "status": "success",
                "time": cost
            }

        else:

            return {
                "task": task_id,
                "status": "failed",
                "time": cost,
                "error": r.text
            }


    except Exception as e:

        cost = time.perf_counter() - start

        return {
            "task": task_id,
            "status": "error",
            "time": cost,
            "error": str(e)
        }



def main():

    parser = argparse.ArgumentParser(
        description="MinerU批量并发性能测试"
    )


    parser.add_argument(
        "-i",
        "--input",
        default="/root/test/test.pdf"
    )


    parser.add_argument(
        "-o",
        "--output",
        default="/root/test/batch_output"
    )


    parser.add_argument(
        "-n",
        "--number",
        type=int,
        default=20,
        help="任务数量"
    )


    parser.add_argument(
        "-c",
        "--concurrency",
        type=int,
        default=5,
        help="并发数"
    )


    parser.add_argument(
        "--url",
        default="http://127.0.0.1:8000/file_parse"
    )


    args = parser.parse_args()


    print("=" * 70)
    print("MinerU 批量性能测试")
    print(f"任务数量 : {args.number}")
    print(f"并发数量 : {args.concurrency}")
    print(f"输入文件 : {args.input}")
    print(f"API地址  : {args.url}")
    print("=" * 70)



    start_all = time.perf_counter()


    results = []


    with ThreadPoolExecutor(
        max_workers=args.concurrency
    ) as pool:


        futures = []


        for i in range(args.number):

            futures.append(
                pool.submit(
                    run_task,
                    i + 1,
                    args.url,
                    args.input,
                    args.output
                )
            )


        for future in as_completed(futures):

            result = future.result()

            results.append(result)

            print(
                f"任务 {result['task']:02d} "
                f"{result['status']} "
                f"{result['time']:.2f}s"
            )


    total = time.perf_counter() - start_all


    success = [
        x for x in results
        if x["status"] == "success"
    ]


    failed = [
        x for x in results
        if x["status"] != "success"
    ]


    print("\n" + "=" * 70)
    print("测试结果")
    print("=" * 70)


    print(f"总任务数 : {len(results)}")
    print(f"成功数量 : {len(success)}")
    print(f"失败数量 : {len(failed)}")

    print(f"总耗时   : {total:.2f}s")


    if success:

        avg = sum(
            x["time"]
            for x in success
        ) / len(success)

        print(
            f"平均单任务耗时 : {avg:.2f}s"
        )


    print(
        f"吞吐量(QPS): {len(success)/total:.2f} task/s"
    )


if __name__ == "__main__":
    main()

本站文章除注明转载/出处外，均为本站原创或翻译，转载前请务必署名,转载请标明出处
最后编辑时间为: 2026/07/17 17:53