Merge pull request #1659 from HKUDS/mineru_integration

MinerU integration

README-zh.md

@@ -4,6 +4,7 @@
 
 ## 🎉 新闻
 
+- [X] [2025.06.05]🎯📢LightRAG现已集成MinerU，支持多模态文档解析与RAG（PDF、图片、Office、表格、公式等）。详见下方多模态处理模块。
 - [X] [2025.03.18]🎯📢LightRAG现已支持引文功能。
 - [X] [2025.02.05]🎯📢我们团队发布了[VideoRAG](https://github.com/HKUDS/VideoRAG)，用于理解超长上下文视频。
 - [X] [2025.01.13]🎯📢我们团队发布了[MiniRAG](https://github.com/HKUDS/MiniRAG)，使用小型模型简化RAG。
@@ -1002,6 +1003,32 @@ rag.merge_entities(
 
 </details>
 
+## 多模态文档处理（MinerU集成）
+
+LightRAG 现已支持通过 [MinerU](https://github.com/opendatalab/MinerU) 实现多模态文档解析与检索增强生成（RAG）。您可以从 PDF、图片、Office 文档中提取结构化内容（文本、图片、表格、公式等），并在 RAG 流程中使用。
+
+**主要特性：**
+- 支持解析 PDF、图片、DOC/DOCX/PPT/PPTX 等多种格式
+- 提取并索引文本、图片、表格、公式及文档结构
+- 在 RAG 中查询和检索多模态内容（文本、图片、表格、公式）
+- 与 LightRAG Core 及 RAGAnything 无缝集成
+
+**快速开始：**
+1. 安装依赖：
+   ```bash
+   pip install "magic-pdf[full]>=1.2.2" huggingface_hub
+   ```
+2. 下载 MinerU 模型权重（详见 [MinerU 集成指南](docs/mineru_integration_zh.md)）
+3. 使用新版 `MineruParser` 或 RAGAnything 的 `process_document_complete` 处理文件：
+   ```python
+   from lightrag.mineru_parser import MineruParser
+   content_list, md_content = MineruParser.parse_pdf('path/to/document.pdf', 'output_dir')
+   # 或自动识别类型：
+   content_list, md_content = MineruParser.parse_document('path/to/file', 'auto', 'output_dir')
+   ```
+4. 使用 LightRAG 查询多模态内容请参见 [docs/mineru_integration_zh.md](docs/mineru_integration_zh.md)。
+
+
 ## Token统计功能
 
 <details>

README.md

@@ -39,7 +39,7 @@
 </div>
 
 ## 🎉 News
-
+- [X] [2025.06.05]🎯📢LightRAG now supports multimodal document parsing and RAG with MinerU integration (PDF, images, Office, tables, formulas, etc.). See the new multimodal section below.
 - [X] [2025.03.18]🎯📢LightRAG now supports citation functionality, enabling proper source attribution.
 - [X] [2025.02.05]🎯📢Our team has released [VideoRAG](https://github.com/HKUDS/VideoRAG) understanding extremely long-context videos.
 - [X] [2025.01.13]🎯📢Our team has released [MiniRAG](https://github.com/HKUDS/MiniRAG) making RAG simpler with small models.
@@ -1051,6 +1051,31 @@ When merging entities:
 
 </details>
 
+## Multimodal Document Processing (MinerU Integration)
+
+LightRAG now supports multimodal document parsing and retrieval-augmented generation (RAG) via [MinerU](https://github.com/opendatalab/MinerU). You can extract structured content (text, images, tables, formulas, etc.) from PDF, images, and Office documents, and use them in your RAG pipeline.
+
+**Key Features:**
+- Parse PDFs, images, DOC/DOCX/PPT/PPTX, and more
+- Extract and index text, images, tables, formulas, and document structure
+- Query and retrieve multimodal content (text, image, table, formula) in RAG
+- Seamless integration with LightRAG core and RAGAnything
+
+**Quick Start:**
+1. Install dependencies:
+   ```bash
+   pip install "magic-pdf[full]>=1.2.2" huggingface_hub
+   ```
+2. Download MinerU model weights (see [MinerU Integration Guide](docs/mineru_integration_en.md))
+3. Use the new `MineruParser` or RAGAnything's `process_document_complete` to process files:
+   ```python
+   from lightrag.mineru_parser import MineruParser
+   content_list, md_content = MineruParser.parse_pdf('path/to/document.pdf', 'output_dir')
+   # or for any file type:
+   content_list, md_content = MineruParser.parse_document('path/to/file', 'auto', 'output_dir')
+   ```
+4. Query multimodal content with LightRAG see [docs/mineru_integration_en.md](docs/mineru_integration_en.md).
+
 ## Token Usage Tracking
 
 <details>

docs/mineru_integration_en.md

@@ -0,0 +1,360 @@
+# MinerU Integration Guide
+
+### About MinerU
+
+MinerU is a powerful open-source tool for extracting high-quality structured data from PDF, image, and office documents. It provides the following features:
+
+- Text extraction while preserving document structure (headings, paragraphs, lists, etc.)
+- Handling complex layouts including multi-column formats
+- Automatic formula recognition and conversion to LaTeX format
+- Image, table, and footnote extraction
+- Automatic scanned document detection and OCR application
+- Support for multiple output formats (Markdown, JSON)
+
+### Installation
+
+#### Installing MinerU Dependencies
+
+If you have already installed LightRAG but don't have MinerU support, you can add MinerU support by installing the magic-pdf package directly:
+
+```bash
+pip install "magic-pdf[full]>=1.2.2" huggingface_hub
+```
+
+These are the MinerU-related dependencies required by LightRAG.
+
+#### MinerU Model Weights
+
+MinerU requires model weight files to function properly. After installation, you need to download the required model weights. You can use either Hugging Face or ModelScope to download the models.
+
+##### Option 1: Download from Hugging Face
+
+```bash
+pip install huggingface_hub
+wget https://github.com/opendatalab/MinerU/raw/master/scripts/download_models_hf.py -O download_models_hf.py
+python download_models_hf.py
+```
+
+##### Option 2: Download from ModelScope (Recommended for users in China)
+
+```bash
+pip install modelscope
+wget https://github.com/opendatalab/MinerU/raw/master/scripts/download_models.py -O download_models.py
+python download_models.py
+```
+
+Both methods will automatically download the model files and configure the model directory in the configuration file. The configuration file is located in your user directory and named `magic-pdf.json`.
+
+> **Note for Windows users**: User directory is at `C:\Users\username`
+> **Note for Linux users**: User directory is at `/home/username`
+> **Note for macOS users**: User directory is at `/Users/username`
+
+#### Optional: LibreOffice Installation
+
+To process Office documents (DOC, DOCX, PPT, PPTX), you need to install LibreOffice:
+
+**Linux/macOS:**
+```bash
+apt-get/yum/brew install libreoffice
+```
+
+**Windows:**
+1. Install LibreOffice
+2. Add the installation directory to your PATH: `install_dir\LibreOffice\program`
+
+### Using MinerU Parser
+
+#### Basic Usage
+
+```python
+from lightrag.mineru_parser import MineruParser
+
+# Parse a PDF document
+content_list, md_content = MineruParser.parse_pdf('path/to/document.pdf', 'output_dir')
+
+# Parse an image
+content_list, md_content = MineruParser.parse_image('path/to/image.jpg', 'output_dir')
+
+# Parse an Office document
+content_list, md_content = MineruParser.parse_office_doc('path/to/document.docx', 'output_dir')
+
+# Auto-detect and parse any supported document type
+content_list, md_content = MineruParser.parse_document('path/to/file', 'auto', 'output_dir')
+```
+
+#### RAGAnything Integration
+
+In RAGAnything, you can directly use file paths as input to the `process_document_complete` method to process documents. Here's a complete configuration example:
+
+```python
+from lightrag.llm.openai import openai_complete_if_cache, openai_embed
+from lightrag.raganything import RAGAnything
+
+
+# Initialize RAGAnything
+rag = RAGAnything(
+    working_dir="./rag_storage",  # Working directory
+    llm_model_func=lambda prompt, system_prompt=None, history_messages=[], **kwargs: openai_complete_if_cache(
+        "gpt-4o-mini",  # Model to use
+        prompt,
+        system_prompt=system_prompt,
+        history_messages=history_messages,
+        api_key="your-api-key",  # Replace with your API key
+        base_url="your-base-url",  # Replace with your API base URL
+        **kwargs,
+    ),
+    vision_model_func=lambda prompt, system_prompt=None, history_messages=[], image_data=None, **kwargs: openai_complete_if_cache(
+        "gpt-4o",  # Vision model
+        "",
+        system_prompt=None,
+        history_messages=[],
+        messages=[
+            {"role": "system", "content": system_prompt} if system_prompt else None,
+            {"role": "user", "content": [
+                {"type": "text", "text": prompt},
+                {
+                    "type": "image_url",
+                    "image_url": {
+                        "url": f"data:image/jpeg;base64,{image_data}"
+                    }
+                }
+            ]} if image_data else {"role": "user", "content": prompt}
+        ],
+        api_key="your-api-key",  # Replace with your API key
+        base_url="your-base-url",  # Replace with your API base URL
+        **kwargs,
+    ) if image_data else openai_complete_if_cache(
+        "gpt-4o-mini",
+        prompt,
+        system_prompt=system_prompt,
+        history_messages=history_messages,
+        api_key="your-api-key",  # Replace with your API key
+        base_url="your-base-url",  # Replace with your API base URL
+        **kwargs,
+    ),
+    embedding_func=lambda texts: openai_embed(
+        texts,
+        model="text-embedding-3-large",
+        api_key="your-api-key",  # Replace with your API key
+        base_url="your-base-url",  # Replace with your API base URL
+    ),
+    embedding_dim=3072,
+    max_token_size=8192
+)
+
+# Process a single file
+await rag.process_document_complete(
+    file_path="path/to/document.pdf",
+    output_dir="./output",
+    parse_method="auto"
+)
+
+# Query the processed document
+result = await rag.query_with_multimodal(
+    "What is the main content of the document?",
+    mode="hybrid"
+)
+
+```
+
+MinerU categorizes document content into text, formulas, images, and tables, processing each with its corresponding ingestion type:
+- Text content: `ingestion_type='text'`
+- Image content: `ingestion_type='image'`
+- Table content: `ingestion_type='table'`
+- Formula content: `ingestion_type='equation'`
+
+#### Query Examples
+
+Here are some common query examples:
+
+```python
+# Query text content
+result = await rag.query_with_multimodal(
+    "What is the main topic of the document?",
+    mode="hybrid"
+)
+
+# Query image-related content
+result = await rag.query_with_multimodal(
+    "Describe the images and figures in the document",
+    mode="hybrid"
+)
+
+# Query table-related content
+result = await rag.query_with_multimodal(
+    "Tell me about the experimental results and data tables",
+    mode="hybrid"
+)
+```
+
+#### Command Line Tool
+
+We also provide a command-line tool for document parsing:
+
+```bash
+python examples/mineru_example.py path/to/document.pdf
+```
+
+Optional parameters:
+- `--output` or `-o`: Specify output directory
+- `--method` or `-m`: Choose parsing method (auto, ocr, txt)
+- `--stats`: Display content statistics
+
+### Output Format
+
+MinerU generates three files for each parsed document:
+
+1. `{filename}.md` - Markdown representation of the document
+2. `{filename}_content_list.json` - Structured JSON content
+3. `{filename}_model.json` - Detailed model parsing results
+
+The `content_list.json` file contains all structured content extracted from the document, including:
+- Text blocks (body text, headings, etc.)
+- Images (paths and optional captions)
+- Tables (table content and optional captions)
+- Lists
+- Formulas
+
+### Troubleshooting
+
+If you encounter issues with MinerU:
+
+1. Check that model weights are correctly downloaded
+2. Ensure you have sufficient RAM (16GB+ recommended)
+3. For CUDA acceleration issues, see [MinerU documentation](https://mineru.readthedocs.io/en/latest/additional_notes/faq.html)
+4. If parsing Office documents fails, verify LibreOffice is properly installed
+5. If you encounter `pickle.UnpicklingError: invalid load key, 'v'.`, it might be due to an incomplete model download. Try re-downloading the models.
+6. For users with newer graphics cards (H100, etc.) and garbled OCR text, try upgrading the CUDA version used by Paddle:
+   ```bash
+   pip install paddlepaddle-gpu==3.0.0b1 -i https://www.paddlepaddle.org.cn/packages/stable/cu123/
+   ```
+7. If you encounter a "filename too long" error, the latest version of MineruParser includes logic to automatically handle this issue.
+
+#### Updating Existing Models
+
+If you have previously downloaded models and need to update them, you can simply run the download script again. The script will update the model directory to the latest version.
+
+### Advanced Configuration
+
+The MinerU configuration file `magic-pdf.json` supports various customization options, including:
+
+- Model directory path
+- OCR engine selection
+- GPU acceleration settings
+- Cache settings
+
+For complete configuration options, refer to the [MinerU official documentation](https://mineru.readthedocs.io/).
+
+### Using Modal Processors Directly
+
+You can also use LightRAG's modal processors directly without going through MinerU. This is useful when you want to process specific types of content or have more control over the processing pipeline.
+
+Each modal processor returns a tuple containing:
+1. A description of the processed content
+2. Entity information that can be used for further processing or storage
+
+The processors support different types of content:
+- `ImageModalProcessor`: Processes images with captions and footnotes
+- `TableModalProcessor`: Processes tables with captions and footnotes
+- `EquationModalProcessor`: Processes mathematical equations in LaTeX format
+- `GenericModalProcessor`: A base processor that can be extended for custom content types
+
+> **Note**: A complete working example can be found in `examples/modalprocessors_example.py`. You can run it using:
+> ```bash
+> python examples/modalprocessors_example.py --api-key YOUR_API_KEY
+> ```
+
+<details>
+<summary> Here's an example of how to use different modal processors: </summary>
+
+```python
+from lightrag.modalprocessors import (
+    ImageModalProcessor,
+    TableModalProcessor,
+    EquationModalProcessor,
+    GenericModalProcessor
+)
+
+# Initialize LightRAG
+lightrag = LightRAG(
+    working_dir="./rag_storage",
+    embedding_func=lambda texts: openai_embed(
+        texts,
+        model="text-embedding-3-large",
+        api_key="your-api-key",
+        base_url="your-base-url",
+    ),
+    llm_model_func=lambda prompt, system_prompt=None, history_messages=[], **kwargs: openai_complete_if_cache(
+        "gpt-4o-mini",
+        prompt,
+        system_prompt=system_prompt,
+        history_messages=history_messages,
+        api_key="your-api-key",
+        base_url="your-base-url",
+        **kwargs,
+    ),
+)
+
+# Process an image
+image_processor = ImageModalProcessor(
+    lightrag=lightrag,
+    modal_caption_func=vision_model_func
+)
+
+image_content = {
+    "img_path": "image.jpg",
+    "img_caption": ["Example image caption"],
+    "img_footnote": ["Example image footnote"]
+}
+
+description, entity_info = await image_processor.process_multimodal_content(
+    modal_content=image_content,
+    content_type="image",
+    file_path="image_example.jpg",
+    entity_name="Example Image"
+)
+
+# Process a table
+table_processor = TableModalProcessor(
+    lightrag=lightrag,
+    modal_caption_func=llm_model_func
+)
+
+table_content = {
+    "table_body": """
+    | Name | Age | Occupation |
+    |------|-----|------------|
+    | John | 25  | Engineer   |
+    | Mary | 30  | Designer   |
+    """,
+    "table_caption": ["Employee Information Table"],
+    "table_footnote": ["Data updated as of 2024"]
+}
+
+description, entity_info = await table_processor.process_multimodal_content(
+    modal_content=table_content,
+    content_type="table",
+    file_path="table_example.md",
+    entity_name="Employee Table"
+)
+
+# Process an equation
+equation_processor = EquationModalProcessor(
+    lightrag=lightrag,
+    modal_caption_func=llm_model_func
+)
+
+equation_content = {
+    "text": "E = mc^2",
+    "text_format": "LaTeX"
+}
+
+description, entity_info = await equation_processor.process_multimodal_content(
+    modal_content=equation_content,
+    content_type="equation",
+    file_path="equation_example.txt",
+    entity_name="Mass-Energy Equivalence"
+)
+```
+
+</details>

docs/mineru_integration_zh.md

@@ -0,0 +1,358 @@
+# MinerU 集成指南
+
+### 关于 MinerU
+
+MinerU 是一个强大的开源工具，用于从 PDF、图像和 Office 文档中提取高质量的结构化数据。它提供以下功能：
+
+- 保留文档结构（标题、段落、列表等）的文本提取
+- 处理包括多列格式在内的复杂布局
+- 自动识别并将公式转换为 LaTeX 格式
+- 提取图像、表格和脚注
+- 自动检测扫描文档并应用 OCR
+- 支持多种输出格式（Markdown、JSON）
+
+### 安装
+
+#### 安装 MinerU 依赖
+
+如果您已经安装了 LightRAG，但没有 MinerU 支持，您可以通过安装 magic-pdf 包来直接添加 MinerU 支持：
+
+```bash
+pip install "magic-pdf[full]>=1.2.2" huggingface_hub
+```
+
+这些是 LightRAG 所需的 MinerU 相关依赖项。
+
+#### MinerU 模型权重
+
+MinerU 需要模型权重文件才能正常运行。安装后，您需要下载所需的模型权重。您可以使用 Hugging Face 或 ModelScope 下载模型。
+
+##### 选项 1：从 Hugging Face 下载
+
+```bash
+pip install huggingface_hub
+wget https://github.com/opendatalab/MinerU/raw/master/scripts/download_models_hf.py -O download_models_hf.py
+python download_models_hf.py
+```
+
+##### 选项 2：从 ModelScope 下载（推荐中国用户使用）
+
+```bash
+pip install modelscope
+wget https://github.com/opendatalab/MinerU/raw/master/scripts/download_models.py -O download_models.py
+python download_models.py
+```
+
+两种方法都会自动下载模型文件并在配置文件中配置模型目录。配置文件位于用户目录中，名为 `magic-pdf.json`。
+
+> **Windows 用户注意**：用户目录位于 `C:\Users\用户名`
+> **Linux 用户注意**：用户目录位于 `/home/用户名`
+> **macOS 用户注意**：用户目录位于 `/Users/用户名`
+
+#### 可选：安装 LibreOffice
+
+要处理 Office 文档（DOC、DOCX、PPT、PPTX），您需要安装 LibreOffice：
+
+**Linux/macOS：**
+```bash
+apt-get/yum/brew install libreoffice
+```
+
+**Windows：**
+1. 安装 LibreOffice
+2. 将安装目录添加到 PATH 环境变量：`安装目录\LibreOffice\program`
+
+### 使用 MinerU 解析器
+
+#### 基本用法
+
+```python
+from lightrag.mineru_parser import MineruParser
+
+# 解析 PDF 文档
+content_list, md_content = MineruParser.parse_pdf('path/to/document.pdf', 'output_dir')
+
+# 解析图像
+content_list, md_content = MineruParser.parse_image('path/to/image.jpg', 'output_dir')
+
+# 解析 Office 文档
+content_list, md_content = MineruParser.parse_office_doc('path/to/document.docx', 'output_dir')
+
+# 自动检测并解析任何支持的文档类型
+content_list, md_content = MineruParser.parse_document('path/to/file', 'auto', 'output_dir')
+```
+
+#### RAGAnything 集成
+
+在 RAGAnything 中，您可以直接使用文件路径作为 `process_document_complete` 方法的输入来处理文档。以下是一个完整的配置示例：
+
+```python
+from lightrag.llm.openai import openai_complete_if_cache, openai_embed
+from lightrag.raganything import RAGAnything
+
+
+# 初始化 RAGAnything
+rag = RAGAnything(
+    working_dir="./rag_storage",  # 工作目录
+    llm_model_func=lambda prompt, system_prompt=None, history_messages=[], **kwargs: openai_complete_if_cache(
+        "gpt-4o-mini",  # 使用的模型
+        prompt,
+        system_prompt=system_prompt,
+        history_messages=history_messages,
+        api_key="your-api-key",  # 替换为您的 API 密钥
+        base_url="your-base-url",  # 替换为您的 API 基础 URL
+        **kwargs,
+    ),
+    vision_model_func=lambda prompt, system_prompt=None, history_messages=[], image_data=None, **kwargs: openai_complete_if_cache(
+        "gpt-4o",  # 视觉模型
+        "",
+        system_prompt=None,
+        history_messages=[],
+        messages=[
+            {"role": "system", "content": system_prompt} if system_prompt else None,
+            {"role": "user", "content": [
+                {"type": "text", "text": prompt},
+                {
+                    "type": "image_url",
+                    "image_url": {
+                        "url": f"data:image/jpeg;base64,{image_data}"
+                    }
+                }
+            ]} if image_data else {"role": "user", "content": prompt}
+        ],
+        api_key="your-api-key",  # 替换为您的 API 密钥
+        base_url="your-base-url",  # 替换为您的 API 基础 URL
+        **kwargs,
+    ) if image_data else openai_complete_if_cache(
+        "gpt-4o-mini",
+        prompt,
+        system_prompt=system_prompt,
+        history_messages=history_messages,
+        api_key="your-api-key",  # 替换为您的 API 密钥
+        base_url="your-base-url",  # 替换为您的 API 基础 URL
+        **kwargs,
+    ),
+    embedding_func=lambda texts: openai_embed(
+        texts,
+        model="text-embedding-3-large",
+        api_key="your-api-key",  # 替换为您的 API 密钥
+        base_url="your-base-url",  # 替换为您的 API 基础 URL
+    ),
+    embedding_dim=3072,
+    max_token_size=8192
+)
+
+# 处理单个文件
+await rag.process_document_complete(
+    file_path="path/to/document.pdf",
+    output_dir="./output",
+    parse_method="auto"
+)
+
+# 查询处理后的文档
+result = await rag.query_with_multimodal(
+    "What is the main content of the document?",
+    mode="hybrid"
+)
+```
+
+MinerU 会将文档内容分类为文本、公式、图像和表格，分别使用相应的摄入类型进行处理：
+- 文本内容：`ingestion_type='text'`
+- 图像内容：`ingestion_type='image'`
+- 表格内容：`ingestion_type='table'`
+- 公式内容：`ingestion_type='equation'`
+
+#### 查询示例
+
+以下是一些常见的查询示例：
+
+```python
+# 查询文本内容
+result = await rag.query_with_multimodal(
+    "What is the main topic of the document?",
+    mode="hybrid"
+)
+
+# 查询图片相关内容
+result = await rag.query_with_multimodal(
+    "Describe the images and figures in the document",
+    mode="hybrid"
+)
+
+# 查询表格相关内容
+result = await rag.query_with_multimodal(
+    "Tell me about the experimental results and data tables",
+    mode="hybrid"
+)
+```
+
+#### 命令行工具
+
+我们还提供了一个用于文档解析的命令行工具：
+
+```bash
+python examples/mineru_example.py path/to/document.pdf
+```
+
+可选参数：
+- `--output` 或 `-o`：指定输出目录
+- `--method` 或 `-m`：选择解析方法（auto、ocr、txt）
+- `--stats`：显示内容统计信息
+
+### 输出格式
+
+MinerU 为每个解析的文档生成三个文件：
+
+1. `{文件名}.md` - 文档的 Markdown 表示
+2. `{文件名}_content_list.json` - 结构化 JSON 内容
+3. `{文件名}_model.json` - 详细的模型解析结果
+
+`content_list.json` 文件包含从文档中提取的所有结构化内容，包括：
+- 文本块（正文、标题等）
+- 图像（路径和可选的标题）
+- 表格（表格内容和可选的标题）
+- 列表
+- 公式
+
+### 疑难解答
+
+如果您在使用 MinerU 时遇到问题：
+
+1. 检查模型权重是否正确下载
+2. 确保有足够的内存（建议 16GB+）
+3. 对于 CUDA 加速问题，请参阅 [MinerU 文档](https://mineru.readthedocs.io/en/latest/additional_notes/faq.html)
+4. 如果解析 Office 文档失败，请验证 LibreOffice 是否正确安装
+5. 如果遇到 `pickle.UnpicklingError: invalid load key, 'v'.`，可能是因为模型下载不完整。尝试重新下载模型。
+6. 对于使用较新显卡（H100 等）并出现 OCR 文本乱码的用户，请尝试升级 Paddle 使用的 CUDA 版本：
+   ```bash
+   pip install paddlepaddle-gpu==3.0.0b1 -i https://www.paddlepaddle.org.cn/packages/stable/cu123/
+   ```
+7. 如果遇到 "文件名太长" 错误，最新版本的 MineruParser 已经包含了自动处理此问题的逻辑。
+
+#### 更新现有模型
+
+如果您之前已经下载了模型并需要更新它们，只需再次运行下载脚本即可。脚本将更新模型目录到最新版本。
+
+### 高级配置
+
+MinerU 配置文件 `magic-pdf.json` 支持多种自定义选项，包括：
+
+- 模型目录路径
+- OCR 引擎选择
+- GPU 加速设置
+- 缓存设置
+
+有关完整的配置选项，请参阅 [MinerU 官方文档](https://mineru.readthedocs.io/)。
+
+### 直接使用模态处理器
+
+您也可以直接使用 LightRAG 的模态处理器，而不需要通过 MinerU。这在您想要处理特定类型的内容或对处理流程有更多控制时特别有用。
+
+每个模态处理器都会返回一个包含以下内容的元组：
+1. 处理后内容的描述
+2. 可用于进一步处理或存储的实体信息
+
+处理器支持不同类型的内容：
+- `ImageModalProcessor`：处理带有标题和脚注的图像
+- `TableModalProcessor`：处理带有标题和脚注的表格
+- `EquationModalProcessor`：处理 LaTeX 格式的数学公式
+- `GenericModalProcessor`：可用于扩展自定义内容类型的基础处理器
+
+> **注意**：完整的可运行示例可以在 `examples/modalprocessors_example.py` 中找到。您可以使用以下命令运行它：
+> ```bash
+> python examples/modalprocessors_example.py --api-key YOUR_API_KEY
+> ```
+
+<details>
+<summary> 使用不同模态处理器的示例 </summary>
+
+```python
+from lightrag.modalprocessors import (
+    ImageModalProcessor,
+    TableModalProcessor,
+    EquationModalProcessor,
+    GenericModalProcessor
+)
+
+# 初始化 LightRAG
+lightrag = LightRAG(
+    working_dir="./rag_storage",
+    embedding_func=lambda texts: openai_embed(
+        texts,
+        model="text-embedding-3-large",
+        api_key="your-api-key",
+        base_url="your-base-url",
+    ),
+    llm_model_func=lambda prompt, system_prompt=None, history_messages=[], **kwargs: openai_complete_if_cache(
+        "gpt-4o-mini",
+        prompt,
+        system_prompt=system_prompt,
+        history_messages=history_messages,
+        api_key="your-api-key",
+        base_url="your-base-url",
+        **kwargs,
+    ),
+)
+
+# 处理图像
+image_processor = ImageModalProcessor(
+    lightrag=lightrag,
+    modal_caption_func=vision_model_func
+)
+
+image_content = {
+    "img_path": "image.jpg",
+    "img_caption": ["示例图像标题"],
+    "img_footnote": ["示例图像脚注"]
+}
+
+description, entity_info = await image_processor.process_multimodal_content(
+    modal_content=image_content,
+    content_type="image",
+    file_path="image_example.jpg",
+    entity_name="示例图像"
+)
+
+# 处理表格
+table_processor = TableModalProcessor(
+    lightrag=lightrag,
+    modal_caption_func=llm_model_func
+)
+
+table_content = {
+    "table_body": """
+    | 姓名 | 年龄 | 职业 |
+    |------|-----|------|
+    | 张三 | 25  | 工程师 |
+    | 李四 | 30  | 设计师 |
+    """,
+    "table_caption": ["员工信息表"],
+    "table_footnote": ["数据更新至2024年"]
+}
+
+description, entity_info = await table_processor.process_multimodal_content(
+    modal_content=table_content,
+    content_type="table",
+    file_path="table_example.md",
+    entity_name="员工表格"
+)
+
+# 处理公式
+equation_processor = EquationModalProcessor(
+    lightrag=lightrag,
+    modal_caption_func=llm_model_func
+)
+
+equation_content = {
+    "text": "E = mc^2",
+    "text_format": "LaTeX"
+}
+
+description, entity_info = await equation_processor.process_multimodal_content(
+    modal_content=equation_content,
+    content_type="equation",
+    file_path="equation_example.txt",
+    entity_name="质能方程"
+)
+```
+</details>

examples/mineru_example.py

@@ -0,0 +1,85 @@
+#!/usr/bin/env python
+"""
+Example script demonstrating the basic usage of MinerU parser
+
+This example shows how to:
+1. Parse different types of documents (PDF, images, office documents)
+2. Use different parsing methods
+3. Display document statistics
+"""
+
+import os
+import argparse
+from lightrag.mineru_parser import MineruParser
+
+
+def parse_document(
+    file_path: str, output_dir: str = None, method: str = "auto", stats: bool = False
+):
+    """
+    Parse a document using MinerU parser
+
+    Args:
+        file_path: Path to the document
+        output_dir: Output directory for parsed results
+        method: Parsing method (auto, ocr, txt)
+        stats: Whether to display content statistics
+    """
+    try:
+        # Parse the document
+        content_list, md_content = MineruParser.parse_document(
+            file_path=file_path, parse_method=method, output_dir=output_dir
+        )
+
+        # Display statistics if requested
+        if stats:
+            print("\nDocument Statistics:")
+            print(f"Total content blocks: {len(content_list)}")
+
+            # Count different types of content
+            content_types = {}
+            for item in content_list:
+                content_type = item.get("type", "unknown")
+                content_types[content_type] = content_types.get(content_type, 0) + 1
+
+            print("\nContent Type Distribution:")
+            for content_type, count in content_types.items():
+                print(f"- {content_type}: {count}")
+
+        return content_list, md_content
+
+    except Exception as e:
+        print(f"Error parsing document: {str(e)}")
+        return None, None
+
+
+def main():
+    """Main function to run the example"""
+    parser = argparse.ArgumentParser(description="MinerU Parser Example")
+    parser.add_argument("file_path", help="Path to the document to parse")
+    parser.add_argument("--output", "-o", help="Output directory path")
+    parser.add_argument(
+        "--method",
+        "-m",
+        choices=["auto", "ocr", "txt"],
+        default="auto",
+        help="Parsing method (auto, ocr, txt)",
+    )
+    parser.add_argument(
+        "--stats", action="store_true", help="Display content statistics"
+    )
+
+    args = parser.parse_args()
+
+    # Create output directory if specified
+    if args.output:
+        os.makedirs(args.output, exist_ok=True)
+
+    # Parse document
+    content_list, md_content = parse_document(
+        args.file_path, args.output, args.method, args.stats
+    )
+
+
+if __name__ == "__main__":
+    main()

examples/modalprocessors_example.py

@@ -0,0 +1,224 @@
+"""
+Example of directly using modal processors
+
+This example demonstrates how to use LightRAG's modal processors directly without going through MinerU.
+"""
+
+import asyncio
+import argparse
+from lightrag.llm.openai import openai_complete_if_cache, openai_embed
+from lightrag.kg.shared_storage import initialize_pipeline_status
+from lightrag import LightRAG
+from lightrag.modalprocessors import (
+    ImageModalProcessor,
+    TableModalProcessor,
+    EquationModalProcessor,
+)
+
+WORKING_DIR = "./rag_storage"
+
+
+def get_llm_model_func(api_key: str, base_url: str = None):
+    return (
+        lambda prompt,
+        system_prompt=None,
+        history_messages=[],
+        **kwargs: openai_complete_if_cache(
+            "gpt-4o-mini",
+            prompt,
+            system_prompt=system_prompt,
+            history_messages=history_messages,
+            api_key=api_key,
+            base_url=base_url,
+            **kwargs,
+        )
+    )
+
+
+def get_vision_model_func(api_key: str, base_url: str = None):
+    return (
+        lambda prompt,
+        system_prompt=None,
+        history_messages=[],
+        image_data=None,
+        **kwargs: openai_complete_if_cache(
+            "gpt-4o",
+            "",
+            system_prompt=None,
+            history_messages=[],
+            messages=[
+                {"role": "system", "content": system_prompt} if system_prompt else None,
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": prompt},
+                        {
+                            "type": "image_url",
+                            "image_url": {
+                                "url": f"data:image/jpeg;base64,{image_data}"
+                            },
+                        },
+                    ],
+                }
+                if image_data
+                else {"role": "user", "content": prompt},
+            ],
+            api_key=api_key,
+            base_url=base_url,
+            **kwargs,
+        )
+        if image_data
+        else openai_complete_if_cache(
+            "gpt-4o-mini",
+            prompt,
+            system_prompt=system_prompt,
+            history_messages=history_messages,
+            api_key=api_key,
+            base_url=base_url,
+            **kwargs,
+        )
+    )
+
+
+async def process_image_example(lightrag: LightRAG, vision_model_func):
+    """Example of processing an image"""
+    # Create image processor
+    image_processor = ImageModalProcessor(
+        lightrag=lightrag, modal_caption_func=vision_model_func
+    )
+
+    # Prepare image content
+    image_content = {
+        "img_path": "image.jpg",
+        "img_caption": ["Example image caption"],
+        "img_footnote": ["Example image footnote"],
+    }
+
+    # Process image
+    description, entity_info = await image_processor.process_multimodal_content(
+        modal_content=image_content,
+        content_type="image",
+        file_path="image_example.jpg",
+        entity_name="Example Image",
+    )
+
+    print("Image Processing Results:")
+    print(f"Description: {description}")
+    print(f"Entity Info: {entity_info}")
+
+
+async def process_table_example(lightrag: LightRAG, llm_model_func):
+    """Example of processing a table"""
+    # Create table processor
+    table_processor = TableModalProcessor(
+        lightrag=lightrag, modal_caption_func=llm_model_func
+    )
+
+    # Prepare table content
+    table_content = {
+        "table_body": """
+        | Name | Age | Occupation |
+        |------|-----|------------|
+        | John | 25  | Engineer   |
+        | Mary | 30  | Designer   |
+        """,
+        "table_caption": ["Employee Information Table"],
+        "table_footnote": ["Data updated as of 2024"],
+    }
+
+    # Process table
+    description, entity_info = await table_processor.process_multimodal_content(
+        modal_content=table_content,
+        content_type="table",
+        file_path="table_example.md",
+        entity_name="Employee Table",
+    )
+
+    print("\nTable Processing Results:")
+    print(f"Description: {description}")
+    print(f"Entity Info: {entity_info}")
+
+
+async def process_equation_example(lightrag: LightRAG, llm_model_func):
+    """Example of processing a mathematical equation"""
+    # Create equation processor
+    equation_processor = EquationModalProcessor(
+        lightrag=lightrag, modal_caption_func=llm_model_func
+    )
+
+    # Prepare equation content
+    equation_content = {"text": "E = mc^2", "text_format": "LaTeX"}
+
+    # Process equation
+    description, entity_info = await equation_processor.process_multimodal_content(
+        modal_content=equation_content,
+        content_type="equation",
+        file_path="equation_example.txt",
+        entity_name="Mass-Energy Equivalence",
+    )
+
+    print("\nEquation Processing Results:")
+    print(f"Description: {description}")
+    print(f"Entity Info: {entity_info}")
+
+
+async def initialize_rag(api_key: str, base_url: str = None):
+    rag = LightRAG(
+        working_dir=WORKING_DIR,
+        embedding_func=lambda texts: openai_embed(
+            texts,
+            model="text-embedding-3-large",
+            api_key=api_key,
+            base_url=base_url,
+        ),
+        llm_model_func=lambda prompt,
+        system_prompt=None,
+        history_messages=[],
+        **kwargs: openai_complete_if_cache(
+            "gpt-4o-mini",
+            prompt,
+            system_prompt=system_prompt,
+            history_messages=history_messages,
+            api_key=api_key,
+            base_url=base_url,
+            **kwargs,
+        ),
+    )
+
+    await rag.initialize_storages()
+    await initialize_pipeline_status()
+
+    return rag
+
+
+def main():
+    """Main function to run the example"""
+    parser = argparse.ArgumentParser(description="Modal Processors Example")
+    parser.add_argument("--api-key", required=True, help="OpenAI API key")
+    parser.add_argument("--base-url", help="Optional base URL for API")
+    parser.add_argument(
+        "--working-dir", "-w", default=WORKING_DIR, help="Working directory path"
+    )
+
+    args = parser.parse_args()
+
+    # Run examples
+    asyncio.run(main_async(args.api_key, args.base_url))
+
+
+async def main_async(api_key: str, base_url: str = None):
+    # Initialize LightRAG
+    lightrag = await initialize_rag(api_key, base_url)
+
+    # Get model functions
+    llm_model_func = get_llm_model_func(api_key, base_url)
+    vision_model_func = get_vision_model_func(api_key, base_url)
+
+    # Run examples
+    await process_image_example(lightrag, vision_model_func)
+    await process_table_example(lightrag, llm_model_func)
+    await process_equation_example(lightrag, llm_model_func)
+
+
+if __name__ == "__main__":
+    main()

examples/raganything_example.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python
+"""
+Example script demonstrating the integration of MinerU parser with RAGAnything
+
+This example shows how to:
+1. Process parsed documents with RAGAnything
+2. Perform multimodal queries on the processed documents
+3. Handle different types of content (text, images, tables)
+"""
+
+import os
+import argparse
+import asyncio
+from lightrag.llm.openai import openai_complete_if_cache, openai_embed
+from lightrag.raganything import RAGAnything
+
+
+async def process_with_rag(
+    file_path: str,
+    output_dir: str,
+    api_key: str,
+    base_url: str = None,
+    working_dir: str = None,
+):
+    """
+    Process document with RAGAnything
+
+    Args:
+        file_path: Path to the document
+        output_dir: Output directory for RAG results
+        api_key: OpenAI API key
+        base_url: Optional base URL for API
+    """
+    try:
+        # Initialize RAGAnything
+        rag = RAGAnything(
+            working_dir=working_dir,
+            llm_model_func=lambda prompt,
+            system_prompt=None,
+            history_messages=[],
+            **kwargs: openai_complete_if_cache(
+                "gpt-4o-mini",
+                prompt,
+                system_prompt=system_prompt,
+                history_messages=history_messages,
+                api_key=api_key,
+                base_url=base_url,
+                **kwargs,
+            ),
+            vision_model_func=lambda prompt,
+            system_prompt=None,
+            history_messages=[],
+            image_data=None,
+            **kwargs: openai_complete_if_cache(
+                "gpt-4o",
+                "",
+                system_prompt=None,
+                history_messages=[],
+                messages=[
+                    {"role": "system", "content": system_prompt}
+                    if system_prompt
+                    else None,
+                    {
+                        "role": "user",
+                        "content": [
+                            {"type": "text", "text": prompt},
+                            {
+                                "type": "image_url",
+                                "image_url": {
+                                    "url": f"data:image/jpeg;base64,{image_data}"
+                                },
+                            },
+                        ],
+                    }
+                    if image_data
+                    else {"role": "user", "content": prompt},
+                ],
+                api_key=api_key,
+                base_url=base_url,
+                **kwargs,
+            )
+            if image_data
+            else openai_complete_if_cache(
+                "gpt-4o-mini",
+                prompt,
+                system_prompt=system_prompt,
+                history_messages=history_messages,
+                api_key=api_key,
+                base_url=base_url,
+                **kwargs,
+            ),
+            embedding_func=lambda texts: openai_embed(
+                texts,
+                model="text-embedding-3-large",
+                api_key=api_key,
+                base_url=base_url,
+            ),
+            embedding_dim=3072,
+            max_token_size=8192,
+        )
+
+        # Process document
+        await rag.process_document_complete(
+            file_path=file_path, output_dir=output_dir, parse_method="auto"
+        )
+
+        # Example queries
+        queries = [
+            "What is the main content of the document?",
+            "Describe the images and figures in the document",
+            "Tell me about the experimental results and data tables",
+        ]
+
+        print("\nQuerying processed document:")
+        for query in queries:
+            print(f"\nQuery: {query}")
+            result = await rag.query_with_multimodal(query, mode="hybrid")
+            print(f"Answer: {result}")
+
+    except Exception as e:
+        print(f"Error processing with RAG: {str(e)}")
+
+
+def main():
+    """Main function to run the example"""
+    parser = argparse.ArgumentParser(description="MinerU RAG Example")
+    parser.add_argument("file_path", help="Path to the document to process")
+    parser.add_argument(
+        "--working_dir", "-w", default="./rag_storage", help="Working directory path"
+    )
+    parser.add_argument(
+        "--output", "-o", default="./output", help="Output directory path"
+    )
+    parser.add_argument(
+        "--api-key", required=True, help="OpenAI API key for RAG processing"
+    )
+    parser.add_argument("--base-url", help="Optional base URL for API")
+
+    args = parser.parse_args()
+
+    # Create output directory if specified
+    if args.output:
+        os.makedirs(args.output, exist_ok=True)
+
+    # Process with RAG
+    asyncio.run(
+        process_with_rag(
+            args.file_path, args.output, args.api_key, args.base_url, args.working_dir
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

lightrag/mineru_parser.py

@@ -0,0 +1,513 @@
+# type: ignore
+"""
+MinerU Document Parser Utility
+
+This module provides functionality for parsing PDF, image and office documents using MinerU library,
+and converts the parsing results into markdown and JSON formats
+"""
+
+from __future__ import annotations
+
+__all__ = ["MineruParser"]
+
+import os
+import json
+import argparse
+from pathlib import Path
+from typing import (
+    Dict,
+    List,
+    Optional,
+    Union,
+    Tuple,
+    Any,
+    TypeVar,
+    cast,
+    TYPE_CHECKING,
+    ClassVar,
+)
+
+# Type stubs for magic_pdf
+FileBasedDataWriter = Any
+FileBasedDataReader = Any
+PymuDocDataset = Any
+InferResult = Any
+PipeResult = Any
+SupportedPdfParseMethod = Any
+doc_analyze = Any
+read_local_office = Any
+read_local_images = Any
+
+if TYPE_CHECKING:
+    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
+    from magic_pdf.data.read_api import read_local_office, read_local_images
+else:
+    # MinerU imports
+    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
+    from magic_pdf.data.read_api import read_local_office, read_local_images
+
+T = TypeVar("T")
+
+
+class MineruParser:
+    """
+    MinerU document parsing utility class
+
+    Supports parsing PDF, image and office documents (like Word, PPT, etc.),
+    converting the content into structured data and generating markdown and JSON output
+    """
+
+    __slots__: ClassVar[Tuple[str, ...]] = ()
+
+    def __init__(self) -> None:
+        """Initialize MineruParser"""
+        pass
+
+    @staticmethod
+    def safe_write(
+        writer: Any,
+        content: Union[str, bytes, Dict[str, Any], List[Any]],
+        filename: str,
+    ) -> None:
+        """
+        Safely write content to a file, ensuring the filename is valid
+
+        Args:
+            writer: The writer object to use
+            content: The content to write
+            filename: The filename to write to
+        """
+        # Ensure the filename isn't too long
+        if len(filename) > 200:  # Most filesystems have limits around 255 characters
+            # Truncate the filename while keeping the extension
+            base, ext = os.path.splitext(filename)
+            filename = base[:190] + ext  # Leave room for the extension and some margin
+
+        # Handle specific content types
+        if isinstance(content, str):
+            # Ensure str content is encoded to bytes if required
+            try:
+                writer.write(content, filename)
+            except TypeError:
+                # If the writer expects bytes, convert string to bytes
+                writer.write(content.encode("utf-8"), filename)
+        else:
+            # For dict/list content, always encode as JSON string first
+            if isinstance(content, (dict, list)):
+                try:
+                    writer.write(
+                        json.dumps(content, ensure_ascii=False, indent=4), filename
+                    )
+                except TypeError:
+                    # If the writer expects bytes, convert JSON string to bytes
+                    writer.write(
+                        json.dumps(content, ensure_ascii=False, indent=4).encode(
+                            "utf-8"
+                        ),
+                        filename,
+                    )
+            else:
+                # Regular content (assumed to be bytes or compatible)
+                writer.write(content, filename)
+
+    @staticmethod
+    def parse_pdf(
+        pdf_path: Union[str, Path],
+        output_dir: Optional[str] = None,
+        use_ocr: bool = False,
+    ) -> Tuple[List[Dict[str, Any]], str]:
+        """
+        Parse PDF document
+
+        Args:
+            pdf_path: Path to the PDF file
+            output_dir: Output directory path
+            use_ocr: Whether to force OCR parsing
+
+        Returns:
+            Tuple[List[Dict[str, Any]], str]: Tuple containing (content list JSON, Markdown text)
+        """
+        try:
+            # Convert to Path object for easier handling
+            pdf_path = Path(pdf_path)
+            name_without_suff = pdf_path.stem
+
+            # Prepare output directories - ensure file name is in path
+            if output_dir:
+                base_output_dir = Path(output_dir)
+                local_md_dir = base_output_dir / name_without_suff
+            else:
+                local_md_dir = pdf_path.parent / name_without_suff
+
+            local_image_dir = local_md_dir / "images"
+            image_dir = local_image_dir.name
+
+            # Create directories
+            os.makedirs(local_image_dir, exist_ok=True)
+            os.makedirs(local_md_dir, exist_ok=True)
+
+            # Initialize writers and reader
+            image_writer = FileBasedDataWriter(str(local_image_dir))  # type: ignore
+            md_writer = FileBasedDataWriter(str(local_md_dir))  # type: ignore
+            reader = FileBasedDataReader("")  # type: ignore
+
+            # Read PDF bytes
+            pdf_bytes = reader.read(str(pdf_path))  # type: ignore
+
+            # Create dataset instance
+            ds = PymuDocDataset(pdf_bytes)  # type: ignore
+
+            # Process based on PDF type and user preference
+            if use_ocr or ds.classify() == SupportedPdfParseMethod.OCR:  # type: ignore
+                infer_result = ds.apply(doc_analyze, ocr=True)  # type: ignore
+                pipe_result = infer_result.pipe_ocr_mode(image_writer)  # type: ignore
+            else:
+                infer_result = ds.apply(doc_analyze, ocr=False)  # type: ignore
+                pipe_result = infer_result.pipe_txt_mode(image_writer)  # type: ignore
+
+            # Draw visualizations
+            try:
+                infer_result.draw_model(
+                    os.path.join(local_md_dir, f"{name_without_suff}_model.pdf")
+                )  # type: ignore
+                pipe_result.draw_layout(
+                    os.path.join(local_md_dir, f"{name_without_suff}_layout.pdf")
+                )  # type: ignore
+                pipe_result.draw_span(
+                    os.path.join(local_md_dir, f"{name_without_suff}_spans.pdf")
+                )  # type: ignore
+            except Exception as e:
+                print(f"Warning: Failed to draw visualizations: {str(e)}")
+
+            # Get data using API methods
+            md_content = pipe_result.get_markdown(image_dir)  # type: ignore
+            content_list = pipe_result.get_content_list(image_dir)  # type: ignore
+
+            # Save files using dump methods (consistent with API)
+            pipe_result.dump_md(md_writer, f"{name_without_suff}.md", image_dir)  # type: ignore
+            pipe_result.dump_content_list(
+                md_writer, f"{name_without_suff}_content_list.json", image_dir
+            )  # type: ignore
+            pipe_result.dump_middle_json(md_writer, f"{name_without_suff}_middle.json")  # type: ignore
+
+            # Save model result - convert JSON string to bytes before writing
+            model_inference_result = infer_result.get_infer_res()  # type: ignore
+            json_str = json.dumps(model_inference_result, ensure_ascii=False, indent=4)
+
+            try:
+                # Try to write to a file manually to avoid FileBasedDataWriter issues
+                model_file_path = os.path.join(
+                    local_md_dir, f"{name_without_suff}_model.json"
+                )
+                with open(model_file_path, "w", encoding="utf-8") as f:
+                    f.write(json_str)
+            except Exception as e:
+                print(
+                    f"Warning: Failed to save model result using file write: {str(e)}"
+                )
+                try:
+                    # If direct file write fails, try using the writer with bytes encoding
+                    md_writer.write(
+                        json_str.encode("utf-8"), f"{name_without_suff}_model.json"
+                    )  # type: ignore
+                except Exception as e2:
+                    print(
+                        f"Warning: Failed to save model result using writer: {str(e2)}"
+                    )
+
+            return cast(Tuple[List[Dict[str, Any]], str], (content_list, md_content))
+
+        except Exception as e:
+            print(f"Error in parse_pdf: {str(e)}")
+            raise
+
+    @staticmethod
+    def parse_office_doc(
+        doc_path: Union[str, Path], output_dir: Optional[str] = None
+    ) -> Tuple[List[Dict[str, Any]], str]:
+        """
+        Parse office document (Word, PPT, etc.)
+
+        Args:
+            doc_path: Path to the document file
+            output_dir: Output directory path
+
+        Returns:
+            Tuple[List[Dict[str, Any]], str]: Tuple containing (content list JSON, Markdown text)
+        """
+        try:
+            # Convert to Path object for easier handling
+            doc_path = Path(doc_path)
+            name_without_suff = doc_path.stem
+
+            # Prepare output directories - ensure file name is in path
+            if output_dir:
+                base_output_dir = Path(output_dir)
+                local_md_dir = base_output_dir / name_without_suff
+            else:
+                local_md_dir = doc_path.parent / name_without_suff
+
+            local_image_dir = local_md_dir / "images"
+            image_dir = local_image_dir.name
+
+            # Create directories
+            os.makedirs(local_image_dir, exist_ok=True)
+            os.makedirs(local_md_dir, exist_ok=True)
+
+            # Initialize writers
+            image_writer = FileBasedDataWriter(str(local_image_dir))  # type: ignore
+            md_writer = FileBasedDataWriter(str(local_md_dir))  # type: ignore
+
+            # Read office document
+            ds = read_local_office(str(doc_path))[0]  # type: ignore
+
+            # Apply chain of operations according to API documentation
+            # This follows the pattern shown in MS-Office example in the API docs
+            ds.apply(doc_analyze, ocr=True).pipe_txt_mode(image_writer).dump_md(
+                md_writer, f"{name_without_suff}.md", image_dir
+            )  # type: ignore
+
+            # Re-execute for getting the content data
+            infer_result = ds.apply(doc_analyze, ocr=True)  # type: ignore
+            pipe_result = infer_result.pipe_txt_mode(image_writer)  # type: ignore
+
+            # Get data for return values and additional outputs
+            md_content = pipe_result.get_markdown(image_dir)  # type: ignore
+            content_list = pipe_result.get_content_list(image_dir)  # type: ignore
+
+            # Save additional output files
+            pipe_result.dump_content_list(
+                md_writer, f"{name_without_suff}_content_list.json", image_dir
+            )  # type: ignore
+            pipe_result.dump_middle_json(md_writer, f"{name_without_suff}_middle.json")  # type: ignore
+
+            # Save model result - convert JSON string to bytes before writing
+            model_inference_result = infer_result.get_infer_res()  # type: ignore
+            json_str = json.dumps(model_inference_result, ensure_ascii=False, indent=4)
+
+            try:
+                # Try to write to a file manually to avoid FileBasedDataWriter issues
+                model_file_path = os.path.join(
+                    local_md_dir, f"{name_without_suff}_model.json"
+                )
+                with open(model_file_path, "w", encoding="utf-8") as f:
+                    f.write(json_str)
+            except Exception as e:
+                print(
+                    f"Warning: Failed to save model result using file write: {str(e)}"
+                )
+                try:
+                    # If direct file write fails, try using the writer with bytes encoding
+                    md_writer.write(
+                        json_str.encode("utf-8"), f"{name_without_suff}_model.json"
+                    )  # type: ignore
+                except Exception as e2:
+                    print(
+                        f"Warning: Failed to save model result using writer: {str(e2)}"
+                    )
+
+            return cast(Tuple[List[Dict[str, Any]], str], (content_list, md_content))
+
+        except Exception as e:
+            print(f"Error in parse_office_doc: {str(e)}")
+            raise
+
+    @staticmethod
+    def parse_image(
+        image_path: Union[str, Path], output_dir: Optional[str] = None
+    ) -> Tuple[List[Dict[str, Any]], str]:
+        """
+        Parse image document
+
+        Args:
+            image_path: Path to the image file
+            output_dir: Output directory path
+
+        Returns:
+            Tuple[List[Dict[str, Any]], str]: Tuple containing (content list JSON, Markdown text)
+        """
+        try:
+            # Convert to Path object for easier handling
+            image_path = Path(image_path)
+            name_without_suff = image_path.stem
+
+            # Prepare output directories - ensure file name is in path
+            if output_dir:
+                base_output_dir = Path(output_dir)
+                local_md_dir = base_output_dir / name_without_suff
+            else:
+                local_md_dir = image_path.parent / name_without_suff
+
+            local_image_dir = local_md_dir / "images"
+            image_dir = local_image_dir.name
+
+            # Create directories
+            os.makedirs(local_image_dir, exist_ok=True)
+            os.makedirs(local_md_dir, exist_ok=True)
+
+            # Initialize writers
+            image_writer = FileBasedDataWriter(str(local_image_dir))  # type: ignore
+            md_writer = FileBasedDataWriter(str(local_md_dir))  # type: ignore
+
+            # Read image
+            ds = read_local_images(str(image_path))[0]  # type: ignore
+
+            # Apply chain of operations according to API documentation
+            # This follows the pattern shown in Image example in the API docs
+            ds.apply(doc_analyze, ocr=True).pipe_ocr_mode(image_writer).dump_md(
+                md_writer, f"{name_without_suff}.md", image_dir
+            )  # type: ignore
+
+            # Re-execute for getting the content data
+            infer_result = ds.apply(doc_analyze, ocr=True)  # type: ignore
+            pipe_result = infer_result.pipe_ocr_mode(image_writer)  # type: ignore
+
+            # Get data for return values and additional outputs
+            md_content = pipe_result.get_markdown(image_dir)  # type: ignore
+            content_list = pipe_result.get_content_list(image_dir)  # type: ignore
+
+            # Save additional output files
+            pipe_result.dump_content_list(
+                md_writer, f"{name_without_suff}_content_list.json", image_dir
+            )  # type: ignore
+            pipe_result.dump_middle_json(md_writer, f"{name_without_suff}_middle.json")  # type: ignore
+
+            # Save model result - convert JSON string to bytes before writing
+            model_inference_result = infer_result.get_infer_res()  # type: ignore
+            json_str = json.dumps(model_inference_result, ensure_ascii=False, indent=4)
+
+            try:
+                # Try to write to a file manually to avoid FileBasedDataWriter issues
+                model_file_path = os.path.join(
+                    local_md_dir, f"{name_without_suff}_model.json"
+                )
+                with open(model_file_path, "w", encoding="utf-8") as f:
+                    f.write(json_str)
+            except Exception as e:
+                print(
+                    f"Warning: Failed to save model result using file write: {str(e)}"
+                )
+                try:
+                    # If direct file write fails, try using the writer with bytes encoding
+                    md_writer.write(
+                        json_str.encode("utf-8"), f"{name_without_suff}_model.json"
+                    )  # type: ignore
+                except Exception as e2:
+                    print(
+                        f"Warning: Failed to save model result using writer: {str(e2)}"
+                    )
+
+            return cast(Tuple[List[Dict[str, Any]], str], (content_list, md_content))
+
+        except Exception as e:
+            print(f"Error in parse_image: {str(e)}")
+            raise
+
+    @staticmethod
+    def parse_document(
+        file_path: Union[str, Path],
+        parse_method: str = "auto",
+        output_dir: Optional[str] = None,
+        save_results: bool = True,
+    ) -> Tuple[List[Dict[str, Any]], str]:
+        """
+        Parse document using MinerU based on file extension
+
+        Args:
+            file_path: Path to the file to be parsed
+            parse_method: Parsing method, supports "auto", "ocr", "txt", default is "auto"
+            output_dir: Output directory path, if None, use the directory of the input file
+            save_results: Whether to save parsing results to files
+
+        Returns:
+            Tuple[List[Dict[str, Any]], str]: Tuple containing (content list JSON, Markdown text)
+        """
+        # Convert to Path object
+        file_path = Path(file_path)
+        if not file_path.exists():
+            raise FileNotFoundError(f"File does not exist: {file_path}")
+
+        # Get file extension
+        ext = file_path.suffix.lower()
+
+        # Choose appropriate parser based on file type
+        if ext in [".pdf"]:
+            return MineruParser.parse_pdf(
+                file_path, output_dir, use_ocr=(parse_method == "ocr")
+            )
+        elif ext in [".jpg", ".jpeg", ".png", ".bmp", ".tiff", ".tif"]:
+            return MineruParser.parse_image(file_path, output_dir)
+        elif ext in [".doc", ".docx", ".ppt", ".pptx"]:
+            return MineruParser.parse_office_doc(file_path, output_dir)
+        else:
+            # For unsupported file types, default to PDF parsing
+            print(
+                f"Warning: Unsupported file extension '{ext}', trying generic PDF parser"
+            )
+            return MineruParser.parse_pdf(
+                file_path, output_dir, use_ocr=(parse_method == "ocr")
+            )
+
+
+def main():
+    """
+    Main function to run the MinerU parser from command line
+    """
+    parser = argparse.ArgumentParser(description="Parse documents using MinerU")
+    parser.add_argument("file_path", help="Path to the document to parse")
+    parser.add_argument("--output", "-o", help="Output directory path")
+    parser.add_argument(
+        "--method",
+        "-m",
+        choices=["auto", "ocr", "txt"],
+        default="auto",
+        help="Parsing method (auto, ocr, txt)",
+    )
+    parser.add_argument(
+        "--stats", action="store_true", help="Display content statistics"
+    )
+
+    args = parser.parse_args()
+
+    try:
+        # Parse the document
+        content_list, md_content = MineruParser.parse_document(
+            file_path=args.file_path, parse_method=args.method, output_dir=args.output
+        )
+
+        # Display statistics if requested
+        if args.stats:
+            print("\nDocument Statistics:")
+            print(f"Total content blocks: {len(content_list)}")
+
+            # Count different types of content
+            content_types = {}
+            for item in content_list:
+                content_type = item.get("type", "unknown")
+                content_types[content_type] = content_types.get(content_type, 0) + 1
+
+            print("\nContent Type Distribution:")
+            for content_type, count in content_types.items():
+                print(f"- {content_type}: {count}")
+
+    except Exception as e:
+        print(f"Error: {str(e)}")
+        return 1
+
+    return 0
+
+
+if __name__ == "__main__":
+    exit(main())

lightrag/modalprocessors.py

@@ -0,0 +1,699 @@
+"""
+Specialized processors for different modalities
+
+Includes:
+- ImageModalProcessor: Specialized processor for image content
+- TableModalProcessor: Specialized processor for table content
+- EquationModalProcessor: Specialized processor for equation content
+- GenericModalProcessor: Processor for other modal content
+"""
+
+import re
+import json
+import time
+import asyncio
+import base64
+from typing import Dict, Any, Tuple, cast
+from pathlib import Path
+
+from lightrag.base import StorageNameSpace
+from lightrag.utils import (
+    logger,
+    compute_mdhash_id,
+)
+from lightrag.lightrag import LightRAG
+from dataclasses import asdict
+from lightrag.kg.shared_storage import get_namespace_data, get_pipeline_status_lock
+
+
+class BaseModalProcessor:
+    """Base class for modal processors"""
+
+    def __init__(self, lightrag: LightRAG, modal_caption_func):
+        """Initialize base processor
+
+        Args:
+            lightrag: LightRAG instance
+            modal_caption_func: Function for generating descriptions
+        """
+        self.lightrag = lightrag
+        self.modal_caption_func = modal_caption_func
+
+        # Use LightRAG's storage instances
+        self.text_chunks_db = lightrag.text_chunks
+        self.chunks_vdb = lightrag.chunks_vdb
+        self.entities_vdb = lightrag.entities_vdb
+        self.relationships_vdb = lightrag.relationships_vdb
+        self.knowledge_graph_inst = lightrag.chunk_entity_relation_graph
+
+        # Use LightRAG's configuration and functions
+        self.embedding_func = lightrag.embedding_func
+        self.llm_model_func = lightrag.llm_model_func
+        self.global_config = asdict(lightrag)
+        self.hashing_kv = lightrag.llm_response_cache
+        self.tokenizer = lightrag.tokenizer
+
+    async def process_multimodal_content(
+        self,
+        modal_content,
+        content_type: str,
+        file_path: str = "manual_creation",
+        entity_name: str = None,
+    ) -> Tuple[str, Dict[str, Any]]:
+        """Process multimodal content"""
+        # Subclasses need to implement specific processing logic
+        raise NotImplementedError("Subclasses must implement this method")
+
+    async def _create_entity_and_chunk(
+        self, modal_chunk: str, entity_info: Dict[str, Any], file_path: str
+    ) -> Tuple[str, Dict[str, Any]]:
+        """Create entity and text chunk"""
+        # Create chunk
+        chunk_id = compute_mdhash_id(str(modal_chunk), prefix="chunk-")
+        tokens = len(self.tokenizer.encode(modal_chunk))
+
+        chunk_data = {
+            "tokens": tokens,
+            "content": modal_chunk,
+            "chunk_order_index": 0,
+            "full_doc_id": chunk_id,
+            "file_path": file_path,
+        }
+
+        # Store chunk
+        await self.text_chunks_db.upsert({chunk_id: chunk_data})
+
+        # Create entity node
+        node_data = {
+            "entity_id": entity_info["entity_name"],
+            "entity_type": entity_info["entity_type"],
+            "description": entity_info["summary"],
+            "source_id": chunk_id,
+            "file_path": file_path,
+            "created_at": int(time.time()),
+        }
+
+        await self.knowledge_graph_inst.upsert_node(
+            entity_info["entity_name"], node_data
+        )
+
+        # Insert entity into vector database
+        entity_vdb_data = {
+            compute_mdhash_id(entity_info["entity_name"], prefix="ent-"): {
+                "entity_name": entity_info["entity_name"],
+                "entity_type": entity_info["entity_type"],
+                "content": f"{entity_info['entity_name']}\n{entity_info['summary']}",
+                "source_id": chunk_id,
+                "file_path": file_path,
+            }
+        }
+        await self.entities_vdb.upsert(entity_vdb_data)
+
+        # Process entity and relationship extraction
+        await self._process_chunk_for_extraction(chunk_id, entity_info["entity_name"])
+
+        # Ensure all storage updates are complete
+        await self._insert_done()
+
+        return entity_info["summary"], {
+            "entity_name": entity_info["entity_name"],
+            "entity_type": entity_info["entity_type"],
+            "description": entity_info["summary"],
+            "chunk_id": chunk_id,
+        }
+
+    async def _process_chunk_for_extraction(
+        self, chunk_id: str, modal_entity_name: str
+    ):
+        """Process chunk for entity and relationship extraction"""
+        chunk_data = await self.text_chunks_db.get_by_id(chunk_id)
+        if not chunk_data:
+            logger.error(f"Chunk {chunk_id} not found")
+            return
+
+        # Create text chunk for vector database
+        chunk_vdb_data = {
+            chunk_id: {
+                "content": chunk_data["content"],
+                "full_doc_id": chunk_id,
+                "tokens": chunk_data["tokens"],
+                "chunk_order_index": chunk_data["chunk_order_index"],
+                "file_path": chunk_data["file_path"],
+            }
+        }
+
+        await self.chunks_vdb.upsert(chunk_vdb_data)
+
+        # Trigger extraction process
+        from lightrag.operate import extract_entities, merge_nodes_and_edges
+
+        pipeline_status = await get_namespace_data("pipeline_status")
+        pipeline_status_lock = get_pipeline_status_lock()
+
+        # Prepare chunk for extraction
+        chunks = {chunk_id: chunk_data}
+
+        # Extract entities and relationships
+        chunk_results = await extract_entities(
+            chunks=chunks,
+            global_config=self.global_config,
+            pipeline_status=pipeline_status,
+            pipeline_status_lock=pipeline_status_lock,
+            llm_response_cache=self.hashing_kv,
+        )
+
+        # Add "belongs_to" relationships for all extracted entities
+        for maybe_nodes, _ in chunk_results:
+            for entity_name in maybe_nodes.keys():
+                if entity_name != modal_entity_name:  # Skip self-relationship
+                    # Create belongs_to relationship
+                    relation_data = {
+                        "description": f"Entity {entity_name} belongs to {modal_entity_name}",
+                        "keywords": "belongs_to,part_of,contained_in",
+                        "source_id": chunk_id,
+                        "weight": 10.0,
+                        "file_path": chunk_data.get("file_path", "manual_creation"),
+                    }
+                    await self.knowledge_graph_inst.upsert_edge(
+                        entity_name, modal_entity_name, relation_data
+                    )
+
+                    relation_id = compute_mdhash_id(
+                        entity_name + modal_entity_name, prefix="rel-"
+                    )
+                    relation_vdb_data = {
+                        relation_id: {
+                            "src_id": entity_name,
+                            "tgt_id": modal_entity_name,
+                            "keywords": relation_data["keywords"],
+                            "content": f"{relation_data['keywords']}\t{entity_name}\n{modal_entity_name}\n{relation_data['description']}",
+                            "source_id": chunk_id,
+                            "file_path": chunk_data.get("file_path", "manual_creation"),
+                        }
+                    }
+                    await self.relationships_vdb.upsert(relation_vdb_data)
+
+        await merge_nodes_and_edges(
+            chunk_results=chunk_results,
+            knowledge_graph_inst=self.knowledge_graph_inst,
+            entity_vdb=self.entities_vdb,
+            relationships_vdb=self.relationships_vdb,
+            global_config=self.global_config,
+            pipeline_status=pipeline_status,
+            pipeline_status_lock=pipeline_status_lock,
+            llm_response_cache=self.hashing_kv,
+        )
+
+    async def _insert_done(self) -> None:
+        await asyncio.gather(
+            *[
+                cast(StorageNameSpace, storage_inst).index_done_callback()
+                for storage_inst in [
+                    self.text_chunks_db,
+                    self.chunks_vdb,
+                    self.entities_vdb,
+                    self.relationships_vdb,
+                    self.knowledge_graph_inst,
+                ]
+            ]
+        )
+
+
+class ImageModalProcessor(BaseModalProcessor):
+    """Processor specialized for image content"""
+
+    def __init__(self, lightrag: LightRAG, modal_caption_func):
+        """Initialize image processor
+
+        Args:
+            lightrag: LightRAG instance
+            modal_caption_func: Function for generating descriptions (supporting image understanding)
+        """
+        super().__init__(lightrag, modal_caption_func)
+
+    def _encode_image_to_base64(self, image_path: str) -> str:
+        """Encode image to base64"""
+        try:
+            with open(image_path, "rb") as image_file:
+                encoded_string = base64.b64encode(image_file.read()).decode("utf-8")
+            return encoded_string
+        except Exception as e:
+            logger.error(f"Failed to encode image {image_path}: {e}")
+            return ""
+
+    async def process_multimodal_content(
+        self,
+        modal_content,
+        content_type: str,
+        file_path: str = "manual_creation",
+        entity_name: str = None,
+    ) -> Tuple[str, Dict[str, Any]]:
+        """Process image content"""
+        try:
+            # Parse image content
+            if isinstance(modal_content, str):
+                try:
+                    content_data = json.loads(modal_content)
+                except json.JSONDecodeError:
+                    content_data = {"description": modal_content}
+            else:
+                content_data = modal_content
+
+            image_path = content_data.get("img_path")
+            captions = content_data.get("img_caption", [])
+            footnotes = content_data.get("img_footnote", [])
+
+            # Build detailed visual analysis prompt
+            vision_prompt = f"""Please analyze this image in detail and provide a JSON response with the following structure:
+
+                {{
+                    "detailed_description": "A comprehensive and detailed visual description of the image following these guidelines:
+                    - Describe the overall composition and layout
+                    - Identify all objects, people, text, and visual elements
+                    - Explain relationships between elements
+                    - Note colors, lighting, and visual style
+                    - Describe any actions or activities shown
+                    - Include technical details if relevant (charts, diagrams, etc.)
+                    - Always use specific names instead of pronouns",
+                    "entity_info": {{
+                        "entity_name": "{entity_name if entity_name else 'unique descriptive name for this image'}",
+                        "entity_type": "image",
+                        "summary": "concise summary of the image content and its significance (max 100 words)"
+                    }}
+                }}
+
+                Additional context:
+                - Image Path: {image_path}
+                - Captions: {captions if captions else 'None'}
+                - Footnotes: {footnotes if footnotes else 'None'}
+
+                Focus on providing accurate, detailed visual analysis that would be useful for knowledge retrieval."""
+
+            # If image path exists, try to encode image
+            image_base64 = ""
+            if image_path and Path(image_path).exists():
+                image_base64 = self._encode_image_to_base64(image_path)
+
+            # Call vision model
+            if image_base64:
+                # Use real image for analysis
+                response = await self.modal_caption_func(
+                    vision_prompt,
+                    image_data=image_base64,
+                    system_prompt="You are an expert image analyst. Provide detailed, accurate descriptions.",
+                )
+            else:
+                # Analyze based on existing text information
+                text_prompt = f"""Based on the following image information, provide analysis:
+
+                    Image Path: {image_path}
+                    Captions: {captions}
+                    Footnotes: {footnotes}
+
+                    {vision_prompt}"""
+
+                response = await self.modal_caption_func(
+                    text_prompt,
+                    system_prompt="You are an expert image analyst. Provide detailed analysis based on available information.",
+                )
+
+            # Parse response
+            enhanced_caption, entity_info = self._parse_response(response, entity_name)
+
+            # Build complete image content
+            modal_chunk = f"""
+                    Image Content Analysis:
+                    Image Path: {image_path}
+                    Captions: {', '.join(captions) if captions else 'None'}
+                    Footnotes: {', '.join(footnotes) if footnotes else 'None'}
+
+                    Visual Analysis: {enhanced_caption}"""
+
+            return await self._create_entity_and_chunk(
+                modal_chunk, entity_info, file_path
+            )
+
+        except Exception as e:
+            logger.error(f"Error processing image content: {e}")
+            # Fallback processing
+            fallback_entity = {
+                "entity_name": entity_name
+                if entity_name
+                else f"image_{compute_mdhash_id(str(modal_content))}",
+                "entity_type": "image",
+                "summary": f"Image content: {str(modal_content)[:100]}",
+            }
+            return str(modal_content), fallback_entity
+
+    def _parse_response(
+        self, response: str, entity_name: str = None
+    ) -> Tuple[str, Dict[str, Any]]:
+        """Parse model response"""
+        try:
+            response_data = json.loads(
+                re.search(r"\{.*\}", response, re.DOTALL).group(0)
+            )
+
+            description = response_data.get("detailed_description", "")
+            entity_data = response_data.get("entity_info", {})
+
+            if not description or not entity_data:
+                raise ValueError("Missing required fields in response")
+
+            if not all(
+                key in entity_data for key in ["entity_name", "entity_type", "summary"]
+            ):
+                raise ValueError("Missing required fields in entity_info")
+
+            entity_data["entity_name"] = (
+                entity_data["entity_name"] + f" ({entity_data['entity_type']})"
+            )
+            if entity_name:
+                entity_data["entity_name"] = entity_name
+
+            return description, entity_data
+
+        except (json.JSONDecodeError, AttributeError, ValueError) as e:
+            logger.error(f"Error parsing image analysis response: {e}")
+            fallback_entity = {
+                "entity_name": entity_name
+                if entity_name
+                else f"image_{compute_mdhash_id(response)}",
+                "entity_type": "image",
+                "summary": response[:100] + "..." if len(response) > 100 else response,
+            }
+            return response, fallback_entity
+
+
+class TableModalProcessor(BaseModalProcessor):
+    """Processor specialized for table content"""
+
+    async def process_multimodal_content(
+        self,
+        modal_content,
+        content_type: str,
+        file_path: str = "manual_creation",
+        entity_name: str = None,
+    ) -> Tuple[str, Dict[str, Any]]:
+        """Process table content"""
+        # Parse table content
+        if isinstance(modal_content, str):
+            try:
+                content_data = json.loads(modal_content)
+            except json.JSONDecodeError:
+                content_data = {"table_body": modal_content}
+        else:
+            content_data = modal_content
+
+        table_img_path = content_data.get("img_path")
+        table_caption = content_data.get("table_caption", [])
+        table_body = content_data.get("table_body", "")
+        table_footnote = content_data.get("table_footnote", [])
+
+        # Build table analysis prompt
+        table_prompt = f"""Please analyze this table content and provide a JSON response with the following structure:
+
+            {{
+                "detailed_description": "A comprehensive analysis of the table including:
+                - Table structure and organization
+                - Column headers and their meanings
+                - Key data points and patterns
+                - Statistical insights and trends
+                - Relationships between data elements
+                - Significance of the data presented
+                Always use specific names and values instead of general references.",
+                "entity_info": {{
+                    "entity_name": "{entity_name if entity_name else 'descriptive name for this table'}",
+                    "entity_type": "table",
+                    "summary": "concise summary of the table's purpose and key findings (max 100 words)"
+                }}
+            }}
+
+            Table Information:
+            Image Path: {table_img_path}
+            Caption: {table_caption if table_caption else 'None'}
+            Body: {table_body}
+            Footnotes: {table_footnote if table_footnote else 'None'}
+
+            Focus on extracting meaningful insights and relationships from the tabular data."""
+
+        response = await self.modal_caption_func(
+            table_prompt,
+            system_prompt="You are an expert data analyst. Provide detailed table analysis with specific insights.",
+        )
+
+        # Parse response
+        enhanced_caption, entity_info = self._parse_table_response(
+            response, entity_name
+        )
+
+        # TODO: Add Retry Mechanism
+
+        # Build complete table content
+        modal_chunk = f"""Table Analysis:
+            Image Path: {table_img_path}
+            Caption: {', '.join(table_caption) if table_caption else 'None'}
+            Structure: {table_body}
+            Footnotes: {', '.join(table_footnote) if table_footnote else 'None'}
+
+            Analysis: {enhanced_caption}"""
+
+        return await self._create_entity_and_chunk(modal_chunk, entity_info, file_path)
+
+    def _parse_table_response(
+        self, response: str, entity_name: str = None
+    ) -> Tuple[str, Dict[str, Any]]:
+        """Parse table analysis response"""
+        try:
+            response_data = json.loads(
+                re.search(r"\{.*\}", response, re.DOTALL).group(0)
+            )
+
+            description = response_data.get("detailed_description", "")
+            entity_data = response_data.get("entity_info", {})
+
+            if not description or not entity_data:
+                raise ValueError("Missing required fields in response")
+
+            if not all(
+                key in entity_data for key in ["entity_name", "entity_type", "summary"]
+            ):
+                raise ValueError("Missing required fields in entity_info")
+
+            entity_data["entity_name"] = (
+                entity_data["entity_name"] + f" ({entity_data['entity_type']})"
+            )
+            if entity_name:
+                entity_data["entity_name"] = entity_name
+
+            return description, entity_data
+
+        except (json.JSONDecodeError, AttributeError, ValueError) as e:
+            logger.error(f"Error parsing table analysis response: {e}")
+            fallback_entity = {
+                "entity_name": entity_name
+                if entity_name
+                else f"table_{compute_mdhash_id(response)}",
+                "entity_type": "table",
+                "summary": response[:100] + "..." if len(response) > 100 else response,
+            }
+            return response, fallback_entity
+
+
+class EquationModalProcessor(BaseModalProcessor):
+    """Processor specialized for equation content"""
+
+    async def process_multimodal_content(
+        self,
+        modal_content,
+        content_type: str,
+        file_path: str = "manual_creation",
+        entity_name: str = None,
+    ) -> Tuple[str, Dict[str, Any]]:
+        """Process equation content"""
+        # Parse equation content
+        if isinstance(modal_content, str):
+            try:
+                content_data = json.loads(modal_content)
+            except json.JSONDecodeError:
+                content_data = {"equation": modal_content}
+        else:
+            content_data = modal_content
+
+        equation_text = content_data.get("text")
+        equation_format = content_data.get("text_format", "")
+
+        # Build equation analysis prompt
+        equation_prompt = f"""Please analyze this mathematical equation and provide a JSON response with the following structure:
+
+            {{
+                "detailed_description": "A comprehensive analysis of the equation including:
+                - Mathematical meaning and interpretation
+                - Variables and their definitions
+                - Mathematical operations and functions used
+                - Application domain and context
+                - Physical or theoretical significance
+                - Relationship to other mathematical concepts
+                - Practical applications or use cases
+                Always use specific mathematical terminology.",
+                "entity_info": {{
+                    "entity_name": "{entity_name if entity_name else 'descriptive name for this equation'}",
+                    "entity_type": "equation",
+                    "summary": "concise summary of the equation's purpose and significance (max 100 words)"
+                }}
+            }}
+
+            Equation Information:
+            Equation: {equation_text}
+            Format: {equation_format}
+
+            Focus on providing mathematical insights and explaining the equation's significance."""
+
+        response = await self.modal_caption_func(
+            equation_prompt,
+            system_prompt="You are an expert mathematician. Provide detailed mathematical analysis.",
+        )
+
+        # Parse response
+        enhanced_caption, entity_info = self._parse_equation_response(
+            response, entity_name
+        )
+
+        # Build complete equation content
+        modal_chunk = f"""Mathematical Equation Analysis:
+            Equation: {equation_text}
+            Format: {equation_format}
+
+            Mathematical Analysis: {enhanced_caption}"""
+
+        return await self._create_entity_and_chunk(modal_chunk, entity_info, file_path)
+
+    def _parse_equation_response(
+        self, response: str, entity_name: str = None
+    ) -> Tuple[str, Dict[str, Any]]:
+        """Parse equation analysis response"""
+        try:
+            response_data = json.loads(
+                re.search(r"\{.*\}", response, re.DOTALL).group(0)
+            )
+
+            description = response_data.get("detailed_description", "")
+            entity_data = response_data.get("entity_info", {})
+
+            if not description or not entity_data:
+                raise ValueError("Missing required fields in response")
+
+            if not all(
+                key in entity_data for key in ["entity_name", "entity_type", "summary"]
+            ):
+                raise ValueError("Missing required fields in entity_info")
+
+            entity_data["entity_name"] = (
+                entity_data["entity_name"] + f" ({entity_data['entity_type']})"
+            )
+            if entity_name:
+                entity_data["entity_name"] = entity_name
+
+            return description, entity_data
+
+        except (json.JSONDecodeError, AttributeError, ValueError) as e:
+            logger.error(f"Error parsing equation analysis response: {e}")
+            fallback_entity = {
+                "entity_name": entity_name
+                if entity_name
+                else f"equation_{compute_mdhash_id(response)}",
+                "entity_type": "equation",
+                "summary": response[:100] + "..." if len(response) > 100 else response,
+            }
+            return response, fallback_entity
+
+
+class GenericModalProcessor(BaseModalProcessor):
+    """Generic processor for other types of modal content"""
+
+    async def process_multimodal_content(
+        self,
+        modal_content,
+        content_type: str,
+        file_path: str = "manual_creation",
+        entity_name: str = None,
+    ) -> Tuple[str, Dict[str, Any]]:
+        """Process generic modal content"""
+        # Build generic analysis prompt
+        generic_prompt = f"""Please analyze this {content_type} content and provide a JSON response with the following structure:
+
+            {{
+                "detailed_description": "A comprehensive analysis of the content including:
+                - Content structure and organization
+                - Key information and elements
+                - Relationships between components
+                - Context and significance
+                - Relevant details for knowledge retrieval
+                Always use specific terminology appropriate for {content_type} content.",
+                "entity_info": {{
+                    "entity_name": "{entity_name if entity_name else f'descriptive name for this {content_type}'}",
+                    "entity_type": "{content_type}",
+                    "summary": "concise summary of the content's purpose and key points (max 100 words)"
+                }}
+            }}
+
+            Content: {str(modal_content)}
+
+            Focus on extracting meaningful information that would be useful for knowledge retrieval."""
+
+        response = await self.modal_caption_func(
+            generic_prompt,
+            system_prompt=f"You are an expert content analyst specializing in {content_type} content.",
+        )
+
+        # Parse response
+        enhanced_caption, entity_info = self._parse_generic_response(
+            response, entity_name, content_type
+        )
+
+        # Build complete content
+        modal_chunk = f"""{content_type.title()} Content Analysis:
+            Content: {str(modal_content)}
+
+            Analysis: {enhanced_caption}"""
+
+        return await self._create_entity_and_chunk(modal_chunk, entity_info, file_path)
+
+    def _parse_generic_response(
+        self, response: str, entity_name: str = None, content_type: str = "content"
+    ) -> Tuple[str, Dict[str, Any]]:
+        """Parse generic analysis response"""
+        try:
+            response_data = json.loads(
+                re.search(r"\{.*\}", response, re.DOTALL).group(0)
+            )
+
+            description = response_data.get("detailed_description", "")
+            entity_data = response_data.get("entity_info", {})
+
+            if not description or not entity_data:
+                raise ValueError("Missing required fields in response")
+
+            if not all(
+                key in entity_data for key in ["entity_name", "entity_type", "summary"]
+            ):
+                raise ValueError("Missing required fields in entity_info")
+
+            entity_data["entity_name"] = (
+                entity_data["entity_name"] + f" ({entity_data['entity_type']})"
+            )
+            if entity_name:
+                entity_data["entity_name"] = entity_name
+
+            return description, entity_data
+
+        except (json.JSONDecodeError, AttributeError, ValueError) as e:
+            logger.error(f"Error parsing generic analysis response: {e}")
+            fallback_entity = {
+                "entity_name": entity_name
+                if entity_name
+                else f"{content_type}_{compute_mdhash_id(response)}",
+                "entity_type": content_type,
+                "summary": response[:100] + "..." if len(response) > 100 else response,
+            }
+            return response, fallback_entity

lightrag/raganything.py

@@ -0,0 +1,686 @@
+"""
+Complete MinerU parsing + multimodal content insertion Pipeline
+
+This script integrates:
+1. MinerU document parsing
+2. Pure text content LightRAG insertion
+3. Specialized processing for multimodal content (using different processors)
+"""
+
+import os
+import asyncio
+import logging
+from pathlib import Path
+from typing import Dict, List, Any, Tuple, Optional, Callable
+import sys
+
+# Add project root directory to Python path
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from lightrag import LightRAG, QueryParam
+from lightrag.utils import EmbeddingFunc, setup_logger
+
+# Import parser and multimodal processors
+from lightrag.mineru_parser import MineruParser
+
+# Import specialized processors
+from lightrag.modalprocessors import (
+    ImageModalProcessor,
+    TableModalProcessor,
+    EquationModalProcessor,
+    GenericModalProcessor,
+)
+
+
+class RAGAnything:
+    """Multimodal Document Processing Pipeline - Complete document parsing and insertion pipeline"""
+
+    def __init__(
+        self,
+        lightrag: Optional[LightRAG] = None,
+        llm_model_func: Optional[Callable] = None,
+        vision_model_func: Optional[Callable] = None,
+        embedding_func: Optional[Callable] = None,
+        working_dir: str = "./rag_storage",
+        embedding_dim: int = 3072,
+        max_token_size: int = 8192,
+    ):
+        """
+        Initialize Multimodal Document Processing Pipeline
+
+        Args:
+            lightrag: Optional pre-initialized LightRAG instance
+            llm_model_func: LLM model function for text analysis
+            vision_model_func: Vision model function for image analysis
+            embedding_func: Embedding function for text vectorization
+            working_dir: Working directory for storage (used when creating new RAG)
+            embedding_dim: Embedding dimension (used when creating new RAG)
+            max_token_size: Maximum token size for embeddings (used when creating new RAG)
+        """
+        self.working_dir = working_dir
+        self.llm_model_func = llm_model_func
+        self.vision_model_func = vision_model_func
+        self.embedding_func = embedding_func
+        self.embedding_dim = embedding_dim
+        self.max_token_size = max_token_size
+
+        # Set up logging
+        setup_logger("RAGAnything")
+        self.logger = logging.getLogger("RAGAnything")
+
+        # Create working directory if needed
+        if not os.path.exists(working_dir):
+            os.makedirs(working_dir)
+
+        # Use provided LightRAG or mark for later initialization
+        self.lightrag = lightrag
+        self.modal_processors = {}
+
+        # If LightRAG is provided, initialize processors immediately
+        if self.lightrag is not None:
+            self._initialize_processors()
+
+    def _initialize_processors(self):
+        """Initialize multimodal processors with appropriate model functions"""
+        if self.lightrag is None:
+            raise ValueError(
+                "LightRAG instance must be initialized before creating processors"
+            )
+
+        # Create different multimodal processors
+        self.modal_processors = {
+            "image": ImageModalProcessor(
+                lightrag=self.lightrag,
+                modal_caption_func=self.vision_model_func or self.llm_model_func,
+            ),
+            "table": TableModalProcessor(
+                lightrag=self.lightrag, modal_caption_func=self.llm_model_func
+            ),
+            "equation": EquationModalProcessor(
+                lightrag=self.lightrag, modal_caption_func=self.llm_model_func
+            ),
+            "generic": GenericModalProcessor(
+                lightrag=self.lightrag, modal_caption_func=self.llm_model_func
+            ),
+        }
+
+        self.logger.info("Multimodal processors initialized")
+        self.logger.info(f"Available processors: {list(self.modal_processors.keys())}")
+
+    async def _ensure_lightrag_initialized(self):
+        """Ensure LightRAG instance is initialized, create if necessary"""
+        if self.lightrag is not None:
+            return
+
+        # Validate required functions
+        if self.llm_model_func is None:
+            raise ValueError(
+                "llm_model_func must be provided when LightRAG is not pre-initialized"
+            )
+        if self.embedding_func is None:
+            raise ValueError(
+                "embedding_func must be provided when LightRAG is not pre-initialized"
+            )
+
+        from lightrag.kg.shared_storage import initialize_pipeline_status
+
+        # Create LightRAG instance with provided functions
+        self.lightrag = LightRAG(
+            working_dir=self.working_dir,
+            llm_model_func=self.llm_model_func,
+            embedding_func=EmbeddingFunc(
+                embedding_dim=self.embedding_dim,
+                max_token_size=self.max_token_size,
+                func=self.embedding_func,
+            ),
+        )
+
+        await self.lightrag.initialize_storages()
+        await initialize_pipeline_status()
+
+        # Initialize processors after LightRAG is ready
+        self._initialize_processors()
+
+        self.logger.info("LightRAG and multimodal processors initialized")
+
+    def parse_document(
+        self,
+        file_path: str,
+        output_dir: str = "./output",
+        parse_method: str = "auto",
+        display_stats: bool = True,
+    ) -> Tuple[List[Dict[str, Any]], str]:
+        """
+        Parse document using MinerU
+
+        Args:
+            file_path: Path to the file to parse
+            output_dir: Output directory
+            parse_method: Parse method ("auto", "ocr", "txt")
+            display_stats: Whether to display content statistics
+
+        Returns:
+            (content_list, md_content): Content list and markdown text
+        """
+        self.logger.info(f"Starting document parsing: {file_path}")
+
+        file_path = Path(file_path)
+        if not file_path.exists():
+            raise FileNotFoundError(f"File not found: {file_path}")
+
+        # Choose appropriate parsing method based on file extension
+        ext = file_path.suffix.lower()
+
+        try:
+            if ext in [".pdf"]:
+                self.logger.info(
+                    f"Detected PDF file, using PDF parser (OCR={parse_method == 'ocr'})..."
+                )
+                content_list, md_content = MineruParser.parse_pdf(
+                    file_path, output_dir, use_ocr=(parse_method == "ocr")
+                )
+            elif ext in [".jpg", ".jpeg", ".png", ".bmp", ".tiff", ".tif"]:
+                self.logger.info("Detected image file, using image parser...")
+                content_list, md_content = MineruParser.parse_image(
+                    file_path, output_dir
+                )
+            elif ext in [".doc", ".docx", ".ppt", ".pptx"]:
+                self.logger.info("Detected Office document, using Office parser...")
+                content_list, md_content = MineruParser.parse_office_doc(
+                    file_path, output_dir
+                )
+            else:
+                # For other or unknown formats, use generic parser
+                self.logger.info(
+                    f"Using generic parser for {ext} file (method={parse_method})..."
+                )
+                content_list, md_content = MineruParser.parse_document(
+                    file_path, parse_method=parse_method, output_dir=output_dir
+                )
+
+        except Exception as e:
+            self.logger.error(f"Error during parsing with specific parser: {str(e)}")
+            self.logger.warning("Falling back to generic parser...")
+            # If specific parser fails, fall back to generic parser
+            content_list, md_content = MineruParser.parse_document(
+                file_path, parse_method=parse_method, output_dir=output_dir
+            )
+
+        self.logger.info(
+            f"Parsing complete! Extracted {len(content_list)} content blocks"
+        )
+        self.logger.info(f"Markdown text length: {len(md_content)} characters")
+
+        # Display content statistics if requested
+        if display_stats:
+            self.logger.info("\nContent Information:")
+            self.logger.info(f"* Total blocks in content_list: {len(content_list)}")
+            self.logger.info(f"* Markdown content length: {len(md_content)} characters")
+
+            # Count elements by type
+            block_types: Dict[str, int] = {}
+            for block in content_list:
+                if isinstance(block, dict):
+                    block_type = block.get("type", "unknown")
+                    if isinstance(block_type, str):
+                        block_types[block_type] = block_types.get(block_type, 0) + 1
+
+            self.logger.info("* Content block types:")
+            for block_type, count in block_types.items():
+                self.logger.info(f"  - {block_type}: {count}")
+
+        return content_list, md_content
+
+    def _separate_content(
+        self, content_list: List[Dict[str, Any]]
+    ) -> Tuple[str, List[Dict[str, Any]]]:
+        """
+        Separate text content and multimodal content
+
+        Args:
+            content_list: Content list from MinerU parsing
+
+        Returns:
+            (text_content, multimodal_items): Pure text content and multimodal items list
+        """
+        text_parts = []
+        multimodal_items = []
+
+        for item in content_list:
+            content_type = item.get("type", "text")
+
+            if content_type == "text":
+                # Text content
+                text = item.get("text", "")
+                if text.strip():
+                    text_parts.append(text)
+            else:
+                # Multimodal content (image, table, equation, etc.)
+                multimodal_items.append(item)
+
+        # Merge all text content
+        text_content = "\n\n".join(text_parts)
+
+        self.logger.info("Content separation complete:")
+        self.logger.info(f"  - Text content length: {len(text_content)} characters")
+        self.logger.info(f"  - Multimodal items count: {len(multimodal_items)}")
+
+        # Count multimodal types
+        modal_types = {}
+        for item in multimodal_items:
+            modal_type = item.get("type", "unknown")
+            modal_types[modal_type] = modal_types.get(modal_type, 0) + 1
+
+        if modal_types:
+            self.logger.info(f"  - Multimodal type distribution: {modal_types}")
+
+        return text_content, multimodal_items
+
+    async def _insert_text_content(
+        self,
+        input: str | list[str],
+        split_by_character: str | None = None,
+        split_by_character_only: bool = False,
+        ids: str | list[str] | None = None,
+        file_paths: str | list[str] | None = None,
+    ):
+        """
+        Insert pure text content into LightRAG
+
+        Args:
+            input: Single document string or list of document strings
+            split_by_character: if split_by_character is not None, split the string by character, if chunk longer than
+            chunk_token_size, it will be split again by token size.
+            split_by_character_only: if split_by_character_only is True, split the string by character only, when
+            split_by_character is None, this parameter is ignored.
+            ids: single string of the document ID or list of unique document IDs, if not provided, MD5 hash IDs will be generated
+            file_paths: single string of the file path or list of file paths, used for citation
+        """
+        self.logger.info("Starting text content insertion into LightRAG...")
+
+        # Use LightRAG's insert method with all parameters
+        await self.lightrag.ainsert(
+            input=input,
+            file_paths=file_paths,
+            split_by_character=split_by_character,
+            split_by_character_only=split_by_character_only,
+            ids=ids,
+        )
+
+        self.logger.info("Text content insertion complete")
+
+    async def _process_multimodal_content(
+        self, multimodal_items: List[Dict[str, Any]], file_path: str
+    ):
+        """
+        Process multimodal content (using specialized processors)
+
+        Args:
+            multimodal_items: List of multimodal items
+            file_path: File path (for reference)
+        """
+        if not multimodal_items:
+            self.logger.debug("No multimodal content to process")
+            return
+
+        self.logger.info("Starting multimodal content processing...")
+
+        file_name = os.path.basename(file_path)
+
+        for i, item in enumerate(multimodal_items):
+            try:
+                content_type = item.get("type", "unknown")
+                self.logger.info(
+                    f"Processing item {i+1}/{len(multimodal_items)}: {content_type} content"
+                )
+
+                # Select appropriate processor
+                processor = self._get_processor_for_type(content_type)
+
+                if processor:
+                    (
+                        enhanced_caption,
+                        entity_info,
+                    ) = await processor.process_multimodal_content(
+                        modal_content=item,
+                        content_type=content_type,
+                        file_path=file_name,
+                    )
+                    self.logger.info(
+                        f"{content_type} processing complete: {entity_info.get('entity_name', 'Unknown')}"
+                    )
+                else:
+                    self.logger.warning(
+                        f"No suitable processor found for {content_type} type content"
+                    )
+
+            except Exception as e:
+                self.logger.error(f"Error processing multimodal content: {str(e)}")
+                self.logger.debug("Exception details:", exc_info=True)
+                continue
+
+        self.logger.info("Multimodal content processing complete")
+
+    def _get_processor_for_type(self, content_type: str):
+        """
+        Get appropriate processor based on content type
+
+        Args:
+            content_type: Content type
+
+        Returns:
+            Corresponding processor instance
+        """
+        # Direct mapping to corresponding processor
+        if content_type == "image":
+            return self.modal_processors.get("image")
+        elif content_type == "table":
+            return self.modal_processors.get("table")
+        elif content_type == "equation":
+            return self.modal_processors.get("equation")
+        else:
+            # For other types, use generic processor
+            return self.modal_processors.get("generic")
+
+    async def process_document_complete(
+        self,
+        file_path: str,
+        output_dir: str = "./output",
+        parse_method: str = "auto",
+        display_stats: bool = True,
+        split_by_character: str | None = None,
+        split_by_character_only: bool = False,
+        doc_id: str | None = None,
+    ):
+        """
+        Complete document processing workflow
+
+        Args:
+            file_path: Path to the file to process
+            output_dir: MinerU output directory
+            parse_method: Parse method
+            display_stats: Whether to display content statistics
+            split_by_character: Optional character to split the text by
+            split_by_character_only: If True, split only by the specified character
+            doc_id: Optional document ID, if not provided MD5 hash will be generated
+        """
+        # Ensure LightRAG is initialized
+        await self._ensure_lightrag_initialized()
+
+        self.logger.info(f"Starting complete document processing: {file_path}")
+
+        # Step 1: Parse document using MinerU
+        content_list, md_content = self.parse_document(
+            file_path, output_dir, parse_method, display_stats
+        )
+
+        # Step 2: Separate text and multimodal content
+        text_content, multimodal_items = self._separate_content(content_list)
+
+        # Step 3: Insert pure text content with all parameters
+        if text_content.strip():
+            file_name = os.path.basename(file_path)
+            await self._insert_text_content(
+                text_content,
+                file_paths=file_name,
+                split_by_character=split_by_character,
+                split_by_character_only=split_by_character_only,
+                ids=doc_id,
+            )
+
+        # Step 4: Process multimodal content (using specialized processors)
+        if multimodal_items:
+            await self._process_multimodal_content(multimodal_items, file_path)
+
+        self.logger.info(f"Document {file_path} processing complete!")
+
+    async def process_folder_complete(
+        self,
+        folder_path: str,
+        output_dir: str = "./output",
+        parse_method: str = "auto",
+        display_stats: bool = False,
+        split_by_character: str | None = None,
+        split_by_character_only: bool = False,
+        file_extensions: Optional[List[str]] = None,
+        recursive: bool = True,
+        max_workers: int = 1,
+    ):
+        """
+        Process all files in a folder in batch
+
+        Args:
+            folder_path: Path to the folder to process
+            output_dir: MinerU output directory
+            parse_method: Parse method
+            display_stats: Whether to display content statistics for each file (recommended False for batch processing)
+            split_by_character: Optional character to split text by
+            split_by_character_only: If True, split only by the specified character
+            file_extensions: List of file extensions to process, e.g. [".pdf", ".docx"]. If None, process all supported formats
+            recursive: Whether to recursively process subfolders
+            max_workers: Maximum number of concurrent workers
+        """
+        # Ensure LightRAG is initialized
+        await self._ensure_lightrag_initialized()
+
+        folder_path = Path(folder_path)
+        if not folder_path.exists() or not folder_path.is_dir():
+            raise ValueError(
+                f"Folder does not exist or is not a valid directory: {folder_path}"
+            )
+
+        # Supported file formats
+        supported_extensions = {
+            ".pdf",
+            ".jpg",
+            ".jpeg",
+            ".png",
+            ".bmp",
+            ".tiff",
+            ".tif",
+            ".doc",
+            ".docx",
+            ".ppt",
+            ".pptx",
+            ".txt",
+            ".md",
+        }
+
+        # Use specified extensions or all supported formats
+        if file_extensions:
+            target_extensions = set(ext.lower() for ext in file_extensions)
+            # Validate if all are supported formats
+            unsupported = target_extensions - supported_extensions
+            if unsupported:
+                self.logger.warning(
+                    f"The following file formats may not be fully supported: {unsupported}"
+                )
+        else:
+            target_extensions = supported_extensions
+
+        # Collect all files to process
+        files_to_process = []
+
+        if recursive:
+            # Recursively traverse all subfolders
+            for file_path in folder_path.rglob("*"):
+                if (
+                    file_path.is_file()
+                    and file_path.suffix.lower() in target_extensions
+                ):
+                    files_to_process.append(file_path)
+        else:
+            # Process only current folder
+            for file_path in folder_path.glob("*"):
+                if (
+                    file_path.is_file()
+                    and file_path.suffix.lower() in target_extensions
+                ):
+                    files_to_process.append(file_path)
+
+        if not files_to_process:
+            self.logger.info(f"No files to process found in {folder_path}")
+            return
+
+        self.logger.info(f"Found {len(files_to_process)} files to process")
+        self.logger.info("File type distribution:")
+
+        # Count file types
+        file_type_count = {}
+        for file_path in files_to_process:
+            ext = file_path.suffix.lower()
+            file_type_count[ext] = file_type_count.get(ext, 0) + 1
+
+        for ext, count in sorted(file_type_count.items()):
+            self.logger.info(f"  {ext}: {count} files")
+
+        # Create progress tracking
+        processed_count = 0
+        failed_files = []
+
+        # Use semaphore to control concurrency
+        semaphore = asyncio.Semaphore(max_workers)
+
+        async def process_single_file(file_path: Path, index: int) -> None:
+            """Process a single file"""
+            async with semaphore:
+                nonlocal processed_count
+                try:
+                    self.logger.info(
+                        f"[{index}/{len(files_to_process)}] Processing: {file_path}"
+                    )
+
+                    # Create separate output directory for each file
+                    file_output_dir = Path(output_dir) / file_path.stem
+                    file_output_dir.mkdir(parents=True, exist_ok=True)
+
+                    # Process file
+                    await self.process_document_complete(
+                        file_path=str(file_path),
+                        output_dir=str(file_output_dir),
+                        parse_method=parse_method,
+                        display_stats=display_stats,
+                        split_by_character=split_by_character,
+                        split_by_character_only=split_by_character_only,
+                    )
+
+                    processed_count += 1
+                    self.logger.info(
+                        f"[{index}/{len(files_to_process)}] Successfully processed: {file_path}"
+                    )
+
+                except Exception as e:
+                    self.logger.error(
+                        f"[{index}/{len(files_to_process)}] Failed to process: {file_path}"
+                    )
+                    self.logger.error(f"Error: {str(e)}")
+                    failed_files.append((file_path, str(e)))
+
+        # Create all processing tasks
+        tasks = []
+        for index, file_path in enumerate(files_to_process, 1):
+            task = process_single_file(file_path, index)
+            tasks.append(task)
+
+        # Wait for all tasks to complete
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+        # Output processing statistics
+        self.logger.info("\n===== Batch Processing Complete =====")
+        self.logger.info(f"Total files: {len(files_to_process)}")
+        self.logger.info(f"Successfully processed: {processed_count}")
+        self.logger.info(f"Failed: {len(failed_files)}")
+
+        if failed_files:
+            self.logger.info("\nFailed files:")
+            for file_path, error in failed_files:
+                self.logger.info(f"  - {file_path}: {error}")
+
+        return {
+            "total": len(files_to_process),
+            "success": processed_count,
+            "failed": len(failed_files),
+            "failed_files": failed_files,
+        }
+
+    async def query_with_multimodal(self, query: str, mode: str = "hybrid") -> str:
+        """
+        Query with multimodal content support
+
+        Args:
+            query: Query content
+            mode: Query mode
+
+        Returns:
+            Query result
+        """
+        if self.lightrag is None:
+            raise ValueError(
+                "No LightRAG instance available. "
+                "Please either:\n"
+                "1. Provide a pre-initialized LightRAG instance when creating RAGAnything, or\n"
+                "2. Process documents first using process_document_complete() or process_folder_complete() "
+                "to create and populate the LightRAG instance."
+            )
+
+        result = await self.lightrag.aquery(query, param=QueryParam(mode=mode))
+
+        return result
+
+    def get_processor_info(self) -> Dict[str, Any]:
+        """Get processor information"""
+        if not self.modal_processors:
+            return {"status": "Not initialized"}
+
+        info = {
+            "status": "Initialized",
+            "processors": {},
+            "models": {
+                "llm_model": "External function"
+                if self.llm_model_func
+                else "Not provided",
+                "vision_model": "External function"
+                if self.vision_model_func
+                else "Not provided",
+                "embedding_model": "External function"
+                if self.embedding_func
+                else "Not provided",
+            },
+        }
+
+        for proc_type, processor in self.modal_processors.items():
+            info["processors"][proc_type] = {
+                "class": processor.__class__.__name__,
+                "supports": self._get_processor_supports(proc_type),
+            }
+
+        return info
+
+    def _get_processor_supports(self, proc_type: str) -> List[str]:
+        """Get processor supported features"""
+        supports_map = {
+            "image": [
+                "Image content analysis",
+                "Visual understanding",
+                "Image description generation",
+                "Image entity extraction",
+            ],
+            "table": [
+                "Table structure analysis",
+                "Data statistics",
+                "Trend identification",
+                "Table entity extraction",
+            ],
+            "equation": [
+                "Mathematical formula parsing",
+                "Variable identification",
+                "Formula meaning explanation",
+                "Formula entity extraction",
+            ],
+            "generic": [
+                "General content analysis",
+                "Structured processing",
+                "Entity extraction",
+            ],
+        }
+        return supports_map.get(proc_type, ["Basic processing"])