How to Use Callback Hooks in RAGAnything for Pipeline Observability and Metrics

Register a ProcessingCallback subclass with RAGAnything.callback_manager to instrument every pipeline stage—parsing, multimodal processing, text insertion, and querying—without modifying core code.

RAGAnything provides a lightweight publish-subscribe callback system that enables comprehensive observability and metrics collection for document processing pipelines. By hooking into the CallbackManager attached to every RAGAnything instance, you can capture timing data, track success rates, and build detailed audit trails for external monitoring systems.

Core Architecture of RAGAnything Callback Hooks

The callback system centers on three main components defined in [raganything/callbacks.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/callbacks.py):

Component	Purpose
ProcessingCallback	Abstract base class with on_* hook methods; subclass and override only what you need
CallbackManager	Thread-safe registry that dispatches events to all registered callbacks; optionally logs every event as a ProcessingEvent
MetricsCallback	Built-in implementation that aggregates counters and provides summary() and reset() methods

Every RAGAnything instance automatically creates a callback_manager field during construction, as defined in [raganything/raganything.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/raganything.py). Pipeline code throughout the repository checks for its presence via getattr(self, "callback_manager", None) before dispatching events.

Available Callback Hooks and Lifecycle Events

The RAGAnything pipeline dispatches granular events at every processing stage. Override these methods in your ProcessingCallback subclass to capture specific observability data:

Document Parsing Hooks

Hook	Trigger Location	Data Provided
on_parse_start	[processor.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/processor.py)	file_path, parser
on_parse_complete	[processor.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/processor.py)	file_path, content_blocks, doc_id, duration_seconds
on_parse_error	[processor.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/processor.py)	file_path, error, parser

Text Insertion Hooks

Hook	Trigger Location	Data Provided
on_text_insert_start	[processor.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/processor.py)	file_path, text_length
on_text_insert_complete	[processor.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/processor.py)	file_path, duration_seconds

Multimodal Processing Hooks

Hook	Trigger Location	Data Provided
on_multimodal_start	[processor.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/processor.py)	file_path, item_count
on_multimodal_item_complete	[processor.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/processor.py)	file_path, item_index, item_type, processed_count
on_multimodal_complete	[processor.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/processor.py)	file_path, duration_seconds

Query Execution Hooks

Hook	Trigger Location	Data Provided
on_query_start	[query.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/query.py)	query, mode
on_query_complete	[query.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/query.py)	query, duration_seconds, result_length
on_query_error	[query.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/query.py)	query, error

Lifecycle Completion Hooks

Hook	Trigger Location	Data Provided
on_document_complete	[processor.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/processor.py)	file_path, doc_id, duration_seconds
on_document_error	[processor.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/processor.py)	file_path, error, stage
on_batch_start	[batch.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch.py)	file_count, total_files
on_batch_complete	[batch.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch.py)	total_files, successful, failed, duration_seconds

Unknown event names are silently ignored, ensuring forward compatibility as the pipeline evolves.

Building a Custom Callback for Observability

Create a subclass of ProcessingCallback and override only the methods relevant to your monitoring needs. The CallbackManager passes all keyword arguments via **kw, so your methods should accept **kw to remain compatible with future event additions.

Example: Timestamp Logger Callback

import time
from raganything.callbacks import ProcessingCallback


class TimingCallback(ProcessingCallback):
    """Emits timestamped events for parsing and query operations."""
    
    def on_parse_start(self, file_path, **kw):
        print(f"[{time.time():.3f}] Parsing started: {file_path}")
    
    def on_parse_complete(self, file_path, duration_seconds=0, **kw):
        print(f"[{time.time():.3f}] Parsing finished ({duration_seconds:.2f}s): {file_path}")
    
    def on_query_start(self, query, **kw):
        print(f"[{time.time():.3f}] Query started: {query}")
    
    def on_query_complete(self, query, duration_seconds=0, **kw):
        print(f"[{time.time():.3f}] Query finished ({duration_seconds:.2f}s): {query}")

Register with your RAGAnything instance:

from raganything import RAGAnything, RAGAnythingConfig

cfg = RAGAnythingConfig()
rag = RAGAnything(config=cfg)

rag.callback_manager.register(TimingCallback())

Once registered, all parsing and query operations automatically trigger your callback methods.

Using the Built-In MetricsCallback for Aggregated Statistics

The MetricsCallback class provides zero-configuration metrics aggregation for pipeline observability. It automatically tracks counters and timing data across all document processing stages.

Metrics Tracked

Metric	Description
documents_processed	Successfully completed documents
documents_failed	Documents that failed at any stage
total_content_blocks	Sum of parsed content blocks
total_multimodal_items	Sum of processed multimodal items
total_parse_time	Cumulative seconds spent parsing
total_insert_time	Cumulative seconds spent in text insertion
total_multimodal_time	Cumulative seconds spent in multimodal processing
queries_executed	Total query operations completed
total_query_time	Cumulative seconds spent querying
errors	List of error records with context

Basic Usage

from raganything import RAGAnything, RAGAnythingConfig
from raganything.callbacks import MetricsCallback

cfg = RAGAnythingConfig()
rag = RAGAnything(config=cfg)

metrics = MetricsCallback()
rag.callback_manager.register(metrics)

# Run your pipeline...

# rag.insert_file("document.pdf")

# result = await rag.aquery("What is the main topic?")

# Inspect collected metrics

print(metrics.summary())

Output Format

The summary() method returns a human-readable formatted string:

Documents: 15 processed, 2 failed
Content blocks: 47
Parse time: 12.45s
Insert time: 3.21s
Multimodal items: 8 (time: 5.67s)
Queries: 23 (time: 8.90s)
Errors: 2

Reset counters for a new run:

metrics.reset()  # All counters return to zero

The MetricsCallback requires no manual method calls—CallbackManager automatically invokes the appropriate hook methods during pipeline execution.

Enabling and Inspecting the Event Log

For complete audit trails or integration with external monitoring systems, enable the full event log. This captures every dispatched event as an immutable ProcessingEvent record.

Activation

rag.callback_manager.enable_event_log(True)

Once enabled, every callback dispatch creates a ProcessingEvent containing:

Field	Description
event_name	Hook name (e.g., "on_parse_complete")
timestamp	ISO-format datetime
file_path	Source document path (if applicable)
duration_seconds	Operation timing (if applicable)
error	Exception message (if applicable)
Additional context	Varies by event type

Accessing the Log

# Iterate through all recorded events

for event in rag.callback_manager.event_log:
    print(event.to_dict())

Export to external systems:

import json

events = [ev.to_dict() for ev in rag.callback_manager.event_log]
with open("pipeline_audit.jsonl", "w") as f:
    for ev in events:
        f.write(json.dumps(ev) + "\n")

Disable logging when no longer needed:

rag.callback_manager.enable_event_log(False)  # Stops recording new events

The event log is read-only; you cannot modify recorded events. This immutability ensures reliable audit trails for compliance and debugging.

Advanced Use Cases and Best Practices

Combining Multiple Callbacks

Register multiple callbacks for different concerns—one for metrics, one for custom logging, one for error alerting:

from raganything.callbacks import MetricsCallback, ProcessingCallback

class ErrorAlertCallback(ProcessingCallback):
    def on_document_error(self, file_path, error, stage, **kw):
        send_alert(f"Document failed at {stage}: {file_path} - {error}")
    
    def on_query_error(self, query, error, **kw):
        send_alert(f"Query failed: {query} - {error}")

metrics = MetricsCallback()
alerts = ErrorAlertCallback()

rag.callback_manager.register(metrics)
rag.callback_manager.register(alerts)

Batch Processing Monitoring

For large-scale document ingestion, use batch-level hooks. These are dispatched from [raganything/batch.py](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch.py):

class BatchProgressCallback(ProcessingCallback):
    def on_batch_start(self, file_count=0, total_files=0, **kw):
        self.start_time = time.time()
        print(f"Starting batch: {file_count}/{total_files} files")
    
    def on_batch_complete(self, total_files=0, successful=0, failed=0,
                          duration_seconds=0, **kw):
        rate = successful / duration_seconds if duration_seconds > 0 else 0
        print(f"Batch complete: {successful}/{total_files} at {rate:.2f} docs/sec")

Integrating with External Observability Platforms

Export MetricsCallback data or event logs to Prometheus, OpenTelemetry, or cloud monitoring:

from prometheus_client import Counter, Histogram, push_to_gateway

class PrometheusCallback(ProcessingCallback):
    def __init__(self):
        self.docs_processed = Counter("raganything_docs_total", "Documents processed")
        self.parse_time = Histogram("raganything_parse_seconds", "Parse duration")
        self.query_time = Histogram("raganything_query_seconds", "Query duration")
    
    def on_parse_complete(self, duration_seconds=0, **kw):
        self.docs_processed.inc()
        self.parse_time.observe(duration_seconds)
    
    def on_query_complete(self, duration_seconds=0, **kw):
        self.query_time.observe(duration_seconds)

Summary

• RAGAnything callback hooks enable non-invasive pipeline observability through the ProcessingCallback abstract base class and CallbackManager registry.

• Register callbacks via rag.callback_manager.register() to capture events at every stage: parsing, text insertion, multimodal processing, and querying.

• Use MetricsCallback for automatic aggregation of document counts, timing data, and error rates—access results with summary() and reset with reset().

• Enable event logging with enable_event_log(True) to capture immutable ProcessingEvent records for complete audit trails and external system integration.

• Combine multiple callbacks for different concerns—metrics, alerting, custom logging—without modifying core pipeline code in processor.py, query.py, or batch.py.

Frequently Asked Questions

What is the difference between MetricsCallback and custom ProcessingCallback subclasses?

MetricsCallback is a ready-made implementation that automatically tracks predefined metrics like document counts, parse times, and query latencies. Custom ProcessingCallback subclasses let you define arbitrary behavior—custom logging, external API calls, or specialized metrics—by overriding only the hook methods you need. Both register with CallbackManager the same way.

Can I use multiple callbacks simultaneously on the same RAGAnything instance?

Yes. CallbackManager.register() accepts any number of callbacks, and all registered callbacks receive every dispatched event. This lets you combine MetricsCallback for aggregated statistics, a custom callback for error alerting, and another for progress logging—all operating independently on the same pipeline executions.

How do I persist callback data across process restarts?

The callback system itself is in-memory and stateless across restarts. To persist data, export MetricsCallback.summary() results or CallbackManager.event_log entries to external storage before shutdown. For ProcessingEvent logs, iterate through event_log and serialize each event.to_dict() to JSON, then write to a file or stream to a database.

What performance overhead does enabling callbacks add?

The overhead is minimal. CallbackManager uses simple iteration over registered callbacks, and event logging is disabled by default. When enabled, ProcessingEvent creation adds a small allocation cost per dispatch. For high-throughput scenarios, use MetricsCallback without event logging, or implement custom callbacks that sample events rather than capturing every one.

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