evals/run_eval.py

import json
import os
import re
import sys
import time
from pathlib import Path
from statistics import mean

from dotenv import load_dotenv

SERVER_ROOT = Path(__file__).resolve().parents[1]
if str(SERVER_ROOT) not in sys.path:
    sys.path.insert(0, str(SERVER_ROOT))

load_dotenv(SERVER_ROOT / ".env")

from src.rag_system import CodebaseRAGSystem

EVAL_SESSION_KEY = "eval-session"
TOP_K = int(os.getenv("CODEBASE_RAG_TOP_K", "8"))
EVAL_SET_PATH = Path(
    os.getenv("CODEBASE_RAG_EVAL_SET", Path(__file__).with_name("sample_eval_set.json"))
)
EVAL_OUTPUT_PATH = os.getenv("CODEBASE_RAG_EVAL_OUTPUT")
ENABLE_FAITHFULNESS = os.getenv("CODEBASE_RAG_ENABLE_FAITHFULNESS", "1") == "1"


def log(message: str):
    print(f"[eval] {message}", file=sys.stderr, flush=True)


def load_eval_set():
    return json.loads(EVAL_SET_PATH.read_text())


def validate_eval_set(repositories):
    errors = []
    seen_ids = set()
    for repo in repositories:
        if not repo.get("github_url"):
            errors.append(f"repo {repo.get('id', '?')}: missing github_url")
        if not repo.get("cases"):
            errors.append(f"repo {repo.get('id', '?')}: has no cases")
        for case in repo.get("cases", []):
            case_id = case.get("id") or case.get("question", "?")
            if case_id in seen_ids:
                errors.append(f"duplicate case id: {case_id}")
            seen_ids.add(case_id)
            if not case.get("question", "").strip():
                errors.append(f"{case_id}: missing question")
            if not case.get("ground_truth", "").strip():
                errors.append(f"{case_id}: missing ground_truth")
            if not case.get("expected_sources"):
                errors.append(f"{case_id}: expected_sources must be a non-empty list")
    return errors


def normalize_path(path: str) -> str:
    return path.strip().lstrip("./").lower()


def tokenize_text(text: str):
    return re.findall(r"[a-z0-9_]+", (text or "").lower())


def matches_expected(actual_path: str, expected_sources) -> bool:
    actual = normalize_path(actual_path)
    for expected in expected_sources:
        expected_norm = normalize_path(expected).rstrip("/")
        is_dir = "." not in expected_norm.rsplit("/", 1)[-1]
        if actual == expected_norm:
            return True
        if is_dir and actual.startswith(expected_norm + "/"):
            return True
    return False


def compute_retrieval_metrics(expected_sources, actual_sources):
    matching_ranks = [
        rank
        for rank, path in enumerate(actual_sources, start=1)
        if matches_expected(path, expected_sources)
    ]
    hit = bool(matching_ranks)
    top1 = bool(matching_ranks and matching_ranks[0] == 1)
    reciprocal_rank = 1.0 / matching_ranks[0] if matching_ranks else 0.0
    return {
        "retrieval_hit": int(hit),
        "top1_hit": int(top1),
        "reciprocal_rank": reciprocal_rank,
    }


def keyword_hits(answer: str, keywords):
    if not keywords:
        return 0, 0
    tokens = set(tokenize_text(answer))
    matched = 0
    for keyword in keywords:
        keyword_tokens = tokenize_text(keyword)
        if keyword_tokens and all(token in tokens for token in keyword_tokens):
            matched += 1
    return matched, len(keywords)


def judge_faithfulness(rag_system, question: str, answer: str, sources: list):
    if not ENABLE_FAITHFULNESS or not answer.strip() or not sources:
        return None
    context = "\n\n".join(
        f"[{i}] {source['file_path']}\n{source['snippet'][:1500]}"
        for i, source in enumerate(sources, start=1)
    )
    system_prompt = (
        "You are a strict grading assistant. Given a question, retrieved code context, and a "
        "generated answer, output ONLY a single number between 0 and 1 for how faithful the "
        "answer is to the context (1.0 = every claim is supported, 0.0 = the answer invents or "
        "contradicts facts not in the context). Output just the number."
    )
    user_prompt = f"Question: {question}\n\nContext:\n{context}\n\nAnswer:\n{answer}\n\nFaithfulness score:"
    try:
        text, _ = rag_system._generate_markdown_response(system_prompt, user_prompt)
        match = re.search(r"(\d(?:\.\d+)?)", text)
        if not match:
            return None
        return max(0.0, min(1.0, float(match.group(1))))
    except Exception as exc:
        log(f"Faithfulness judge failed: {exc}")
        return None


def index_repo(rag_system, github_url: str, name: str):
    repo = rag_system.create_or_reset_repository(github_url, EVAL_SESSION_KEY)
    log(f"Indexing {name} ({github_url}), repo_id={repo.id}")
    rag_system.index_repository(repo.id)
    repo_state = rag_system.get_repository_for_session(repo.id, EVAL_SESSION_KEY)
    if not repo_state or repo_state["status"] != "indexed":
        detail = repo_state.get("error_message") if repo_state else "repository disappeared"
        raise RuntimeError(f"Failed to index {name}: {detail}")
    log(
        f"Indexed {name}: {repo_state['file_count']} files, "
        f"{repo_state['chunk_count']} chunks"
    )
    return repo.id


def run_case(rag_system, repo_id: int, repo_name: str, case: dict):
    start = time.time()
    result = rag_system.answer_question(
        repo_id=repo_id,
        session_key=EVAL_SESSION_KEY,
        question=case["question"],
        top_k=TOP_K,
        history=case.get("turns", []),
        debug_retrieval=True,
    )
    elapsed_ms = (time.time() - start) * 1000

    sources = result.get("sources", [])
    cited_paths = [source["file_path"] for source in sources]
    retrieval = compute_retrieval_metrics(case.get("expected_sources", []), cited_paths)
    matched, total_keywords = keyword_hits(result.get("answer", ""), case.get("must_include_any", []))
    has_citations = bool(result.get("citations"))
    expected_source_grounded = (
        retrieval["retrieval_hit"] == 1
        and has_citations
        and (total_keywords == 0 or matched > 0)
    )

    retrieval_debug = result.get("retrieval_debug", [])
    for item in retrieval_debug:
        item["expected_source"] = matches_expected(
            item.get("file_path", ""),
            case.get("expected_sources", []),
        )

    return {
        "id": case.get("id", case["question"]),
        "repo": repo_name,
        "category": case.get("category", "general"),
        "question": case["question"],
        "answer": result.get("answer", ""),
        "citations": result.get("citations", []),
        "expected_sources": case.get("expected_sources", []),
        "retrieved_sources": cited_paths,
        "retrieval_hit": retrieval["retrieval_hit"],
        "top1_hit": retrieval["top1_hit"],
        "reciprocal_rank": round(retrieval["reciprocal_rank"], 4),
        "expected_source_grounded": int(expected_source_grounded),
        # Backward-compatible alias for older report consumers.
        "grounded": int(expected_source_grounded),
        "retrieval_debug": retrieval_debug,
        "faithfulness": judge_faithfulness(rag_system, case["question"], result.get("answer", ""), sources),
        "latency_ms": round(elapsed_ms, 1),
    }


def summarize(details):
    if not details:
        return {}
    latencies = sorted(item["latency_ms"] for item in details)
    p95_index = min(len(latencies) - 1, int(len(latencies) * 0.95))
    faith_scores = [item["faithfulness"] for item in details if item["faithfulness"] is not None]
    return {
        "case_count": len(details),
        "retrieval_hit_rate": round(mean(item["retrieval_hit"] for item in details), 4),
        "top1_hit_rate": round(mean(item["top1_hit"] for item in details), 4),
        "mrr": round(mean(item["reciprocal_rank"] for item in details), 4),
        "expected_source_grounded_rate": round(
            mean(item["expected_source_grounded"] for item in details), 4
        ),
        # Backward-compatible alias.
        "grounded_answer_rate": round(mean(item["grounded"] for item in details), 4),
        "faithfulness": round(mean(faith_scores), 4) if faith_scores else None,
        "latency_p95_ms": round(latencies[p95_index], 1),
    }


def summarize_by_repo(details):
    grouped = {}
    for item in details:
        grouped.setdefault(item["repo"], []).append(item)
    return {repo: summarize(items) for repo, items in grouped.items()}


def summarize_by_category(details):
    grouped = {}
    for item in details:
        grouped.setdefault(item["category"], []).append(item)
    return {category: summarize(items) for category, items in sorted(grouped.items())}


def run():
    eval_set = load_eval_set()
    repositories = eval_set["repositories"]

    errors = validate_eval_set(repositories)
    if errors:
        raise RuntimeError("Eval set validation failed: " + "; ".join(errors))

    total_cases = sum(len(repo["cases"]) for repo in repositories)
    log(f"Loaded eval set: {len(repositories)} repositories, {total_cases} cases")

    rag_system = CodebaseRAGSystem()
    log(f"LLM provider={rag_system.llm_provider} model={rag_system.llm_model}")

    details = []
    try:
        for repo_config in repositories:
            repo_id = index_repo(rag_system, repo_config["github_url"], repo_config["name"])
            cases = repo_config["cases"]
            for index, case in enumerate(cases, start=1):
                log(f"[{repo_config['id']} {index}/{len(cases)}] {case['id']}")
                details.append(run_case(rag_system, repo_id, repo_config["name"], case))
    finally:
        rag_system.end_session(EVAL_SESSION_KEY)

    report = {
        "config": {
            "llm_provider": rag_system.llm_provider,
            "llm_model": rag_system.llm_model,
            "top_k": TOP_K,
            "eval_set": str(EVAL_SET_PATH),
            "repositories": [
                {"id": repo["id"], "name": repo["name"], "github_url": repo["github_url"]}
                for repo in repositories
            ],
        },
        "headline_metrics": summarize(details),
        "repo_breakdown": summarize_by_repo(details),
        "category_breakdown": summarize_by_category(details),
        "cases": details,
    }

    if EVAL_OUTPUT_PATH:
        target = Path(EVAL_OUTPUT_PATH)
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_text(json.dumps(report, indent=2))
        log(f"Wrote JSON report to {target}")

    log("Eval complete. Printing JSON report.")
    print(json.dumps(report, indent=2))


if __name__ == "__main__":
    run()