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

> Instrument RAGAnything pipelines with callback hooks for enhanced observability and metrics. Easily track parsing, multimodal processing, and querying stages without core code changes.

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

---

**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)](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)](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/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/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/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/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/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/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/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/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/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/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/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/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/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/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/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

```python
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:

```python
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

```python
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:

```python
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

```python
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

```python

# Iterate through all recorded events

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

```

Export to external systems:

```python
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:

```python
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:

```python
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)](https://github.com/HKUDS/RAG-Anything/blob/main/raganything/batch.py):

```python
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:

```python
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`](https://github.com/HKUDS/RAG-Anything/blob/main/processor.py), [`query.py`](https://github.com/HKUDS/RAG-Anything/blob/main/query.py), or [`batch.py`](https://github.com/HKUDS/RAG-Anything/blob/main/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.