app/tests/eval_rag.py

import argparse
import asyncio
import csv
import json
import sys
from datetime import datetime, timezone
from pathlib import Path
from time import perf_counter
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import answer_relevancy, faithfulness, context_precision, context_recall, answer_correctness
from ragas.run_config import RunConfig
from langchain_openai import ChatOpenAI

from app.core.config import settings
from app.engine.indexer import dense_embeddings
from app.graph.workflow import compile_workflow

DATASET_PATH = Path("datasets/golden_qa.csv")
REPORT_PATH = Path("reports/eval_report.md")

def load_golden_questions(path: Path = DATASET_PATH) -> list[dict]:
    if not path.exists():
        return []
    with path.open(newline="", encoding="utf-8") as handle:
        return list(csv.DictReader(handle))

def source_hit(expected_sources: str, sources: list[dict]) -> bool:
    expected = {source.strip() for source in expected_sources.split("|") if source.strip()}
    actual = {source.get("source") for source in sources}
    return bool(expected & actual)

async def run_local_evaluation(
    assert_faithfulness: float = None,
    assert_precision: float = None,
    assert_relevancy: float = None,
) -> dict:
    agent = compile_workflow()
    rows = load_golden_questions()
    results = []
    
    questions = []
    answers = []
    contexts = []
    ground_truths = []

    for row in rows:
        started = perf_counter()
        output_state = await agent.ainvoke({"question": row["question"], "chat_history": [], "run_count": 0})
        latency_ms = round((perf_counter() - started) * 1000, 2)
        answer = output_state.get("generation", "")
        sources_dicts = output_state.get("sources", [])
        docs = output_state.get("documents", [])
        
        questions.append(row["question"])
        answers.append(answer)
        contexts.append([doc.page_content for doc in docs])
        ground_truths.append(row["expected_answer"])
        
        results.append(
            {
                "question": row["question"],
                "answer": answer,
                "latency_ms": latency_ms,
                "source_hit": source_hit(row["expected_sources"], sources_dicts),
                "retrieved_contexts": len(docs),
            }
        )

    # RAGAS Two-Tier Judge Configuration
    fast_llm = ChatOpenAI(
        model=getattr(settings, "FAST_LLM_MODEL", settings.LLM_MODEL),
        temperature=0,
        openai_api_key=getattr(settings, "FAST_LLM_API_KEY", "") or settings.OPENROUTER_API_KEY,
        openai_api_base=getattr(settings, "FAST_LLM_BASE_URL", "") or settings.OPENROUTER_BASE_URL,
        default_headers={"HTTP-Referer": "https://localhost:3000", "X-Title": "Support Docs Copilot"},
    )
    slow_llm = ChatOpenAI(
        model=getattr(settings, "SLOW_LLM_MODEL", settings.LLM_MODEL),
        temperature=0,
        openai_api_key=settings.OPENROUTER_API_KEY,
        openai_api_base=settings.OPENROUTER_BASE_URL,
        default_headers={"HTTP-Referer": "https://localhost:3000", "X-Title": "Support Docs Copilot"},
    )

    # Assign fast LLM to structural metrics and slow LLM to reasoning metrics
    answer_relevancy.llm = fast_llm
    context_precision.llm = fast_llm
    context_recall.llm = fast_llm
    faithfulness.llm = slow_llm
    answer_correctness.llm = slow_llm

    ragas_dataset = Dataset.from_dict({
        "question": questions,
        "answer": answers,
        "contexts": contexts,
        "ground_truth": ground_truths,
    })
    
    # Use global singleton ONNX embedder (eliminates 2.5s reload)
    embeddings = dense_embeddings()
    try:
        ragas_result = evaluate(
            ragas_dataset,
            metrics=[answer_relevancy, faithfulness, context_precision, context_recall, answer_correctness],
            llm=fast_llm,
            embeddings=embeddings,
            run_config=RunConfig(max_workers=4, max_wait=60, max_retries=2),
        )
        ragas_scores = ragas_result
    except Exception as e:
        ragas_scores = {"error": str(e)}

    source_hit_rate = round(sum(1 for result in results if result["source_hit"]) / max(len(results), 1), 3)
    average_latency_ms = round(sum(result["latency_ms"] for result in results) / max(len(results), 1), 2)
    
    summary = {
        "questions": len(results),
        "source_hit_rate": source_hit_rate,
        "average_latency_ms": average_latency_ms,
        "ragas_scores": ragas_scores,
        "results": results,
    }
    write_report(summary)

    # CI/CD Quality Gate Assertions
    if isinstance(ragas_scores, dict) and "error" not in ragas_scores:
        if assert_faithfulness is not None:
            val = ragas_scores.get("faithfulness", 0.0)
            if val < assert_faithfulness:
                print(f"❌ CI Quality Gate FAILED: faithfulness score {val:.3f} < {assert_faithfulness}", file=sys.stderr)
                sys.exit(1)
        if assert_precision is not None:
            val = ragas_scores.get("context_precision", 0.0)
            if val < assert_precision:
                print(f"❌ CI Quality Gate FAILED: context_precision score {val:.3f} < {assert_precision}", file=sys.stderr)
                sys.exit(1)
        if assert_relevancy is not None:
            val = ragas_scores.get("answer_relevancy", 0.0)
            if val < assert_relevancy:
                print(f"❌ CI Quality Gate FAILED: answer_relevancy score {val:.3f} < {assert_relevancy}", file=sys.stderr)
                sys.exit(1)

    return summary

def write_report(summary: dict) -> None:
    REPORT_PATH.parent.mkdir(parents=True, exist_ok=True)
    lines = [
        "# RAG Evaluation Report",
        "",
        f"Generated: {datetime.now(timezone.utc).isoformat()}",
        "",
        "## Overall Metrics",
        f"- **Questions Evaluated:** {summary['questions']}",
        f"- **Source Hit Rate:** {summary['source_hit_rate']}",
        f"- **Average Latency:** {summary['average_latency_ms']} ms",
        "",
        "### Ragas Scores",
        "```json",
        json.dumps(summary.get("ragas_scores", {}), indent=2, default=str),
        "```",
        "",
        "## Question Results",
        "",
    ]
    for result in summary["results"]:
        lines.extend(
            [
                f"### Q: {result['question']}",
                f"**A:** {result['answer']}",
                f"- Source hit: {result['source_hit']}",
                f"- Retrieved contexts: {result['retrieved_contexts']}",
                f"- Latency ms: {result['latency_ms']}",
                "",
            ]
        )
    REPORT_PATH.write_text("\n".join(lines), encoding="utf-8")

def generate_synthetic_testset(output_path: Path = Path("datasets/synthetic_qa.json"), test_size: int = 5) -> list[dict]:
    from ragas.testset.generator import TestsetGenerator
    from langchain_community.document_loaders import DirectoryLoader, TextLoader
    from langchain_text_splitters import RecursiveCharacterTextSplitter
    
    print(f"Generating synthetic testset of size {test_size} from {settings.DATA_DIR}...")
    loader = DirectoryLoader(settings.DATA_DIR, glob="**/*.*", loader_cls=TextLoader, loader_kwargs={"encoding": "utf-8"})
    docs = loader.load()
    splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
    split_docs = splitter.split_documents(docs)
    
    generator = TestsetGenerator.from_langchain_docs(
        docs=split_docs,
        llm=ChatOpenAI(model=settings.LLM_MODEL, temperature=0.7, openai_api_key=settings.OPENROUTER_API_KEY, openai_api_base=settings.OPENROUTER_BASE_URL),
        embeddings=dense_embeddings(),
    )
    testset = generator.generate_with_langchain_docs(split_docs, test_size=test_size)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    df = testset.to_pandas()
    df.to_json(output_path, orient="records", indent=2)
    print(f"Saved {len(df)} synthetic QA pairs to {output_path}")
    return df.to_dict(orient="records")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Support Docs Copilot RAG Evaluation Suite")
    parser.add_argument("--generate-testset", action="store_true", help="Generate synthetic QA testset from documentation")
    parser.add_argument("--test-size", type=int, default=5, help="Number of synthetic QA pairs to generate")
    parser.add_argument("--assert-faithfulness", type=float, default=None, help="Minimum required faithfulness score")
    parser.add_argument("--assert-precision", type=float, default=None, help="Minimum required context precision score")
    parser.add_argument("--assert-relevancy", type=float, default=None, help="Minimum required answer relevancy score")
    args = parser.parse_args()

    if args.generate_testset:
        generate_synthetic_testset(test_size=args.test_size)
    else:
        print("Running evaluation...")
        summary = asyncio.run(run_local_evaluation(
            assert_faithfulness=args.assert_faithfulness,
            assert_precision=args.assert_precision,
            assert_relevancy=args.assert_relevancy,
        ))
        print("Evaluation report generated at:", REPORT_PATH)