scripts/ragas_eval.py

"""Real RAGAS evaluation against the live RICS report-generation pipeline.

This script exercises the *running* FastAPI server (no mocks, no synthetic
data, no stubbed retrieval) against an actual surveyor PDF that is already
ingested into the tenant's FAISS index. It:

1. Creates a fresh report on an existing tenant + ingested document so we
   evaluate the current code path end-to-end.
2. Triggers ``POST /reports/{rid}/generate`` for a representative slice of
   RICS sections (mix of intro, fabric, services, risks).
3. Polls ``/reports/{rid}/sections`` until the generations land in the DB.
4. Re-runs ``retrieve_tenant_evidence`` with the *same* tier-aware fan-out
   the inspector loop uses, so ``contexts`` reflects what the LLM actually
   saw.
5. Extracts ground-truth section text from the source PDF using a
   heading-aware splitter (regex on RICS section codes).
6. Feeds (question, contexts, answer, ground_truth) into RAGAS using
   ``faithfulness``, ``answer_relevancy``, ``context_precision`` and
   ``context_recall``.
7. Saves per-section scores + aggregates as JSON under ``eval_runs/``.

Run with:
    python scripts/ragas_eval.py
"""

from __future__ import annotations

import argparse
import asyncio
import json
import logging
import os
import re
import sys
import time
import uuid
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parent.parent
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import httpx  # noqa: E402

from app.agentic.tools import retrieve_tenant_evidence  # noqa: E402
from app.config import settings  # noqa: E402
from app.templates.registry import get_template  # noqa: E402

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("ragas_eval")

# Defaults (override via CLI flags; see `--help`)
DEFAULT_TENANT_ID = os.environ.get("RAGAS_TENANT_ID", "tenant_pwto0nrd")
DEFAULT_SOURCE_DOCUMENT_ID = os.environ.get("RAGAS_DOCUMENT_ID", "7b6379ea-fbeb-439a-a65f-e1a48df2b900")
DEFAULT_SOURCE_PDF_NAME = os.environ.get(
    "RAGAS_SOURCE_PDF_NAME",
    "Building Survey - 37 Elms Crescent London SW4 8QE.pdf",
)
DEFAULT_SURVEY_LEVEL = int(os.environ.get("RAGAS_SURVEY_LEVEL", "3") or "3")

API_BASE = os.environ.get("RAGAS_API_BASE", "http://127.0.0.1:8000")

# A representative cross-section — enough to surface real quality issues
# without burning $50 of OpenAI credit. Picks deliberately span:
#   • Narrative / non-rated   (A, D)
#   • External fabric         (E2 Roof, E4 Main walls)
#   • Internal fabric         (F1 Roof structure, F4 Floors)
#   • Services                (G6 Drainage)
#   • Risk                    (J1 Risks to the building)
EVAL_SECTIONS: list[str] = ["A", "D", "E2", "E4", "F1", "F4", "G6", "J1"]


# ─────────────────────────────────────────────────────────────────────────────
# Source-PDF section splitter (ground truth extraction)
# ─────────────────────────────────────────────────────────────────────────────


_RICS_HEADING_RE = re.compile(
    r"""
    (?:^|\n)\s*                         # line start
    (?P<code>(?:E|F|G|H|I|J|K)\s?\d{1,2}|[A-L])   # E2 / F4 / J / D
    \s+                                  # required whitespace
    (?P<title>[A-Z][A-Za-z][A-Za-z\s,'’/&\-]{3,80})  # human title
    (?:\n|\s*$)                          # ends at newline
    """,
    re.VERBOSE | re.MULTILINE,
)


def extract_pdf_text(pdf_path: Path) -> str:
    """Return raw text from the entire PDF (line-preserving)."""
    import pypdf

    reader = pypdf.PdfReader(str(pdf_path))
    pages: list[str] = []
    for p in reader.pages:
        try:
            pages.append(p.extract_text() or "")
        except Exception:  # noqa: BLE001
            pages.append("")
    return "\n".join(pages)


def split_pdf_into_sections(full_text: str) -> dict[str, str]:
    """Heading-aware split of full PDF text into ``{code: section_text}``.

    The regex is generous (matches any ``[A-L]`` or ``[EFGHIJK]\\d{1,2}``)
    so a non-RICS heading like "B7 Crescent Road" can produce noise, but
    the eval tolerates this — we just want the text under each canonical
    section heading. Section text spans up to the *next* matched heading.
    """
    matches = list(_RICS_HEADING_RE.finditer(full_text))
    if not matches:
        return {}

    sections: dict[str, str] = {}
    for i, m in enumerate(matches):
        code = m.group("code").replace(" ", "")
        # Filter junk codes that are clearly not RICS sections
        if code not in {"A", "B", "C", "D", "L"} and not re.match(r"^[EFGHIJK]\d+$", code):
            continue
        start = m.end()
        end = matches[i + 1].start() if i + 1 < len(matches) else len(full_text)
        body = full_text[start:end].strip()
        # If we already grabbed this code, only keep the longest occurrence —
        # the source PDF tends to put the most detail in the body, not in
        # the contents table at the front.
        if code in sections and len(body) <= len(sections[code]):
            continue
        sections[code] = body
    return sections


# ─────────────────────────────────────────────────────────────────────────────
# Bullet extraction (input to /generate) — taken from the actual PDF section
# ─────────────────────────────────────────────────────────────────────────────


def bullets_from_section_text(section_text: str, max_bullets: int = 14) -> list[str]:
    """Split a PDF section block into bullet-sized lines.

    Aims to mimic what a surveyor would paste into the notes box. Drops
    trivial 1-2 token lines (page numbers, photo callouts) and clamps to
    ``max_bullets`` so the LLM call cost stays bounded.
    """
    out: list[str] = []
    for raw in section_text.splitlines():
        line = raw.strip()
        if not line:
            continue
        if line.lower().startswith("photo -") or re.fullmatch(r"\d{1,3}", line):
            continue
        if len(line.split()) < 3:
            continue
        # Light de-duplication — surveyor PDFs sometimes repeat headings.
        if line in out:
            continue
        out.append(line)
        if len(out) >= max_bullets:
            break
    return out


# ─────────────────────────────────────────────────────────────────────────────
# HTTP helpers (talk to the live FastAPI server)
# ─────────────────────────────────────────────────────────────────────────────


def create_report(client: httpx.Client, *, tenant_id: str, source_document_id: str, survey_level: int) -> str:
    """Create a fresh report against the pinned ingested document."""
    r = client.post(
        f"{API_BASE}/reports",
        params={
            "document_id": source_document_id,
            "survey_level": survey_level,
            "confirm_tier_mismatch": "true",
        },
        headers={"X-Tenant-ID": tenant_id},
    )
    r.raise_for_status()
    rid = r.json()["report_id"]
    log.info("created report %s on tenant=%s doc=%s", rid, tenant_id, source_document_id)
    return rid


def trigger_generate(
    client: httpx.Client,
    *,
    tenant_id: str,
    report_id: str,
    code: str,
    bullets: list[str],
) -> None:
    body = {
        "template_id": code,
        "bullets": bullets,
        "mode": "generate",
        "ai_level": 3,
        "ai_percent": 50,
        "force_regenerate": True,
    }
    r = client.post(
        f"{API_BASE}/reports/{report_id}/generate",
        json=body,
        headers={"X-Tenant-ID": tenant_id},
        timeout=30,
    )
    r.raise_for_status()


def poll_section(
    client: httpx.Client,
    *,
    tenant_id: str,
    report_id: str,
    code: str,
    timeout_s: int = 600,
    interval_s: float = 10.0,
) -> dict[str, Any] | None:
    """Wait until ``report_sections`` row for ``code`` exists and is non-empty.

    Generations can take 60–180 s each because the inspector tool loop runs
    up to 32 rounds and the OpenAI call inside the loop is synchronous (it
    blocks the FastAPI event loop), so we use a long timeout + a long
    interval to avoid hammering the server while it is busy with the LLM.
    """
    deadline = time.time() + timeout_s
    last_text_len = 0
    while time.time() < deadline:
        try:
            r = client.get(
                f"{API_BASE}/reports/{report_id}/sections",
                headers={"X-Tenant-ID": tenant_id},
                timeout=httpx.Timeout(connect=10, read=120, write=30, pool=10),
            )
            r.raise_for_status()
            data = r.json()
        except httpx.ReadTimeout:
            # Likely the event loop is still blocked by an active LLM round.
            # Back off and retry — the section may already be persisted by
            # the time the next request gets through.
            log.info("[%s] /sections read-timeout, backing off", code)
            time.sleep(interval_s)
            continue
        sections_map = data.get("sections") or {}
        s = sections_map.get(code)
        if isinstance(s, dict):
            text = (s.get("text") or "").strip()
            if text and len(text) > 80:
                return s
            last_text_len = len(text)
        time.sleep(interval_s)
    log.warning("section %s did not finish — last text len = %d", code, last_text_len)
    return None


# ─────────────────────────────────────────────────────────────────────────────
# Re-run the inspector's seed retrieval to capture ``contexts`` honestly
# ─────────────────────────────────────────────────────────────────────────────


def capture_contexts(
    *,
    tenant_id: str,
    source_document_id: str,
    query: str,
    level: int,
) -> list[str]:
    """Mirror the tier-aware seed retrieval the inspector loop uses.

    inspector_loop.py opens with:
        L1 → k=12, rerank_top_n=5
        L2 → k=18, rerank_top_n=8
        L3 → k=24, rerank_top_n=10

    We deliberately use the same fan-out so the captured ``contexts`` is
    representative of what the LLM actually had at draft time.
    """
    if level <= 1:
        k, rerank = 12, 5
    elif level == 2:
        k, rerank = 18, 8
    else:
        k, rerank = 24, 10
    hits = retrieve_tenant_evidence(
        query=query,
        tenant_id=tenant_id,
        primary_document_id=source_document_id,
        secondary_document_ids=None,
        k=k,
        rerank_top_n=rerank,
    )
    return [h.text for h in hits if h and h.text]


# ─────────────────────────────────────────────────────────────────────────────
# Main eval orchestration
# ─────────────────────────────────────────────────────────────────────────────


def build_query(code: str, level: int, bullets: list[str]) -> str:
    """Compose a deterministic ``question`` for RAGAS.

    Strives to match how a surveyor actually phrases the section ask, so
    answer_relevancy + context_precision are scored on a realistic prompt
    rather than the raw RAG snippets.
    """
    tpl = get_template(code, level)
    title = tpl.title if tpl else code
    head = bullets[0] if bullets else ""
    return (
        f"Write the {title} ({code}) section of a RICS Level {level} Building "
        f"Survey for the property in the source notes. {head}"
    ).strip()


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--limit", type=int, default=len(EVAL_SECTIONS), help="cap eval to first N sections")
    parser.add_argument("--out", type=str, default=None)
    parser.add_argument("--tenant-id", type=str, default=DEFAULT_TENANT_ID)
    parser.add_argument("--document-id", type=str, default=DEFAULT_SOURCE_DOCUMENT_ID)
    parser.add_argument("--survey-level", type=int, default=DEFAULT_SURVEY_LEVEL, choices=[1, 2, 3])
    parser.add_argument(
        "--pdf-path",
        type=str,
        default=None,
        help=(
            "Optional path to the source PDF on disk to extract ground-truth references. "
            "If omitted (or missing), the run still produces answer_relevancy + context_precision "
            "(and skips faithfulness/context_recall which require reference text)."
        ),
    )
    parser.add_argument(
        "--sections",
        type=str,
        default=",".join(EVAL_SECTIONS),
        help="Comma-separated section codes to evaluate (default: built-in representative set).",
    )
    args = parser.parse_args()

    if not settings.openai_api_key:
        log.error("OPENAI_API_KEY is not set — RAGAS judge needs it. aborting.")
        return 2

    eval_codes = [c.strip().upper() for c in str(args.sections).split(",") if c.strip()]
    eval_codes = eval_codes[: args.limit]
    log.info("evaluating %d sections: %s", len(eval_codes), ", ".join(eval_codes))

    tenant_id = str(args.tenant_id)
    source_document_id = str(args.document_id)
    survey_level = int(args.survey_level)

    pdf_sections: dict[str, str] = {}
    pdf_path: Path | None = None
    if args.pdf_path:
        pdf_path = Path(args.pdf_path)
    else:
        # Backwards-compatible default: expects a local Behrang corpus checkout.
        pdf_path = ROOT / "Behrang RICS Documents" / DEFAULT_SOURCE_PDF_NAME

    have_reference = bool(pdf_path and pdf_path.is_file())
    if have_reference:
        full_text = extract_pdf_text(pdf_path)  # type: ignore[arg-type]
        pdf_sections = split_pdf_into_sections(full_text)
        log.info(
            "extracted %d ground-truth sections from PDF (%d chars total)",
            len(pdf_sections),
            len(full_text),
        )
        log.info("ground-truth coverage for eval set:")
        for code in eval_codes:
            gt = pdf_sections.get(code, "")
            log.info("  %-4s : %d chars", code, len(gt))
    else:
        log.warning(
            "source PDF not found; running without ground-truth reference (will score only answer_relevancy + context_precision). "
            "missing: %s",
            pdf_path,
        )

    rows: list[dict[str, Any]] = []
    with httpx.Client(timeout=httpx.Timeout(connect=10, read=120, write=30, pool=10)) as client:
        report_id = create_report(
            client,
            tenant_id=tenant_id,
            source_document_id=source_document_id,
            survey_level=survey_level,
        )

        for code in eval_codes:
            gt = pdf_sections.get(code, "").strip()
            bullets = bullets_from_section_text(gt) if gt else []
            if not bullets:
                # fall back to template skeleton bullets so we still hit the
                # endpoint with realistic context — happens for narrative
                # sections like A and L where the heading detector misses.
                tpl = get_template(code, survey_level)
                bullets = [
                    f"This is the {tpl.title} section." if tpl else f"Section {code}.",
                ]
            log.info("[%s] %d bullets, %d chars ground truth", code, len(bullets), len(gt))

            trigger_generate(client, tenant_id=tenant_id, report_id=report_id, code=code, bullets=bullets)
            section = poll_section(client, tenant_id=tenant_id, report_id=report_id, code=code, timeout_s=300)
            if not section:
                log.warning("[%s] generation never landed; skipping", code)
                continue
            answer = (section.get("text") or "").strip()
            query = build_query(code, survey_level, bullets)
            contexts = capture_contexts(
                tenant_id=tenant_id,
                source_document_id=source_document_id,
                query=query,
                level=survey_level,
            )
            log.info("[%s] answer=%d chars, contexts=%d", code, len(answer), len(contexts))

            rows.append({
                "section_code": code,
                "user_input": query,
                "retrieved_contexts": contexts,
                "response": answer,
                "reference": gt,
                "bullets": bullets,
                "provenance": section.get("provenance") or {},
            })

    if not rows:
        log.error("no sections evaluated — aborting before RAGAS")
        return 3

    out_path = Path(args.out) if args.out else (
        ROOT / "eval_runs" / f"ragas_run_{int(time.time())}.json"
    )
    out_path.parent.mkdir(parents=True, exist_ok=True)
    raw_path = out_path.with_suffix(".raw.json")
    with raw_path.open("w", encoding="utf-8") as f:
        json.dump(
            {
                "meta": {
                    "tenant_id": tenant_id,
                    "report_id": report_id,
                    "doc_id": source_document_id,
                    "pdf_path": str(pdf_path) if pdf_path else None,
                    "level": survey_level,
                    "api_base": API_BASE,
                },
                "rows": rows,
            },
            f,
            indent=2,
            ensure_ascii=False,
        )
    log.info("saved raw eval set: %s", raw_path)

    # ── RAGAS scoring ───────────────────────────────────────────────────────
    log.info("running RAGAS metrics …")
    from datasets import Dataset
    from langchain_openai import ChatOpenAI, OpenAIEmbeddings
    from ragas import evaluate
    from ragas.embeddings import LangchainEmbeddingsWrapper
    from ragas.llms import LangchainLLMWrapper
    from ragas.metrics import (
        answer_relevancy,
        context_precision,
        context_recall,
        faithfulness,
    )

    ds_rows = [
        {
            "user_input": r["user_input"],
            "retrieved_contexts": r["retrieved_contexts"],
            "response": r["response"],
            "reference": r["reference"],
        }
        for r in rows
        if r.get("retrieved_contexts") and r.get("response")
    ]
    ds = Dataset.from_list(ds_rows)
    log.info("dataset rows: %d", len(ds))
    if len(ds) == 0:
        log.error("dataset is empty (no ground truth) — aborting RAGAS")
        return 4

    judge_llm = LangchainLLMWrapper(ChatOpenAI(
        model="gpt-4o-mini", temperature=0.0, api_key=settings.openai_api_key
    ))
    judge_emb = LangchainEmbeddingsWrapper(OpenAIEmbeddings(
        model="text-embedding-3-small", api_key=settings.openai_api_key
    ))

    metrics = [answer_relevancy, context_precision]
    if any(r.get("reference") for r in ds_rows):
        metrics = [faithfulness, answer_relevancy, context_precision, context_recall]
    else:
        log.warning("no reference text available; skipping faithfulness + context_recall")

    result = evaluate(
        dataset=ds,
        metrics=metrics,
        llm=judge_llm,
        embeddings=judge_emb,
    )
    log.info("RAGAS aggregate: %s", result)

    df = result.to_pandas()
    summary = {
        "aggregate": {col: float(df[col].mean()) for col in df.columns if df[col].dtype.kind in "fi"},
        "per_section": [],
    }
    code_iter = iter([r["section_code"] for r in rows if r["reference"]])
    for _, row in df.iterrows():
        scores = {col: (float(row[col]) if isinstance(row[col], (int, float)) else None)
                  for col in df.columns if col not in ("user_input", "retrieved_contexts", "response", "reference")}
        summary["per_section"].append({
            "section_code": next(code_iter, "?"),
            "scores": scores,
        })

    with out_path.open("w", encoding="utf-8") as f:
        json.dump(summary, f, indent=2, ensure_ascii=False)
    log.info("RAGAS scored output saved: %s", out_path)

    print()
    print("=" * 72)
    print("RAGAS aggregate scores")
    print("=" * 72)
    for k, v in summary["aggregate"].items():
        print(f"  {k:24s} : {v:.3f}")
    print()
    print("Per-section:")
    for entry in summary["per_section"]:
        sc = entry["scores"]
        line = f"  {entry['section_code']:5s}  " + "  ".join(
            f"{k}={v:.2f}" for k, v in sc.items() if v is not None
        )
        print(line)
    return 0


if __name__ == "__main__":
    sys.exit(main())