# How to Utilize Batch Processing for Concurrent Document Processing in RAG-Anything

> Learn to use batch processing in RAG-Anything for concurrent document parsing with thread pools. Features progress tracking, error handling, and seamless RAG pipeline integration.

- Repository: [✨Data Intelligence Lab@HKU✨/RAG-Anything](https://github.com/HKUDS/RAG-Anything)
- Tags: how-to-guide
- Published: 2026-04-22

---

**Batch processing in RAG-Anything enables parallel parsing of hundreds of documents using thread pools, with built-in progress tracking, error handling, and optional direct RAG pipeline integration.**

The RAG-Anything library provides a production-ready batch processing subsystem that eliminates the bottleneck of sequential document parsing. Whether you're processing a directory of PDFs, OCR-scanned images, or mixed office formats, the system scales horizontally through configurable worker threads while maintaining granular visibility into successes and failures.

## Core Architecture of RAG-Anything Batch Processing

The batch processing capability rests on three integrated components implemented in [`raganything/batch.py`](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch.py) and [`raganything/batch_parser.py`](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch_parser.py):

| Component | Responsibility | Source Location |
|-----------|---------------|---------------|
| **`BatchMixin`** | High-level API methods (`process_documents_batch`, `process_documents_batch_async`, `process_documents_with_rag_batch`) | [`raganything/batch.py`](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch.py) |
| **`BatchParser`** | Thread pool orchestration, file discovery, progress visualization, timeout handling | [`raganything/batch_parser.py`](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch_parser.py) |
| **`BatchProcessingResult`** | Result aggregation with `successful_files`, `failed_files`, `success_rate`, and `summary()` | [`raganything/batch_parser.py`](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch_parser.py) |

### Execution Flow

When you invoke batch processing in RAG-Anything, the system executes this pipeline:

1. **Configuration resolution** — `BatchMixin` extracts defaults from `RAGAnythingConfig` (worker count, parser type, output directory)
2. **Parser instantiation** — `BatchParser` creates the underlying parser (`mineru`, `docling`, or `paddleocr`) via `get_parser`
3. **File discovery** — `filter_supported_files` expands directories and filters by supported extensions: `OFFICE_FORMATS`, `IMAGE_FORMATS`, `TEXT_FORMATS`, and `.pdf`
4. **Parallel execution** — `ThreadPoolExecutor` distributes `process_single_file` across workers; each worker creates document-specific output subdirectories
5. **Result aggregation** — `BatchProcessingResult` captures timing, success rates, and error messages

For RAG-integrated batches, the mixin additionally iterates `result.successful_files` and calls `process_document_complete` for each parsed document.

## Batch Processing Methods in RAG-Anything

### Synchronous Batch Processing

The foundational method for concurrent document processing in RAG-Anything uses `BatchParser` directly:

```python
from raganything.batch_parser import BatchParser

# Initialize with worker pool and progress display

batch = BatchParser(
    parser_type="mineru",
    max_workers=4,
    show_progress=True,
    timeout_per_file=120,
    skip_installation_check=True,
)

# Define input paths—mix files and directories

paths = [
    "data/report.pdf",
    "data/meeting_notes/",
    "data/presentation.pptx",
]

# Execute batch with recursive directory traversal

result = batch.process_batch(
    file_paths=paths,
    output_dir="parsed_output",
    parse_method="auto",
    recursive=True,
)

print(result.summary())
print(f"Success rate: {result.success_rate:.1f}%")

```

**Source:** `batch_parser.process_batch` implementation in [`raganything/batch_parser.py`](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch_parser.py) (lines 122-176).

### Asynchronous Batch Processing

For integration with async application code, RAG-Anything provides `process_batch_async`:

```python
import asyncio
from raganything.batch_parser import BatchParser

async def async_batch():
    batch = BatchParser(
        parser_type="mineru",
        max_workers=3,
        show_progress=True,
        skip_installation_check=True,
    )

    # Glob patterns are expanded automatically

    result = await batch.process_batch_async(
        file_paths=["data/**/*.pdf"],
        output_dir="async_parsed",
        parse_method="auto",
        recursive=True,
    )
    print(result.summary())

asyncio.run(async_batch())

```

**Source:** `process_batch_async` wrapper in [`raganything/batch_parser.py`](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch_parser.py) (lines 390-405), which delegates to `asyncio.get_event_loop().run_in_executor`.

### High-Level RAG-Anything API

For typical workflows, use the convenience methods mixed into `RAGAnything`:

```python
from raganything import RAGAnything, RAGAnythingConfig

cfg = RAGAnythingConfig(
    working_dir="rag_workspace",
    max_concurrent_files=4,   # Workers for batch processing

)

rag = RAGAnything(config=cfg)

# Direct batch invocation

batch_result = rag.process_documents_batch(
    file_paths=["docs/"],
    output_dir="rag_parsed",
    max_workers=4,
    show_progress=True,
)

print(f"Parsed {len(batch_result.successful_files)} files")

```

**Source:** `BatchMixin.process_documents_batch` in [`raganything/batch.py`](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch.py) (lines 174-225).

## Full RAG Integration: Parse and Ingest

The complete pipeline for concurrent document processing in RAG-Anything combines parsing with automatic RAG ingestion:

```python
import asyncio
from raganything import RAGAnything, RAGAnythingConfig

cfg = RAGAnythingConfig(
    working_dir="full_rag_ws",
    max_concurrent_files=2,
)

rag = RAGAnything(config=cfg)

async def rag_batch():
    results = await rag.process_documents_with_rag_batch(
        file_paths=["papers/"],
        output_dir="rag_full_output",
        max_workers=2,
        show_progress=True,
    )
    print(f"RAG ingest success: {results['successful_rag_files']} / "
          f"{results['total_processing_time']:.2f}s")

asyncio.run(rag_batch())

```

**Source:** `BatchMixin.process_documents_with_rag_batch` in [`raganything/batch.py`](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch.py) (lines 302-421). This method:
- Executes `process_documents_batch_async` for parsing
- Iterates `result.successful_files` for RAG ingestion via `process_document_complete`
- Dispatches optional callbacks: `on_batch_start`, `on_batch_complete`

## Directory-Wide Processing with Recursion

Process entire directory trees using the `recursive` parameter:

```python
from raganything.batch_parser import BatchParser

batch = BatchParser(parser_type="mineru", max_workers=2, show_progress=True)

result = batch.process_batch(
    file_paths=["my_project/docs/"],  # Single directory

    output_dir="dir_output",
    parse_method="auto",
    recursive=True,                   # Walk sub-folders

)

print(result.summary())

```

**Source:** The `recursive` flag controls glob behavior in `process_folder_complete` and `filter_supported_files` ([`batch_parser.py`](https://github.com/HKUDS/RAG-Anything/blob/main/batch_parser.py), lines 83-88).

## Error Handling and Retry Patterns

Robust concurrent document processing in RAG-Anything leverages `BatchProcessingResult` for failure recovery:

```python
from raganything.batch_parser import BatchParser

batch = BatchParser(parser_type="mineru", max_workers=2, show_progress=True)

# First run (some files may fail)

first = batch.process_batch(
    file_paths=["mixed/"],
    output_dir="first_run",
    parse_method="auto",
)

# Retry only the failures

if first.failed_files:
    retry = batch.process_batch(
        file_paths=first.failed_files,
        output_dir="retry_run",
        parse_method="auto",
    )
    print("Retry summary:", retry.summary())

```

**Source:** `BatchProcessingResult.failed_files` property defined at the top of [`batch_parser.py`](https://github.com/HKUDS/RAG-Anything/blob/main/batch_parser.py) (lines 21-30).

## Key Configuration Parameters

| Parameter | Default | Description |
|-----------|---------|-------------|
| `max_workers` / `max_concurrent_files` | 4 | Thread pool size for parallel processing |
| `timeout_per_file` | 120 | Seconds before individual file processing is aborted |
| `show_progress` | True | Display `tqdm` progress bar during batch execution |
| `recursive` | False | Traverse subdirectories when processing folder paths |
| `skip_installation_check` | False | Bypass parser dependency validation (useful for testing) |

## Summary

- **Batch processing in RAG-Anything** is implemented through `BatchParser` for low-level parallel execution and `BatchMixin` for high-level RAG integration
- **Three access patterns** exist: direct `BatchParser` usage, `RAGAnything` convenience methods, and full async/await support
- **Thread-based parallelism** via `ThreadPoolExecutor` enables concurrent document processing without GIL contention for I/O-bound parsing operations
- **Built-in resilience** through `BatchProcessingResult` provides granular success/failure tracking and enables retry workflows
- **RAG pipeline integration** via `process_documents_with_rag_batch` automates the transition from parsing to vector store ingestion

## Frequently Asked Questions

### What is the maximum number of concurrent files RAG-Anything can process?

The practical limit depends on your system's CPU cores, memory, and the parser's resource requirements. The default `max_workers=4` suits most workstations; GPU-accelerated parsers like `mineru` may benefit from lower concurrency to prevent VRAM exhaustion. Monitor `result.total_processing_time` to identify your optimal configuration.

### Does RAG-Anything batch processing support cloud storage paths?

The `BatchParser` accepts local file paths and directories. For cloud storage (S3, GCS, Azure Blob), mount buckets as local filesystems using `s3fs`, `gcsfuse`, or equivalent tools, then pass the mounted paths to `process_batch`. Native cloud integration is not implemented in the current version.

### How does RAG-Anything handle corrupted or unsupported files during batch processing?

Corrupted files, password-protected PDFs, and unsupported formats are captured in `BatchProcessingResult.failed_files` with descriptive error messages. The batch continues processing remaining files without interruption. Use the `failed_files` list to implement retry logic or quarantine problematic documents for manual inspection.

### Can I customize the progress bar or disable it entirely in RAG-Anything batch processing?

Set `show_progress=False` when instantiating `BatchParser` or calling `process_documents_batch` to suppress the `tqdm` progress bar. For custom progress reporting, pass callback functions via `on_batch_start` and `on_batch_complete` in `process_documents_with_rag_batch`, or wrap `BatchParser` with your own progress instrumentation.