""" evaluation/metrics.py ---------------------- Metrics aggregation for AutoDevAgent benchmark results. Takes the list of BenchmarkResult objects from the runner and computes summary statistics for display in the README, LinkedIn posts, and W&B dashboards. Metrics computed: - Overall success rate (fraction of tasks that reached SUCCESS) - Average debug iterations per task - Average token usage per task - Average execution time per task - Per-language breakdown (Python vs SQL) - Per-category breakdown Design: - Pure functions — no side effects, easy to test. - Returns typed dataclasses for clean serialization to W&B. - Formatted output functions produce README-ready Markdown tables. Usage: from evaluation.metrics import compute_metrics, MetricsSummary summary = compute_metrics(results) print(summary.success_rate) print(summary.to_markdown_table()) print(summary.to_dict()) # for W&B logging """ import logging from dataclasses import dataclass, field from typing import Any logger = logging.getLogger(__name__) # ------------------------------------------------------------------ # # Summary dataclass # # ------------------------------------------------------------------ # @dataclass class MetricsSummary: """ Aggregated metrics across all benchmark tasks. Attributes: total_tasks: Total number of tasks attempted. successful_tasks: Number of tasks that reached SUCCESS. success_rate: Fraction of tasks that succeeded (0.0 – 1.0). avg_iterations: Mean debug iterations across all tasks. avg_tokens: Mean total Groq tokens per task. avg_exec_time: Mean wall-clock seconds per task. max_iterations: Maximum iterations any single task required. max_tokens: Maximum tokens any single task consumed. per_language: Dict of language → LanguageMetrics. per_category: Dict of category → success rate float. task_results: List of per-task result dicts for W&B table. """ total_tasks: int = 0 successful_tasks: int = 0 success_rate: float = 0.0 avg_iterations: float = 0.0 avg_tokens: float = 0.0 avg_exec_time: float = 0.0 max_iterations: int = 0 max_tokens: int = 0 per_language: dict[str, Any] = field(default_factory=dict) per_category: dict[str, float] = field(default_factory=dict) task_results: list[dict[str, Any]] = field(default_factory=list) def to_dict(self) -> dict[str, Any]: """ Serialize to a flat dict for W&B logging. Returns: Dict with all scalar metrics suitable for wandb.log(). """ d = { "total_tasks": self.total_tasks, "successful_tasks": self.successful_tasks, "success_rate": self.success_rate, "avg_iterations": self.avg_iterations, "avg_tokens": self.avg_tokens, "avg_exec_time": self.avg_exec_time, "max_iterations": self.max_iterations, "max_tokens": self.max_tokens, } # Flatten per-language metrics for lang, lang_metrics in self.per_language.items(): for k, v in lang_metrics.items(): d[f"{lang}_{k}"] = v return d def to_markdown_table(self) -> str: """ Format per-task results as a Markdown table for the README. Returns: Multi-line Markdown string with one row per benchmark task. """ if not self.task_results: return "*No benchmark results available.*" header = ( "| Task | Language | Status | Iterations | Tokens | Time |\n" "|------|----------|--------|------------|--------|------|" ) rows = [] for r in self.task_results: status_emoji = "✅" if r.get("success") else "❌" rows.append( f"| {r.get('task_name', '?')} " f"| {r.get('language', '?')} " f"| {status_emoji} " f"| {r.get('iterations', 0)} " f"| {r.get('total_tokens', 0):,} " f"| {r.get('exec_time', 0):.1f}s |" ) summary_row = ( f"\n**Summary:** {self.successful_tasks}/{self.total_tasks} tasks passed " f"· avg {self.avg_iterations:.1f} iterations " f"· avg {self.avg_tokens:,.0f} tokens " f"· avg {self.avg_exec_time:.1f}s" ) return "\n".join([header, *rows]) + summary_row def to_summary_string(self) -> str: """ Return a one-line summary suitable for LinkedIn posts. Example: "5/5 tasks passed · avg 1.4 debug iterations · avg 980 tokens/task" Returns: Formatted summary string. """ return ( f"{self.successful_tasks}/{self.total_tasks} tasks passed " f"· avg {self.avg_iterations:.1f} debug iteration(s) " f"· avg {self.avg_tokens:,.0f} tokens/task" ) # ------------------------------------------------------------------ # # Core compute function # # ------------------------------------------------------------------ # def compute_metrics(results: list) -> MetricsSummary: """ Compute aggregated metrics from a list of BenchmarkResult objects. Args: results: List of BenchmarkResult objects from BenchmarkRunner. Returns: Populated MetricsSummary with all computed metrics. """ if not results: logger.warning("compute_metrics: received empty results list") return MetricsSummary() total = len(results) successful = sum(1 for r in results if r.success) # ── Scalar aggregates ─────────────────────────────────────────── # avg_iterations = _safe_mean([r.iterations for r in results]) avg_tokens = _safe_mean([r.total_tokens for r in results]) avg_exec_time = _safe_mean([r.exec_time for r in results]) max_iterations = max((r.iterations for r in results), default=0) max_tokens = max((r.total_tokens for r in results), default=0) # ── Per-language breakdown ────────────────────────────────────── # per_language: dict[str, dict[str, Any]] = {} for lang in {"python", "sql"}: lang_results = [r for r in results if r.language == lang] if not lang_results: continue per_language[lang] = { "total": len(lang_results), "successful": sum(1 for r in lang_results if r.success), "success_rate": round( sum(1 for r in lang_results if r.success) / len(lang_results), 3 ), "avg_iterations": _safe_mean([r.iterations for r in lang_results]), "avg_tokens": _safe_mean([r.total_tokens for r in lang_results]), } # ── Per-category breakdown ────────────────────────────────────── # categories: dict[str, list] = {} for r in results: categories.setdefault(r.category, []).append(r.success) per_category = { cat: round(sum(successes) / len(successes), 3) for cat, successes in categories.items() } # ── Task-level results for W&B table ─────────────────────────── # task_results = [r.to_dict() for r in results] summary = MetricsSummary( total_tasks = total, successful_tasks = successful, success_rate = round(successful / total, 3), avg_iterations = avg_iterations, avg_tokens = avg_tokens, avg_exec_time = avg_exec_time, max_iterations = max_iterations, max_tokens = max_tokens, per_language = per_language, per_category = per_category, task_results = task_results, ) logger.info( "compute_metrics: %d/%d passed · avg %.1f iterations · avg %.0f tokens", successful, total, avg_iterations, avg_tokens, ) return summary # ------------------------------------------------------------------ # # Helpers # # ------------------------------------------------------------------ # def _safe_mean(values: list[float | int]) -> float: """ Compute the mean of a list, returning 0.0 for an empty list. Args: values: List of numeric values. Returns: Rounded mean, or 0.0 if the list is empty. """ if not values: return 0.0 return round(sum(values) / len(values), 2) def format_results_for_display(results: list) -> str: """ Format benchmark results as a plain-text table for Gradio display. Args: results: List of BenchmarkResult objects. Returns: Multi-line string with aligned columns. """ if not results: return "No benchmark results to display." lines = [ f"{'Task':<28} {'Lang':<8} {'Status':<8} {'Iters':<7} {'Tokens':<8} {'Time':<6}", "-" * 70, ] for r in results: status = "PASS" if r.success else "FAIL" lines.append( f"{r.task_name:<28} " f"{r.language:<8} " f"{status:<8} " f"{r.iterations:<7} " f"{r.total_tokens:<8,} " f"{r.exec_time:.1f}s" ) lines.append("-" * 70) summary = compute_metrics(results) lines.append( f"{'SUMMARY':<28} " f"{'all':<8} " f"{summary.successful_tasks}/{summary.total_tasks:<6} " f"{summary.avg_iterations:<7.1f} " f"{summary.avg_tokens:<8,.0f} " f"{summary.avg_exec_time:.1f}s" ) return "\n".join(lines)