Spaces:
Paused
Paused
File size: 6,301 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 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 | """Agent evaluation framework."""
from __future__ import annotations
import logging
import time
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from collections.abc import Callable
logger = logging.getLogger(__name__)
@dataclass
class EvalResult:
"""Evaluation result."""
metric_name: str
value: float
passed: bool
threshold: float
details: str = ""
@dataclass
class EvalReport:
"""Complete evaluation report."""
task_id: str
results: list[EvalResult] = field(default_factory=list)
overall_score: float = 0.0
passed: bool = False
duration_seconds: float = 0.0
def calculate_overall(self) -> None:
"""Calculate overall score and pass/fail."""
if self.results:
self.overall_score = sum(r.value for r in self.results) / len(self.results)
self.passed = all(r.passed for r in self.results)
class AgentEvaluator:
"""Evaluate agent performance."""
def __init__(self) -> None:
self._metrics: dict[str, Callable] = {}
self._register_default_metrics()
def _register_default_metrics(self) -> None:
"""Register default evaluation metrics."""
self._metrics["task_completion"] = self._evaluate_task_completion
self._metrics["success_rate"] = self._evaluate_success_rate
self._metrics["latency"] = self._evaluate_latency
self._metrics["cost"] = self._evaluate_cost
self._metrics["tool_efficiency"] = self._evaluate_tool_efficiency
self._metrics["hallucination_rate"] = self._evaluate_hallucination_rate
async def evaluate(
self,
task_id: str,
agent_output: dict[str, Any],
expected_output: dict[str, Any] | None = None,
metrics: list[str] | None = None,
) -> EvalReport:
"""Run evaluation on agent output."""
start_time = time.monotonic()
report = EvalReport(task_id=task_id)
metrics_to_run = metrics or list(self._metrics.keys())
for metric_name in metrics_to_run:
if metric_name in self._metrics:
result = await self._metrics[metric_name](agent_output, expected_output)
report.results.append(result)
report.duration_seconds = time.monotonic() - start_time
report.calculate_overall()
logger.info(
f"Evaluation complete: {task_id} - Score: {report.overall_score:.2f}, "
f"Passed: {report.passed}"
)
return report
async def _evaluate_task_completion(
self, output: dict[str, Any], expected: dict[str, Any] | None
) -> EvalResult:
"""Evaluate task completion."""
completed = output.get("status") == "completed" or output.get("result") is not None
return EvalResult(
metric_name="task_completion",
value=1.0 if completed else 0.0,
passed=completed,
threshold=1.0,
details="Task was completed successfully" if completed else "Task was not completed",
)
async def _evaluate_success_rate(
self, output: dict[str, Any], expected: dict[str, Any] | None
) -> EvalResult:
"""Evaluate success rate."""
success = output.get("success", True)
return EvalResult(
metric_name="success_rate",
value=1.0 if success else 0.0,
passed=success,
threshold=1.0,
details="Agent execution succeeded" if success else "Agent execution failed",
)
async def _evaluate_latency(
self, output: dict[str, Any], expected: dict[str, Any] | None
) -> EvalResult:
"""Evaluate execution latency."""
latency_ms = output.get("latency_ms", 0)
threshold_ms = 30000
passed = latency_ms <= threshold_ms
score = max(0.0, 1.0 - (latency_ms / (threshold_ms * 2)))
return EvalResult(
metric_name="latency",
value=score,
passed=passed,
threshold=threshold_ms,
details=f"Latency: {latency_ms}ms (threshold: {threshold_ms}ms)",
)
async def _evaluate_cost(
self, output: dict[str, Any], expected: dict[str, Any] | None
) -> EvalResult:
"""Evaluate execution cost."""
tokens_used = output.get("tokens_used", 0)
max_tokens = 100000
score = max(0.0, 1.0 - (tokens_used / max_tokens))
passed = tokens_used <= max_tokens
return EvalResult(
metric_name="cost",
value=score,
passed=passed,
threshold=max_tokens,
details=f"Tokens used: {tokens_used}",
)
async def _evaluate_tool_efficiency(
self, output: dict[str, Any], expected: dict[str, Any] | None
) -> EvalResult:
"""Evaluate tool usage efficiency."""
tool_calls = output.get("tool_calls", [])
successful_calls = sum(1 for tc in tool_calls if tc.get("success", True))
total_calls = len(tool_calls) if tool_calls else 1
efficiency = successful_calls / total_calls
passed = efficiency >= 0.7
return EvalResult(
metric_name="tool_efficiency",
value=efficiency,
passed=passed,
threshold=0.7,
details=f"Tool efficiency: {efficiency:.2%} ({successful_calls}/{total_calls})",
)
async def _evaluate_hallucination_rate(
self, output: dict[str, Any], expected: dict[str, Any] | None
) -> EvalResult:
"""Evaluate hallucination rate."""
output.get("result", "")
has_citations = output.get("has_citations", True)
confidence = output.get("confidence", 0.8)
hallucination_score = 1.0 if has_citations else max(0.5, confidence)
passed = hallucination_score >= 0.7
return EvalResult(
metric_name="hallucination_rate",
value=hallucination_score,
passed=passed,
threshold=0.7,
details=f"Hallucination score: {hallucination_score:.2f}",
)
|