How to Implement Custom Modal Processors in RAGAnything for New Content Types

To implement custom modal processors in RAGAnything, create a subclass of BaseModalProcessor in modalprocessors.py, implement generate_description_only and process_multimodal_content, then wire it into the pipeline via RAGAnything._initialize_processors and utils.get_processor_for_type.

RAGAnything is a multimodal RAG framework that processes heterogeneous content—images, tables, equations, and more—through a unified pipeline. Each content type is handled by a dedicated modal processor that transforms raw data into searchable text chunks and knowledge graph entities. This guide walks you through extending RAGAnything with custom modal processors for new content types, using the actual source architecture from HKUDS/RAG-Anything.

Understanding the BaseModalProcessor Architecture

The foundation of every modal processor is BaseModalProcessor, defined in raganything/modalprocessors.py (lines 360–462). This abstract base class provides:

• Shared infrastructure: References to the LightRAG instance, vector stores, and LLM functions
• Context extraction: The _get_context_for_item method leverages ContextExtractor to pull surrounding text, captions, or metadata
• Robust parsing: Utility methods like _robust_json_parse and _strip_thinking_tags for handling LLM responses
• Abstract API: Two methods you must implement—generate_description_only and process_multimodal_content

All existing processors (ImageModalProcessor, TableModalProcessor, EquationModalProcessor, GenericModalProcessor) inherit from this base and follow the same contract.

Leveraging ContextExtractor for Rich Modal Processing

The ContextExtractor class (lines 49–338 in modalprocessors.py) enables processors to gather contextual information from surrounding document content. Key capabilities include:

• Configurable extraction: Controlled by ContextConfig in raganything/config.py with settings for context_window, include_captions, and more
• Flexible source handling: Works with minerU output lists, plain text, and dictionary structures
• Simple processor integration: Call self._get_context_for_item(item_info) from within your processor to obtain a context string

This context can be interpolated into your LLM prompts to generate more accurate, grounded descriptions for your custom content type.

Step-by-Step: Creating a Custom Modal Processor

Step 1: Subclass BaseModalProcessor

Create a new file or add to modalprocessors.py. Here's a complete AudioModalProcessor implementation:

# examples/audio_processor.py

from typing import Any, Dict, Tuple
from raganything.modalprocessors import BaseModalProcessor
from raganything.prompt import PROMPTS

class AudioModalProcessor(BaseModalProcessor):
    """Processor for audio content (e.g., speech or music)."""

    async def generate_description_only(
        self,
        modal_content: Any,
        content_type: str,
        item_info: Dict[str, Any] = None,
        entity_name: str = None,
    ) -> Tuple[str, Dict[str, Any]]:
        """
        Turn an audio file into a textual description and entity record.
        `modal_content` is expected to be a dict with ``audio_path`` and optional
        ``metadata`` fields.
        """
        # Parse the incoming payload

        if isinstance(modal_content, str):
            import json
            modal_content = json.loads(modal_content)

        audio_path = modal_content.get("audio_path")
        metadata = modal_content.get("metadata", {})

        if not audio_path:
            raise ValueError("AudioModalProcessor requires an ``audio_path`` key")

        # Grab surrounding context if available

        context = ""
        if item_info:
            context = self._get_context_for_item(item_info)

        # Build the LLM prompt

        if context:
            prompt = PROMPTS.get(
                "generic_prompt_with_context", PROMPTS["generic_prompt"]
            ).format(
                context=context,
                content_type=content_type,
                entity_name=entity_name or "audio_clip",
                content=f"Audio file: {audio_path}\nMetadata: {metadata}",
            )
        else:
            prompt = PROMPTS["generic_prompt"].format(
                content_type=content_type,
                entity_name=entity_name or "audio_clip",
                content=f"Audio file: {audio_path}\nMetadata: {metadata}",
            )

        # Call the LLM

        response = await self.modal_caption_func(
            prompt,
            system_prompt=PROMPTS["GENERIC_ANALYSIS_SYSTEM"].format(content_type=content_type),
        )

        # Parse using the base class utility

        description, entity_info = self._parse_generic_response(response, entity_name, content_type)

        return description, entity_info

    async def process_multimodal_content(
        self,
        modal_content: Any,
        content_type: str,
        file_path: str = "manual_creation",
        entity_name: str = None,
        item_info: Dict[str, Any] = None,
        batch_mode: bool = False,
        doc_id: str = None,
        chunk_order_index: int = 0,
    ) -> Tuple[str, Dict[str, Any]]:
        """Full pipeline – generate description then store the chunk."""
        description, entity_info = await self.generate_description_only(
            modal_content, content_type, item_info, entity_name
        )

        # Build the final chunk for indexing

        modal_chunk = PROMPTS["generic_chunk"].format(
            content_type=content_type.title(),
            content=str(modal_content),
            enhanced_caption=description,
        )

        return await self._create_entity_and_chunk(
            modal_chunk,
            entity_info,
            file_path,
            batch_mode,
            doc_id,
            chunk_order_index,
        )

Step 2: Register in _initialize_processors

Add your processor to raganything/raganything.py (lines 14–42):

# Inside RAGAnything._initialize_processors

if self.config.enable_audio_processing:  # Add this config option

    self.modal_processors["audio"] = AudioModalProcessor(
        lightrag=self.lightrag,
        modal_caption_func=self.llm_model_func,
        context_extractor=self.context_extractor,
    )

Step 3: Update get_processor_for_type

Extend raganything/utils.py (lines 28–48):

# Inside get_processor_for_type

elif content_type == "audio":
    return modal_processors.get("audio")

Step 4: Add Configuration Toggle

Add to raganything/config.py:

enable_audio_processing: bool = field(
    default=get_env_value("ENABLE_AUDIO_PROCESSING", True, bool)
)
"""Enable audio content processing."""

Using Your Custom Processor

Once registered, the core pipeline automatically invokes your processor for matching content types:

import asyncio
from raganything.raganything import RAGAnything
from raganything.config import RAGAnythingConfig

async def demo():
    rag = RAGAnything(
        config=RAGAnythingConfig(enable_audio_processing=True)
    )
    await rag._ensure_lightrag_initialized()

    # Simulate parsed multimodal payload

    multimodal = [
        {
            "type": "audio",
            "audio_path": "samples/lecture.wav",
            "metadata": {"duration_s": 360, "speaker": "Prof. X"},
            "page_idx": 2,
        }
    ]

    # Process – RAGAnything selects AudioModalProcessor automatically

    await rag.process_multimodal_content(multimodal, file_path="lecture.pdf")

asyncio.run(demo())

The pipeline inserts a text chunk describing the audio, creates an entity node (e.g., lecture.wav (audio)), and links it to surrounding document context using the same graph-building logic as images and tables.

Key Files Reference

File	Role	Link
raganything/modalprocessors.py	Core class hierarchy (BaseModalProcessor, ContextExtractor, built-in processors)	[modalprocessors.py]
raganything/raganything.py	Orchestrates processor creation, stores modal_processors dict, runs the pipeline	[raganything.py]
raganything/utils.py	Helper get_processor_for_type that selects a processor at runtime	[utils.py]
raganything/config.py	Configuration flags, including optional enable_*_processing switches	[config.py]
examples/modalprocessors_example.py	Usage examples of existing processors; reference for new implementations	[modalprocessors_example.py]
tests/test_strip_thinking_tags.py	Unit tests for base class utilities – template for testing new processors	[test_strip_thinking_tags.py]

Summary

• Subclass BaseModalProcessor in modalprocessors.py to implement generate_description_only and process_multimodal_content for your new content type.
• Wire into the pipeline by instantiating in RAGAnything._initialize_processors and extending utils.get_processor_for_type to recognize your content_type string.
• Leverage ContextExtractor via _get_context_for_item to enrich descriptions with surrounding document context.
• Add configuration toggles in config.py for optional processor enablement.
• Follow existing patterns in ImageModalProcessor and TableModalProcessor for consistent behavior.

Frequently Asked Questions

What is the minimum code needed to implement a custom modal processor?

The minimum implementation requires a class inheriting from BaseModalProcessor with two async methods: generate_description_only to produce a text description from your raw content, and process_multimodal_content to orchestrate storage via _create_entity_and_chunk. The base class handles vector database operations, graph creation, and context extraction—you only write the modality-specific logic.

How does RAGAnything decide which processor to use for each content item?

The selection happens in utils.get_processor_for_type (lines 28–48), which receives the content_type string from a parsed document item and returns the matching processor from the modal_processors dictionary. This dictionary is populated during RAGAnything._initialize_processors based on configuration flags, so adding your processor to both locations ensures automatic routing.

Can I access surrounding text or captions from my custom processor?

Yes. The BaseModalProcessor provides _get_context_for_item(item_info), which internally uses the shared ContextExtractor instance. Call this method with the item_info dict (containing keys like page_idx, bbox, or custom metadata) to retrieve a formatted string of relevant surrounding content. This context can then be interpolated into your LLM prompts.

What configuration options should I expose for my custom processor?

Follow the pattern used by built-in processors: add a boolean flag like enable_yourmodality_processing to RAGAnythingConfig in config.py, with a default value and environment variable override via get_env_value. Then check this flag in RAGAnything._initialize_processors before instantiating your processor. This keeps your integration consistent with the RAGAnything configuration system.

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