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

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 and 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
BatchParser	Thread pool orchestration, file discovery, progress visualization, timeout handling	raganything/batch_parser.py
BatchProcessingResult	Result aggregation with successful_files, failed_files, success_rate, and summary()	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:

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 (lines 122-176).

Asynchronous Batch Processing

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

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 (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:

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 (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:

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 (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:

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, lines 83-88).

Error Handling and Retry Patterns

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

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 (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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}