Ragas on Kubernetes: Continuous RAG Evaluation in Production (2026)

# Dockerfile
FROM python:3.11-slim

RUN pip install --no-cache-dir \
    ragas==0.2.8 \
    langfuse==2.54.0 \
    langchain-openai==0.2.0 \
    langchain-community==0.3.0 \
    datasets==3.0.0 \
    prometheus-client==0.21.0

COPY eval/ /app/
WORKDIR /app
ENTRYPOINT ["python", "-m", "eval.runner"]

# eval/runner.py
import asyncio
import os
from datetime import datetime, timedelta, timezone

from datasets import Dataset
from langchain_openai import ChatOpenAI
from langfuse import Langfuse
from ragas import evaluate
from ragas.metrics import (
    answer_relevancy,
    faithfulness,
    context_precision,
    context_recall,
)

EVALUATOR_LLM = ChatOpenAI(
    model="gpt-4o-uae-primary",
    openai_api_base=os.environ["LITELLM_URL"],
    openai_api_key=os.environ["LITELLM_EVAL_KEY"],
    temperature=0.0,
    max_tokens=1024,
    # Tag evaluations in Langfuse
    default_headers={"langfuse-tags": "ragas-eval"},
)

lf = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
    host=os.environ["LANGFUSE_HOST"],
)

async def pull_recent_traces(hours: int = 1, sample_rate: float = 0.02):
    since = datetime.now(timezone.utc) - timedelta(hours=hours)
    traces = lf.fetch_traces(
        from_timestamp=since,
        tags=["rag", "prod"],
        limit=5000,
    ).data

    # Sample rather than evaluate all
    import random
    sampled = [t for t in traces if random.random() < sample_rate]
    return sampled

def extract_eval_row(trace):
    """Pull question, retrieved context, answer from a Langfuse trace."""
    q = trace.input.get("message") or trace.input.get("question")
    # Context retrieved is typically in a child span named "retrieve"
    ctx_span = next((s for s in trace.observations if s.name == "retrieve"), None)
    contexts = (ctx_span.output or []) if ctx_span else []
    contexts = [c.get("text", "") for c in contexts] if isinstance(contexts, list) else []
    answer = trace.output.get("answer") if isinstance(trace.output, dict) else str(trace.output)
    return {
        "trace_id": trace.id,
        "question": q,
        "contexts": contexts,
        "answer": answer,
    }

def write_scores_back(trace_id: str, scores: dict):
    for metric, value in scores.items():
        if value is None or value != value:        # skip NaN
            continue
        lf.score(
            trace_id=trace_id,
            name=f"ragas.{metric}",
            value=float(value),
            data_type="NUMERIC",
        )

async def main():
    traces = await pull_recent_traces(hours=1, sample_rate=0.02)
    rows = [extract_eval_row(t) for t in traces if t.output]
    rows = [r for r in rows if r["question"] and r["answer"] and r["contexts"]]

    if not rows:
        print("No eval rows after filtering")
        return

    ds = Dataset.from_list(rows)
    result = evaluate(
        dataset=ds,
        metrics=[faithfulness, answer_relevancy],     # no ground truth needed
        llm=EVALUATOR_LLM,
        raise_exceptions=False,
    )

    for i, row in enumerate(rows):
        write_scores_back(
            row["trace_id"],
            {
                "faithfulness": result["faithfulness"][i],
                "answer_relevancy": result["answer_relevancy"][i],
            },
        )
    lf.flush()
    print(f"Evaluated {len(rows)} traces. Mean faithfulness: {result['faithfulness'].mean():.3f}")

if __name__ == "__main__":
    asyncio.run(main())

apiVersion: batch/v1
kind: CronJob
metadata:
  name: ragas-online-eval
  namespace: evaluation
spec:
  schedule: "0 * * * *"                # hourly
  concurrencyPolicy: Forbid
  successfulJobsHistoryLimit: 3
  failedJobsHistoryLimit: 7
  jobTemplate:
    spec:
      backoffLimit: 2
      activeDeadlineSeconds: 1800      # fail hard if taking > 30 min
      template:
        spec:
          restartPolicy: OnFailure
          containers:
            - name: eval
              image: registry.example.ae/eval/ragas-runner:0.2.8
              env:
                - name: LITELLM_URL
                  value: "http://litellm.llm-gateway.svc.cluster.local:4000"
                - name: LANGFUSE_HOST
                  value: "http://langfuse-web.langfuse.svc.cluster.local:3000"
                - name: SAMPLE_RATE
                  value: "0.02"
                - name: WINDOW_HOURS
                  value: "1"
              envFrom:
                - secretRef:
                    name: ragas-secrets
              resources:
                requests: {cpu: "1", memory: "2Gi"}
                limits: {memory: "4Gi"}

# eval-suite.yaml
- id: onboarding-multi-hop
  question: "Which APIs do I need to call to onboard a new enterprise customer?"
  reference_answer: |
    The onboarding flow requires calls to (1) /v1/accounts to create the account,
    (2) /v1/entitlements to grant plan-tier access, and (3) /v1/provision to
    initialize the customer's tenant environment. Optionally, /v1/sso-setup for
    SAML integration.
  ground_truth_contexts:
    - "POST /v1/accounts creates a new enterprise account..."
    - "POST /v1/entitlements grants plan-tier access..."
    - "POST /v1/provision initializes the customer's tenant..."

- id: policy-lookup
  question: "What is our data retention policy for customer audit logs?"
  reference_answer: "Audit logs are retained for 7 years, with the last 12 months in hot storage..."
  ground_truth_contexts:
    - "Section 4.2 Audit Log Retention: ..."

# eval/offline.py
import yaml
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
    faithfulness, answer_relevancy,
    context_precision, context_recall,
)

with open("eval-suite.yaml") as f:
    suite = yaml.safe_load(f)

# Run each question through the production RAG endpoint
rows = []
for item in suite:
    resp = rag_client.ask(item["question"])
    rows.append({
        "question": item["question"],
        "answer": resp["answer"],
        "contexts": resp["retrieved_contexts"],
        "ground_truth": item["reference_answer"],
        "reference_contexts": item["ground_truth_contexts"],
    })

result = evaluate(
    dataset=Dataset.from_list(rows),
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
    llm=EVALUATOR_LLM,
)

# Fail the CI build if any metric regresses by > 5%
thresholds = {
    "faithfulness": 0.85,
    "answer_relevancy": 0.80,
    "context_precision": 0.75,
    "context_recall": 0.70,
}
for metric, threshold in thresholds.items():
    score = result[metric].mean()
    print(f"{metric}: {score:.3f} (threshold {threshold})")
    if score < threshold:
        raise SystemExit(f"REGRESSION: {metric} below threshold")

groups:
  - name: rag-quality
    rules:
      - alert: RAGFaithfulnessDropped
        expr: avg_over_time(ragas_faithfulness_hourly[24h]) < 0.80
        for: 2h
        labels:
          severity: warning
        annotations:
          summary: "RAG faithfulness score dropped below 0.80 over 24h"
          runbook: "https://runbooks.example.ae/rag-faithfulness-drop"

      - alert: RAGAnswerRelevancyDropped
        expr: avg_over_time(ragas_answer_relevancy_hourly[24h]) < 0.75
        for: 2h
        labels:
          severity: critical
        annotations:
          summary: "RAG answer relevancy critical - investigate model or prompt changes"

from ragas.metrics import AspectCritic

hallucination_critic = AspectCritic(
    name="hallucination",
    definition="Does the answer contain factual claims not supported by the retrieved context?",
    llm=EVALUATOR_LLM,
)

refusal_critic = AspectCritic(
    name="unnecessary_refusal",
    definition="Did the answer refuse to answer despite the context containing the required information?",
    llm=EVALUATOR_LLM,
)