Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
90 changes: 90 additions & 0 deletions agentx/testing.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
"""pytest-friendly assertions over evaluation runs.

The DeepEval-style dev loop (``assert_test`` inside a pytest suite) on top of AgentX's existing
run + CI-gate primitives: run the evaluation however you like, then make the test fail with a
readable verdict when quality drops. No plugin registration needed - it's a plain function that
raises ``AssertionError``, so it works in any test runner and any CI.

Usage::

from agentx import AgentX
from agentx.testing import assert_evaluation

def test_support_agent_quality():
client = AgentX.from_env()
report = (
client.evaluations
.run(dataset_id=DATASET_ID, scorer_id=SCORER_ID, subject=SUBJECT)
.execute(my_agent)
.finalize()
)
assert_evaluation(report, min_rating=7.0, no_regression=True)

The check rides the engine's CI gate, so every pytest verdict is also recorded in the
dashboard's gate history (CI Gates tab) with ``caller="pytest"`` - a red test and the
dashboard's gate row are the same event, not two systems drifting apart.
"""

from typing import Any, List, Optional


class EvaluationAssertionError(AssertionError):
"""Raised when an evaluation run fails its quality checks.

Subclasses ``AssertionError`` so pytest renders it as a plain test failure; carries the
``gate`` result for programmatic inspection in test hooks.
"""

def __init__(self, message: str, gate: Any = None):
super().__init__(message)
self.gate = gate


def _format_failures(gate: Any) -> str:
lines: List[str] = []
checks = getattr(gate, "checks", None) or []
for check in checks:
get = check.get if isinstance(check, dict) else lambda k, d=None: getattr(check, k, d)
status = "PASS" if get("passed") else "FAIL"
lines.append(f" [{status}] {get('name', 'check')}: {get('detail', '')}")
average = getattr(gate, "average_rating", None)
if average is not None:
lines.append(f" average rating: {average}")
return "\n".join(lines) if lines else f" gate: {gate!r}"


def assert_evaluation(
report: Any,
*,
min_rating: Optional[float] = None,
no_regression: bool = False,
tolerance: Optional[float] = None,
caller: str = "pytest",
) -> Any:
"""Assert a finalized evaluation run meets its quality floor.

``report`` is the finalized :class:`~agentx.evaluations.runner.EvaluationRunContext`
returned by ``.execute(...).finalize()`` (or any object exposing the same ``.gate()``).

- ``min_rating`` - fail when the run's average judge rating is below this floor (0-10).
- ``no_regression`` - fail when the average dropped more than ``tolerance`` (default 0.5;
judge scores are noisy) below the dataset's previous completed run.

At least one check is required. Returns the ``GateResult`` on success; raises
:class:`EvaluationAssertionError` with a per-check verdict on failure.
"""
if min_rating is None and not no_regression:
raise ValueError("assert_evaluation needs at least one check: min_rating and/or no_regression=True")
gate = report.gate(
fail_under=min_rating,
no_regression=no_regression,
tolerance=tolerance,
caller=caller,
)
if getattr(gate, "passed", False):
return gate
run_id = getattr(report, "run_id", None) or getattr(getattr(report, "_run", None), "run_id", "?")
raise EvaluationAssertionError(
f"Evaluation run {run_id} failed its quality gate:\n{_format_failures(gate)}",
gate=gate,
)
2 changes: 1 addition & 1 deletion agentx/version.py
Original file line number Diff line number Diff line change
@@ -1 +1 @@
VERSION = "0.6.39"
VERSION = "0.1.0"
49 changes: 49 additions & 0 deletions tests/test_testing.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,49 @@
"""agentx.testing.assert_evaluation - the pytest-native quality gate wrapper."""

import pytest

from agentx.evaluations.runner import GateResult
from agentx.testing import EvaluationAssertionError, assert_evaluation


class FakeReport:
def __init__(self, gate_payload):
self.run_id = "run-123"
self.gate_kwargs = None
self._payload = gate_payload

def gate(self, **kwargs):
self.gate_kwargs = kwargs
return GateResult(self._payload)


def test_passing_gate_returns_result_and_forwards_checks():
report = FakeReport({"passed": True, "averageRating": 8.2, "checks": [{"name": "floor", "passed": True}]})
gate = assert_evaluation(report, min_rating=7.0, no_regression=True, tolerance=0.3)
assert gate.passed is True
assert report.gate_kwargs == {"fail_under": 7.0, "no_regression": True, "tolerance": 0.3, "caller": "pytest"}


def test_failing_gate_raises_assertion_error_with_verdict():
report = FakeReport({
"passed": False,
"averageRating": 5.1,
"checks": [
{"name": "floor", "passed": False, "detail": "average 5.1 below fail_under 7"},
{"name": "regression", "passed": True, "detail": "no baseline"},
],
})
with pytest.raises(EvaluationAssertionError) as excinfo:
assert_evaluation(report, min_rating=7.0)
message = str(excinfo.value)
assert "run-123" in message
assert "[FAIL] floor" in message
assert "average 5.1 below fail_under 7" in message
# AssertionError subclass, so pytest treats it as a normal test failure.
assert isinstance(excinfo.value, AssertionError)
assert excinfo.value.gate.average_rating == 5.1


def test_requires_at_least_one_check():
with pytest.raises(ValueError):
assert_evaluation(FakeReport({"passed": True}))
Loading