From ef8bab59628fb6097a7da734b977ea8b6b8d911c Mon Sep 17 00:00:00 2001 From: Mahdi Bazrafshan Date: Mon, 27 Jul 2026 14:14:40 +0330 Subject: [PATCH] feat(benchmarking): add evaluation, benchmark service, and report MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: - Need LLM-as-Judge evaluation for automated scoring - Need benchmark orchestration to run questions × strategies - Need HTML report generation with two views (managerial/technical) Changes: - evaluation.py: LLM-as-Judge scoring on 4 metrics (context, similarity, faithfulness, hallucination) - benchmark_service.py: Orchestration with per-strategy failure isolation - report.py: Dual-view HTML reports with dark mode, charts, and strategy cards --- src/benchmarking/benchmark_service.py | 358 +++++++++++ src/benchmarking/evaluation.py | 208 +++++++ src/benchmarking/report.py | 815 ++++++++++++++++++++++++++ 3 files changed, 1381 insertions(+) create mode 100644 src/benchmarking/benchmark_service.py create mode 100644 src/benchmarking/evaluation.py create mode 100644 src/benchmarking/report.py diff --git a/src/benchmarking/benchmark_service.py b/src/benchmarking/benchmark_service.py new file mode 100644 index 0000000..d34f208 --- /dev/null +++ b/src/benchmarking/benchmark_service.py @@ -0,0 +1,358 @@ +"""Benchmark service: orchestrates full benchmark runs. + +Runs multiple questions against multiple strategies, evaluates answers, +and stores results for comparison. +""" + +from __future__ import annotations + +import json +import logging +import time +from pathlib import Path +from typing import Any + +from src.benchmarking.evaluation import evaluate_single +from src.benchmarking.query_service import run_query +from src.core.config import settings +from src.core.exceptions import BenchmarkError +from src.core.models import StrategyName +from src.storage import sqlite as db + +logger = logging.getLogger(__name__) + +# ── Questions Loading ───────────────────────────────────────────── + +def load_questions(file_path: str | Path) -> list[dict]: + """Load questions from a JSON file. + + Args: + file_path: Path to questions.json + + Returns: + List of question dicts + + Raises: + BenchmarkError: If file not found or invalid + """ + path = Path(file_path) + + if not path.exists(): + raise BenchmarkError(f"Questions file not found: {path}") + + try: + with open(path, encoding="utf-8") as f: + data = json.load(f) + + if "questions" not in data: + raise BenchmarkError("Invalid questions file: missing 'questions' key") + + questions = data["questions"] + logger.info("Loaded %d questions from %s", len(questions), path.name) + + return questions + + except json.JSONDecodeError as exc: + raise BenchmarkError(f"Invalid JSON in questions file: {exc}") from exc + + +def load_questions_from_string(questions_json: str) -> list[dict]: + """Load questions from a JSON string. + + Args: + questions_json: JSON string containing questions + + Returns: + List of question dicts + """ + try: + data = json.loads(questions_json) + + if isinstance(data, list): + return data + elif "questions" in data: + return data["questions"] + else: + raise BenchmarkError("Invalid questions format") + + except json.JSONDecodeError as exc: + raise BenchmarkError(f"Invalid JSON string: {exc}") from exc + + +# ── Cost Estimation ─────────────────────────────────────────────── + +def estimate_cost( + num_questions: int, + num_strategies: int, +) -> dict[str, Any]: + """Estimate the cost of running a benchmark. + + Args: + num_questions: Number of questions + num_strategies: Number of strategies + + Returns: + Cost estimate dict + """ + # Rough estimates based on GPT-4o-mini pricing + embedding_cost_per_call = 0.0001 + query_cost_per_call = 0.001 + evaluation_cost_per_call = 0.001 + + total_queries = num_questions * num_strategies + total_evaluations = total_queries + + embedding_cost = total_queries * embedding_cost_per_call + query_cost = total_queries * query_cost_per_call + evaluation_cost = total_evaluations * evaluation_cost_per_call + total_cost = embedding_cost + query_cost + evaluation_cost + + # Estimate tokens + avg_input_tokens = 500 + avg_output_tokens = 200 + total_input_tokens = total_queries * avg_input_tokens + total_evaluations * avg_input_tokens + total_output_tokens = total_queries * avg_output_tokens + total_evaluations * avg_output_tokens + + return { + "num_questions": num_questions, + "num_strategies": num_strategies, + "total_queries": total_queries, + "total_evaluations": total_evaluations, + "estimated_tokens": { + "input": total_input_tokens, + "output": total_output_tokens, + }, + "estimated_cost_usd": round(total_cost, 4), + "cost_breakdown": { + "embedding": round(embedding_cost, 4), + "queries": round(query_cost, 4), + "evaluation": round(evaluation_cost, 4), + }, + } + + +# ── Benchmark Runner ────────────────────────────────────────────── + +def run_benchmark( + *, + document_id: str, + strategies: list[StrategyName], + questions: list[dict], + top_k: int = 5, +) -> dict[str, Any]: + """Run a full benchmark across questions and strategies. + + Args: + document_id: Document to benchmark against + strategies: List of strategies to test + questions: List of question dicts + top_k: Number of chunks to retrieve per query + + Returns: + Complete benchmark results + """ + t_start = time.time() + logger.info("=" * 80) + logger.info("[BENCHMARK] Starting benchmark") + logger.info("[BENCHMARK] Document: %s", document_id) + logger.info("[BENCHMARK] Strategies: %s", [s.value for s in strategies]) + logger.info("[BENCHMARK] Questions: %d", len(questions)) + + per_question_results = [] + total_cost = 0.0 + + for q_idx, question in enumerate(questions, 1): + question_text = question.get("question", "") + expected_answer = question.get("expected_answer", "") + question_id = question.get("id", f"q{q_idx}") + + logger.info("[BENCHMARK] Question %d/%d: %s", q_idx, len(questions), question_text[:50]) + + question_results = { + "question_id": question_id, + "question": question_text, + "expected_answer": expected_answer, + "strategies": {}, + } + + for strategy in strategies: + logger.info("[BENCHMARK] Strategy: %s", strategy.value) + + try: + # Run query + t0 = time.time() + query_result = run_query( + document_id=document_id, + strategy_name=strategy, + question=question_text, + top_k=top_k, + ) + t_query = time.time() - t0 + + # Evaluate + t1 = time.time() + eval_scores = evaluate_single( + question=question_text, + retrieved_chunks=query_result["retrieved_chunks"], + expected_answer=expected_answer, + generated_answer=query_result["answer"], + ) + t_eval = time.time() - t1 + + # Track cost + query_tokens = query_result.get("token_usage", {}).get("total_tokens", 0) + total_cost += query_tokens * 0.000001 # rough estimate + + question_results["strategies"][strategy.value] = { + "answer": query_result["answer"], + "retrieved_chunks": query_result["retrieved_chunks"], + "scores": eval_scores, + "latency": { + "query_seconds": round(t_query, 3), + "evaluation_seconds": round(t_eval, 3), + }, + "token_usage": query_result.get("token_usage", {}), + } + + logger.info( + "[BENCHMARK] Scores: context=%d, similarity=%d, faithfulness=%d, hallucination=%s", + eval_scores["context_relevance"], + eval_scores["answer_similarity"], + eval_scores["faithfulness"], + eval_scores["hallucination"], + ) + + except Exception as exc: + logger.error("[BENCHMARK] Strategy %s failed: %s", strategy.value, exc) + question_results["strategies"][strategy.value] = { + "error": str(exc), + "scores": { + "context_relevance": 0, + "answer_similarity": 0, + "faithfulness": 0, + "hallucination": True, + "reasoning": f"Error: {exc}", + }, + } + + per_question_results.append(question_results) + + # Aggregate metrics per strategy + aggregate = _aggregate_metrics(per_question_results, strategies) + + t_total = time.time() - t_start + + # Find best strategy + best_strategy = _find_best_strategy(aggregate) + + # Store experiment + experiment = db.save_experiment( + document_id=document_id, + benchmark_config={ + "strategies": [s.value for s in strategies], + "num_questions": len(questions), + "top_k": top_k, + }, + questions=questions, + per_question=per_question_results, + aggregate_metrics=aggregate, + strategies_used=[s.value for s in strategies], + ) + + logger.info("[BENCHMARK] Completed in %.1fs", t_total) + logger.info("[BENCHMARK] Best strategy: %s", best_strategy) + logger.info("=" * 80) + + return { + "experiment_id": experiment["id"], + "document_id": document_id, + "strategies_used": [s.value for s in strategies], + "questions_count": len(questions), + "aggregate_metrics": aggregate, + "best_strategy": best_strategy, + "total_latency_seconds": round(t_total, 2), + "estimated_cost_usd": round(total_cost, 4), + "created_at": experiment["created_at"], + } + + +def _aggregate_metrics( + per_question_results: list[dict], + strategies: list[StrategyName], +) -> dict[str, dict]: + """Aggregate metrics across all questions for each strategy.""" + aggregate = {} + + for strategy in strategies: + strategy_name = strategy.value + scores_list = [] + hallucination_count = 0 + total_count = 0 + + for qr in per_question_results: + strat_result = qr.get("strategies", {}).get(strategy_name, {}) + if "error" in strat_result: + continue + + scores = strat_result.get("scores", {}) + if scores: + scores_list.append(scores) + total_count += 1 + if scores.get("hallucination", False): + hallucination_count += 1 + + if scores_list: + avg_context = sum(s.get("context_relevance", 0) for s in scores_list) / len(scores_list) + avg_similarity = sum(s.get("answer_similarity", 0) for s in scores_list) / len(scores_list) + avg_faithfulness = sum(s.get("faithfulness", 0) for s in scores_list) / len(scores_list) + hallucination_rate = hallucination_count / total_count if total_count > 0 else 0 + else: + avg_context = 0 + avg_similarity = 0 + avg_faithfulness = 0 + hallucination_rate = 0 + + aggregate[strategy_name] = { + "avg_context_relevance": round(avg_context, 2), + "avg_answer_similarity": round(avg_similarity, 2), + "avg_faithfulness": round(avg_faithfulness, 2), + "hallucination_rate": round(hallucination_rate, 2), + "total_questions": total_count, + "failed_questions": len(per_question_results) - total_count, + } + + return aggregate + + +def _find_best_strategy(aggregate: dict) -> str: + """Find the best strategy based on overall score.""" + best_strategy = None + best_score = -1 + + for strategy_name, metrics in aggregate.items(): + # Calculate overall score (weighted average) + overall = ( + metrics.get("avg_context_relevance", 0) * 0.3 + + metrics.get("avg_answer_similarity", 0) * 0.4 + + metrics.get("avg_faithfulness", 0) * 0.3 + ) + # Penalize hallucination + overall *= (1 - metrics.get("hallucination_rate", 0)) + + if overall > best_score: + best_score = overall + best_strategy = strategy_name + + return best_strategy or "unknown" + + +# ── Experiment Retrieval ────────────────────────────────────────── + +def get_experiment(experiment_id: str) -> dict[str, Any] | None: + """Retrieve an experiment by ID.""" + return db.get_experiment(experiment_id) + + +def list_experiments(document_id: str | None = None) -> dict[str, Any]: + """List experiments, optionally filtered by document.""" + return db.list_experiments(document_id=document_id) diff --git a/src/benchmarking/evaluation.py b/src/benchmarking/evaluation.py new file mode 100644 index 0000000..537f831 --- /dev/null +++ b/src/benchmarking/evaluation.py @@ -0,0 +1,208 @@ +"""LLM-as-Judge evaluation for benchmarking. + +Uses GPT-4o-mini to evaluate answer quality on 4 metrics: +- Context Relevance (1-10): How relevant are the retrieved chunks? +- Answer Similarity (1-10): How similar is the answer to expected? +- Faithfulness (1-10): Is the answer grounded in context? +- Hallucination (bool): Did the LLM invent information? +""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +from openai import OpenAI + +from src.core.config import settings +from src.core.dependencies import get_openai_client +from src.core.exceptions import BenchmarkError + +logger = logging.getLogger(__name__) + +# ── Evaluation System Prompt ────────────────────────────────────── + +_EVALUATION_SYSTEM_PROMPT = """You are an expert evaluator for a RAG (Retrieval-Augmented Generation) system. +You will evaluate the quality of an answer based on the retrieved context and the expected answer. + +Score each metric on a scale of 1-10: +- Context Relevance: How relevant are the retrieved chunks to answering the question? +- Answer Similarity: How similar is the generated answer to the expected answer? +- Faithfulness: Is the generated answer grounded in the retrieved context (no hallucination)? + +Also determine if there is hallucination (true/false): +- Hallucination = true if the answer contains information not found in the context +- Hallucination = false if the answer is fully grounded in the context + +IMPORTANT: Return ONLY valid JSON, no markdown, no explanation.""" + +# ── Evaluation User Prompt Template ─────────────────────────────── + +_EVALUATION_USER_PROMPT = """Evaluate this RAG system output: + +**Question:** {question} + +**Retrieved Context:** +{context} + +**Expected Answer:** {expected_answer} + +**Generated Answer:** {generated_answer} + +Return JSON with these exact keys: +{{ + "context_relevance": <1-10>, + "answer_similarity": <1-10>, + "faithfulness": <1-10>, + "hallucination": , + "reasoning": "" +}}""" + + +# ── Evaluation Functions ────────────────────────────────────────── + +def _build_context_for_evaluation(retrieved_chunks: list[dict]) -> str: + """Build a readable context string from retrieved chunks.""" + parts = [] + for i, chunk in enumerate(retrieved_chunks, 1): + score = chunk.get("score", 0) + text = chunk.get("text", "") + parts.append(f"[Chunk {i} (score: {score:.3f})]\n{text}") + return "\n\n".join(parts) + + +def evaluate_single( + *, + question: str, + retrieved_chunks: list[dict], + expected_answer: str, + generated_answer: str, +) -> dict[str, Any]: + """Evaluate a single question-answer pair using LLM-as-Judge. + + Args: + question: The original question + retrieved_chunks: Chunks retrieved from vector search + expected_answer: The golden/expected answer + generated_answer: The answer generated by the system + + Returns: + dict with context_relevance, answer_similarity, faithfulness, + hallucination, and reasoning + """ + client = get_openai_client() + + context = _build_context_for_evaluation(retrieved_chunks) + + user_prompt = _EVALUATION_USER_PROMPT.format( + question=question, + context=context, + expected_answer=expected_answer, + generated_answer=generated_answer, + ) + + try: + response = client.chat.completions.create( + model=settings.llm_model, + temperature=0.0, + max_tokens=500, + response_format={"type": "json_object"}, + messages=[ + {"role": "system", "content": _EVALUATION_SYSTEM_PROMPT}, + {"role": "user", "content": user_prompt}, + ], + ) + + content = response.choices[0].message.content or "{}" + + # Parse JSON response + try: + scores = json.loads(content) + except json.JSONDecodeError: + # Try to extract JSON from markdown code block + if "```json" in content: + json_str = content.split("```json")[1].split("```")[0].strip() + scores = json.loads(json_str) + elif "```" in content: + json_str = content.split("```")[1].split("```")[0].strip() + scores = json.loads(json_str) + else: + raise BenchmarkError(f"Failed to parse evaluation response: {content}") + + # Ensure all required fields exist with defaults + result = { + "context_relevance": scores.get("context_relevance", 5), + "answer_similarity": scores.get("answer_similarity", 5), + "faithfulness": scores.get("faithfulness", 5), + "hallucination": scores.get("hallucination", False), + "reasoning": scores.get("reasoning", ""), + } + + # Validate ranges + for metric in ["context_relevance", "answer_similarity", "faithfulness"]: + val = result[metric] + if isinstance(val, (int, float)): + result[metric] = max(1, min(10, int(val))) + else: + result[metric] = 5 + + # Validate hallucination is bool + hall = result["hallucination"] + if isinstance(hall, str): + result["hallucination"] = hall.lower() == "true" + else: + result["hallucination"] = bool(hall) + + logger.info( + "Evaluation: context_relevance=%d, answer_similarity=%d, faithfulness=%d, hallucination=%s", + result["context_relevance"], + result["answer_similarity"], + result["faithfulness"], + result["hallucination"], + ) + + return result + + except Exception as exc: + logger.error("Evaluation failed: %s", exc) + # Return default scores on failure + return { + "context_relevance": 5, + "answer_similarity": 5, + "faithfulness": 5, + "hallucination": False, + "reasoning": f"Evaluation failed: {exc}", + } + + +def evaluate_batch( + evaluation_items: list[dict], +) -> list[dict[str, Any]]: + """Evaluate a batch of question-answer pairs. + + Args: + evaluation_items: List of dicts, each containing: + - question: str + - retrieved_chunks: list[dict] + - expected_answer: str + - generated_answer: str + + Returns: + List of evaluation results + """ + results = [] + + for i, item in enumerate(evaluation_items, 1): + logger.info("Evaluating item %d/%d", i, len(evaluation_items)) + + scores = evaluate_single( + question=item["question"], + retrieved_chunks=item["retrieved_chunks"], + expected_answer=item["expected_answer"], + generated_answer=item["generated_answer"], + ) + + results.append(scores) + + return results diff --git a/src/benchmarking/report.py b/src/benchmarking/report.py new file mode 100644 index 0000000..53b5ba4 --- /dev/null +++ b/src/benchmarking/report.py @@ -0,0 +1,815 @@ +"""Enhanced HTML report generator with two views: +1. Managerial: Decision-focused, high-level insights +2. Technical: Full observability with detailed data + +Design: Dark mode, amber/teal accents, Inter + JetBrains Mono. +""" + +from __future__ import annotations + +import json +from typing import Any + + +# ── Design Tokens ───────────────────────────────────────────────── + +COLORS = { + "bg": "#0F1419", + "surface": "#1A2332", + "surface-hover": "#243044", + "border": "#2D3A4A", + "text": "#E8EDF2", + "text-muted": "#7A8BA0", + "accent": "#F59E0B", + "accent-dim": "#D97706", + "teal": "#14B8A6", + "rose": "#F43F5E", + "violet": "#8B5CF6", + "emerald": "#10B981", +} + +CHART_COLORS = ["#F59E0B", "#14B8A6", "#8B5CF6", "#F43F5E", "#3B82F6"] + +FONT_DISPLAY = "'Inter', -apple-system, sans-serif" +FONT_DATA = "'JetBrains Mono', 'Fira Code', monospace" + + +# ── Main Entry Points ───────────────────────────────────────────── + +def generate_report(experiment: dict, view: str = "managerial") -> str: + """Generate report for specified view. + + Args: + experiment: Experiment data from SQLite + view: "managerial" or "technical" + """ + if view == "technical": + return generate_technical_report(experiment) + return generate_managerial_report(experiment) + + +def generate_managerial_report(experiment: dict) -> str: + """Managerial view: Decision-focused, high-level insights.""" + config = experiment.get("benchmark_config", {}) + aggregate = experiment.get("aggregate_metrics", {}) + per_question = experiment.get("per_question", []) + strategies = experiment.get("strategies_used", []) + rankings = _calculate_rankings(aggregate, strategies) + + # Calculate cost from token usage in per-question data + total_prompt_tokens = 0 + total_completion_tokens = 0 + for qr in per_question: + for strategy in strategies: + strat = qr.get("strategies", {}).get(strategy, {}) + usage = strat.get("token_usage", {}) + total_prompt_tokens += usage.get("prompt_tokens", 0) + total_completion_tokens += usage.get("completion_tokens", 0) + + estimated_cost = (total_prompt_tokens * 0.15 + total_completion_tokens * 0.60) / 1_000_000 + + html = _base_html(experiment, "Managerial View", f""" + +
+
Benchmark Results
+

Strategy Comparison

+
+ {config.get('num_questions', 0)} questions · + {len(strategies)} strategies · + ${estimated_cost:.4f} cost +
+
+ + +
+
+
Best Strategy
+
{rankings[0][0] if rankings else 'N/A'}
+
+
+
Overall Score
+
{rankings[0][1]:.2f}
+
+
+
Questions
+
{config.get('num_questions', 0)}
+
+
+
Cost
+
${estimated_cost:.4f}
+
+
+ + +
+
🏆
+
+

Recommendation: {rankings[0][0] if rankings else 'N/A'}

+

{_get_recommendation_text(rankings, aggregate)}

+
+
{rankings[0][1]:.2f}
+
+ + +
+

How Strategies Compare

+
+
+
+ {_build_strategy_cards(rankings, aggregate)} +
+ + +
+

Visual Comparison

+
+
+
+
+

Performance Radar

+
+ +
+
+
+

Score Comparison

+
+ +
+
+
+ + +
+

Decision Guide

+
+
+
+ {_generate_decision_insights(rankings, aggregate)} +
+ + +
+

Details

+
+
+
+ + 🔍 + View Full Technical Report + → + +
+ """) + + return html + + +def generate_technical_report(experiment: dict) -> str: + """Technical view: Full observability with detailed data.""" + config = experiment.get("benchmark_config", {}) + aggregate = experiment.get("aggregate_metrics", {}) + per_question = experiment.get("per_question", []) + strategies = experiment.get("strategies_used", []) + rankings = _calculate_rankings(aggregate, strategies) + + # Calculate metrics from available data + total_queries = len(per_question) * len(strategies) + total_questions = config.get("num_questions", len(per_question)) + top_k = config.get("top_k", 5) + + # Calculate token usage from per-question data + total_prompt_tokens = 0 + total_completion_tokens = 0 + for qr in per_question: + for strategy in strategies: + strat = qr.get("strategies", {}).get(strategy, {}) + usage = strat.get("token_usage", {}) + total_prompt_tokens += usage.get("prompt_tokens", 0) + total_completion_tokens += usage.get("completion_tokens", 0) + + total_tokens = total_prompt_tokens + total_completion_tokens + + # Estimate cost based on token usage (GPT-4o-mini pricing) + # Input: $0.15/1M tokens, Output: $0.60/1M tokens + estimated_cost = (total_prompt_tokens * 0.15 + total_completion_tokens * 0.60) / 1_000_000 + + # Estimate latency (rough: ~0.5s per query + ~1s per evaluation) + estimated_latency = total_queries * 1.5 + + html = _base_html(experiment, "Technical View", f""" + +
+
Technical Report
+

Full Observability

+
+ Experiment: {experiment.get('id', 'N/A')[:16]}... · + {total_questions} questions · + {len(strategies)} strategies +
+
+ + +
+
+
Total Queries
+
{total_queries}
+
+
+
Total Tokens
+
{total_tokens:,}
+
+
+
Cost
+
${estimated_cost:.4f}
+
+
+
Tokens per Query
+
{total_tokens // max(total_queries, 1):,}
+
+
+ + +
+

Aggregate Metrics

+
+
+
+ + + + + + + + + + + + + + + {_build_technical_aggregate_rows(rankings, aggregate)} + +
StrategyContext RelevanceAnswer SimilarityFaithfulnessHallucinationQuestionsFailedOverall
+
+ + +
+

Visual Analysis

+
+
+
+
+

Multi-Metric Radar

+
+ +
+
+
+

Score Distribution

+
+ +
+
+
+ + +
+

Per-Question Results

+
+
+
+ + + + + + + + {_build_strategy_headers(strategies)} + + + + {_build_per_question_rows(per_question, strategies)} + +
IDQuestionCategoryDifficulty
+
+ + +
+

Detailed Scores

+
+
+
+ + + + + + + + + + + + + + {_build_detailed_rows(per_question, strategies)} + +
Q-IDStrategyContextSimilarityFaithfulnessHallucinationAnswer Preview
+
+ + +
+

Token Usage

+
+
+
+ + + + + + + + + + + + {_build_token_rows(per_question, strategies)} + +
Q-IDStrategyPrompt TokensCompletion TokensTotal Tokens
+
+ + +
+

Navigation

+
+
+
+ + 📊 + Switch to Managerial View + → + +
+ """) + + return html + + +# ── Base HTML Template ──────────────────────────────────────────── + +def _base_html(experiment: dict, title: str, content: str) -> str: + """Base HTML wrapper with styles and scripts.""" + strategies = experiment.get("strategies_used", []) + aggregate = experiment.get("aggregate_metrics", {}) + + return f""" + + + + + {title} + + + + + + +
+ {content} + +
+ Generated by RAG Chunking Benchmarker · {experiment.get('created_at', 'N/A')} +
+
+ + + +""" + + +# ── Helper Functions ────────────────────────────────────────────── + +def _calculate_rankings(aggregate: dict, strategies: list) -> list: + """Calculate overall scores and rank strategies.""" + rankings = [] + for strategy in strategies: + m = aggregate.get(strategy, {}) + overall = ( + m.get("avg_context_relevance", 0) * 0.3 + + m.get("avg_answer_similarity", 0) * 0.4 + + m.get("avg_faithfulness", 0) * 0.3 + ) * (1 - m.get("hallucination_rate", 0)) + rankings.append((strategy, round(overall, 2))) + rankings.sort(key=lambda x: x[1], reverse=True) + return rankings + + +def _get_recommendation_text(rankings: list, aggregate: dict) -> str: + """Generate recommendation text for the winner.""" + if not rankings: + return "No data available." + + best = rankings[0][0] + m = aggregate.get(best, {}) + + parts = [] + if m.get("avg_context_relevance", 0) >= 8: + parts.append("strong context retrieval") + if m.get("avg_faithfulness", 0) >= 8: + parts.append("high faithfulness") + if m.get("hallucination_rate", 0) < 0.1: + parts.append("low hallucination") + + if parts: + return f"{best} excels in {', '.join(parts)}, making it the most reliable choice for your use case." + return f"{best} achieved the highest overall score across all evaluation metrics." + + +def _build_strategy_cards(rankings: list, aggregate: dict) -> str: + """Build strategy comparison cards.""" + cards = "" + for rank, (strategy, overall) in enumerate(rankings, 1): + m = aggregate.get(strategy, {}) + rank_class = f"rank-{rank}" if rank <= 3 else "" + + cards += f""" +
+
{rank}
+
{strategy}
+ {_metric_bar("Context", m.get("avg_context_relevance", 0), "teal")} + {_metric_bar("Similarity", m.get("avg_answer_similarity", 0), "accent")} + {_metric_bar("Faithfulness", m.get("avg_faithfulness", 0), "violet")} + {_metric_bar("No Hallucination", (1 - m.get("hallucination_rate", 0)) * 10, "teal" if m.get("hallucination_rate", 0) <= 0.1 else "rose")} +
""" + + return cards + + +def _metric_bar(label: str, value: float, color_class: str) -> str: + """Build a single metric bar.""" + width = min(value * 10, 100) + return f""" +
+
{label}
+
+
{value:.1f}
+
""" + + +def _generate_decision_insights(rankings: list, aggregate: dict) -> str: + """Generate decision-focused insights.""" + insights = [] + + if rankings: + best = rankings[0][0] + worst = rankings[-1][0] + gap = (1 - rankings[-1][1] / rankings[0][1]) * 100 if rankings[0][1] > 0 else 0 + + insights.append(f""" +
+
Recommended Strategy
+
{best} is the best choice with {rankings[0][1]:.2f} overall score.
+
""") + + if gap > 20: + insights.append(f""" +
+
Avoid
+
{worst} underperforms by {gap:.0f}%. Use only if specific constraints require it.
+
""") + + # Cost vs quality + best_m = aggregate.get(best, {}) + if best_m.get("avg_context_relevance", 0) >= 8 and best_m.get("hallucination_rate", 0) < 0.1: + insights.append(f""" +
+
Quality Assessment
+
{best} delivers high-quality answers with reliable context retrieval.
+
""") + else: + insights.append(f""" +
+
Quality Note
+
Consider tuning parameters or adding more context for better results.
+
""") + + return "\n".join(insights) + + +def _build_technical_aggregate_rows(rankings: list, aggregate: dict) -> str: + """Build technical aggregate table rows.""" + rows = "" + for rank, (strategy, overall) in enumerate(rankings, 1): + m = aggregate.get(strategy, {}) + rows += f""" + + {strategy} + {_pill(m.get('avg_context_relevance', 0))} + {_pill(m.get('avg_answer_similarity', 0))} + {_pill(m.get('avg_faithfulness', 0))} + {_hall_pill(m.get('hallucination_rate', 0))} + {m.get('total_questions', 0)} + {m.get('failed_questions', 0)} + {overall:.2f} + """ + return rows + + +def _build_strategy_headers(strategies: list) -> str: + """Build table headers.""" + return "".join(f'{s[:12]}' for s in strategies) + + +def _build_per_question_rows(per_question: list, strategies: list) -> str: + """Build per-question rows.""" + rows = "" + for qr in per_question: + q_id = qr.get("question_id", "") + q_text = qr.get("question", "")[:40] + category = qr.get("category", "") + difficulty = qr.get("difficulty", "") + + cells = "" + for strategy in strategies: + strat = qr.get("strategies", {}).get(strategy, {}) + scores = strat.get("scores", {}) + sim = scores.get("answer_similarity", 0) + cells += f"{_pill(sim)}" + + rows += f""" + + {q_id} + {q_text}... + {category} + {difficulty} + {cells} + """ + + return rows + + +def _build_detailed_rows(per_question: list, strategies: list) -> str: + """Build detailed score rows.""" + rows = "" + for qr in per_question: + q_id = qr.get("question_id", "") + for strategy in strategies: + strat = qr.get("strategies", {}).get(strategy, {}) + scores = strat.get("scores", {}) + answer = strat.get("answer", "")[:60] + + rows += f""" + + {q_id} + {strategy} + {_pill(scores.get('context_relevance', 0))} + {_pill(scores.get('answer_similarity', 0))} + {_pill(scores.get('faithfulness', 0))} + {'✓' if not scores.get('hallucination', False) else '✗'} + {answer}... + """ + + return rows + + +def _build_token_rows(per_question: list, strategies: list) -> str: + """Build token usage rows.""" + rows = "" + for qr in per_question: + q_id = qr.get("question_id", "") + for strategy in strategies: + strat = qr.get("strategies", {}).get(strategy, {}) + usage = strat.get("token_usage", {}) + + rows += f""" + + {q_id} + {strategy} + {usage.get('prompt_tokens', 0)} + {usage.get('completion_tokens', 0)} + {usage.get('total_tokens', 0)} + """ + + return rows + + +def _pill(score: float) -> str: + """Create a score pill.""" + if score >= 8: + css = "high" + elif score >= 6: + css = "mid" + else: + css = "low" + return f'{score:.1f}' + + +def _hall_pill(rate: float) -> str: + """Create hallucination rate pill.""" + if rate <= 0.1: + css = "high" + elif rate <= 0.2: + css = "mid" + else: + css = "low" + return f'{rate*100:.0f}%' + + +# ── Legacy compatibility ────────────────────────────────────────── + +def generate_enhanced_report(experiment: dict) -> str: + """Backward-compatible entry point (defaults to managerial).""" + return generate_managerial_report(experiment)