From 62eae630bdbbeb74eb8656452358f7053ba7f01a Mon Sep 17 00:00:00 2001 From: TheCreeperrr Date: Thu, 8 Oct 2026 16:07:41 +0530 Subject: [PATCH] feat(metrics): add TokenEfficiencyMetric --- src/harness/metrics/__init__.py | 2 + src/harness/metrics/token_efficiency.py | 53 ++++++++++ tests/unit/metrics/test_token_efficiency.py | 103 ++++++++++++++++++++ 3 files changed, 158 insertions(+) create mode 100644 src/harness/metrics/token_efficiency.py create mode 100644 tests/unit/metrics/test_token_efficiency.py diff --git a/src/harness/metrics/__init__.py b/src/harness/metrics/__init__.py index 50d9512..39e283c 100644 --- a/src/harness/metrics/__init__.py +++ b/src/harness/metrics/__init__.py @@ -6,6 +6,7 @@ from harness.metrics.latency import LatencyMetric from harness.metrics.reliability import ReliabilityMetric from harness.metrics.success_rate import SuccessRateMetric +from harness.metrics.token_efficiency import TokenEfficiencyMetric __all__ = [ "Metric", @@ -16,4 +17,5 @@ "LatencyMetric", "ReliabilityMetric", "SuccessRateMetric", + "TokenEfficiencyMetric", ] diff --git a/src/harness/metrics/token_efficiency.py b/src/harness/metrics/token_efficiency.py new file mode 100644 index 0000000..2c8fed6 --- /dev/null +++ b/src/harness/metrics/token_efficiency.py @@ -0,0 +1,53 @@ +"""Token consumption efficiency metric for agent evaluations.""" + +from __future__ import annotations + +from collections.abc import Sequence + +from harness.core.experiment import MetricResult +from harness.core.result import TestCaseResult +from harness.metrics.base import MetricDirection, MetricRegistry + + +@MetricRegistry.register("token_efficiency") +class TokenEfficiencyMetric: + """Calculates token consumption per passed test case and per test run.""" + + name: str = "token_efficiency" + direction: MetricDirection = MetricDirection.LOWER_IS_BETTER + + def calculate(self, results: Sequence[TestCaseResult]) -> MetricResult: + total_tokens = 0 + prompt_tokens = 0 + completion_tokens = 0 + + for result in results: + token_usage = result.output.token_usage + if token_usage is None: + continue + total_tokens += token_usage.total_tokens + prompt_tokens += token_usage.prompt_tokens + completion_tokens += token_usage.completion_tokens + + passed_tests = sum(1 for result in results if result.passed) + tokens_per_pass = total_tokens / passed_tests if passed_tests else 0.0 + tokens_per_run = total_tokens / len(results) if results else 0.0 + formatted_value = ( + f"{tokens_per_pass:.1f} tokens/pass" if tokens_per_pass else "0 tokens/pass" + ) + + return MetricResult( + name=self.name, + value=tokens_per_pass, + formatted_value=formatted_value, + direction=self.direction.value, + description="Average token consumption per passed test case", + details={ + "total_tokens": total_tokens, + "prompt_tokens": prompt_tokens, + "completion_tokens": completion_tokens, + "passed_tests": passed_tests, + "tokens_per_pass": tokens_per_pass, + "tokens_per_run": tokens_per_run, + }, + ) diff --git a/tests/unit/metrics/test_token_efficiency.py b/tests/unit/metrics/test_token_efficiency.py new file mode 100644 index 0000000..34c5f4d --- /dev/null +++ b/tests/unit/metrics/test_token_efficiency.py @@ -0,0 +1,103 @@ +"""Unit tests for the token efficiency metric.""" + +from __future__ import annotations + +from harness.core.agent import AgentOutput, TokenUsage +from harness.core.result import TestCaseResult +from harness.core.testcase import TestCase +from harness.metrics.base import MetricDirection, MetricRegistry +from harness.metrics.token_efficiency import TokenEfficiencyMetric + + +def _result( + *, + passed: bool, + prompt_tokens: int = 0, + completion_tokens: int = 0, +) -> TestCaseResult: + usage = TokenUsage( + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + total_tokens=prompt_tokens + completion_tokens, + ) + return TestCaseResult( + test_case=TestCase(id="test", name="test", input="input"), + output=AgentOutput(output="output", token_usage=usage), + passed=passed, + ) + + +def test_token_efficiency_metric_calculates_token_usage() -> None: + results = [ + _result(passed=True, prompt_tokens=100, completion_tokens=50), + _result(passed=True, prompt_tokens=200, completion_tokens=100), + _result(passed=False, prompt_tokens=300, completion_tokens=150), + ] + + metric = TokenEfficiencyMetric() + result = metric.calculate(results) + + assert metric.direction == MetricDirection.LOWER_IS_BETTER + assert result.value == 450.0 + assert result.formatted_value == "450.0 tokens/pass" + assert result.details == { + "total_tokens": 900, + "prompt_tokens": 600, + "completion_tokens": 300, + "passed_tests": 2, + "tokens_per_pass": 450.0, + "tokens_per_run": 300.0, + } + + +def test_token_efficiency_metric_handles_failed_tests() -> None: + results = [ + _result(passed=False, prompt_tokens=100, completion_tokens=50), + _result(passed=False, prompt_tokens=200, completion_tokens=100), + ] + + result = TokenEfficiencyMetric().calculate(results) + + assert result.value == 0.0 + assert result.formatted_value == "0 tokens/pass" + assert result.details["passed_tests"] == 0 + assert result.details["tokens_per_pass"] == 0.0 + assert result.details["tokens_per_run"] == 225.0 + + +def test_token_efficiency_metric_handles_empty_results() -> None: + result = TokenEfficiencyMetric().calculate([]) + + assert result.value == 0.0 + assert result.formatted_value == "0 tokens/pass" + assert result.details == { + "total_tokens": 0, + "prompt_tokens": 0, + "completion_tokens": 0, + "passed_tests": 0, + "tokens_per_pass": 0.0, + "tokens_per_run": 0.0, + } + + +def test_token_efficiency_metric_handles_missing_token_usage() -> None: + missing_usage_result = TestCaseResult( + test_case=TestCase(id="test", name="test", input="input"), + output=AgentOutput(output="output"), + passed=True, + ) + + result = TokenEfficiencyMetric().calculate([missing_usage_result]) + + assert result.value == 0.0 + assert result.formatted_value == "0 tokens/pass" + assert result.details["total_tokens"] == 0 + assert result.details["passed_tests"] == 1 + assert result.details["tokens_per_pass"] == 0.0 + assert result.details["tokens_per_run"] == 0.0 + + +def test_token_efficiency_metric_is_registered() -> None: + metric = MetricRegistry.get("token_efficiency") + + assert isinstance(metric, TokenEfficiencyMetric)