RAG Observability Framework

A generic, configurable observability SDK for RAG (Retrieval-Augmented Generation) applications. Works with any operation names your application uses — no hardcoded metric names anywhere in the SDK.

---

Table of Contents

1. Prerequisites
2. Installation
3. How metrics are captured
4. Quick start
5. YAML configuration reference
6. Custom metrics — declarative capture
7. Grafana dashboard generation
8. Evaluation (optional)
9. Architecture
10. Testing
11. Publishing

---

Prerequisites

Before using this SDK, ensure the following infrastructure is in place.

Metrics export target (required)

The SDK exports metrics and traces via OpenTelemetry. You need at least one of the following:

Option A — Use your existing setup

If you already have an OpenTelemetry Collector, Prometheus, or compatible backend running, point the SDK at it via the exporters section of your YAML config (see YAML configuration reference).

Option B — Spin up the full observability stack with Docker

A ready-made, fully dockerised observability pipeline is available at:

github.com/CasperLabs/observability-pipeline (branch: feat/ibm-dev)

This installs and wires together:

Component	Purpose
OpenTelemetry Collector	Receives OTLP traces and metrics from the SDK
Prometheus	Scrapes and stores metrics
Victoria Metrics	Long-term metrics storage
Grafana	Dashboards and visualisation
Jaeger	Distributed trace visualisation

Follow the repo's README to bring the stack up with docker compose.

MinIO object storage (optional — for runtime data capture)

If you intend to use the runtime data capture feature, the SDK automatically selects a storage backend:

Condition	Storage used
MINIO_ENDPOINT env var set or minio_endpoint in config	MinIO — files stored as objects; export_path is used as the object key prefix inside the bucket
No MinIO credentials present	Local filesystem — export_path is used as the local directory (default: evaluation_data/captured/)

The filesystem fallback requires no additional setup and is ideal for local development and testing.

To use MinIO, set the following environment variables (or provide them in the evaluation.capture section of your YAML config):

MINIO_ENDPOINT=<host>:<port>       # e.g. minio:9000
MINIO_ACCESS_KEY=<access-key>
MINIO_SECRET_KEY=<secret-key>
MINIO_BUCKET=<bucket-name>         # e.g. rag-evaluation-data

MinIO can also be run locally via Docker:

docker run -p 9000:9000 -p 9001:9001 \
  -e MINIO_ROOT_USER=minioadmin \
  -e MINIO_ROOT_PASSWORD=minioadmin \
  quay.io/minio/minio server /data --console-address ":9001"

---

Installation

# Core SDK
pip install rag-observability

# Development
pip install rag-observability[dev]

---

How metrics are captured

Understanding this flow prevents the most common integration mistakes.

Every @obs.observe call auto-records two things

@obs.observe(operation="<name>")

1. Latency histogram — rag.<name>.latency (milliseconds, recorded on every call, successful or not).
2. Result count — rag.<name>.count (only when capture_result_count=True and the return value is a list/tuple).

These metrics are auto-registered in the in-memory registry the first time the decorated function is called. No YAML configuration is required for this to happen.

Types of metrics captured

YAML section	Effect on metric capture
metrics.custom	Yes — SDK hooks into decorated functions and records the metric declaratively from arguments/results
metrics.standard	None — metrics are already captured by @obs.observe regardless; this section is only used by the Grafana dashboard generator

Prometheus metric name produced

The SDK applies no prefix of its own. The final Prometheus metric name is shaped by two things:

1. The internal metric name — dots replaced with underscores, unit appended by the OTel SDK.
2. The namespace setting in your OTel Collector's Prometheus exporter config — this becomes the prefix. It is entirely under your control and defaults to empty (no prefix) if not set.

# otel-collector-config.yaml
exporters:
  prometheus:
    namespace: llm        # ← controls the prefix; change or omit as needed

With namespace: llm the name transforms as:

internal name:  rag.embed_time.latency   (unit = ms)
Prometheus:     llm_rag_embed_time_latency_milliseconds_bucket
                llm_rag_embed_time_latency_milliseconds_sum
                llm_rag_embed_time_latency_milliseconds_count

With no namespace set:

internal name:  rag.embed_time.latency   (unit = ms)
Prometheus:     rag_embed_time_latency_milliseconds_bucket
                rag_embed_time_latency_milliseconds_sum
                rag_embed_time_latency_milliseconds_count

Rule: dots → underscores, namespace prefix prepended (if configured), unit appended, histogram suffix added.

---

Quick start

1. Initialise the framework

from rag_observability import RAGObservability

# Option A — from a YAML config file (recommended)
obs = RAGObservability.from_config("config/rag_obs.yaml")

# Option B — defaults only (console exporter)
# Metrics are printed to stdout only. The following features are NOT available:
#   - Prometheus / Victoria Metrics export
#   - Distributed tracing in Jaeger
#   - Grafana dashboards
#   - Runtime data capture and evaluation
#   - Custom metrics from YAML
obs = RAGObservability()

Recommended: Option A. For any real deployment — development, staging, or production — use a YAML config file. Option B is only suitable for quick local debugging.

2. Decorate every RAG pipeline stage

Use @obs.observe(operation="<name>") on each function. The operation string becomes part of the Prometheus metric name — choose it to be descriptive and consistent across deployments.

@obs.observe(operation="embed_time")
async def generate_embedding(query: str) -> list[float]:
    return await embedding_model.embed(query)

@obs.observe(operation="retrieval")
async def retrieve_context(query: str, top_k: int = 5) -> list[str]:
    return await vector_store.search(query, top_k=top_k)

@obs.observe(operation="llm_generation")
async def generate_response(query: str, context: list[str]) -> str:
    return await llm.complete(query, context=context)

# capture_result_count=True also records rag.retrieval.count
@obs.observe(operation="retrieval", capture_result_count=True)
async def retrieve_context(query: str) -> list[str]:
    ...

There are no reserved or required operation names. Use whatever names make sense for your application.

3. Record custom metrics manually (optional)

# Record any value at any time
obs.record_metric("my.custom.counter", 1.0, labels={"user_id": "u123"})

# Use the trace context manager for ad-hoc spans
with obs.trace("reranking") as span:
    span["set_attribute"]("model", "cross-encoder-v1")
    results = reranker.rerank(candidates)

3a. Create a root request span

If you want multiple spans to appear under one trace in Jaeger, create one real root span at the request boundary with obs.trace(...), then execute observed calls inside that span.

@router.post("/message")
async def send_chat_message(chat_request: ChatRequest):
    with obs.trace("chat_request") as span:
        span.set_attribute("session_id", chat_request.session_id)
        span.set_attribute("user_id", chat_request.user_id)

        await retrieve_context(chat_request.message)
        await generate_response(chat_request.message, [])

This produces:

• one root request span
• child spans nested correctly under that root in Jaeger
• searchable metadata on the root span via attributes like session_id

For more detail, see TRACING_GUIDE.md and examples/tracing_usage.py.

4. Inspect what was recorded

# See all auto-registered metrics after traffic has flowed
for m in obs.registry.list_all():
    print(m.name, m.type.value, m.unit)

# Registry statistics
print(obs.registry.get_stats())
# {
#   "total_metrics": 4,
#   "auto_registered": 4,
#   "explicitly_registered": 0,
#   "counters": 1, "gauges": 0, "histograms": 3
# }

---

YAML configuration reference

rag_observability:
  service_name: "my-rag-service"    # used in OTel resource attributes
  environment: "production"
  enabled: true

  # ── Instrumentation ────────────────────────────────────────────────────────
  instrumentation:
    auto_instrument: true
    trace_sampling_rate: 1.0        # 0.0–1.0
    capture_inputs: true            # attach function args to traces
    capture_outputs: true           # attach return values to traces
    capture_context: true
    max_context_length: 1000        # max chars stored per input/output

  # ── Exporters ──────────────────────────────────────────────────────────────
  exporters:
    - type: otel
      enabled: true
      config:
        endpoint: "http://localhost:4317"
        insecure: true
        service_name: "my-rag-service"
        export_interval_ms: 60000   # how often metrics are flushed (ms)
        include_metric_timestamps: false  # adds timestamp labels to Prometheus metrics; high cardinality

    - type: console                 # useful for local development
      enabled: true
      config:
        format: "json"

  # ── Metrics ────────────────────────────────────────────────────────────────
  metrics:
    # custom: declarative metrics extracted from function arguments/results.
    # See "Custom metrics" section below.
    custom:
      - name: "ecommerce.user.query"
        type: "counter"
        description: "Queries per user"
        labels: ["user_id", "session_id"]
        operations: ["rag_pipeline"]
        extract_from: "kwargs"
        fields:
          user_id: "kwargs.user_id"
          session_id: "kwargs.session_id"
        condition: "user_id is not None"

  # ── Evaluation (optional) ──────────────────────────────────────────────────
  # Required only when using RAGEvaluator or the runtime data capture feature.
  evaluation:
    enabled: true
    framework: "deepeval"           # only supported framework

    model:
      provider: "vllm"              # "vllm" or "bedrock"
      model_name: "mistral-7b"
      api_base: "http://host:8001/v1"   # required for vllm; omit for bedrock
      api_key: "dummy"              # vllm only; default "dummy"
      timeout: 120                  # seconds; default 120

      # Bedrock-specific (used when provider: bedrock)
      # aws_region: "us-east-1"
      # aws_profile: "my-aws-profile"

    metrics:
      retrieval:
        - "ContextualPrecisionMetric"
        - "ContextualRecallMetric"
        - "ContextualRelevancyMetric"
      response:
        - "AnswerRelevancyMetric"
        - "FaithfulnessMetric"
        - "HallucinationMetric"

    thresholds:
      contextual_precision: 0.7
      contextual_recall: 0.7
      contextual_relevancy: 0.7
      answer_relevancy: 0.7
      faithfulness: 0.7
      hallucination: 0.3            # lower is better for hallucination

    # ── Runtime data capture ─────────────────────────────────────────────────
    capture:
      enabled: true
      capture_inputs: true          # store query / input in captured record
      capture_outputs: true         # store generated response in captured record
      capture_context: true         # store retrieved context in captured record
      buffer_size: 1000             # max records held in memory before flush
      auto_export_interval: 300     # seconds between automatic flushes; 0 disables
      export_path: "evaluation_data/captured"  # filesystem: local directory path
                                               # MinIO: object key prefix inside the bucket
      purge_enabled: false          # delete exported files on a schedule
      purge_interval: 0             # seconds between purges; 0 disables

      # ── Storage backend ───────────────────────────────────────────────────
      # MinIO is used when minio_endpoint is set (or MINIO_ENDPOINT env var).
      # Falls back to local filesystem when no MinIO credentials are present.
      minio_endpoint: ""            # e.g. "minio:9000"; leave blank for filesystem
      minio_access_key: ""          # or set MINIO_ACCESS_KEY env var
      minio_secret_key: ""          # or set MINIO_SECRET_KEY env var
      minio_bucket: ""              # e.g. "rag-evaluation-data"
      minio_secure: false           # set true for TLS

      # ── Sampling ─────────────────────────────────────────────────────────
      sampling_enabled: true
      sampling_strategy: "adaptive" # "all" | "random" | "adaptive" | "error_based"
      sampling_rate: 0.1            # fraction of requests sampled (random strategy)

      # Adaptive sampling — adjusts rate based on observed error rate
      adaptive_min_rate: 0.05       # floor: never drop below 5%
      adaptive_max_rate: 0.5        # ceiling: never exceed 50%
      adaptive_error_boost: 2.0     # multiply rate by this factor when errors spike
      adaptive_window_size: 100     # re-evaluate rate every N requests

      # Error-based sampling
      always_sample_errors: true    # always capture failed requests regardless of rate
      error_sample_rate: 1.0        # sample 100% of errors

  # ── Visualization (optional) ────────────────────────────────────────────────
  # Only needed when using DashboardGenerator.from_config().
  # When using DashboardGenerator.from_framework(), these are read automatically.
  visualization:
    GRAFANA_ENDPOINT: "http://localhost:3000"
    grafana_token: "glsa_xxxx"
    datasource: "Prometheus"
    metric_prefix: "llm"            # must match the namespace in your OTel Collector
                                    # Prometheus exporter config (see "Prometheus metric
                                    # name produced" section)

What you do NOT need in the YAML

• metrics.standard — this section is optional. The SDK discovers instrumented metrics automatically from the live registry. You only need it if you want to override the auto-discovered set or provide explicit display titles for dashboard panels.

---

Custom metrics — declarative capture

Custom metrics let the SDK automatically extract counter/gauge/histogram values from the arguments or return values of any decorated function — without writing obs.record_metric() calls yourself.

Minimal counter example

metrics:
  custom:
    - name: "app.user.query"
        type: "counter"
        description: "Total queries per user"
        labels: ["user_id"]
        operations: ["rag_pipeline"]   # function decorated with @obs.observe(operation="rag_pipeline")
        fields:
          user_id: "kwargs.user_id"    # extract user_id from keyword arguments
        condition: "user_id is not None"
        value: 1                       # static increment

Extracting a value from the result

- name: "app.retrieval.docs_returned"
  type: "gauge"
  description: "Number of documents returned"
  operations: ["retrieval"]
  extract_from: "result"
  value: "result.count"              # dot-notation path into the return value

Fields reference

Field	Type	Description
name	string	Metric name (any dot-separated string)
type	counter | gauge | histogram	Prometheus metric type
description	string	Human-readable description
labels	list of strings	Label key names
operations	list of strings	Operation names (must match @obs.observe(operation=…))
extract_from	kwargs | args | result	Source for field extraction
fields	dict	Maps label name → dot-notation path
condition	string	Python expression; metric only recorded when True
value	number or string	Static value or dot-notation path to extract value from

---

Full end-to-end example

import asyncio
from rag_observability import RAGObservability
from rag_observability.visualization import DashboardGenerator

# 1. Initialise the framework
obs = RAGObservability.from_config("config/rag_obs.yaml")

# 2. Decorate every pipeline stage — use your own names
@obs.observe(operation="embed_time")
async def embed(query: str) -> list:
    ...

@obs.observe(operation="retrieval", capture_result_count=True)
async def retrieve(query: str) -> list:
    ...

@obs.observe(operation="llm_generation")
async def generate(query: str, context: list) -> str:
    ...

@obs.observe(operation="rag_pipeline")
async def full_pipeline(query: str, user_id: str) -> dict:
    embedding  = await embed(query)
    context    = await retrieve(query)
    response   = await generate(query, context)
    return {"response": response, "context": context}

# 3. Handle some traffic (metrics are auto-registered on first call)
async def main():
    await full_pipeline("What laptops do you recommend?", user_id="u1")

asyncio.run(main())

---

Evaluation (optional)

Requires pip install rag-observability[evaluation].

Capture data for evaluation

@obs.capture_for_evaluation(operation="rag_pipeline")
async def process_query(query: str) -> dict:
    context  = await retrieve(query)
    response = await generate(query, context)
    return {"response": response, "context": context, "success": True}

Run evaluation with Prometheus bridging

Pass obs= to RAGEvaluator to automatically forward every evaluation score to Prometheus after each evaluate() call. The scores are written as deepeval.<metric_name> gauges (e.g. deepeval.contextual_precision) and become visible in the eval_quality Grafana dashboard with no additional code.

from rag_observability.evaluation import RAGEvaluator

# Pass obs= to enable automatic score bridging to Prometheus
evaluator = RAGEvaluator(
    model_config={
        "provider": "vllm",              # or "bedrock"
        "api_base": "http://host:8001/v1",
        "model_name": "mistral-7b",
        "api_key": "dummy",
        "timeout": 120,
    },
    obs=obs,   # ← bridge: scores flow to Prometheus automatically
)

result = await evaluator.evaluate(
    query="Best laptops under $1000",
    retrieved_context=["Gaming laptop with RTX 4060..."],
    generated_response="I recommend the HP Victus...",
    expected_answer="Should recommend gaming laptops",
    ground_truth_context=["Gaming laptops under $1000"],
    query_type="product_search",   # optional — added as a Prometheus label
)

print(result.retrieval_metrics)   # precision, recall, relevancy
print(result.response_metrics)    # answer_relevancy, faithfulness, hallucination
# deepeval.contextual_precision, deepeval.faithfulness, deepeval.success, … now in Prometheus

Omit obs= if you only need the EvaluationResult object and do not want scores forwarded to Prometheus.

Evaluation model configuration supports both vLLM (any OpenAI-compatible endpoint) and Amazon Bedrock — see examples/rag_observability.yaml for the full configuration reference.

Captured data storage

The @obs.capture_for_evaluation decorator buffers request/response pairs in memory and flushes them to storage automatically. The storage backend is chosen at runtime:

Condition	Where files are written
MINIO_ENDPOINT env var or minio_endpoint config field is set	MinIO bucket — export_path used as the object key prefix inside the bucket
No MinIO credentials	Local filesystem — export_path used as the directory path (default: evaluation_data/captured/)

No configuration change is needed to switch between backends — set or unset the MinIO credentials and the SDK adapts automatically. The filesystem backend is zero-dependency and works out of the box for local development.

Evaluation metrics written to Prometheus

Metric name	Type	Value
deepeval.contextual_precision	gauge	0.0 – 1.0
deepeval.contextual_recall	gauge	0.0 – 1.0
deepeval.contextual_relevancy	gauge	0.0 – 1.0
deepeval.answer_relevancy	gauge	0.0 – 1.0
deepeval.faithfulness	gauge	0.0 – 1.0
deepeval.hallucination	gauge	0.0 – 1.0 (lower is better)
deepeval.success	gauge	1.0 = all metrics passed, 0.0 = any failed

Only the metrics enabled in evaluation.metrics in your YAML are written. Labels: framework="deepeval", query_type (when supplied to evaluate()).

---

Architecture

RAGObservability
├── ConfigManager          — loads and validates YAML / JSON config
├── MetricsRegistry        — thread-safe store of MetricDefinition objects
│                            auto-populated by record_metric() on first use
├── ExporterRegistry
│   ├── OTELExporter       — OTLP gRPC export to any OTel collector
│   └── ConsoleExporter    — stdout/stderr, useful for development
└── @observe decorator     — wraps sync and async functions;
                             records latency histogram + optional count

RAGEvaluator (evaluation optional)
├── evaluate() / evaluate_batch()  — runs DeepEval judge model
├── _bridge_to_obs()               — forwards scores to obs.record_metric()
│                                    writes deepeval.<metric> gauges to Prometheus
└── from_config_dict(config, obs=) — construct from YAML config dict

MetricNameMapper
├── get_all_standard_metrics()      — YAML standard list, falls back to registry
├── get_observed_metrics()          — auto-registered metrics from live registry
└── get_metric_type(name)           — explicit type from config, or name heuristics

---

Testing

The test suite uses pytest. All tests run without any external services — the DeepEvalAdapter and other network-dependent components are fully mocked.

1. Install dev dependencies

pip install -e .[dev]

2. Run the full test suite

pytest

3. Run with verbose output

pytest -v

4. Run with coverage report

python -m pytest --cov=rag_observability --cov-report=term-missing

5. Run a specific file or test

# Single file
pytest tests/test_config.py -v

# Single test class
pytest tests/test_registry.py::TestMetricsRegistry -v

# Single test
pytest tests/test_registry.py::TestMetricsRegistry::test_thread_safety -v

Test files

File	What it covers
tests/test_config.py	Config models, validators, YAML/JSON round-trips, ConfigManager
tests/test_registry.py	MetricDefinition, MetricsRegistry (including thread-safety)
tests/test_exporters_base.py	MetricData, TraceData, BaseExporter, ExporterRegistry
tests/test_exporters_console.py	ConsoleExporter — all formats, lifecycle, error resilience
tests/test_evaluation_evaluator.py	RAGEvaluator — construction, evaluate, batch, aggregation, Prometheus bridging

---

Publishing

Steps to build and publish the package — use TestPyPI for POC/pre-release validation and PyPI for production.

Prerequisites

pip install build twine

1. Build the distribution

uv build

This produces two artefacts inside dist/:

• rag_observability-<version>-py3-none-any.whl — wheel (preferred)
• rag_observability-<version>.tar.gz — source distribution

2. Publish to TestPyPI (POC)

twine upload --repository-url https://test.pypi.org/legacy/ dist/rag_observability-<version>-py3-none-any.whl

You will be prompted for your TestPyPI username (__token__) and password (your TestPyPI API token).

To also upload the source distribution:

twine upload --repository-url https://test.pypi.org/legacy/ dist/rag_observability-<version>-py3-none-any.whl dist/rag_observability-<version>.tar.gz

Verify the install from TestPyPI:

pip install --index-url https://test.pypi.org/simple/ \
            --extra-index-url https://pypi.org/simple/ \
            rag-observability==<version>

--extra-index-url is required because the package's dependencies are hosted on the real PyPI, not TestPyPI.

3. Publish to PyPI (production)

twine upload dist/rag_observability-<version>-py3-none-any.whl

Or upload both artefacts:

twine upload dist/rag_observability-<version>-py3-none-any.whl dist/rag_observability-<version>.tar.gz

You will be prompted for your PyPI username (__token__) and password (your PyPI API token).

Verify the install from PyPI:

pip install rag-observability==<version>

Notes:

• Each publish requires a unique version number. Bump version in pyproject.toml before every release.
• Store credentials in ~/.pypirc or via the TWINE_USERNAME / TWINE_PASSWORD environment variables to avoid interactive prompts in CI/CD.
• API tokens can be created at test.pypi.org/manage/account/token/ and pypi.org/manage/account/token/.

---

Dependencies

Core

• opentelemetry-api >= 1.20.0
• opentelemetry-sdk >= 1.20.0
• opentelemetry-exporter-otlp-proto-grpc >= 1.20.0
• pydantic >= 2.0.0
• pyyaml >= 6.0

Evaluation (optional)

• deepeval >= 0.20.0
• httpx >= 0.24.0
• boto3 (Bedrock provider only)

---