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. Installation
2. How metrics are captured
3. Quick start
4. YAML configuration reference
5. Custom metrics — declarative capture
6. Grafana dashboard generation
7. Evaluation (optional)
8. Architecture
9. Publishing

---

Installation

# Core SDK
pip install rag-observability

# With evaluation support (DeepEval)
pip install rag-observability[evaluation]

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

Prometheus metric name produced

The OTel SDK exports through a meter named "llm" (default prefix). The Prometheus exporter transforms the internal name as follows:

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

Rule: dots → underscores, prefix prepended, unit appended, histogram suffix added.

What the YAML metrics section does (and does not do)

Section	Effect on metric capture	Effect on dashboards
metrics.custom	Yes — SDK hooks into decorated functions and records the metric declaratively from arguments/results	Panels in the "Custom Metrics" dashboard
metrics.standard	None — metrics are already captured by @obs.observe regardless	Lets the dashboard generator know which operations to draw panels for (optional since v0.1.16 — see dashboard section)

---

Quick start

1. Initialise the framework

from rag_observability import RAGObservability

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

# Option B — defaults only (console exporter, no Grafana)
obs = RAGObservability()

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"

  # ── 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 OTel meter name (default: "llm")

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

---

Grafana dashboard generation

The SDK generates Grafana dashboards from a DashboardGenerator. There are two ways to create one depending on whether the application is running.

Method A — from a live framework instance (recommended)

Use this when your application is running and has already received traffic. The generator reads the live registry to discover every metric that was recorded via @obs.observe — no metrics.standard section in the YAML is needed.

from rag_observability import RAGObservability
from rag_observability.visualization import DashboardGenerator

obs = RAGObservability.from_config("config/rag_obs.yaml")

# ... application handles some traffic so metrics are auto-registered ...

gen = DashboardGenerator.from_framework(
    obs,
    GRAFANA_ENDPOINT="http://localhost:3000",
    grafana_token="glsa_xxxx",
)

# Generate and upload all standard dashboards in one call
gen.delete_and_republish(folder_name="RAG Observability")

Method B — from a config file (offline / CI/CD)

Use this when generating dashboards at deploy time without a running application. The metrics.standard section in the YAML tells the generator which operations to render panels for.

# config/rag_obs.yaml
metrics:
  standard:
    - embed_time          # name heuristic: contains "time" → histogram panel
    - retrieval_latency   # name heuristic: contains "latency" → histogram panel
    - llm_generation      # no keyword match → gauge panel
    - req_count           # name heuristic: contains "count" → counter panel

    # Dict form for explicit type and display title overrides:
    - name: my_custom_op
      type: histogram
      title: "My Custom Operation Latency"

gen = DashboardGenerator.from_config(
    "config/rag_obs.yaml",
    GRAFANA_ENDPOINT="http://localhost:3000",
    grafana_token="glsa_xxxx",
)
gen.delete_and_republish(folder_name="RAG Observability")

Name heuristics for type inference

When a standard entry has no explicit type field, the type is inferred from the metric name:

Name contains	Panel type
latency, duration, time	Histogram → P50/P95/P99 time-series
count, rate, usage, total	Counter → rate stat panel
(anything else)	Gauge → current-value stat panel

Available dashboard templates

Template name	Description
cost_analysis	Daily/hourly cost, token breakdown — requires token custom metrics
custom_metrics	One panel per metrics.custom entry, grouped by name prefix
eval_quality	Evaluation quality scores (precision, recall, faithfulness, hallucination, …) — auto-populated when RAGEvaluator is initialised with obs=

# Generate a single dashboard to a local JSON file (no Grafana required)
dashboard = gen.generate_dashboard("cost_analysis")
gen.export_to_file(dashboard, "dashboards/rag_performance.json")

# Upload a specific dashboard
gen.generate_and_upload("cost_analysis", folder_name="RAG Observability")

# List all available templates
print(gen.list_available_templates())

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")

    # 4. Generate dashboards from the live registry — no YAML standard list needed
    gen = DashboardGenerator.from_framework(
        obs,
        GRAFANA_ENDPOINT="http://localhost:3000",
        grafana_token="glsa_xxxx",
    )
    gen.delete_and_republish(
        folder_name="RAG Observability",
        export_dir="dashboards/",    # also saves JSON files locally
    )

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.

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

DashboardGenerator
├── from_framework(obs)    — wires to live registry; no YAML standard list needed
├── from_config(path)      — offline generation driven by metrics.standard YAML
└── Templates
    ├── LLMCostAnalysisDashboard    — cost tracking from token custom metrics
    ├── CustomMetricsDashboard      — one panel per metrics.custom entry
    └── EvalQualityDashboard        — gauge + trend + bar-gauge for deepeval.* metrics
                                      sources: evaluation.metrics YAML or live registry

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

---

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

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)

Visualization (optional)

• grafana-client (for Grafana API upload)

---

License

MIT