智能文档处理框架，结合版面分析与视觉语言模型（VLM）实现文档理解与内容提取。
- 模块化架构：可插拔的版面分析器、VLM 分析器和内容处理器
- 多模型融合：支持同时运行多个版面检测模型，通过 IoU 和优先级智能融合结果
- 异步设计：所有 I/O 和推理操作均为异步，支持高并发处理
- 灵活的 Pipeline：通过组合不同的 ProcessingStep 构建自定义处理流程
- 多层内容提取：针对不同类型区域（文本/表格/公式/图片）使用专门的处理器
- 阅读顺序恢复：使用 XY-cut 算法恢复文档的自然阅读顺序
- Debug 可视化：生成 HTML 调试页面，支持原图与识别结果的对比查看
# 安装 uv（如未安装）
curl -LsSf https://astral.sh/uv/install.sh | sh
# 克隆并安装项目
cd oculodoc
uv sync
# 安装开发依赖
uv sync --group dev
# 安装 OpenAI SDK（用于 VLM）
uv pip install openai# 基础处理（仅版面分析）
python scripts/multi_layer_example.py document.pdf
# 使用 VLM 进行完整处理
python scripts/multi_layer_example.py document.pdf \
--vlm-base-url https://api.openai.com/v1 \
--vlm-api-key $OPENAI_API_KEY \
--vlm-model gpt-4o
# 指定输出目录
python scripts/multi_layer_example.py document.pdf -o output/import asyncio
from pathlib import Path
from oculodoc.config import OculodocConfig, MultiModelConfig, RegionAnalyzerSpec
from oculodoc.analyzers import create_and_initialize_analyzers, cleanup_analyzers
from oculodoc.processor import ProcessingPipeline
from oculodoc.processor.steps import MultiModelDetectionStep, RegionFusionStep, RegionFilterStep
from oculodoc.content_extraction import MultiLayerExtractionStep
from oculodoc.vlm import OpenAICompatibleVLMAnalyzer
async def process_document(pdf_path: Path):
# 1. 配置版面分析模型
multi_config = MultiModelConfig(
analyzers=[
RegionAnalyzerSpec(
enabled=True,
model_type="layout",
model_name="PP-DocLayout-L",
priority=1,
),
],
)
# 2. 初始化分析器
analyzers = await create_and_initialize_analyzers(multi_config)
# 3. 配置 VLM（可选）
vlm_analyzer = OpenAICompatibleVLMAnalyzer(
base_url="https://api.openai.com/v1",
api_key="your-api-key",
model="gpt-4o",
)
await vlm_analyzer.initialize({})
# 4. 构建 Pipeline
pipeline = ProcessingPipeline(steps=[
MultiModelDetectionStep(analyzers=analyzers),
RegionFusionStep(),
RegionFilterStep(min_confidence=0.3),
MultiLayerExtractionStep.create_default(),
])
# 5. 处理文档
config = OculodocConfig()
result = await pipeline.process_document(
pdf_path,
config,
vlm_analyzer=vlm_analyzer,
)
print(f"处理了 {result.total_pages} 页，耗时 {result.processing_time:.2f}s")
# 6. 清理资源
await cleanup_analyzers(analyzers)
await vlm_analyzer.cleanup()
return result
# 运行
asyncio.run(process_document(Path("document.pdf")))┌─────────────────────────────────────────────────────────────────┐
│ ProcessingPipeline │
├─────────────────────────────────────────────────────────────────┤
│ Step 1: MultiModelDetectionStep │
│ └── 运行多个 IRegionAnalyzer，收集所有检测结果 │
├─────────────────────────────────────────────────────────────────┤
│ Step 2: RegionFusionStep │
│ └── 基于 IoU 和优先级融合重叠区域 │
├─────────────────────────────────────────────────────────────────┤
│ Step 3: RegionFilterStep │
│ └── 过滤低置信度或不需要的区域类型 │
├─────────────────────────────────────────────────────────────────┤
│ Step 4: MultiLayerExtractionStep │
│ └── 按类型路由到不同的 IContentProcessor │
│ ├── TextContentProcessor (文本 → VLM OCR) │
│ ├── TableContentProcessor (表格 → PaddleOCR/VLM) │
│ ├── FormulaContentProcessor (公式 → PaddleOCR/VLM) │
│ └── FigureContentProcessor (图片 → VLM 描述) │
├─────────────────────────────────────────────────────────────────┤
│ Step 5: ReadingOrderStep (可选) │
│ └── 使用 XY-cut 算法恢复阅读顺序 │
└─────────────────────────────────────────────────────────────────┘
oculodoc/
├── interfaces/ # 抽象接口定义
│ ├── layout_analyzer.py # ILayoutAnalyzer, LayoutDetection
│ ├── vlm_analyzer.py # IVLMAnalyzer, VLMAnalysisResult
│ ├── region_analyzer.py # IRegionAnalyzer（通用区域分析器）
│ └── document_processor.py
├── config/ # 配置管理
│ ├── schema.py # 配置数据类
│ └── loader.py # 配置加载器
├── layout/ # 版面分析实现
│ ├── doclayout_yolo_analyzer.py # DocLayout-YOLO
│ └── paddle_layout_analyzer.py # PaddleX PP-DocLayout
├── vlm/ # VLM 实现
│ ├── openai_compatible_analyzer.py # OpenAI 兼容 API
│ └── sglang_ocrflux_analyzer.py # SGLang OCRFlux
├── analyzers/ # 专用分析器
│ ├── paddle_formula_analyzer.py # 公式识别
│ └── paddle_table_analyzer.py # 表格结构识别
├── content_extraction/ # 内容提取模块
│ ├── processors.py # IContentProcessor 实现
│ ├── fusion.py # 多层结果融合
│ ├── step.py # MultiLayerExtractionStep
│ └── formatters.py # 输出格式转换
├── processor/ # 处理器
│ ├── pipeline.py # ProcessingPipeline
│ └── steps/ # Pipeline 步骤
│ ├── base.py # ProcessingStep, PageData, ProcessingContext
│ ├── layout.py # LayoutDetectionStep
│ ├── multi_model.py # MultiModelDetectionStep
│ ├── fusion.py # RegionFusionStep
│ ├── filter.py # RegionFilterStep
│ ├── vlm.py # VLMExtractionStep
│ └── reading_order.py # ReadingOrderStep
├── debug/ # 调试工具
│ └── html_generator.py # Debug HTML 生成器
├── utils/ # 工具函数
│ └── reading_order.py # XY-cut 阅读顺序算法
└── errors/ # 异常定义
└── exceptions.py
支持任何 OpenAI 兼容的 API：
| 服务 | base_url | 示例 model |
|---|---|---|
| OpenAI | https://api.openai.com/v1 |
gpt-4o, gpt-4o-mini |
| 火山引擎 ARK | https://ark.cn-beijing.volces.com/api/v3 |
doubao-seed-1-6-251015 |
| 本地 vLLM | http://localhost:8000/v1 |
Qwen/Qwen2-VL-7B-Instruct |
| 本地 SGLang | http://localhost:30000/v1 |
你的模型名 |
| 模型类型 | model_name | 说明 |
|---|---|---|
layout |
PP-DocLayout-L |
PaddleX 大模型，精度高 |
layout |
PP-DocLayout-M |
PaddleX 中模型，平衡 |
layout |
PP-DocLayout-S |
PaddleX 小模型，速度快 |
layout |
自定义路径 | DocLayout-YOLO 自训练模型 |
from oculodoc.config import MultiModelConfig, RegionAnalyzerSpec, RegionFusionConfig
config = MultiModelConfig(
analyzers=[
# 通用版面分析（低优先级）
RegionAnalyzerSpec(
enabled=True,
model_type="layout",
model_name="PP-DocLayout-L",
priority=1,
),
# 公式识别（高优先级）
RegionAnalyzerSpec(
enabled=True,
model_type="formula",
model_name="PP-FormulaNet-L",
priority=10,
),
],
fusion=RegionFusionConfig(
iou_threshold=0.5, # IoU 阈值
prefer_specialized=True, # 优先使用专用模型结果
containment_threshold=0.8, # 包含关系阈值
),
parallel_inference=True,
)from oculodoc.content_extraction import OutputFormatter
markdown = OutputFormatter.to_markdown(detections)html = OutputFormatter.to_html(detections, include_styles=True)json_data = OutputFormatter.to_structured_json(detections)生成按页调试 HTML，便于对比原图与识别结果：
from oculodoc.debug import generate_debug_html_per_page
# 生成按页的调试 HTML
debug_files = generate_debug_html_per_page(
result=processing_result,
document_path=Path("document.pdf"),
output_dir=Path("output/"), # 生成 debug_page_1.html, debug_page_2.html, ...
)
# 或者生成单个合并的 HTML（所有页面在一个文件中）
from oculodoc.debug import generate_debug_html
generate_debug_html(
result=processing_result,
document_path=Path("document.pdf"),
output_path=Path("debug.html"),
)CLI 使用：
# 处理文档时自动生成按页调试 HTML
python scripts/multi_layer_example.py document.pdf
# 仅运行版面分析并生成调试 HTML
python scripts/debug_layout.py document.pdf -o output/debug
# 通过 CLI 生成调试 HTML
python -m oculodoc debug --input middle.json --pdf document.pdf --output output/ --per-pagefrom oculodoc.processor.steps.base import ProcessingStep, PageData, ProcessingContext
class MyCustomStep(ProcessingStep):
async def process(self, page: PageData, ctx: ProcessingContext) -> PageData:
# 访问版面检测结果
for detection in page.layout_elements:
print(f"Category: {detection.category_name}")
print(f"BBox: {detection.bbox}")
print(f"Content: {detection.metadata.get('content', '')}")
# 修改或添加检测结果
page.layout_elements = [...]
return pagefrom oculodoc.content_extraction.processors import (
IContentProcessor, ContentResult, ContentFormat
)
class MyContentProcessor(IContentProcessor):
@property
def supported_categories(self) -> list[str]:
return ["my_category"]
@property
def processor_name(self) -> str:
return "MyContentProcessor"
async def process(
self,
image: Image.Image,
category: str,
ctx: ProcessingContext,
**kwargs,
) -> ContentResult:
# 你的处理逻辑
content = await my_extraction_logic(image)
return ContentResult(
content=content,
content_format=ContentFormat.MARKDOWN,
confidence=0.9,
)from oculodoc.interfaces import ILayoutAnalyzer, LayoutDetection, LayoutAnalyzerOptions
class MyLayoutAnalyzer(ILayoutAnalyzer):
async def initialize(self, config: dict) -> None:
# 加载模型
pass
async def analyze(
self,
image: Image.Image,
options: LayoutAnalyzerOptions | None = None,
**kwargs,
) -> list[LayoutDetection]:
# 运行推理
results = []
# ...
return results
async def cleanup(self) -> None:
# 释放资源
pass
@property
def supported_categories(self) -> dict[int, str]:
return {
0: "text",
1: "title",
2: "table",
# ...
}from oculodoc.interfaces import IVLMAnalyzer, VLMAnalysisResult
class MyVLMAnalyzer(IVLMAnalyzer):
async def initialize(self, config: dict) -> None:
pass
async def analyze(
self,
image_data: str, # base64 encoded
prompt: str,
**kwargs,
) -> list[VLMAnalysisResult]:
# 调用你的 VLM API
response = await my_vlm_api(image_data, prompt)
return [VLMAnalysisResult(
content_type="text",
content=response,
confidence=0.9,
)]
async def batch_analyze(
self,
image_data_list: list[str],
prompts: list[str],
**kwargs,
) -> list[list[VLMAnalysisResult]]:
# 批量处理
pass# 运行所有测试
uv run pytest
# 运行带覆盖率报告
uv run pytest --cov=oculodoc
# 运行特定测试文件
uv run pytest oculodoc/tests/test_pipeline_steps.py安装 PaddleOCR：
uv pip install paddleocr paddlex确保安装了 OpenAI SDK：
uv pip install openaiPaddleX 会自动检测 GPU。确保安装了正确的 CUDA 版本：
# 检查 PaddlePaddle GPU 支持
python -c "import paddle; print(paddle.device.is_compiled_with_cuda())"MIT