Create rag_eval_metrics.py

rag_eval_metrics.py

@@ -0,0 +1,421 @@
+#!/usr/bin/env python3
+"""
+rag_eval_metrics.py
+
+Evaluate RAG retrieval quality by comparing app logs (JSONL) with a gold file (CSV).
+
+Inputs (CLI):
+  --gold_csv       Path to gold CSV.
+  --logs_jsonl     Path to app JSONL logs (rag_logs.jsonl).
+  --k              Cutoff for metrics (default: 8).
+  --out_dir        Output directory for metrics files (default: rag_artifacts).
+
+Outputs (written into out_dir):
+  - metrics_per_question.csv
+  - metrics_aggregate.json
+
+Gold CSV accepted schemas (case-insensitive headers):
+  Minimal (doc-level):
+    question, doc
+    (multiple rows per question allowed)
+  With page info (page-level optional):
+    question, doc, page
+  List-in-a-cell also supported:
+    question, relevant_docs   # semicolon/comma separated; page matching disabled in this column
+
+Notes:
+  - Matching is case-insensitive on question and doc filename.
+  - Page-level metrics only computed when GOLD includes a concrete page for that question.
+  - Logs are produced by app.py and contain 'retrieval'->'hits' with 'doc' and 'page'.
+"""
+
+import argparse
+import json
+import os
+import sys
+from pathlib import Path
+from typing import Dict, List, Tuple, Any, Optional
+
+import pandas as pd
+import numpy as np
+
+
+# ----------------------------- IO Helpers ----------------------------- #
+
+def read_logs(jsonl_path: Path) -> pd.DataFrame:
+    """Read JSONL logs and return a DataFrame with columns: question, hits(list[dict])."""
+    rows = []
+    with open(jsonl_path, "r", encoding="utf-8") as f:
+        for line in f:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                rec = json.loads(line)
+            except Exception:
+                continue
+            q = (((rec.get("inputs") or {}).get("question")) or "").strip()
+            retr = (rec.get("retrieval") or {})
+            hits = retr.get("hits", [])
+            # Normalize fields we need
+            norm_hits = []
+            for h in hits or []:
+                doc = (h.get("doc") or "").strip()
+                page = str(h.get("page") or "").strip()
+                try:
+                    # Try int page if it looks numeric
+                    page_int = int(page)
+                except Exception:
+                    page_int = None
+                norm_hits.append({"doc": doc, "page": page_int})
+            rows.append({"question": q, "hits": norm_hits})
+    df = pd.DataFrame(rows)
+    if df.empty:
+        return pd.DataFrame(columns=["question", "hits"])
+    # Keep last occurrence per question (latest run), but also allow multiple – we aggregate by question
+    # For stability, group and keep the last non-empty hit list.
+    def _pick_last_non_empty(hit_lists: List[List[dict]]) -> List[dict]:
+        for lst in reversed(hit_lists):
+            if lst:
+                return lst
+        return []
+    df = (
+        df.groupby(df["question"].str.casefold().str.strip(), as_index=False)
+          .agg({"question": "last", "hits": _pick_last_non_empty})
+    )
+    return df
+
+
+def read_gold(csv_path: Path) -> pd.DataFrame:
+    """Read gold CSV, normalize columns, and return rows with:
+       question(cf), question_raw, doc (lowercased filename), page (optional, int or NaN).
+    """
+    df = pd.read_csv(csv_path)
+    # Normalize headers
+    cols = {c.lower().strip(): c for c in df.columns}
+    # Find question column
+    q_col = None
+    for cand in ["question", "query", "q"]:
+        if cand in cols:
+            q_col = cols[cand]
+            break
+    if q_col is None:
+        raise ValueError("Gold CSV must contain a 'question' column (case-insensitive).")
+
+    # Accept either (doc[, page]) rows or a 'relevant_docs' list column
+    rel_list_col = None
+    for cand in ["relevant_docs", "relevant", "docs"]:
+        if cand in cols:
+            rel_list_col = cols[cand]
+            break
+
+    doc_col = None
+    for cand in ["doc", "document", "file", "doc_name"]:
+        if cand in cols:
+            doc_col = cols[cand]
+            break
+
+    page_col = None
+    for cand in ["page", "page_num", "page_number"]:
+        if cand in cols:
+            page_col = cols[cand]
+            break
+
+    rows = []
+    if rel_list_col and doc_col is None:
+        # Each row may contain a list of docs (comma/semicolon separated)
+        for _, r in df.iterrows():
+            q_raw = str(r[q_col]).strip()
+            q_norm = q_raw.casefold().strip()
+            rel_val = str(r[rel_list_col]) if pd.notna(r[rel_list_col]) else ""
+            if not rel_val:
+                # still create an empty row (no gold docs)
+                rows.append({"question_raw": q_raw, "question": q_norm, "doc": None, "page": np.nan})
+                continue
+            # split by semicolon or comma
+            parts = [p.strip() for p in re_split_sc(rel_val)]
+            # one row per doc (page-level off for list column)
+            for d in parts:
+                rows.append({"question_raw": q_raw, "question": q_norm, "doc": filename_key(d), "page": np.nan})
+    elif doc_col:
+        # Standard long form: one doc (+/- page) per row
+        for _, r in df.iterrows():
+            q_raw = str(r[q_col]).strip()
+            q_norm = q_raw.casefold().strip()
+            d = str(r[doc_col]).strip() if pd.notna(r[doc_col]) else ""
+            p = r[page_col] if page_col and pd.notna(r[page_col]) else np.nan
+            try:
+                p = int(p)
+            except Exception:
+                p = np.nan
+            rows.append({"question_raw": q_raw, "question": q_norm, "doc": filename_key(d), "page": p})
+    else:
+        raise ValueError("Gold CSV must contain either a 'doc' column or a 'relevant_docs' column.")
+
+    gold = pd.DataFrame(rows)
+    # drop fully empty doc rows (when no gold docs listed)
+    gold["has_doc"] = gold["doc"].apply(lambda x: isinstance(x, str) and len(x) > 0)
+    if gold["has_doc"].any():
+        gold = gold[gold["has_doc"]].copy()
+    gold.drop(columns=["has_doc"], inplace=True, errors="ignore")
+    # Deduplicate
+    gold = gold.drop_duplicates(subset=["question", "doc", "page"])
+    return gold
+
+
+def filename_key(s: str) -> str:
+    """Normalize document name to just the basename, lowercased."""
+    s = (s or "").strip()
+    s = s.replace("\\", "/")
+    s = s.split("/")[-1]
+    return s.casefold()
+
+
+def re_split_sc(s: str) -> List[str]:
+    """Split on semicolons or commas."""
+    import re
+    return re.split(r"[;,]", s)
+
+
+# ----------------------------- Metric Core ----------------------------- #
+
+def dcg_at_k(relevances: List[int]) -> float:
+    """Binary DCG with log2 discounts; ranks are 1-indexed in denominator."""
+    dcg = 0.0
+    for i, rel in enumerate(relevances, start=1):
+        if rel > 0:
+            dcg += 1.0 / np.log2(i + 1.0)
+    return float(dcg)
+
+
+def ndcg_at_k(relevances: List[int]) -> float:
+    dcg = dcg_at_k(relevances)
+    ideal = sorted(relevances, reverse=True)
+    idcg = dcg_at_k(ideal)
+    if idcg == 0.0:
+        return 0.0
+    return float(dcg / idcg)
+
+
+def compute_metrics_for_question(
+    gold_docs: List[str],
+    gold_pages: List[Optional[int]],
+    hits: List[Dict[str, Any]],
+    k: int
+) -> Dict[str, Any]:
+    """
+    Returns per-question metrics at cutoff k for:
+      - doc-level: match on doc only
+      - page-level: match on (doc,page) where page is provided in GOLD
+    """
+    top = hits[:k] if hits else []
+    pred_docs = [filename_key(h.get("doc", "")) for h in top]
+    pred_pairs = [(filename_key(h.get("doc", "")), h.get("page", None)) for h in top]
+
+    # --- DOC-LEVEL ---
+    gold_doc_set = set([d for d in gold_docs if isinstance(d, str) and d])
+    rel_bin_doc = [1 if d in gold_doc_set else 0 for d in pred_docs]
+    hitk_doc = 1 if any(rel_bin_doc) else 0
+    prec_doc = (sum(rel_bin_doc) / max(1, len(pred_docs))) if pred_docs else 0.0
+    rec_doc = (sum(rel_bin_doc) / max(1, len(gold_doc_set))) if gold_doc_set else 0.0
+    ndcg_doc = ndcg_at_k(rel_bin_doc)
+
+    # --- PAGE-LEVEL (only if at least one GOLD page specified) ---
+    gold_pairs = set()
+    for d, p in zip(gold_docs, gold_pages):
+        if isinstance(d, str) and d and (p is not None) and (not (isinstance(p, float) and np.isnan(p))):
+            try:
+                p_int = int(p)
+            except Exception:
+                continue
+            gold_pairs.add((d, p_int))
+
+    if gold_pairs:
+        rel_bin_page = [1 if ((d, (p if p is not None else -1)) in gold_pairs) else 0
+                        for (d, p) in [(d, (p if isinstance(p, int) else -1)) for (d, p) in pred_pairs]]
+        hitk_page = 1 if any(rel_bin_page) else 0
+        prec_page = (sum(rel_bin_page) / max(1, len(pred_pairs))) if pred_pairs else 0.0
+        rec_page = (sum(rel_bin_page) / max(1, len(gold_pairs))) if gold_pairs else 0.0
+        ndcg_page = ndcg_at_k(rel_bin_page)
+    else:
+        hitk_page = prec_page = rec_page = ndcg_page = np.nan
+
+    return {
+        "hit@k_doc": hitk_doc,
+        "precision@k_doc": prec_doc,
+        "recall@k_doc": rec_doc,
+        "ndcg@k_doc": ndcg_doc,
+        "hit@k_page": hitk_page,
+        "precision@k_page": prec_page,
+        "recall@k_page": rec_page,
+        "ndcg@k_page": ndcg_page,
+        "n_gold_docs": int(len(gold_doc_set)),
+        "n_gold_doc_pages": int(len(gold_pairs)),
+        "n_pred": int(len(pred_docs))
+    }
+
+
+# ----------------------------- Orchestration ----------------------------- #
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--gold_csv", required=True, type=str)
+    ap.add_argument("--logs_jsonl", required=True, type=str)
+    ap.add_argument("--k", type=int, default=8)
+    ap.add_argument("--out_dir", type=str, default="rag_artifacts")
+    args = ap.parse_args()
+
+    out_dir = Path(args.out_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+
+    gold_path = Path(args.gold_csv)
+    logs_path = Path(args.logs_jsonl)
+
+    if not gold_path.exists():
+        print(f"❌ gold.csv not found at {gold_path}", file=sys.stderr)
+        sys.exit(0)
+    if not logs_path.exists() or logs_path.stat().st_size == 0:
+        print(f"❌ logs JSONL not found or empty at {logs_path}", file=sys.stderr)
+        sys.exit(0)
+
+    # Load data
+    try:
+        gold = read_gold(gold_path)
+    except Exception as e:
+        print(f"❌ Failed to read gold: {e}", file=sys.stderr)
+        sys.exit(0)
+    logs = read_logs(logs_path)
+
+    if gold.empty:
+        print("❌ Gold file contains no usable rows.", file=sys.stderr)
+        sys.exit(0)
+    if logs.empty:
+        print("❌ Logs file contains no usable entries.", file=sys.stderr)
+        sys.exit(0)
+
+    # Build gold dict: question -> list of (doc, page)
+    gdict: Dict[str, List[Tuple[str, Optional[int]]]] = {}
+    for _, r in gold.iterrows():
+        q = str(r["question"]).strip()
+        d = r["doc"]
+        p = r["page"] if "page" in r else np.nan
+        gdict.setdefault(q, []).append((d, p))
+
+    # Align on questions (casefolded)
+    logs["q_norm"] = logs["question"].astype(str).str.casefold().str.strip()
+    perq_rows = []
+    not_in_logs, not_in_gold = [], []
+
+    for q_norm, pairs in gdict.items():
+        # Pairs is list of (doc, page)
+        q_gold_variants = [q_norm]  # already normalized
+        # Find logs row with same normalized question
+        row = logs[logs["q_norm"] == q_norm]
+        if row.empty:
+            not_in_logs.append(q_norm)
+            # Still record a row with zeros/NaNs
+            gdocs = [d for (d, _) in pairs]
+            gpages = [p for (_, p) in pairs]
+            metrics = {
+                "hit@k_doc": 0, "precision@k_doc": 0.0, "recall@k_doc": 0.0, "ndcg@k_doc": 0.0,
+                "hit@k_page": np.nan, "precision@k_page": np.nan, "recall@k_page": np.nan, "ndcg@k_page": np.nan,
+                "n_gold_docs": int(len(set([d for d in gdocs if isinstance(d, str) and d]))),
+                "n_gold_doc_pages": int(len([(d, p) for (d, p) in zip(gdocs, gpages) if isinstance(d, str) and d and pd.notna(p)])),
+                "n_pred": 0
+            }
+            perq_rows.append({
+                "question": q_norm,
+                "covered_in_logs": 0,
+                **metrics
+            })
+            continue
+
+        # Use the last row (grouping ensured one row per question)
+        hits = row.iloc[0]["hits"] or []
+        # Prepare gold lists for metric function
+        gdocs = [d for (d, _) in pairs]
+        gpages = [p for (_, p) in pairs]
+        metrics = compute_metrics_for_question(gdocs, gpages, hits, args.k)
+
+        perq_rows.append({
+            "question": q_norm,
+            "covered_in_logs": 1,
+            **metrics
+        })
+
+    # Detect questions present in logs but not in gold (for reporting)
+    gold_qs = set(gdict.keys())
+    for qn in logs["q_norm"].tolist():
+        if qn not in gold_qs:
+            not_in_gold.append(qn)
+
+    perq = pd.DataFrame(perq_rows)
+
+    # Aggregates over questions that are covered_in_logs == 1
+    covered = perq[perq["covered_in_logs"] == 1].copy()
+    agg = {
+        "questions_total_gold": int(len(gdict)),
+        "questions_covered_in_logs": int(covered.shape[0]),
+        "questions_missing_in_logs": int(len(not_in_logs)),
+        "questions_in_logs_not_in_gold": int(len(set(not_in_gold))),
+        "k": int(args.k),
+        # DOC-level
+        "mean_hit@k_doc": float(covered["hit@k_doc"].mean()) if not covered.empty else 0.0,
+        "mean_precision@k_doc": float(covered["precision@k_doc"].mean()) if not covered.empty else 0.0,
+        "mean_recall@k_doc": float(covered["recall@k_doc"].mean()) if not covered.empty else 0.0,
+        "mean_ndcg@k_doc": float(covered["ndcg@k_doc"].mean()) if not covered.empty else 0.0,
+        # PAGE-level (skip NaNs)
+        "mean_hit@k_page": float(covered["hit@k_page"].dropna().mean()) if covered["hit@k_page"].notna().any() else None,
+        "mean_precision@k_page": float(covered["precision@k_page"].dropna().mean()) if covered["precision@k_page"].notna().any() else None,
+        "mean_recall@k_page": float(covered["recall@k_page"].dropna().mean()) if covered["recall@k_page"].notna().any() else None,
+        "mean_ndcg@k_page": float(covered["ndcg@k_page"].dropna().mean()) if covered["ndcg@k_page"].notna().any() else None,
+        # Distribution hints
+        "avg_gold_docs_per_q": float(perq["n_gold_docs"].mean()) if not perq.empty else 0.0,
+        "avg_preds_per_q": float(perq["n_pred"].mean()) if not perq.empty else 0.0,
+        # Listings (truncated for readability)
+        "examples_missing_in_logs": list(not_in_logs[:10]),
+        "examples_in_logs_not_in_gold": list(dict.fromkeys(not_in_gold))[:10],
+    }
+
+    # Write outputs
+    perq_path = out_dir / "metrics_per_question.csv"
+    agg_path = out_dir / "metrics_aggregate.json"
+    perq.to_csv(perq_path, index=False)
+    with open(agg_path, "w", encoding="utf-8") as f:
+        json.dump(agg, f, ensure_ascii=False, indent=2)
+
+    # Console summary (stdout) for app display
+    print("RAG Evaluation Summary")
+    print("----------------------")
+    print(f"Gold questions: {agg['questions_total_gold']}")
+    print(f"Covered in logs: {agg['questions_covered_in_logs']}")
+    print(f"Missing in logs: {agg['questions_missing_in_logs']}")
+    print(f"In logs but not in gold: {agg['questions_in_logs_not_in_gold']}")
+    print(f"k = {agg['k']}")
+    print()
+    print(f"Doc-level:  Hit@k={_fmt(agg['mean_hit@k_doc'])}  "
+          f"Precision@k={_fmt(agg['mean_precision@k_doc'])}  "
+          f"Recall@k={_fmt(agg['mean_recall@k_doc'])}  "
+          f"nDCG@k={_fmt(agg['mean_ndcg@k_doc'])}")
+    if agg["mean_hit@k_page"] is not None:
+        print(f"Page-level: Hit@k={_fmt(agg['mean_hit@k_page'])}  "
+              f"Precision@k={_fmt(agg['mean_precision@k_page'])}  "
+              f"Recall@k={_fmt(agg['mean_recall@k_page'])}  "
+              f"nDCG@k={_fmt(agg['mean_ndcg@k_page'])}")
+    else:
+        print("Page-level: (no page labels in gold)")
+
+    print()
+    print(f"Wrote per-question CSV → {perq_path}")
+    print(f"Wrote aggregate JSON   → {agg_path}")
+
+
+def _fmt(x: Any) -> str:
+    try:
+        return f"{float(x):.3f}"
+    except Exception:
+        return "-"
+
+
+if __name__ == "__main__":
+    main()