feat: add token_efficiency, tool_efficiency, time_efficiency evaluators ⁠

.gitignore

@@ -1 +1 @@
-.venv/
+.venv/evaluators/__pycache__

evaluators/time_efficiency/evaluator.yaml

@@ -0,0 +1,6 @@
+name: time_efficiency
+description: Scores how quickly the agent resolved relative to a time budget
+language: python
+entrypoint: time_efficiency.py
+tags: [performance, time, latency, efficiency, budget]
+author: henrikrexed

evaluators/time_efficiency/time_efficiency.py

@@ -0,0 +1,68 @@
+"""Community evaluator: time_efficiency
+
+Scores how quickly the agent resolved relative to a time budget.
+Uses performance_metrics.duration_s from trace data when available.
+
+Config options:
+  max_duration_s (float): Time budget in seconds (default: 120)
+"""
+
+from agentevals_evaluator_sdk import EvalInput, EvalResult, evaluator
+
+
+def _extract_duration(inv) -> float | None:
+    """Extract duration_s from an invocation's performance_metrics."""
+    perf = getattr(inv, "performance_metrics", None)
+    if perf is None and hasattr(inv, "__getitem__"):
+        try:
+            perf = inv["performance_metrics"]
+        except (KeyError, TypeError):
+            perf = None
+
+    if isinstance(perf, dict):
+        duration = perf.get("duration_s") or perf.get("duration")
+        if duration is not None:
+            return float(duration)
+
+    return None
+
+
+@evaluator
+def time_efficiency(input: EvalInput) -> EvalResult:
+    max_duration = input.config.get("max_duration_s", 120.0)
+
+    scores: list[float] = []
+    details_items: list[str] = []
+    has_data = False
+
+    for inv in input.invocations:
+        duration = _extract_duration(inv)
+
+        if duration is None:
+            # No timing data — assign neutral score
+            scores.append(0.5)
+            details_items.append(f"{inv.invocation_id}: no duration data available")
+            continue
+
+        has_data = True
+        score = max(0.0, min(1.0, 1.0 - (duration / max_duration)))
+        scores.append(score)
+        details_items.append(
+            f"{inv.invocation_id}: {duration:.1f}s / {max_duration:.1f}s budget (score: {score:.2f})"
+        )
+
+    overall = sum(scores) / len(scores) if scores else 0.0
+
+    return EvalResult(
+        score=overall,
+        per_invocation_scores=scores,
+        details={
+            "time_details": details_items,
+            "has_trace_data": has_data,
+            "max_duration_s": max_duration,
+        },
+    )
+
+
+if __name__ == "__main__":
+    time_efficiency.run()

evaluators/token_efficiency/evaluator.yaml

@@ -0,0 +1,6 @@
+name: token_efficiency
+description: Scores how efficiently the agent used tokens relative to a budget
+language: python
+entrypoint: token_efficiency.py
+tags: [performance, tokens, efficiency, budget]
+author: henrikrexed

evaluators/token_efficiency/token_efficiency.py

@@ -0,0 +1,104 @@
+"""Community evaluator: token_efficiency
+
+Scores how efficiently the agent used tokens relative to a budget.
+Uses performance_metrics from trace data when available, falls back to
+counting tool calls as a rough proxy.
+
+Config options:
+  max_tokens (int): Token budget (default: 200000)
+  weight_input (float): Weight for input tokens in scoring (default: 0.7)
+  weight_output (float): Weight for output tokens in scoring (default: 0.3)
+"""
+
+from agentevals_evaluator_sdk import EvalInput, EvalResult, evaluator
+
+
+def _extract_tokens(inv) -> dict | None:
+    """Extract token counts from an invocation's performance_metrics or metadata."""
+    # Check performance_metrics (primary source from OTel trace data)
+    perf = getattr(inv, "performance_metrics", None)
+    if perf is None and hasattr(inv, "__getitem__"):
+        try:
+            perf = inv["performance_metrics"]
+        except (KeyError, TypeError):
+            perf = None
+
+    if isinstance(perf, dict):
+        input_t = perf.get("input_tokens") or perf.get("prompt_tokens")
+        output_t = perf.get("output_tokens") or perf.get("completion_tokens")
+        if input_t is not None or output_t is not None:
+            return {
+                "input_tokens": int(input_t or 0),
+                "output_tokens": int(output_t or 0),
+            }
+
+    # Check performance_budget on invocation (eval_set integration)
+    budget = getattr(inv, "performance_budget", None)
+    if budget is None and hasattr(inv, "__getitem__"):
+        try:
+            budget = inv["performance_budget"]
+        except (KeyError, TypeError):
+            pass
+
+    # No token data available
+    return None
+
+
+@evaluator
+def token_efficiency(input: EvalInput) -> EvalResult:
+    max_tokens = input.config.get("max_tokens", 200000)
+    weight_input = input.config.get("weight_input", 0.7)
+    weight_output = input.config.get("weight_output", 0.3)
+
+    scores: list[float] = []
+    details_items: list[str] = []
+    has_data = False
+
+    for inv in input.invocations:
+        tokens = _extract_tokens(inv)
+
+        if tokens is None:
+            # No token data — score based on tool call count as rough proxy
+            # More tool calls ≈ more tokens used
+            tool_count = len(inv.intermediate_steps.tool_calls) if inv.intermediate_steps else 0
+            if tool_count == 0:
+                scores.append(0.5)  # No data, neutral score
+                details_items.append(f"{inv.invocation_id}: no token data available")
+            else:
+                # Rough heuristic: assume ~5000 tokens per tool call
+                estimated = tool_count * 5000
+                score = max(0.0, min(1.0, 1.0 - (estimated / max_tokens)))
+                scores.append(score)
+                details_items.append(
+                    f"{inv.invocation_id}: estimated ~{estimated} tokens from {tool_count} tool calls"
+                )
+            continue
+
+        has_data = True
+        input_t = tokens["input_tokens"]
+        output_t = tokens["output_tokens"]
+        weighted_total = (input_t * weight_input) + (output_t * weight_output)
+        weighted_budget = max_tokens * 1.0  # Budget applies to weighted total
+
+        score = max(0.0, min(1.0, 1.0 - (weighted_total / weighted_budget)))
+        scores.append(score)
+        details_items.append(
+            f"{inv.invocation_id}: {input_t} input + {output_t} output = "
+            f"{input_t + output_t} total (weighted: {weighted_total:.0f}/{weighted_budget:.0f})"
+        )
+
+    overall = sum(scores) / len(scores) if scores else 0.0
+
+    return EvalResult(
+        score=overall,
+        per_invocation_scores=scores,
+        details={
+            "token_details": details_items,
+            "has_trace_data": has_data,
+            "max_tokens": max_tokens,
+        },
+    )
+
+
+if __name__ == "__main__":
+    token_efficiency.run()

evaluators/tool_efficiency/evaluator.yaml

@@ -0,0 +1,6 @@
+name: tool_efficiency
+description: Scores whether the agent used tools effectively — penalizes waste, duplicates, and errors
+language: python
+entrypoint: tool_efficiency.py
+tags: [performance, tools, efficiency, budget]
+author: henrikrexed

evaluators/tool_efficiency/tool_efficiency.py

@@ -0,0 +1,114 @@
+"""Community evaluator: tool_efficiency
+
+Scores whether the agent used tools effectively. Penalizes duplicate calls
+(same tool + same args), error calls, and budget overruns.
+
+Config options:
+  max_tool_calls (int): Tool call budget (default: 15)
+  penalize_duplicates (bool): Penalize repeated identical calls (default: true)
+  penalize_errors (bool): Penalize failed tool calls (default: true)
+"""
+
+import json
+from agentevals_evaluator_sdk import EvalInput, EvalResult, evaluator
+
+
+def _call_signature(call) -> str:
+    """Create a hashable signature for a tool call (name + sorted args)."""
+    name = call.get("name", "") if isinstance(call, dict) else getattr(call, "name", "")
+    args = call.get("args", {}) if isinstance(call, dict) else getattr(call, "args", {})
+    try:
+        args_str = json.dumps(args, sort_keys=True, default=str)
+    except (TypeError, ValueError):
+        args_str = str(args)
+    return f"{name}::{args_str}"
+
+
+def _is_error_response(response) -> bool:
+    """Check if a tool response indicates an error."""
+    output = response.get("output", "") if isinstance(response, dict) else getattr(response, "output", "")
+    output_str = str(output).lower()
+    # Check common error indicators
+    if any(marker in output_str for marker in ["error", "failed", "exception", "traceback"]):
+        return True
+    status = response.get("status", "") if isinstance(response, dict) else getattr(response, "status", "")
+    if str(status).lower() in ("error", "failed"):
+        return True
+    return False
+
+
+@evaluator
+def tool_efficiency(input: EvalInput) -> EvalResult:
+    max_tool_calls = input.config.get("max_tool_calls", 15)
+    penalize_duplicates = input.config.get("penalize_duplicates", True)
+    penalize_errors = input.config.get("penalize_errors", True)
+
+    scores: list[float] = []
+    details_items: list[str] = []
+
+    for inv in input.invocations:
+        tool_calls = inv.intermediate_steps.tool_calls if inv.intermediate_steps else []
+        tool_responses = (
+            inv.intermediate_steps.tool_responses if inv.intermediate_steps else []
+        )
+        total = len(tool_calls)
+
+        if total == 0:
+            scores.append(1.0)  # No tools needed = perfectly efficient
+            details_items.append(f"{inv.invocation_id}: no tool calls (score: 1.0)")
+            continue
+
+        # Count duplicates
+        seen_signatures: dict[str, int] = {}
+        duplicate_count = 0
+        for call in tool_calls:
+            sig = _call_signature(call)
+            seen_signatures[sig] = seen_signatures.get(sig, 0) + 1
+
+        if penalize_duplicates:
+            duplicate_count = sum(count - 1 for count in seen_signatures.values() if count > 1)
+
+        # Count errors
+        error_count = 0
+        if penalize_errors and tool_responses:
+            for resp in tool_responses:
+                if _is_error_response(resp):
+                    error_count += 1
+
+        # Calculate useful calls
+        wasted = duplicate_count + error_count
+        useful = max(0, total - wasted)
+
+        # Efficiency ratio: useful / total
+        efficiency_ratio = useful / total if total > 0 else 1.0
+
+        # Budget penalty: how much over budget
+        budget_overrun = max(0, total - max_tool_calls) / max_tool_calls
+        budget_factor = max(0.0, 1.0 - budget_overrun)
+
+        score = max(0.0, min(1.0, efficiency_ratio * budget_factor))
+        scores.append(score)
+
+        parts = [f"total={total}", f"useful={useful}"]
+        if duplicate_count > 0:
+            parts.append(f"duplicates={duplicate_count}")
+        if error_count > 0:
+            parts.append(f"errors={error_count}")
+        if total > max_tool_calls:
+            parts.append(f"over_budget={total - max_tool_calls}")
+        details_items.append(f"{inv.invocation_id}: {', '.join(parts)} (score: {score:.2f})")
+
+    overall = sum(scores) / len(scores) if scores else 0.0
+
+    return EvalResult(
+        score=overall,
+        per_invocation_scores=scores,
+        details={
+            "tool_details": details_items,
+            "max_tool_calls": max_tool_calls,
+        },
+    )
+
+
+if __name__ == "__main__":
+    tool_efficiency.run()