import json import os import sys import asyncio import re import time from pathlib import Path from collections import Counter, defaultdict from statistics import mean import requests 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.bedrock_claude import create_bedrock_runtime_client, generate_bedrock_claude_text from src.embeddings import EmbeddingGenerator API_URL = os.getenv("CODEBASE_RAG_API_URL", "http://localhost:8000") REPO_ID = int(os.getenv("CODEBASE_RAG_REPO_ID", "1")) SESSION_ID = os.getenv("CODEBASE_RAG_SESSION_ID", "eval-session") TOP_K = int(os.getenv("CODEBASE_RAG_TOP_K", "8")) QUERY_TIMEOUT_SECONDS = int(os.getenv("CODEBASE_RAG_QUERY_TIMEOUT_SECONDS", "180")) QUERY_MAX_RETRIES = int(os.getenv("CODEBASE_RAG_QUERY_MAX_RETRIES", "5")) QUERY_RETRY_BASE_SECONDS = float(os.getenv("CODEBASE_RAG_QUERY_RETRY_BASE_SECONDS", "2")) ENABLE_RAGAS = os.getenv("CODEBASE_RAG_ENABLE_RAGAS", "1").lower() not in {"0", "false", "no"} RAGAS_ASYNC = os.getenv("CODEBASE_RAG_RAGAS_ASYNC", "0").lower() in {"1", "true", "yes"} RAGAS_RAISE_EXCEPTIONS = os.getenv("CODEBASE_RAG_RAGAS_RAISE_EXCEPTIONS", "0").lower() in { "1", "true", "yes", } MIN_REFERENCE_OVERLAP = float(os.getenv("CODEBASE_RAG_MIN_REFERENCE_OVERLAP", "0.2")) MIN_REFERENCE_TERM_MATCHES = int(os.getenv("CODEBASE_RAG_MIN_REFERENCE_TERM_MATCHES", "2")) EVAL_SET_PATH = Path( os.getenv( "CODEBASE_RAG_EVAL_SET", Path(__file__).with_name("sample_eval_set.json"), ) ) def log(message: str): print(f"[eval] {message}", file=sys.stderr, flush=True) def get_app_model_config(): llm_provider = os.getenv("LLM_PROVIDER", "bedrock").lower() if llm_provider == "groq": llm_model = os.getenv("GROQ_MODEL", "llama-3.3-70b-versatile") elif llm_provider == "bedrock": llm_model = os.getenv( "BEDROCK_LLM_MODEL", "anthropic.claude-sonnet-4-20250514-v1:0", ) elif llm_provider == "vertex_ai": llm_model = os.getenv("VERTEX_LLM_MODEL", "claude-sonnet-4@20250514") else: llm_model = "unknown" embedding_provider = os.getenv("EMBEDDING_PROVIDER", "auto").lower() if embedding_provider == "bedrock": embedding_model = os.getenv("BEDROCK_EMBEDDING_MODEL", "cohere.embed-v4:0") elif embedding_provider == "vertex_ai": embedding_model = os.getenv("VERTEX_EMBEDDING_MODEL", "gemini-embedding-001") elif embedding_provider == "openai": embedding_model = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small") elif embedding_provider == "local": embedding_model = os.getenv("EMBEDDING_MODEL") or os.getenv( "LOCAL_EMBEDDING_MODEL", "nomic-ai/CodeRankEmbed" ) else: embedding_model = os.getenv("EMBEDDING_MODEL") or "auto" eval_model = os.getenv( "EVAL_MODEL", os.getenv("BEDROCK_EVAL_MODEL", "anthropic.claude-opus-4-20250514-v1:0"), ) return { "llm_provider": llm_provider, "llm_model": llm_model, "embedding_provider": embedding_provider, "embedding_model": embedding_model, "eval_model": eval_model, } def load_eval_rows(): return json.loads(EVAL_SET_PATH.read_text()) def post_query(row): payload = { "repo_id": REPO_ID, "question": row["question"], "top_k": TOP_K, "history": row.get("turns", []), } case_id = row.get("id", row["question"]) for attempt in range(1, QUERY_MAX_RETRIES + 1): response = requests.post( f"{API_URL}/api/query", json=payload, headers={"X-Session-Id": SESSION_ID}, timeout=QUERY_TIMEOUT_SECONDS, ) if response.ok: return response.json() detail = response.text try: parsed = response.json() detail = parsed.get("detail") or parsed except Exception: pass detail_text = str(detail) is_retryable = response.status_code in {429, 500, 502, 503, 504} and any( marker in detail_text for marker in [ "ThrottlingException", "throttled", "Too many requests", "timed out", "timeout", "ServiceUnavailable", "temporarily unavailable", ] ) if is_retryable and attempt < QUERY_MAX_RETRIES: retry_after = response.headers.get("Retry-After") try: wait_seconds = ( float(retry_after) if retry_after else QUERY_RETRY_BASE_SECONDS * (2 ** (attempt - 1)) ) except ValueError: wait_seconds = QUERY_RETRY_BASE_SECONDS * (2 ** (attempt - 1)) log( f"Retrying case {case_id} after transient query failure " f"(attempt {attempt}/{QUERY_MAX_RETRIES}, wait={wait_seconds:.1f}s): {detail_text}" ) time.sleep(wait_seconds) continue raise RuntimeError( f"Query failed for eval case {case_id!r} " f"with status {response.status_code}: {detail}" ) raise RuntimeError(f"Query failed for eval case {case_id!r}: exhausted retries") def normalize_path(path: str) -> str: return path.strip().lstrip("./").lower() STOPWORDS = { "a", "an", "and", "are", "as", "at", "be", "by", "for", "from", "how", "in", "into", "is", "it", "its", "of", "on", "or", "that", "the", "their", "this", "to", "what", "when", "where", "which", "with", } def tokenize_text(text: str): tokens = [] for raw_token in re.findall(r"[A-Za-z0-9_./+-]+", text or ""): token = raw_token.lower() tokens.append(token) camel_parts = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", " ", raw_token).split() split_parts = re.split(r"[._/+-]+", token) for part in [*camel_parts, *split_parts]: normalized = part.strip().lower() if normalized and normalized != token: tokens.append(normalized) return tokens def normalize_keywords(keywords): normalized = [] seen = set() for keyword in keywords or []: phrase = " ".join(tokenize_text(str(keyword))) if not phrase or phrase in seen: continue seen.add(phrase) normalized.append(phrase) return normalized def compute_retrieval_metrics(expected_sources, actual_sources): expected = {normalize_path(path) for path in expected_sources} actual = [normalize_path(path) for path in actual_sources] unique_actual = list(dict.fromkeys(actual)) def matches_expected(actual_path: str) -> bool: for expected_path in expected: expected_is_directory = ( expected_path.endswith("/") or "." not in expected_path.rsplit("/", 1)[-1] ) normalized_expected = expected_path.rstrip("/") if actual_path == expected_path: return True if expected_is_directory and actual_path.startswith(normalized_expected + "/"): return True return False hit = 1 if any(matches_expected(path) for path in actual) else 0 recall = 0.0 if expected: matched_expected = set() for expected_path in expected: expected_is_directory = ( expected_path.endswith("/") or "." not in expected_path.rsplit("/", 1)[-1] ) normalized_expected = expected_path.rstrip("/") for actual_path in actual: if actual_path == expected_path or ( expected_is_directory and actual_path.startswith(normalized_expected + "/") ): matched_expected.add(expected_path) break recall = len(matched_expected) / len(expected) mrr = 0.0 for index, path in enumerate(actual, start=1): if matches_expected(path): mrr = 1.0 / index break return { "retrieval_hit": hit, "source_recall": recall, "mrr": mrr, "top1_hit": 1 if actual and matches_expected(actual[0]) else 0, "unique_source_precision": ( sum(1 for path in unique_actual if matches_expected(path)) / len(unique_actual) if unique_actual else 0.0 ), "duplicate_source_rate": ( (len(actual) - len(unique_actual)) / len(actual) if actual else 0.0 ), } def keyword_match_details(row, answer: str): keywords = normalize_keywords(row.get("must_include_any", [])) if not keywords: return None answer_tokens = tokenize_text(answer) if not answer_tokens: return { "coverage": 0.0, "matched_count": 0, "total_keywords": len(keywords), "matched_keywords": [], "missing_keywords": keywords, } matched_keywords = [] for keyword in keywords: keyword_tokens = keyword.split() window = len(keyword_tokens) if window == 1: if keyword_tokens[0] in answer_tokens: matched_keywords.append(keyword) continue for index in range(0, len(answer_tokens) - window + 1): if answer_tokens[index : index + window] == keyword_tokens: matched_keywords.append(keyword) break matched_set = set(matched_keywords) missing_keywords = [keyword for keyword in keywords if keyword not in matched_set] matched_count = len(matched_set) return { "coverage": matched_count / len(keywords), "matched_count": matched_count, "total_keywords": len(keywords), "matched_keywords": sorted(matched_set), "missing_keywords": missing_keywords, } def keyword_pass(row, keyword_details): if keyword_details is None: return None minimum = int(row.get("min_keyword_matches", 1)) return 1 if keyword_details["matched_count"] >= minimum else 0 def answer_length_metrics(answer: str): tokens = tokenize_text(answer) return { "answer_word_count": len(tokens), "has_substantive_answer": 1 if len(tokens) >= 40 else 0, } def reference_support_details(reference: str, candidate: str): reference_terms = { token for token in tokenize_text(reference) if len(token) > 2 and token not in STOPWORDS } if not reference_terms: return None candidate_terms = set(tokenize_text(candidate)) matched_terms = sorted(token for token in reference_terms if token in candidate_terms) matched_count = len(matched_terms) return { "ratio": matched_count / len(reference_terms), "matched_count": matched_count, "reference_term_count": len(reference_terms), "matched_terms": matched_terms, } def reference_support_pass(reference_details): if reference_details is None: return None return 1 if ( reference_details["ratio"] >= MIN_REFERENCE_OVERLAP and reference_details["matched_count"] >= MIN_REFERENCE_TERM_MATCHES ) else 0 def validate_eval_rows(rows): errors = [] warnings = [] category_counts = Counter() id_counts = Counter() id_prefix_counts = Counter() expected_source_counts = [] keyword_counts = [] conversation_cases = 0 benchmark_scope = { "type": "mixed_or_unknown", "dominant_prefix": None, "dominant_prefix_fraction": 0.0, } for index, row in enumerate(rows, start=1): row_id = row.get("id") or f"row-{index}" id_counts[row_id] += 1 prefix = row_id.split("-", 1)[0].lower() if prefix: id_prefix_counts[prefix] += 1 category_counts[row.get("category", "general")] += 1 question = str(row.get("question", "")).strip() ground_truth = str(row.get("ground_truth", "")).strip() expected_sources = row.get("expected_sources", []) must_include_any = row.get("must_include_any", []) if not question: errors.append(f"{row_id}: missing question") if not ground_truth: errors.append(f"{row_id}: missing ground_truth") if not isinstance(expected_sources, list) or not expected_sources: errors.append(f"{row_id}: expected_sources must be a non-empty list") if must_include_any and not isinstance(must_include_any, list): errors.append(f"{row_id}: must_include_any must be a list when present") if isinstance(must_include_any, list): normalized_keywords = normalize_keywords(must_include_any) if len(normalized_keywords) != len([keyword for keyword in must_include_any if str(keyword).strip()]): warnings.append( f"{row_id}: duplicate or case-variant keywords were normalized; " "resume metrics are stricter than the raw checklist wording." ) if row.get("turns"): conversation_cases += 1 expected_source_counts.append(len(expected_sources) if isinstance(expected_sources, list) else 0) keyword_counts.append(len(must_include_any) if isinstance(must_include_any, list) else 0) duplicate_ids = sorted(row_id for row_id, count in id_counts.items() if count > 1) if duplicate_ids: errors.append(f"duplicate ids found: {', '.join(duplicate_ids)}") if len(rows) < 25: warnings.append( "Eval set has fewer than 25 cases. Good for iteration, but light for resume-grade benchmarking." ) if len(category_counts) < 4: warnings.append("Eval set covers fewer than 4 categories, so breadth is limited.") if conversation_cases < 2: warnings.append("Eval set has very little multi-turn coverage.") if category_counts and min(category_counts.values()) < 2: sparse = sorted(category for category, count in category_counts.items() if count < 2) warnings.append(f"Some categories are underrepresented: {', '.join(sparse)}.") if id_prefix_counts: dominant_prefix, dominant_count = id_prefix_counts.most_common(1)[0] dominant_prefix_fraction = dominant_count / len(rows) if dominant_prefix_fraction >= 0.8: benchmark_scope = { "type": "single_repository", "dominant_prefix": dominant_prefix, "dominant_prefix_fraction": round(dominant_prefix_fraction, 4), } return { "case_count": len(rows), "category_counts": dict(sorted(category_counts.items())), "conversation_case_count": conversation_cases, "average_expected_sources": round(mean(expected_source_counts), 2) if expected_source_counts else 0.0, "average_keywords_per_case": round(mean(keyword_counts), 2) if keyword_counts else 0.0, "benchmark_scope": benchmark_scope, "errors": errors, "warnings": warnings, "is_valid": not errors, } def summarize_custom_metrics(details): keyword_coverages = [item["keyword_coverage"] for item in details if item["keyword_coverage"] is not None] keyword_passes = [item["keyword_pass"] for item in details if item["keyword_pass"] is not None] reference_support_passes = [ item["reference_support_pass"] for item in details if item["reference_support_pass"] is not None ] grounded_answer_passes = [ 1 for item in details if item["retrieval_hit"] == 1 and item["has_substantive_answer"] == 1 and (item["keyword_pass"] in {None, 1}) and (item["reference_support_pass"] in {None, 1}) ] exact_source_recall_cases = [1 for item in details if item["source_recall"] == 1.0] return { "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), "source_recall": round(mean(item["source_recall"] for item in details), 4), "mrr": round(mean(item["mrr"] for item in details), 4), "unique_source_precision": round(mean(item["unique_source_precision"] for item in details), 4), "duplicate_source_rate": round(mean(item["duplicate_source_rate"] for item in details), 4), "keyword_coverage": round(mean(keyword_coverages), 4) if keyword_coverages else None, "keyword_pass_rate": round(mean(keyword_passes), 4) if keyword_passes else None, "reference_support_rate": round(mean(reference_support_passes), 4) if reference_support_passes else None, "ground_truth_lexical_overlap": round( mean(item["ground_truth_lexical_overlap"] for item in details if item["ground_truth_lexical_overlap"] is not None), 4, ) if any(item["ground_truth_lexical_overlap"] is not None for item in details) else None, "substantive_answer_rate": round(mean(item["has_substantive_answer"] for item in details), 4), "grounded_answer_rate": round(sum(grounded_answer_passes) / len(details), 4) if details else 0.0, "exact_source_recall_rate": round(sum(exact_source_recall_cases) / len(details), 4) if details else 0.0, } def summarize_by_category(details): grouped = defaultdict(list) for item in details: grouped[item["category"]].append(item) summary = {} for category, items in sorted(grouped.items()): keyword_passes = [item["keyword_pass"] for item in items if item["keyword_pass"] is not None] summary[category] = { "case_count": len(items), "retrieval_hit_rate": round(mean(item["retrieval_hit"] for item in items), 4), "top1_hit_rate": round(mean(item["top1_hit"] for item in items), 4), "source_recall": round(mean(item["source_recall"] for item in items), 4), "mrr": round(mean(item["mrr"] for item in items), 4), "keyword_pass_rate": round(mean(keyword_passes), 4) if keyword_passes else None, "reference_support_rate": round( mean( item["reference_support_pass"] for item in items if item["reference_support_pass"] is not None ), 4, ) if any(item["reference_support_pass"] is not None for item in items) else None, "grounded_answer_rate": round( mean( 1 if item["retrieval_hit"] == 1 and item["has_substantive_answer"] == 1 and item["keyword_pass"] in {None, 1} and item["reference_support_pass"] in {None, 1} else 0 for item in items ), 4, ), } return summary def build_headline_metrics(custom_metrics, audit): return { "sample_size": audit["case_count"], "category_count": len(audit["category_counts"]), "retrieval_hit_rate": custom_metrics["retrieval_hit_rate"], "top1_hit_rate": custom_metrics["top1_hit_rate"], "mrr": custom_metrics["mrr"], "source_recall": custom_metrics["source_recall"], "grounded_answer_rate": custom_metrics["grounded_answer_rate"], "keyword_pass_rate": custom_metrics["keyword_pass_rate"], "reference_support_rate": custom_metrics["reference_support_rate"], } def build_metric_guidance(custom_metrics, ragas_report): retrieval_gate_thresholds = { "retrieval_hit_rate": 0.8, "top1_hit_rate": 0.8, "mrr": 0.75, } retrieval_gate_pass = all( custom_metrics[key] >= threshold for key, threshold in retrieval_gate_thresholds.items() ) next_focus = [] if custom_metrics["source_recall"] < 0.7: next_focus.append("Improve multi-source recall for cross-file and implementation questions.") if custom_metrics["duplicate_source_rate"] > 0.15: next_focus.append("Reduce duplicate or near-duplicate source chunks before answer generation.") if custom_metrics["grounded_answer_rate"] < 0.75: next_focus.append("Tighten answer grounding and checklist coverage before presenting this as a broad benchmark.") if ragas_report and ragas_report.get("context_precision", 1.0) < 0.7: next_focus.append("Treat low RAGAS context precision as a context-selection signal, not as the primary pass/fail gate.") return { "primary_gate": "pass" if retrieval_gate_pass else "needs_work", "primary_gate_basis": "deterministic_retrieval", "primary_gate_thresholds": retrieval_gate_thresholds, "ragas_role": "supporting_signal_not_primary_gate", "next_focus": next_focus, } def build_resume_summary(custom_metrics, audit, ragas_report, ragas_error): lines = [ ( f"Evaluated on {audit['case_count']} repo-QA cases across " f"{len(audit['category_counts'])} categories." ), ( f"Deterministic retrieval metrics: hit@{TOP_K} {custom_metrics['retrieval_hit_rate']:.1%}, " f"top-1 hit {custom_metrics['top1_hit_rate']:.1%}, MRR {custom_metrics['mrr']:.3f}, " f"source recall {custom_metrics['source_recall']:.1%}." ), ( f"Strict answer quality checks: grounded answer rate {custom_metrics['grounded_answer_rate']:.1%}" + ( f", keyword/checklist pass rate {custom_metrics['keyword_pass_rate']:.1%}" + ( f", reference-support pass rate {custom_metrics['reference_support_rate']:.1%}." if custom_metrics["reference_support_rate"] is not None else "." ) if custom_metrics["keyword_pass_rate"] is not None else "." ) ), ] if ragas_report and not ragas_error: lines.append( "LLM-judge metrics (supporting signal, not primary headline): " f"faithfulness {ragas_report.get('faithfulness', 0.0):.3f}, " f"answer relevancy {ragas_report.get('answer_relevancy', 0.0):.3f}, " f"context precision {ragas_report.get('context_precision', 0.0):.3f}." ) else: lines.append("LLM-judge metrics were skipped or unstable, so headline metrics rely on deterministic checks.") scope = audit.get("benchmark_scope", {}) if scope.get("type") == "single_repository": lines.append( "Benchmark scope: single-repository benchmark " f"({scope.get('dominant_prefix')}); use it to judge this target repo, not cross-repo generalization." ) if audit["warnings"]: lines.append( "Benchmark caveat: " + " ".join(audit["warnings"][:2]) ) return " ".join(lines) def benchmark_readiness(audit, ragas_error, metric_guidance=None): reasons = [] if audit["case_count"] < 25: reasons.append("small_sample") if len(audit["category_counts"]) < 4: reasons.append("limited_category_coverage") if audit["conversation_case_count"] < 2: reasons.append("limited_multi_turn_coverage") if audit["warnings"]: reasons.append("eval_set_warnings") if ragas_error not in {None, "disabled"}: reasons.append("ragas_instability") if metric_guidance and metric_guidance.get("primary_gate") != "pass": reasons.append("primary_gate_failed") if reasons: status = "single_repo_benchmark_needs_work" if audit.get("benchmark_scope", {}).get("type") != "single_repository": status = "internal_or_demo_benchmark" return { "status": status, "reasons": reasons, } if audit.get("benchmark_scope", {}).get("type") == "single_repository": return { "status": "single_repo_benchmark_ready", "reasons": [], } return { "status": "presentation_ready", "reasons": [], } def maybe_write_report(report): output_path = os.getenv("CODEBASE_RAG_EVAL_OUTPUT") if not output_path: return None target = Path(output_path) target.parent.mkdir(parents=True, exist_ok=True) target.write_text(json.dumps(report, indent=2)) return str(target) def build_bedrock_ragas_llm(run_config): from langchain_core.outputs import Generation, LLMResult from ragas.llms.base import BaseRagasLLM class BedrockRagasLLM(BaseRagasLLM): def __init__(self, model: str, run_config): self.client = create_bedrock_runtime_client() self.model = model self.set_run_config(run_config) def _prompt_to_text(self, prompt): prefix = ( "Return only valid JSON or the exact structured output requested by the prompt. " "Do not add markdown fences, explanations, or extra prose.\n\n" ) if hasattr(prompt, "to_string"): return prefix + prompt.to_string() return prefix + str(prompt) def _generate_once(self, prompt, n=1, temperature=1e-8, stop=None, callbacks=None): prompt_text = self._prompt_to_text(prompt) text, _ = generate_bedrock_claude_text( self.client, self.model, "Return only valid JSON or the exact structured output requested.", prompt_text, max_tokens=int(os.getenv("EVAL_MAX_OUTPUT_TOKENS", "2048")), temperature=0.0, ) generations = [Generation(text=text)] if text else [] if not generations: raise RuntimeError("Bedrock Claude judge returned an empty response.") return LLMResult(generations=[generations]) def generate_text(self, prompt, n=1, temperature=1e-8, stop=None, callbacks=None): return self._generate_once( prompt=prompt, n=n, temperature=temperature, stop=stop, callbacks=callbacks, ) async def agenerate_text(self, prompt, n=1, temperature=1e-8, stop=None, callbacks=None): return await asyncio.to_thread( self._generate_once, prompt, n, temperature, stop, callbacks, ) model = os.getenv( "EVAL_MODEL", os.getenv("BEDROCK_EVAL_MODEL", "anthropic.claude-opus-4-20250514-v1:0"), ) return BedrockRagasLLM(model=model, run_config=run_config) def build_ragas_embeddings(run_config): from ragas.embeddings.base import BaseRagasEmbeddings class AppEmbeddingWrapper(BaseRagasEmbeddings): def __init__(self, generator, run_config): self.generator = generator self.set_run_config(run_config) def embed_query(self, text): return self.generator.embed_text(text).tolist() def embed_documents(self, texts): vectors = self.generator.embed_batch(list(texts)) return vectors.tolist() async def aembed_query(self, text): return await asyncio.to_thread(self.embed_query, text) async def aembed_documents(self, texts): return await asyncio.to_thread(self.embed_documents, texts) return AppEmbeddingWrapper(EmbeddingGenerator(), run_config=run_config) def run_ragas(rows, outputs): if not ENABLE_RAGAS: log("RAGAS disabled via CODEBASE_RAG_ENABLE_RAGAS=0. Reporting custom metrics only.") return None, "disabled" try: from datasets import Dataset from ragas import evaluate from ragas.metrics import answer_relevancy, context_precision, faithfulness from ragas.run_config import RunConfig except Exception as exc: log(f"Skipping RAGAS because the evaluation dependencies could not be loaded: {exc}") return None, f"import_error: {exc}" def build_ragas_dataset(): samples = [] for row, result in zip(rows, outputs): samples.append( { "question": row["question"], "answer": result["answer"], "contexts": [source["snippet"] for source in result.get("sources", [])], "ground_truth": row["ground_truth"], } ) return Dataset.from_list(samples) log("Running RAGAS metrics. This can take a while.") try: timeout_seconds = int(os.getenv("EVAL_TIMEOUT_SECONDS", "180")) thread_timeout_seconds = float(os.getenv("EVAL_THREAD_TIMEOUT_SECONDS", str(max(timeout_seconds, 240)))) max_workers = int(os.getenv("EVAL_MAX_WORKERS", "2")) run_config = RunConfig( timeout=timeout_seconds, thread_timeout=thread_timeout_seconds, max_workers=max_workers, max_retries=int(os.getenv("EVAL_MAX_RETRIES", "3")), max_wait=int(os.getenv("EVAL_MAX_WAIT_SECONDS", "60")), ) log( "Using Bedrock for RAGAS judge model " f"({os.getenv('EVAL_MODEL', os.getenv('BEDROCK_EVAL_MODEL', 'anthropic.claude-opus-4-20250514-v1:0'))})" ) log( f"RAGAS runtime: async={RAGAS_ASYNC}, raise_exceptions={RAGAS_RAISE_EXCEPTIONS}, " f"timeout={timeout_seconds}s, thread_timeout={thread_timeout_seconds}s, max_workers={max_workers}" ) llm = build_bedrock_ragas_llm(run_config) embeddings = build_ragas_embeddings(run_config) ragas_report = evaluate( build_ragas_dataset(), metrics=[faithfulness, answer_relevancy, context_precision], llm=llm, embeddings=embeddings, run_config=run_config, is_async=RAGAS_ASYNC, raise_exceptions=RAGAS_RAISE_EXCEPTIONS, ) return {key: float(value) for key, value in ragas_report.items()}, None except Exception as exc: log(f"RAGAS evaluation failed: {exc}") return None, str(exc) def run(): log(f"Loading eval set from {EVAL_SET_PATH}") rows = load_eval_rows() audit = validate_eval_rows(rows) model_config = get_app_model_config() if audit["errors"]: raise RuntimeError("Eval set validation failed: " + "; ".join(audit["errors"])) for warning in audit["warnings"]: log(f"Eval set warning: {warning}") log( "Eval model config: " f"qna_provider={model_config['llm_provider']}, " f"qna_model={model_config['llm_model']}, " f"embedding_provider={model_config['embedding_provider']}, " f"embedding_model={model_config['embedding_model']}, " f"judge_model={model_config['eval_model']}" ) log( f"Starting eval with api_url={API_URL}, repo_id={REPO_ID}, " f"session_id={SESSION_ID}, top_k={TOP_K}, cases={len(rows)}" ) outputs = [] details = [] for index, row in enumerate(rows, start=1): case_id = row.get("id", row["question"]) log(f"[{index}/{len(rows)}] Querying case {case_id}") result = post_query(row) outputs.append(result) log( f"[{index}/{len(rows)}] Received answer for {case_id} " f"with {len(result.get('sources', []))} sources" ) cited_paths = [source["file_path"] for source in result.get("sources", [])] metrics = compute_retrieval_metrics(row.get("expected_sources", []), cited_paths) keyword_details = keyword_match_details(row, result.get("answer", "")) keyword_coverage = keyword_details["coverage"] if keyword_details else None keyword_gate = keyword_pass(row, keyword_details) length_metrics = answer_length_metrics(result.get("answer", "")) reference_details = reference_support_details(row.get("ground_truth", ""), result.get("answer", "")) overlap = reference_details["ratio"] if reference_details else None reference_gate = reference_support_pass(reference_details) details.append( { "id": row.get("id", row["question"]), "category": row.get("category", "general"), "question": row["question"], "answer": result.get("answer", ""), "expected_sources": row.get("expected_sources", []), "retrieved_sources": cited_paths, "retrieval_hit": metrics["retrieval_hit"], "source_recall": metrics["source_recall"], "mrr": metrics["mrr"], "top1_hit": metrics["top1_hit"], "unique_source_precision": metrics["unique_source_precision"], "duplicate_source_rate": metrics["duplicate_source_rate"], "keyword_coverage": keyword_coverage, "keyword_pass": keyword_gate, "matched_keyword_count": keyword_details["matched_count"] if keyword_details else None, "total_keywords": keyword_details["total_keywords"] if keyword_details else None, "matched_keywords": keyword_details["matched_keywords"] if keyword_details else [], "missing_keywords": keyword_details["missing_keywords"] if keyword_details else [], "ground_truth_lexical_overlap": overlap, "reference_support_pass": reference_gate, "reference_term_match_count": reference_details["matched_count"] if reference_details else None, "reference_term_count": reference_details["reference_term_count"] if reference_details else None, "matched_reference_terms": reference_details["matched_terms"] if reference_details else [], **length_metrics, } ) log("Finished query loop. Computing aggregate metrics.") custom_metrics = summarize_custom_metrics(details) category_breakdown = summarize_by_category(details) ragas_report, ragas_error = run_ragas(rows, outputs) headline_metrics = build_headline_metrics(custom_metrics, audit) metric_guidance = build_metric_guidance(custom_metrics, ragas_report) resume_summary = build_resume_summary(custom_metrics, audit, ragas_report, ragas_error) readiness = benchmark_readiness(audit, ragas_error, metric_guidance) report = { "config": { "api_url": API_URL, "repo_id": REPO_ID, "session_id": SESSION_ID, "top_k": TOP_K, "qna_provider": model_config["llm_provider"], "qna_model": model_config["llm_model"], "embedding_provider": model_config["embedding_provider"], "embedding_model": model_config["embedding_model"], "eval_model": model_config["eval_model"], "query_timeout_seconds": QUERY_TIMEOUT_SECONDS, "query_max_retries": QUERY_MAX_RETRIES, "query_retry_base_seconds": QUERY_RETRY_BASE_SECONDS, "eval_set": str(EVAL_SET_PATH), "min_reference_overlap": MIN_REFERENCE_OVERLAP, "min_reference_term_matches": MIN_REFERENCE_TERM_MATCHES, }, "eval_set_audit": audit, "headline_metrics": headline_metrics, "benchmark_readiness": readiness, "metric_guidance": metric_guidance, "ragas": ragas_report, "ragas_error": ragas_error, "custom_metrics": custom_metrics, "category_breakdown": category_breakdown, "resume_summary": resume_summary, "cases": details, } output_path = maybe_write_report(report) if output_path: log(f"Wrote JSON report to {output_path}") log("Eval complete. Printing JSON report.") print(json.dumps(report, indent=2)) if __name__ == "__main__": run()