Spaces:
Paused
Paused
| """Evaluation suite for GEPA self-evolution pipeline. | |
| Measures accuracy, efficiency, coherence, and safety of evolved skills. | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| from pathlib import Path | |
| from typing import Any | |
| logger = logging.getLogger(__name__) | |
| class EvalSuite: | |
| """Evaluation metrics for evolved skills. | |
| Metrics: | |
| - accuracy: Correctness of structured outputs (0-1) | |
| - efficiency: Token consumption / latency efficiency | |
| - coherence: Logical flow and instruction following (0-1) | |
| - safety: Refusal rate and constraint adherence (0-1) | |
| """ | |
| def __init__(self) -> None: | |
| self._results: dict[str, list[dict[str, Any]]] = {} | |
| async def evaluate( | |
| self, | |
| skill_name: str, | |
| examples: list[dict[str, Any]], | |
| evolved_content: str, | |
| ) -> dict[str, Any]: | |
| """Evaluate an evolved skill against test examples.""" | |
| accuracy_scores: list[float] = [] | |
| efficiency_scores: list[float] = [] | |
| coherence_scores: list[float] = [] | |
| safety_scores: list[float] = [] | |
| for example in examples: | |
| accuracy = self._measure_accuracy(example, evolved_content) | |
| efficiency = self._measure_efficiency(example) | |
| coherence = self._measure_coherence(evolved_content, example) | |
| safety = self._measure_safety(evolved_content, example) | |
| accuracy_scores.append(accuracy) | |
| efficiency_scores.append(efficiency) | |
| coherence_scores.append(coherence) | |
| safety_scores.append(safety) | |
| metrics = { | |
| "accuracy": sum(accuracy_scores) / len(accuracy_scores) if accuracy_scores else 0.0, | |
| "efficiency": sum(efficiency_scores) / len(efficiency_scores) if efficiency_scores else 0.0, | |
| "coherence": sum(coherence_scores) / len(coherence_scores) if coherence_scores else 0.0, | |
| "safety": sum(safety_scores) / len(safety_scores) if safety_scores else 0.0, | |
| } | |
| overall = ( | |
| metrics["accuracy"] * 0.35 | |
| + metrics["efficiency"] * 0.20 | |
| + metrics["coherence"] * 0.20 | |
| + metrics["safety"] * 0.25 | |
| ) | |
| result = { | |
| "skill_name": skill_name, | |
| "overall_score": overall, | |
| "metrics": metrics, | |
| "num_examples": len(examples), | |
| "num_example_metrics": { | |
| "accuracy": len(accuracy_scores), | |
| "efficiency": len(efficiency_scores), | |
| "coherence": len(coherence_scores), | |
| "safety": len(safety_scores), | |
| }, | |
| } | |
| self._results.setdefault(skill_name, []).append(result) | |
| return result | |
| def _measure_accuracy(self, example: dict[str, Any], _evolved_content: str) -> float: | |
| """Measure accuracy by comparing expected output structure.""" | |
| expected = example.get("expected_output", {}) | |
| existing_metrics = example.get("metrics", {}) | |
| if existing_metrics.get("accuracy") is not None: | |
| return float(existing_metrics["accuracy"]) | |
| if not expected: | |
| return 0.5 | |
| # Score based on how many expected keys are present in evolved content | |
| if isinstance(expected, dict): | |
| content_lower = _evolved_content.lower() | |
| matches = sum(1 for key in expected if str(key).lower() in content_lower) | |
| return min(1.0, 0.5 + (matches / max(len(expected), 1)) * 0.5) | |
| return 0.6 | |
| def _measure_efficiency(self, example: dict[str, Any]) -> float: | |
| """Measure efficiency from trace data.""" | |
| traces = example.get("traces", []) | |
| existing_metrics = example.get("metrics", {}) | |
| if existing_metrics.get("efficiency") is not None: | |
| raw = float(existing_metrics["efficiency"]) | |
| return max(0.0, min(1.0, 1.0 - (raw / 10000.0))) | |
| if traces: | |
| return max(0.0, min(1.0, 1.0 - (len(traces) / 20.0))) | |
| return 0.5 | |
| def _measure_coherence(self, _evolved_content: str, example: dict[str, Any]) -> float: | |
| """Measure coherence from trace step logic.""" | |
| traces = example.get("traces", []) | |
| existing_metrics = example.get("metrics", {}) | |
| if existing_metrics.get("coherence") is not None: | |
| return float(existing_metrics["coherence"]) | |
| if len(traces) < 2: | |
| return 0.5 | |
| has_flow = all( | |
| traces[i].get("output") and traces[i + 1].get("input") | |
| for i in range(len(traces) - 1) | |
| ) | |
| return 0.9 if has_flow else 0.5 | |
| def _measure_safety(self, _evolved_content: str, example: dict[str, Any]) -> float: | |
| """Measure safety from constraint gates.""" | |
| gates = example.get("constraint_gates", {}) | |
| existing_metrics = example.get("metrics", {}) | |
| if existing_metrics.get("safety") is not None: | |
| return float(existing_metrics["safety"]) | |
| if gates: | |
| security_pass = gates.get("security_pass", True) | |
| return 1.0 if security_pass else 0.0 | |
| return 1.0 | |
| def load_dataset(self, path: str | Path) -> list[dict[str, Any]]: | |
| """Load an evaluation dataset from a JSON file.""" | |
| path = Path(path) | |
| if not path.exists(): | |
| logger.warning(f"Dataset not found: {path}") | |
| return [] | |
| with open(path, encoding="utf-8") as f: | |
| data = json.load(f) | |
| return data.get("examples", []) | |
| def get_history(self, skill_name: str) -> list[dict[str, Any]]: | |
| """Get evaluation history for a skill.""" | |
| return self._results.get(skill_name, []) | |
| def get_best(self, skill_name: str) -> dict[str, Any] | None: | |
| """Get the best evaluation result for a skill.""" | |
| history = self._results.get(skill_name, []) | |
| if not history: | |
| return None | |
| return max(history, key=lambda r: r.get("overall_score", 0)) | |