Spaces:
Paused
Paused
File size: 5,951 Bytes
0d3f7cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 | """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))
|