RAG PIPELINE:
Query → Retrieval → Rerank → Context → Generation → Answer

EVALUATION LAYERS:
1. Retrieval Quality    (precision@k, recall@k, MRR, nDCG)
2. Generation Quality   (faithfulness, relevance, hallucination)
3. End-to-End          (answer correctness, user satisfaction)
4. System Metrics       (latency, cost, throughput, errors)

# rag_eval/retrieval_metrics.py
from typing import List, Dict, Set
import numpy as np
from rank_bm25 import BM25Okapi

def precision_at_k(retrieved: List[str], relevant: Set[str], k: int) -> float:
    """Precision@K: fraction of top-K retrieved that are relevant."""
    if k == 0:
        return 0.0
    top_k = retrieved[:k]
    relevant_retrieved = sum(1 for doc in top_k if doc in relevant)
    return relevant_retrieved / k

def recall_at_k(retrieved: List[str], relevant: Set[str], k: int) -> float:
    """Recall@K: fraction of relevant docs retrieved in top-K."""
    if len(relevant) == 0:
        return 0.0
    top_k = retrieved[:k]
    relevant_retrieved = sum(1 for doc in top_k if doc in relevant)
    return relevant_retrieved / len(relevant)

def mrr(retrieved: List[str], relevant: Set[str]) -> float:
    """Mean Reciprocal Rank: 1/rank of first relevant doc."""
    for i, doc in enumerate(retrieved, 1):
        if doc in relevant:
            return 1.0 / i
    return 0.0

def ndcg_at_k(retrieved: List[str], relevant: Dict[str, float], k: int) -> float:
    """nDCG@K: normalized discounted cumulative gain with graded relevance."""
    def dcg(retrieved_docs: List[str], rel_scores: Dict[str, float], k: int) -> float:
        score = 0.0
        for i, doc in enumerate(retrieved_docs[:k], 1):
            rel = rel_scores.get(doc, 0.0)
            score += (2**rel - 1) / np.log2(i + 1)
        return score

    ideal_dcg = dcg(sorted(relevant.keys(), key=relevant.get, reverse=True), relevant, k)
    if ideal_dcg == 0:
        return 0.0
    return dcg(retrieved, relevant, k) / ideal_dcg

def hit_rate(retrieved: List[str], relevant: Set[str], k: int) -> float:
    """Hit Rate@K: whether at least one relevant doc in top-K."""
    top_k = set(retrieved[:k])
    return 1.0 if top_k & relevant else 0.0

# Batch evaluation
def evaluate_retrieval(
    queries: List[str],
    retrieval_fn,  # query -> List[doc_ids]
    ground_truth: Dict[str, Set[str]],  # query -> relevant doc_ids
    k_values: List[int] = [1, 3, 5, 10, 20]
) -> Dict:
    """Evaluate retrieval across multiple queries."""
    results = {k: {"precision": [], "recall": [], "mrr": [], "hit_rate": []} for k in k_values}

    for query in queries:
        retrieved = retrieval_fn(query)
        relevant = ground_truth.get(query, set())

        for k in k_values:
            results[k]["precision"].append(precision_at_k(retrieved, relevant, k))
            results[k]["recall"].append(recall_at_k(retrieved, relevant, k))
            results[k]["hit_rate"].append(hit_rate(retrieved, relevant, k))

        results[k_values[0]]["mrr"].append(mrr(retrieved, relevant))  # MRR uses full list

    # Aggregate
    aggregated = {}
    for k in k_values:
        aggregated[f"precision@{k}"] = np.mean(results[k]["precision"])
        aggregated[f"recall@{k}"] = np.mean(results[k]["recall"])
        aggregated[f"hit_rate@{k}"] = np.mean(results[k]["hit_rate"])
    aggregated["mrr"] = np.mean(results[k_values[0]]["mrr"])

    return aggregated

# rag_eval/retrieval_ablation.py
def ablation_study(
    base_retriever,
    query_set: List[str],
    ground_truth: Dict[str, Set[str]],
    variants: Dict[str, dict]
) -> Dict:
    """Compare retrieval variants: chunk size, embedding model, reranker, etc."""

    results = {}
    for name, config in variants.items():
        retriever = base_retriever.clone_with_config(config)
        metrics = evaluate_retrieval(query_set, retriever.retrieve, ground_truth)
        results[name] = metrics
        print(f"{name}: {metrics}")

    return results

# Example variants
ABLATION_VARIANTS = {
    "baseline": {"chunk_size": 512, "overlap": 50, "embed_model": "bge-small", "rerank": False},
    "larger_chunks": {"chunk_size": 1024, "overlap": 100, "embed_model": "bge-small", "rerank": False},
    "better_embedder": {"chunk_size": 512, "overlap": 50, "embed_model": "bge-large", "rerank": False},
    "with_reranker": {"chunk_size": 512, "overlap": 50, "embed_model": "bge-small", "rerank": "cross-encoder"},
    "full_upgrade": {"chunk_size": 1024, "overlap": 100, "embed_model": "bge-large", "rerank": "cross-encoder"},
}

# rag_eval/generation_metrics.py
from transformers import AutoModelForSequenceClassification, AutoTokenizer
import torch

class FaithfulnessEvaluator:
    """NLI-based faithfulness: does answer entail from context?"""

    def __init__(self, model_name: str = "microsoft/deberta-v3-large-mnli"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
        self.model.eval()

    def score(self, context: str, answer: str) -> float:
        """Entailment probability: context → answer."""
        # Premise: context, Hypothesis: answer
        inputs = self.tokenizer(
            context, answer,
            return_tensors="pt",
            truncation=True,
            max_length=512,
            padding=True
        )
        with torch.no_grad():
            logits = self.model(**inputs).logits
            probs = torch.softmax(logits, dim=-1)
            # MNLI: 0=entailment, 1=neutral, 2=contradiction
            entailment_prob = probs[0, 0].item()
        return entailment_prob

class HallucinationDetector:
    """Detect unsupported claims in generated answer."""

    def __init__(self):
        self.nli = FaithfulnessEvaluator()

    def detect(self, context: str, answer: str, threshold: float = 0.5) -> Dict:
        """Return hallucination score and flagged spans."""
        sentences = self._split_sentences(answer)
        results = []

        for sent in sentences:
            score = self.nli.score(context, sent)
            results.append({
                "sentence": sent,
                "faithfulness": score,
                "hallucinated": score < threshold
            })

        overall = np.mean([r["faithfulness"] for r in results])
        hallucinated_spans = [r["sentence"] for r in results if r["hallucinated"]]

        return {
            "overall_faithfulness": overall,
            "hallucination_rate": len(hallucinated_spans) / len(sentences),
            "flagged_spans": hallucinated_spans,
            "per_sentence": results
        }

# rag_eval/relevance_metrics.py
from sentence_transformers import SentenceTransformer, util

class RelevanceEvaluator:
    """Semantic similarity between query and answer."""

    def __init__(self, model_name: str = "sentence-transformers/all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name)

    def query_answer_relevance(self, query: str, answer: str) -> float:
        """Cosine similarity between query and answer embeddings."""
        q_emb = self.model.encode(query, convert_to_tensor=True)
        a_emb = self.model.encode(answer, convert_to_tensor=True)
        return util.cos_sim(q_emb, a_emb).item()

    def context_answer_relevance(self, context: str, answer: str) -> float:
        """How well answer addresses context."""
        c_emb = self.model.encode(context, convert_to_tensor=True)
        a_emb = self.model.encode(answer, convert_to_tensor=True)
        return util.cos_sim(c_emb, a_emb).item()

class AnswerCorrectnessEvaluator:
    """LLM-as-judge for answer correctness."""

    def __init__(self, judge_model: str = "gpt-4o-mini"):
        self.judge_model = judge_model
        # Use OpenAI API or local model

    def evaluate(
        self,
        query: str,
        ground_truth: str,
        generated: str,
        context: str = None
    ) -> Dict:
        """Score 1-5 on correctness, completeness, conciseness."""
        prompt = f"""Rate the generated answer (1-5) on:
1. Correctness: factually accurate vs ground truth
2. Completeness: covers all key points from ground truth
3. Conciseness: no fluff, direct answer

Query: {query}
Ground Truth: {ground_truth}
Generated: {generated}
Context: {context or 'N/A'}

Return JSON: {{"correctness": 1-5, "completeness": 1-5, "conciseness": 1-5, "reasoning": "..."}}"""

        # Call judge model
        return self._call_judge(prompt)

# rag_eval/ragas_integration.py
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    context_relevancy,
    answer_correctness,
    answer_similarity,
)
from datasets import Dataset

def evaluate_rag_with_ragas(
    questions: List[str],
    answers: List[str],
    contexts: List[List[str]],
    ground_truths: List[str] = None
) -> Dict:
    """Evaluate RAG using Ragas metrics."""

    data = {
        "question": questions,
        "answer": answers,
        "contexts": contexts,
    }
    if ground_truths:
        data["ground_truth"] = ground_truths

    dataset = Dataset.from_dict(data)

    # Select metrics based on available data
    metrics = [
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
    ]
    if ground_truths:
        metrics.extend([answer_correctness, answer_similarity])

    result = evaluate(dataset, metrics=metrics)
    return result.to_pandas().to_dict(orient="records")[0]

# Ragas metric explanations
RAGAS_METRICS = {
    "faithfulness": "Answer grounded in context? (0-1, higher better)",
    "answer_relevancy": "Answer relevant to question? (0-1)",
    "context_precision": "Retrieved contexts ranked by relevance? (0-1)",
    "context_recall": "All needed info retrieved? (0-1)",
    "context_relevancy": "Context relevant to question? (0-1)",
    "answer_correctness": "Answer matches ground truth? (0-1, needs GT)",
    "answer_similarity": "Semantic similarity to ground truth? (0-1, needs GT)",
}

# rag_eval/deepeval_integration.py
from deepeval import evaluate
from deepeval.metrics import (
    AnswerRelevancyMetric,
    FaithfulnessMetric,
    ContextualPrecisionMetric,
    ContextualRecallMetric,
    ContextualRelevancyMetric,
    HallucinationMetric,
    ToxicityMetric,
    BiasMetric,
)
from deepeval.test_case import LLMTestCase

def evaluate_rag_with_deepeval(
    test_cases: List[LLMTestCase],
    threshold: float = 0.7
) -> Dict:
    """Evaluate with DeepEval's comprehensive metrics."""

    metrics = [
        AnswerRelevancyMetric(threshold=threshold),
        FaithfulnessMetric(threshold=threshold),
        ContextualPrecisionMetric(threshold=threshold),
        ContextualRecallMetric(threshold=threshold),
        ContextualRelevancyMetric(threshold=threshold),
        HallucinationMetric(threshold=threshold),
        ToxicityMetric(threshold=0.5),
    ]

    results = evaluate(test_cases, metrics)
    return {
        "passed": all(r.success for r in results),
        "scores": {m.__class__.__name__: r.score for r in results for m in r.metrics},
        "details": results
    }

# Test case creation
def create_test_case(
    query: str,
    answer: str,
    retrieval_context: List[str],
    expected_output: str = None
) -> LLMTestCase:
    return LLMTestCase(
        input=query,
        actual_output=answer,
        retrieval_context=retrieval_context,
        expected_output=expected_output,
    )

# rag_eval/trulens_integration.py
from trulens_eval import Tru, Feedback, OpenAI as fOpenAI
from trulens_eval.feedback import Groundedness
import numpy as np

def setup_trulens():
    """Configure TruLens feedback functions."""
    provider = fOpenAI()

    # Groundedness: answer supported by context
    grounded = Groundedness(groundedness_provider=provider)
    f_groundedness = Feedback(
        grounded.groundedness_measure_with_cot_reasons,
        name="Groundedness"
    ).on_input_output()

    # Answer relevance
    f_answer_relevance = Feedback(
        provider.relevance,
        name="Answer Relevance"
    ).on_input_output()

    # Context relevance
    f_context_relevance = Feedback(
        provider.context_relevance,
        name="Context Relevance"
    ).on_input()

    return [f_groundedness, f_answer_relevance, f_context_relevance]

def instrument_rag_chain(rag_chain, feedback_funcs):
    """Wrap RAG chain for TruLens logging."""
    from trulens_eval import TruChain

    tru_chain = TruChain(
        rag_chain,
        app_id="rag_qa",
        feedbacks=feedback_funcs
    )
    return tru_chain

# rag_monitoring/production_metrics.py
from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime
import time
import asyncio
from prometheus_client import Counter, Histogram, Gauge, Summary

# Prometheus metrics
RAG_REQUESTS_TOTAL = Counter(
    "rag_requests_total",
    "Total RAG requests",
    ["status"]  # success, error, timeout
)

RAG_LATENCY = Histogram(
    "rag_latency_seconds",
    "RAG request latency",
    buckets=[0.5, 1, 2, 5, 10, 30, 60]
)

RETRIEVAL_LATENCY = Histogram(
    "rag_retrieval_latency_seconds",
    "Retrieval step latency"
)

GENERATION_LATENCY = Histogram(
    "rag_generation_latency_seconds",
    "Generation step latency"
)

RAG_TOKENS = Counter(
    "rag_tokens_total",
    "Total tokens used",
    ["type"]  # prompt, completion
)

RAG_COST = Counter(
    "rag_cost_usd_total",
    "Total cost in USD"
)

ACTIVE_REQUESTS = Gauge(
    "rag_active_requests",
    "Currently processing requests"
)

QUALITY_SCORE = Histogram(
    "rag_quality_score",
    "Quality scores from evaluators",
    ["metric"],  # faithfulness, relevance, etc.
    buckets=[0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
)

@dataclass
class RAGRequestLog:
    request_id: str
    timestamp: datetime
    query: str
    retrieved_docs: List[str]
    answer: str
    latency_ms: float
    retrieval_latency_ms: float
    generation_latency_ms: float
    prompt_tokens: int
    completion_tokens: int
    cost_usd: float
    quality_scores: dict = field(default_factory=dict)
    error: Optional[str] = None

class ProductionMonitor:
    """Production RAG monitoring with alerting."""

    def __init__(
        self,
        alert_webhook: str = None,
        quality_threshold: float = 0.6,
        latency_p99_threshold: float = 30.0,
        error_rate_threshold: float = 0.05
    ):
        self.alert_webhook = alert_webhook
        self.quality_threshold = quality_threshold
        self.latency_threshold = latency_p99_threshold
        self.error_rate_threshold = error_rate_threshold
        self.request_logs: List[RAGRequestLog] = []
        self.window_logs: List[RAGRequestLog] = []

    async def log_request(self, log: RAGRequestLog):
        """Log request and update metrics."""
        self.request_logs.append(log)
        self.window_logs.append(log)

        # Update Prometheus
        RAG_REQUESTS_TOTAL.labels(
            status="error" if log.error else "success"
        ).inc()

        RAG_LATENCY.observe(log.latency_ms / 1000)
        RETRIEVAL_LATENCY.observe(log.retrieval_latency_ms / 1000)
        GENERATION_LATENCY.observe(log.generation_latency_ms / 1000)

        RAG_TOKENS.labels(type="prompt").inc(log.prompt_tokens)
        RAG_TOKENS.labels(type="completion").inc(log.completion_tokens)
        RAG_COST.inc(log.cost_usd)

        for metric, score in log.quality_scores.items():
            QUALITY_SCORE.labels(metric=metric).observe(score)

        # Check alerts
        await self._check_alerts(log)

    async def _check_alerts(self, log: RAGRequestLog):
        """Check and fire alerts."""
        alerts = []

        # Quality alert
        for metric, score in log.quality_scores.items():
            if score < self.quality_threshold:
                alerts.append(f"Quality alert: {metric}={score:.2f} < {self.quality_threshold}")

        # Latency alert
        if log.latency_ms > self.latency_threshold * 1000:
            alerts.append(f"Latency alert: {log.latency_ms}ms > {self.latency_threshold}s")

        # Error rate alert (rolling window)
        await self._check_error_rate()

        for alert in alerts:
            await self._send_alert(alert)

    async def _check_error_rate(self):
        """Check error rate in last 100 requests."""
        if len(self.window_logs) >= 100:
            recent = self.window_logs[-100:]
            errors = sum(1 for r in recent if r.error)
            rate = errors / 100
            if rate > self.error_rate_threshold:
                await self._send_alert(f"Error rate alert: {rate:.1%} > {self.error_rate_threshold:.1%}")

    async def _send_alert(self, message: str):
        """Send alert to webhook."""
        if self.alert_webhook:
            # Implement webhook call
            print(f"ALERT: {message}")

    def get_dashboard_data(self, hours: int = 24) -> Dict:
        """Get data for Grafana dashboard."""
        cutoff = datetime.now() - timedelta(hours=hours)
        recent = [r for r in self.request_logs if r.timestamp > cutoff]

        if not recent:
            return {}

        return {
            "total_requests": len(recent),
            "success_rate": sum(1 for r in recent if not r.error) / len(recent),
            "avg_latency_ms": np.mean([r.latency_ms for r in recent]),
            "p99_latency_ms": np.percentile([r.latency_ms for r in recent], 99),
            "avg_cost_per_request": np.mean([r.cost_usd for r in recent]),
            "total_cost": sum(r.cost_usd for r in recent),
            "avg_quality": {
                metric: np.mean([r.quality_scores.get(metric, 0) for r in recent])
                for metric in ["faithfulness", "relevance", "correctness"]
            },
        }

# rag_monitoring/tracing.py
from opentelemetry import trace
from opentelemetry.exporter.jaeger import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

def setup_tracing(service_name: str = "rag-service"):
    """Setup OpenTelemetry tracing."""

    trace.set_tracer_provider(TracerProvider())
    jaeger_exporter = JaegerExporter(
        agent_host_name="jaeger",
        agent_port=6831,
    )
    trace.get_tracer_provider().add_span_processor(
        BatchSpanProcessor(jaeger_exporter)
    )

    return trace.get_tracer(__name__)

class TracedRAGChain:
    """RAG chain with full distributed tracing."""

    def __init__(self, retriever, generator, tracer):
        self.retriever = retriever
        self.generator = generator
        self.tracer = tracer

    async def answer(self, query: str) -> dict:
        with self.tracer.start_as_current_span("rag_query") as span:
            span.set_attribute("query", query)

            # Retrieval span
            with self.tracer.start_as_current_span("retrieval") as retr_span:
                start = time.time()
                docs = await self.retriever.retrieve(query)
                retr_span.set_attribute("num_docs", len(docs))
                retr_span.set_attribute("latency_ms", (time.time() - start) * 1000)

            # Generation span
            with self.tracer.start_as_current_span("generation") as gen_span:
                start = time.time()
                answer = await self.generator.generate(query, docs)
                gen_span.set_attribute("answer_length", len(answer))
                gen_span.set_attribute("latency_ms", (time.time() - start) * 1000)

            span.set_attribute("total_latency_ms", (time.time() - start) * 1000)
            return {"answer": answer, "sources": docs}

# rag_eval/ci_pipeline.py
import pytest
from ragas import evaluate
from datasets import Dataset

class RAGEvalTestSuite:
    """Pytest-style RAG evaluation tests."""

    @pytest.fixture(scope="class")
    def test_data(self):
        return [
            {
                "question": "What is LoRA?",
                "ground_truth": "LoRA is Low-Rank Adaptation for efficient fine-tuning...",
                "contexts": ["LoRA freezes base weights and adds low-rank adapters..."],
            },
            # ... more test cases
        ]

    def test_faithfulness_threshold(self, rag_chain, test_data):
        """Faithfulness must exceed threshold."""
        results = []
        for case in test_data:
            answer = rag_chain.answer(case["question"])
            score = self._compute_faithfulness(case["contexts"][0], answer)
            results.append(score)
            assert score >= 0.7, f"Faithfulness {score:.2f} below threshold for: {case['question']}"

    def test_retrieval_precision(self, rag_chain, test_data):
        """Retrieval precision@5 must exceed threshold."""
        for case in test_data:
            retrieved = rag_chain.retrieve(case["question"])
            relevant = self._get_relevant_docs(case["question"])
            p5 = precision_at_k(retrieved, relevant, 5)
            assert p5 >= 0.6, f"Precision@5 {p5:.2f} below threshold for: {case['question']}"

    def test_latency_sla(self, rag_chain, test_data):
        """P99 latency must be under 10 seconds."""
        latencies = []
        for case in test_data:
            start = time.time()
            rag_chain.answer(case["question"])
            latencies.append(time.time() - start)

        p99 = np.percentile(latencies, 99)
        assert p99 <= 10.0, f"P99 latency {p99:.1f}s exceeds 10s SLA"

    def test_cost_per_query(self, rag_chain, test_data):
        """Average cost per query under $0.01."""
        costs = [rag_chain.get_last_cost() for _ in test_data]
        avg_cost = np.mean(costs)
        assert avg_cost <= 0.01, f"Avg cost ${avg_cost:.4f} exceeds $0.01"

# Regression detection
def detect_regression(
    current_scores: Dict,
    baseline_scores: Dict,
    threshold: float = 0.05
) -> List[str]:
    """Detect metric regressions vs baseline."""
    regressions = []
    for metric, current in current_scores.items():
        baseline = baseline_scores.get(metric)
        if baseline and current < baseline - threshold:
            regressions.append(
                f"REGRESSION: {metric} dropped from {baseline:.3f} to {current:.3f} "
                f"(delta: {current-baseline:.3f})"
            )
    return regressions

# rag_eval/ab_testing.py
import random
from dataclasses import dataclass
from typing: Callable, Dict, List
import numpy as np
from scipy import stats

@dataclass
class ABTestResult:
    variant_a: str
    variant_b: str
    metric: str
    a_mean: float
    b_mean: float
    p_value: float
    significant: bool
    winner: str

class RAGABTester:
    """A/B test RAG variants in production."""

    def __init__(self, traffic_split: float = 0.5):
        self.traffic_split = traffic_split
        self.variants: Dict[str, Callable] = {}
        self.results: Dict[str, List[float]] = {}

    def register_variant(self, name: str, chain_factory: Callable):
        self.variants[name] = chain_factory
        self.results[name] = []

    def route_request(self, query: str) -> tuple:
        """Route to variant based on traffic split."""
        variant_name = "A" if random.random() < self.traffic_split else "B"
        chain = self.variants[variant_name]()
        return variant_name, chain

    def record_result(self, variant: str, score: float):
        self.results[variant].append(score)

    def analyze(self, metric: str = "quality") -> ABTestResult:
        """Statistical significance test."""
        a_scores = self.results["A"]
        b_scores = self.results["B"]

        if len(a_scores) < 30 or len(b_scores) < 30:
            return ABTestResult(
                "A", "B", metric,
                np.mean(a_scores), np.mean(b_scores),
                1.0, False, "insufficient_data"
            )

        # Welch's t-test (unequal variance)
        t_stat, p_value = stats.ttest_ind(a_scores, b_scores, equal_var=False)

        a_mean, b_mean = np.mean(a_scores), np.mean(b_scores)
        winner = "A" if a_mean > b_mean else "B"

        return ABTestResult(
            "A", "B", metric,
            a_mean, b_mean,
            p_value, p_value < 0.05, winner
        )

# rag_monitoring/drift_detection.py
from sklearn.decomposition import PCA
from sklearn.manifold import TSNE
import numpy as np
from scipy.spatial.distance import jensenshannon

class EmbeddingDriftDetector:
    """Detect drift in query/document embeddings."""

    def __init__(self, reference_embeddings: np.ndarray, window_size: int = 1000):
        self.reference = reference_embeddings
        self.window_size = window_size
        self.recent_embeddings: List[np.ndarray] = []

        # Fit PCA on reference for visualization
        self.pca = PCA(n_components=2)
        self.pca.fit(reference_embeddings)

    def add_embeddings(self, embeddings: np.ndarray):
        self.recent_embeddings.append(embeddings)
        if len(self.recent_embeddings) > self.window_size:
            self.recent_embeddings.pop(0)

    def compute_drift_score(self) -> Dict:
        """Compute multiple drift metrics."""
        if not self.recent_embeddings:
            return {"drift_detected": False}

        recent = np.vstack(self.recent_embeddings)

        # 1. Mean shift (L2 distance between centroids)
        ref_centroid = self.reference.mean(axis=0)
        recent_centroid = recent.mean(axis=0)
        mean_shift = np.linalg.norm(ref_centroid - recent_centroid)

        # 2. Covariance shift (Frobenius norm)
        ref_cov = np.cov(self.reference.T)
        recent_cov = np.cov(recent.T)
        cov_shift = np.linalg.norm(ref_cov - recent_cov, 'fro')

        # 3. JS divergence on PCA projections
        ref_pca = self.pca.transform(self.reference)
        recent_pca = self.pca.transform(recent)

        # 2D histogram
        hist_range = [[ref_pca[:,0].min(), ref_pca[:,0].max()],
                      [ref_pca[:,1].min(), ref_pca[:,1].max()]]
        ref_hist, _, _ = np.histogram2d(ref_pca[:,0], ref_pca[:,1], bins=20, range=hist_range, density=True)
        recent_hist, _, _ = np.histogram2d(recent_pca[:,0], recent_pca[:,1], bins=20, range=hist_range, density=True)

        # Flatten and add epsilon
        ref_flat = ref_hist.flatten() + 1e-10
        recent_flat = recent_hist.flatten() + 1e-10

        js_div = jensenshannon(ref_flat, recent_flat)

        return {
            "drift_detected": js_div > 0.1 or mean_shift > 0.5,
            "mean_shift": float(mean_shift),
            "cov_shift": float(cov_shift),
            "js_divergence": float(js_div),
            "ref_centroid": ref_centroid.tolist(),
            "recent_centroid": recent_centroid.tolist(),
        }

# rag_monitoring/data_quality.py
def validate_rag_inputs(
    query: str,
    retrieved_docs: List[str],
    answer: str
) -> Dict:
    """Validate inputs for quality issues."""

    issues = []

    # Query checks
    if len(query.strip()) < 3:
        issues.append("Query too short")
    if len(query) > 2000:
        issues.append("Query extremely long")

    # Retrieval checks
    if not retrieved_docs:
        issues.append("No documents retrieved")
    else:
        avg_doc_len = np.mean([len(d) for d in retrieved_docs])
        if avg_doc_len < 50:
            issues.append("Retrieved docs very short")
        if avg_doc_len > 5000:
            issues.append("Retrieved docs very long")

        # Check for duplicate docs
        unique_docs = set(retrieved_docs)
        if len(unique_docs) < len(retrieved_docs) * 0.8:
            issues.append("High duplicate rate in retrieval")

    # Answer checks
    if not answer.strip():
        issues.append("Empty answer")
    if len(answer) < 10:
        issues.append("Answer too short")
    if answer.count("I don't know") > 0:
        issues.append("Model abstained")

    # Repetition check
    words = answer.split()
    if len(words) > 10:
        unique_ratio = len(set(words)) / len(words)
        if unique_ratio < 0.3:
            issues.append("High repetition in answer")

    return {
        "valid": len(issues) == 0,
        "issues": issues,
        "severity": "high" if any("Empty" in i or "No documents" in i for i in issues) else "low"
    }

# rag_monitoring/grafana_dashboard.json
{
  "dashboard": {
    "title": "RAG Production Monitoring",
    "panels": [
      {
        "title": "Request Rate",
        "targets": [{"expr": "rate(rag_requests_total[5m])"}]
      },
      {
        "title": "Error Rate",
        "targets": [{"expr": "rate(rag_requests_total{status=\"error\"}[5m]) / rate(rag_requests_total[5m])"}]
      },
      {
        "title": "Latency P50/P95/P99",
        "targets": [
          {"expr": "histogram_quantile(0.50, rag_latency_seconds)"},
          {"expr": "histogram_quantile(0.95, rag_latency_seconds)"},
          {"expr": "histogram_quantile(0.99, rag_latency_seconds)"}
        ]
      },
      {
        "title": "Quality Scores",
        "targets": [
          {"expr": "avg by (metric) (rag_quality_score)"}
        ]
      },
      {
        "title": "Cost per Request",
        "targets": [{"expr": "rate(rag_cost_usd_total[1h]) / rate(rag_requests_total[1h])"}]
      },
      {
        "title": "Tokens per Request",
        "targets": [
          {"expr": "rate(rag_tokens_total{type=\"prompt\"}[5m]) / rate(rag_requests_total[5m])"},
          {"expr": "rate(rag_tokens_total{type=\"completion\"}[5m]) / rate(rag_requests_total[5m])"}
        ]
      }
    ]
  }
}

# rag_monitoring/alerts.yaml
groups:
- name: rag_alerts
  rules:
  - alert: RAGHighErrorRate
    expr: rate(rag_requests_total{status="error"}[5m]) / rate(rag_requests_total[5m]) > 0.05
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "RAG error rate > 5%"

  - alert: RAGHighLatency
    expr: histogram_quantile(0.99, rag_latency_seconds) > 30
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "RAG P99 latency > 30s"

  - alert: RAGLowFaithfulness
    expr: avg(rag_quality_score{metric="faithfulness"}) < 0.6
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "RAG faithfulness dropped below 0.6"

  - alert: RAGHighCost
    expr: rate(rag_cost_usd_total[1h]) > 10
    for: 15m
    labels:
      severity: warning
    annotations:
      summary: "RAG cost > $10/hour"

# rag_eval/improvement_loop.py
class RAGImprovementLoop:
    """Automated RAG improvement via evaluation feedback."""

    def __init__(self, rag_system, evaluator, optimizer):
        self.rag = rag_system
        self.evaluator = evaluator
        self.optimizer = optimizer
        self.history = []

    async def run_iteration(self, eval_queries: List[str]):
        """One improvement cycle."""

        # 1. Evaluate current system
        current_metrics = await self.evaluator.evaluate(self.rag, eval_queries)

        # 2. Identify weak spots
        weak_spots = self._identify_weaknesses(current_metrics)

        # 3. Generate optimization hypotheses
        hypotheses = self.optimizer.generate_hypotheses(weak_spots)

        # 4. Test hypotheses (A/B or shadow)
        best_variant = await self._test_hypotheses(hypotheses, eval_queries)

        # 5. Deploy if significant improvement
        if best_variant and best_variant.improvement > 0.02:
            await self._deploy_variant(best_variant)
            self.history.append({
                "iteration": len(self.history),
                "metrics": current_metrics,
                "deployed": best_variant.name,
                "improvement": best_variant.improvement
            })

        return current_metrics

    def _identify_weaknesses(self, metrics: Dict) -> List[str]:
        """Identify which metrics need improvement."""
        thresholds = {
            "faithfulness": 0.8,
            "answer_relevancy": 0.75,
            "context_precision": 0.7,
            "context_recall": 0.7,
        }

        return [
            metric for metric, threshold in thresholds.items()
            if metrics.get(metric, 0) < threshold
        ]

# rag_monitoring/cost_optimization.py
class CostOptimizer:
    """Optimize RAG costs while maintaining quality."""

    def __init__(self, pricing: Dict):
        self.pricing = pricing  # {"gpt-4o": {"input": 5, "output": 15}, ...}

    def estimate_cost(self, prompt_tokens: int, completion_tokens: int, model: str) -> float:
        """Estimate cost in USD."""
        rates = self.pricing.get(model, {"input": 0, "output": 0})
        return (prompt_tokens * rates["input"] + completion_tokens * rates["output"]) / 1_000_000

    def optimize_model_selection(
        self,
        query: str,
        complexity_score: float,  # 0-1
        quality_requirement: float  # 0-1
    ) -> str:
        """Route to cheapest model meeting quality."""

        models = [
            ("gpt-4o-mini", 0.7, 0.15),   # quality, cost_factor
            ("gpt-4o", 0.9, 1.0),
            ("gpt-4-turbo", 0.95, 2.0),
        ]

        for model, quality, cost_factor in models:
            if quality >= quality_requirement:
                # Check if we can use cheaper for simple queries
                if complexity_score < 0.3 and model != "gpt-4o-mini":
                    continue
                return model

        return "gpt-4o"  # fallback

    def optimize_retrieval(self, retrieval_config: Dict) -> Dict:
        """Find optimal retrieval config for cost/quality."""
        # Fewer docs = lower prompt tokens = lower cost
        # But too few = lower quality
        # Use Bayesian optimization
        pass