基于源码级分析 — 核心模块源码片段 · 关键模块依赖关系图 · 数据流架构
下图展示了 MinerU 各核心模块之间的调用与依赖关系。箭头方向表示"调用/依赖"关系。
这是 Pipeline 引擎的入口,负责文档分类、模型初始化、流式处理窗口调度。
mineru/backend/pipeline/pipeline_analyze.py# Copyright (c) Opendatalab. All rights reserved. import os import time from typing import List, Tuple import pypdfium2 as pdfium from PIL import Image from loguru import logger from tqdm import tqdm from .model_init import MineruPipelineModel, PIPELINE_MODEL_INIT_LOCK from .model_json_to_middle_json import ( apply_server_side_postprocess, append_batch_results_to_middle_json, finalize_middle_json, init_middle_json, ) from mineru.utils.config_reader import get_device, get_processing_window_size from ...utils.enum_class import ImageType from ...utils.pdf_classify import classify from ...utils.pdf_image_tools import load_images_from_pdf_doc from ...utils.model_utils import get_vram, clean_memory from ...utils.pdfium_guard import ( close_pdfium_document, get_pdfium_document_page_count, open_pdfium_document, ) os.environ['PYTORCH_ENABLE_MPS_FALLBACK'] = '1' # 让mps可以fallback class ModelSingleton: """Pipeline 模型单例管理器 — 线程安全的懒加载模式""" _instance = None _models = {} _lock = PIPELINE_MODEL_INIT_LOCK def __new__(cls, *args, **kwargs): with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def get_model(self, lang=None, formula_enable=None, table_enable=None): key = (lang, formula_enable, table_enable) with self._lock: if key not in self._models: self._models[key] = custom_model_init( lang=lang, formula_enable=formula_enable, table_enable=table_enable, ) return self._models[key] def _get_ocr_enable(pdf_bytes, parse_method: str) -> bool: """根据解析方法和PDF分类结果判断是否启用OCR""" if parse_method == 'auto': return classify(pdf_bytes) == 'ocr' if parse_method == 'ocr': return True return False def doc_analyze_streaming( pdf_bytes_list, image_writer_list, lang_list, on_doc_ready, parse_method: str = 'auto', formula_enable=True, table_enable=True, client_side_output_generation=False, ): """Pipeline 流式文档分析主入口 — 按处理窗口分批处理页面""" # 1. 为每个文档创建上下文 doc_contexts = [] for doc_index, (pdf_bytes, image_writer, lang) in enumerate( zip(pdf_bytes_list, image_writer_list, lang_list) ): _ocr_enable = _get_ocr_enable(pdf_bytes, parse_method) pdf_doc = open_pdfium_document(pdfium.PdfDocument, pdf_bytes) page_count = get_pdfium_document_page_count(pdf_doc) context = { 'doc_index': doc_index, 'pdf_bytes': pdf_bytes, 'pdf_doc': pdf_doc, 'page_count': page_count, 'next_page_idx': 0, 'middle_json': init_middle_json(), # ← 初始化中间JSON 'model_list': [], 'image_writer': image_writer, 'lang': lang, 'ocr_enable': _ocr_enable, 'closed': False, } doc_contexts.append(context) # 2. 按处理窗口分批推理 → 3. 转换为Middle JSON → 4. 回调输出
模型初始化模块管理所有 AI 模型的生命周期,通过双重单例模式(ModelSingleton + AtomModelSingleton)和细粒度推理锁保证线程安全。
mineru/backend/pipeline/model_init.py# Copyright (c) Opendatalab. All rights reserved. import os import threading import torch from loguru import logger from .model_list import AtomicModel from ...model.layout.pp_doclayoutv2 import PPDocLayoutV2LayoutModel from ...model.mfr.unimernet.Unimernet import UnimernetModel from ...model.mfr.pp_formulanet_plus_m.predict_formula import FormulaRecognizer from mineru.model.ocr.pytorch_paddle import PytorchPaddleOCR from ...model.table.cls.paddle_table_cls import PaddleTableClsModel from ...model.table.rec.slanet_plus.main import PaddleTableModel from ...model.table.rec.unet_table.main import UnetTableModel # 线程安全锁 — 保护 Pipeline 与 Hybrid 共享的模型推理 PIPELINE_MODEL_INIT_LOCK = threading.RLock() PIPELINE_LAYOUT_INFERENCE_LOCK = threading.RLock() PIPELINE_MFR_INFERENCE_LOCK = threading.RLock() PIPELINE_OCR_INFERENCE_LOCK = threading.RLock() PIPELINE_INFERENCE_LOCKS_ENABLED = os.getenv( 'MINERU_ENABLE_PIPELINE_INFERENCE_LOCKS', 'False' ).lower() in ['true', '1', 'yes'] def _run_with_inference_lock(inference_lock, inference_callable, *args, **kwargs): """按实验开关决定是否在指定推理锁内执行模型调用""" if not PIPELINE_INFERENCE_LOCKS_ENABLED: return inference_callable(*args, **kwargs) with inference_lock: return inference_callable(*args, **kwargs) def run_layout_inference(inference_callable, *args, **kwargs): return _run_with_inference_lock(PIPELINE_LAYOUT_INFERENCE_LOCK, ...) def run_mfr_inference(inference_callable, *args, **kwargs): return _run_with_inference_lock(PIPELINE_MFR_INFERENCE_LOCK, ...) def run_ocr_inference(inference_callable, *args, **kwargs): return _run_with_inference_lock(PIPELINE_OCR_INFERENCE_LOCK, ...) # 公式模型选择 — 通过环境变量切换中文公式支持 MFR_MODEL = os.getenv('MINERU_FORMULA_CH_SUPPORT', 'False') if MFR_MODEL.lower() in ['true', '1', 'yes']: MFR_MODEL = "pp_formulanet_plus_m" # 中文公式模型 else: MFR_MODEL = "unimernet_small" # 通用公式模型 def mfr_model_init(weight_dir, device='cpu'): if MFR_MODEL == "unimernet_small": mfr_model = UnimernetModel(weight_dir, device) elif MFR_MODEL == "pp_formulanet_plus_m": mfr_model = FormulaRecognizer(weight_dir, device) return mfr_model def ocr_model_init(det_db_box_thresh=0.5, lang=None, ...): if lang in [None, "ch"]: use_dilation = True det_db_unclip_ratio = 1.8 else: use_dilation = False model = PytorchPaddleOCR( det_db_box_thresh=det_db_box_thresh, lang=lang, use_dilation=use_dilation, det_db_unclip_ratio=det_db_unclip_ratio, enable_merge_det_boxes=enable_merge_det_boxes, ) return model class AtomModelSingleton: """原子模型单例 — 管理 OCR、Table 等可复用的子模型""" _instance = None _models = {} _lock = PIPELINE_MODEL_INIT_LOCK def __new__(cls, *args, **kwargs): with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) return cls._instance def get_atom_model(self, atom_model_name: str, **kwargs): key = (atom_model_name, kwargs.get('lang'), ...) with self._lock: if key not in self._models: self._models[key] = self._create_model(atom_model_name, **kwargs) return self._models[key]
VLM 引擎通过视觉语言模型(如 Qwen2-VL)端到端理解文档页面,支持多种推理后端。
mineru/backend/vlm/vlm_analyze.py# Copyright (c) Opendatalab. All rights reserved. import asyncio, atexit, gc, os, time, json, threading from contextlib import asynccontextmanager, contextmanager import pypdfium2 as pdfium from loguru import logger from tqdm import tqdm from .utils import (enable_custom_logits_processors, set_default_gpu_memory_utilization, set_default_batch_size, set_lmdeploy_backend, mod_kwargs_by_device_type) from .model_output_to_middle_json import ( append_page_blocks_to_middle_json, finalize_middle_json, init_middle_json, ) from mineru.data.data_reader_writer import DataWriter from mineru_vl_utils import MinerUClient from packaging import version class ModelSingleton: """VLM 模型单例 — 支持多种推理后端""" _instance = None _models = {} _lock = threading.RLock() def get_model(self, backend: str, model_path: str | None, server_url: str | None, **kwargs) -> MinerUClient: key = (backend, model_path, server_url) with self._lock: if key not in self._models: # 根据 backend 类型选择不同的推理引擎 if backend == "transformers": model = Qwen2VLForConditionalGeneration.from_pretrained( model_path, device_map={"": device}, **{dtype_key: "auto"}) processor = AutoProcessor.from_pretrained(model_path) elif backend == "vllm-engine": import vllm kwargs = mod_kwargs_by_device_type(kwargs, vllm_mode="sync_engine") # vLLM 高性能推理引擎 elif backend == "lmdeploy-engine": from lmdeploy import pipeline # LMDeploy 推理引擎 elif backend == "http-client": # 远程 HTTP API 调用 pass return self._models[key]
| 后端 | 类型 | 适用场景 | 依赖 |
|---|---|---|---|
| transformers | 本地推理 | 开发调试、小批量 | transformers, torch |
| vllm-engine | 本地推理 | 高性能批量推理 | vllm |
| lmdeploy-engine | 本地推理 | 高性能推理(TurboMind) | lmdeploy |
| mlx-engine | 本地推理 | Apple Silicon Mac | mlx |
| http-client | 远程调用 | 分布式部署 | httpx |
Hybrid 引擎融合 Pipeline 和 VLM 的能力:先用 Pipeline 模型获取版面结构,再用 VLM 对复杂区域补充理解。
mineru/backend/hybrid/hybrid_analyze.py# Copyright (c) Opendatalab. All rights reserved. import asyncio, os, time from collections import defaultdict import cv2 import numpy as np import pypdfium2 as pdfium from loguru import logger from mineru_vl_utils import MinerUClient from mineru_vl_utils.structs import BlockType # ← 关键:Hybrid 同时依赖 Pipeline 和 VLM 两个引擎 from mineru.backend.pipeline.model_init import ( HybridModelSingleton, run_layout_inference, run_mfr_inference, run_ocr_inference, ) from mineru.backend.vlm.vlm_analyze import ( ModelSingleton, aio_predictor_execution_guard, predictor_execution_guard, _maybe_enable_serial_execution, _get_model_async, ) from mineru.data.data_reader_writer import DataWriter from mineru.utils.ocr_utils import ( get_adjusted_mfdetrec_res, get_ocr_result_list, sorted_boxes, merge_det_boxes, update_det_boxes, OcrConfidence) from mineru.utils.pdf_classify import classify LAYOUT_BASE_BATCH_SIZE = 1 MFR_BASE_BATCH_SIZE = 16 OCR_DET_BASE_BATCH_SIZE = 8 def ocr_classify(pdf_bytes, parse_method: str = 'auto') -> bool: """确定OCR设置 — 复用 Pipeline 的分类逻辑""" _ocr_enable = False if parse_method == 'auto': if classify(pdf_bytes) == 'ocr': _ocr_enable = True elif parse_method == 'ocr': _ocr_enable = True return _ocr_enable def ocr_det(hybrid_pipeline_model, np_images, model_list, mfd_res, _ocr_enable, batch_ratio: int = 1, *, fill_text: bool = True): """Hybrid 专用 OCR 检测 — 对 Pipeline 版面结果进行行级 OCR 补充""" ocr_res_list = [] if not hybrid_pipeline_model.enable_ocr_det_batch: # 非批处理模式 — 逐页逐块处理 for np_image, page_mfd_res, page_results in tqdm( zip(np_images, mfd_res, model_list), desc="OCR-det" ): for res in page_results: if not _is_hybrid_ocr_det_candidate(res): continue # 裁剪区域 → OCR 检测 → 合并结果 new_image, useful_list = crop_img(res, np_image, ...) bgr_image = cv2.cvtColor(new_image, cv2.COLOR_RGB2BGR) ocr_res = run_ocr_inference( hybrid_pipeline_model.ocr_model.ocr, bgr_image, mfd_res=adjusted_mfdetrec_res, rec=False )[0] else: # 批处理模式 — 按语言和分辨率分组批量推理 pass
Office 引擎的 MagicModel 将 Office 文档的原生结构化数据转换为统一的 Block 体系。
mineru/backend/office/office_magic_model.py# Copyright (c) Opendatalab. All rights reserved. import html as html_lib import re from typing import Literal from urllib.parse import urlparse from loguru import logger from mineru.utils.enum_class import ContentType, BlockType from mineru.utils.magic_model_utils import tie_up_category_by_index class MagicModel: """Office 文档的 MagicModel — 将原生结构转为统一 Block 表示""" def __init__(self, page_blocks: list): self.page_blocks = page_blocks blocks = [] self.all_spans = [] # 对 caption 块分类 → image_caption / table_caption / chart_caption page_blocks = classify_caption_blocks(page_blocks) for index, block_info in enumerate(page_blocks): block_type = block_info["type"] block_content = block_info.get("content", "") if block_type in ["text", "title", "image_caption", ...]: span = parse_text_block_spans(block_content) elif block_type in ["image"]: block_type = BlockType.IMAGE_BODY span = {"type": ContentType.IMAGE, "image_base64": block_content} elif block_type in ["table"]: block_type = BlockType.TABLE_BODY span = {"type": ContentType.TABLE, "html": clean_table_html(block_content)} elif block_type in ["equation"]: block_type = BlockType.INTERLINE_EQUATION span = {"type": ContentType.INTERLINE_EQUATION, "content": block_content} elif block_type in ["list"]: parsed_list = parse_list_block(block_info) if parsed_list: blocks.append(parsed_list) continue # 构建 Block → Line → Span 三级结构 block = { "type": block_type, "lines": [{"spans": [span]}], "index": index, } if block_type == BlockType.TITLE: block["is_numbered_style"] = block_info.get("is_numbered_style", False) block["level"] = block_info.get("level", 1) blocks.append(block) # 按 BlockType 分类存储 self.image_blocks = [] self.table_blocks = [] self.text_blocks = [] self.title_blocks = [] self.discarded_blocks = [] for block in blocks: if block["type"] in [BlockType.IMAGE_BODY, BlockType.IMAGE_CAPTION, ...]: self.image_blocks.append(block) elif block["type"] in [BlockType.TABLE_BODY, BlockType.TABLE_CAPTION, ...]: self.table_blocks.append(block) # ...
这是 Pipeline 引擎中连接模型推理和内容生成的关键桥梁,将各模型原始输出转换为统一的中间表示。
mineru/backend/pipeline/model_json_to_middle_json.py# Copyright (c) Opendatalab. All rights reserved. import copy from tqdm import tqdm from mineru.backend.utils.html_image_utils import replace_inline_table_images from mineru.backend.utils.runtime_utils import cross_page_table_merge from mineru.backend.pipeline.model_init import AtomModelSingleton, run_ocr_inference from mineru.backend.pipeline.para_split import para_split from mineru.utils.enum_class import ContentType, BlockType from mineru.backend.pipeline.pipeline_magic_model import MagicModel from mineru.utils.ocr_utils import OcrConfidence, rotate_vertical_crop_if_needed def page_model_info_to_page_info(page_model_info, image_dict, page, image_writer, page_index, ocr_enable=False): """将单页模型输出转换为 page_info 结构""" scale = image_dict["scale"] page_pil_img = image_dict["img_pil"] page_img_md5 = bytes_md5(page_pil_img.tobytes()) # 1. 构建 MagicModel — 封装模型检测结果 magic_model = MagicModel( page_model_info, page, scale, page_pil_img, page_w, page_h, ocr_enable ) # 2. 提取各类区块 preproc_blocks = magic_model.get_preproc_blocks() discarded_blocks = magic_model.get_discarded_blocks() all_image_spans = magic_model.get_all_image_spans() # 3. 对 image/table/chart/interline_equation 的 span 截图 for span in all_image_spans: if span["type"] in [ContentType.IMAGE, ContentType.TABLE, ContentType.CHART, ContentType.INTERLINE_EQUATION]: span = cut_image_and_table(span, page_pil_img, page_img_md5, page_index, image_writer, scale=scale) # 4. 替换内联表格图片 → 构造 page_info replace_inline_table_images(preproc_blocks, image_writer, page_index) page_info = make_page_info_dict(preproc_blocks, page_index, page_w, page_h, discarded_blocks) return page_info def append_batch_results_to_middle_json(middle_json, batch_results, images_list, pdf_doc, image_writer, page_start_index=0, ocr_enable=False, model_list=None, progress_bar=None): """将批量推理结果追加到 Middle JSON""" page_model_infos = [] for offset, (image_dict, page_layout_dets) in enumerate( zip(images_list, batch_results) ): page_model_info = build_page_model_info( page_layout_dets, page_start_index + offset, image_dict["img_pil"]) page_model_infos.append(page_model_info) # 逐页转换 → 追加到 middle_json["pdf_info"] append_page_model_infos_to_middle_json( middle_json, page_model_infos, images_list, pdf_doc, image_writer, page_start_index, ocr_enable, progress_bar)
定义了数据读写的抽象接口,所有存储后端实现此接口即可无缝接入。
mineru/data/data_reader_writer/base.py# Copyright (c) Opendatalab. All rights reserved. from abc import ABC, abstractmethod class DataReader(ABC): def read(self, path: str) -> bytes: """Read the file.""" return self.read_at(path) @abstractmethod def read_at(self, path: str, offset: int = 0, limit: int = -1) -> bytes: """Read the file at offset and limit.""" pass class DataWriter(ABC): @abstractmethod def write(self, path: str, data: bytes) -> None: """Write the data to the file.""" pass def write_string(self, path: str, data: str) -> None: """Write string data with safe encoding (utf-8 / ascii).""" def safe_encode(data: str, method: str): try: bit_data = data.encode(encoding=method, errors='replace') return bit_data, True except: return None, False for method in ['utf-8', 'ascii']: bit_data, flag = safe_encode(data, method) if flag: self.write(path, bit_data) break
枚举类定义了整个系统共享的类型常量,是各模块间数据交换的"契约"。
mineru/utils/enum_class.py# Copyright (c) Opendatalab. All rights reserved. from enum import Enum class BlockType: """文档版面元素类型 — 所有引擎共享的 Block 类型体系""" IMAGE = 'image' TABLE = 'table' CHART = 'chart' IMAGE_BODY = 'image_body' TABLE_BODY = 'table_body' CHART_BODY = 'chart_body' CAPTION = 'caption' IMAGE_CAPTION = 'image_caption' TABLE_CAPTION = 'table_caption' CHART_CAPTION = 'chart_caption' ALGORITHM_CAPTION = 'algorithm_caption' FOOTNOTE = 'footnote' TEXT = 'text' TITLE = 'title' INTERLINE_EQUATION = 'interline_equation' EQUATION = "equation" # 行间公式(独立行) LIST = 'list' INDEX = 'index' DISCARDED = 'discarded' CODE = "code" CODE_BODY = "code_body" CODE_CAPTION = "code_caption" ALGORITHM = "algorithm" ABSTRACT = "abstract" DOC_TITLE = "doc_title" PARAGRAPH_TITLE = "paragraph_title" class ContentType: """Span 内容类型 — Block 内部最小内容单元的类型""" IMAGE = 'image' TABLE = 'table' CHART = 'chart' TEXT = 'text' INTERLINE_EQUATION = 'interline_equation' INLINE_EQUATION = 'inline_equation' EQUATION = 'equation' HYPERLINK = 'hyperlink' class MakeMode: """输出格式模式""" MM_MD = 'mm_markdown' NLP_MD = 'nlp_markdown' CONTENT_LIST = 'content_list' CONTENT_LIST_V2 = 'content_list_v2' class ModelPath: """模型权重路径 — HuggingFace / ModelScope""" vlm_root_hf = "opendatalab/MinerU2.5-Pro-2605-1.2B" vlm_root_modelscope = "OpenDataLab/MinerU2.5-Pro-2605-1.2B" pipeline_root_modelscope = "OpenDataLab/PDF-Extract-Kit-1.0" pp_doclayout_v2 = "models/Layout/PP-DocLayoutV2" unimernet_small = "models/MFR/unimernet_hf_small_2503" pp_formulanet_plus_m = "models/MFR/pp_formulanet_plus_m" pytorch_paddle = "models/OCR/paddleocr_torch" slanet_plus = "models/TabRec/SLANetPlus/slanet-plus.onnx" unet_structure = "models/TabRec/UnetStructure/unet.onnx"
CLI 客户端是用户交互的主入口,负责任务规划、文档分类、引擎调度和结果输出。
mineru/cli/client.py# Copyright (c) Opendatalab. All rights reserved. import asyncio, multiprocessing, os, sys, threading from concurrent.futures import Future, ProcessPoolExecutor from dataclasses import dataclass from pathlib import Path from typing import Awaitable, Callable, Optional, TextIO import click import httpx import pypdfium2 as pdfium from loguru import logger from mineru.cli.common import ( HybridDependencyError, ensure_backend_dependencies, image_suffixes, office_suffixes, pdf_suffixes, uniquify_task_stems, ) @dataclass(frozen=True) class InputDocument: path: Path suffix: str stem: str effective_pages: int order: int @dataclass class PlannedTask: index: int documents: list[InputDocument] total_pages: int @dataclass class TaskExecutionProgress: total_tasks: int total_pages: int completed_tasks: int completed_pages: int lock: asyncio.Lock # 实时任务状态渲染器 — 支持终端进度条显示 class LiveTaskStatusRenderer: BAR_WIDTH = 12 RUNNER_WIDTH = 4 ACTIVE_STATUSES = {"pending", "processing"} def register_task(self, task: PlannedTask, task_id: str, ...): """注册新任务到渲染器""" ... def update_status(self, task_id: str, status_update, ...): """更新任务状态并重新渲染""" ...
下表展示了各核心模块之间的直接依赖关系。
| 共享模块 | 被依赖次数 | 主要消费者 |
|---|---|---|
| enum_class (BlockType/ContentType) | 所有引擎 | Pipeline, VLM, Hybrid, Office, Utils |
| config_reader | 所有引擎 | model_init, pipeline_analyze, vlm_analyze, hybrid_analyze, client |
| pdfium_guard | PDF 引擎 | pipeline_analyze, vlm_analyze, hybrid_analyze, client |
| pdf_classify | PDF 引擎 | pipeline_analyze, hybrid_analyze |
| pdf_image_tools | PDF 引擎 | pipeline_analyze, vlm_analyze, hybrid_analyze |
| DataReader/Writer | 所有引擎 | vlm_analyze, hybrid_analyze, fast_api |
| ocr_utils | Pipeline+Hybrid | model_json_to_middle_json, hybrid_analyze |
| magic_model_utils | Pipeline+Office | pipeline_magic_model, office_magic_model |
尽管三种 PDF 解析引擎的内部实现不同,但它们都遵循相同的 Model → Middle JSON → Content 三阶段范式。
Middle JSON 结构示意{ "pdf_info": [ // 每页一个 page_info { "page_no": 0, "width": 595, "height": 842, "preproc_blocks": [ // 预处理后的 Block 列表 { "type": "text", // BlockType "lines": [ // Line → Span 结构 { "spans": [ { "type": "text", "content": "..." }, { "type": "inline_equation", "content": "E=mc^2" } ] } ], "index": 0 }, { "type": "table_body", "lines": [{ "spans": [{ "type": "table", "html": "<table>..." }] }] } ], "discarded_blocks": [...] // 页眉/页脚/页码等 } ] }
ModelSingleton 管理组合模型,AtomModelSingleton 管理原子模型。线程安全懒加载,按参数组合缓存实例。
PIPELINE_*_INFERENCE_LOCK 保护 Pipeline/Hybrid 共享的模型推理调用,防止多线程并发进入同一模型对象。
doc_analyze_streaming 按处理窗口分批加载页面,控制 GPU 显存占用,支持超长文档处理。
Middle JSON 作为三种引擎的统一输出格式,解耦模型推理与内容生成,支持灵活的后处理流水线。
DataReader/DataWriter ABC 定义统一接口,FileBase/S3/MultiBucketS3 实现可插拔存储后端。
所有引擎输出统一的 Block(Line(Span)) 层次结构,Span 是最小内容单元(文本/公式/图片/表格)。