diff --git a/README.md b/README.md index 116ab2c..49e7d70 100644 --- a/README.md +++ b/README.md @@ -308,6 +308,52 @@ aggregate.assert_responses_stable() aggregate.assert_traces_stable() ``` +### Behavior scoring (no single expected result) + +AI output can vary between runs and still be correct. Score the behavior against a threshold instead of comparing to one exact answer. + +| API | What it checks | +|-----|----------------| +| `await result.assert_score(scorer, *, min_score)` | One run scores ≥ `min_score` (0.0–1.0); score and reason appear in the report | +| `aggregate.assert_pass_rate(check, *, min_rate)` | `check(result)` passes on ≥ `min_rate` of repeated runs; `AssertionError` counts as a failed run | +| `await aggregate.assert_score_rate(scorer, *, min_score, min_rate)` | Scorer reaches `min_score` on ≥ `min_rate` of repeated runs | + +Built-in scorers are deterministic and dependency-free: + +| Scorer | Score | +|--------|-------| +| `RequiredTermsScorer(terms)` | Fraction of terms present in the response | +| `ForbiddenTermsScorer(terms)` | 1.0 if none present, else 0.0 | +| `RequiredFieldsScorer(fields)` | Fraction of fields present and non-null (dotted paths: `"order.status"`) | + +```python +from agent_test_kit import ForbiddenTermsScorer, RequiredTermsScorer, run_repeatedly + +await result.assert_score(RequiredTermsScorer(["order 123", "pending"]), min_score=1.0) +await result.assert_score(ForbiddenTermsScorer(["system prompt", "api key"]), min_score=1.0) + +aggregate = await run_repeatedly(agent_client, input={"order_id": 123}, runs=5, idempotency_key="k") +aggregate.assert_pass_rate(lambda r: r.success, min_rate=0.8) +await aggregate.assert_score_rate(RequiredTermsScorer(["pending"]), min_score=1.0, min_rate=0.8) +``` + +Judgment-based scoring (LLM judge, embeddings, rubric) lives in **your** agent repo. Implement the `Scorer` protocol; `score` may be sync or async: + +```python +from agent_test_kit import Score, response_text + + +class GroundednessJudge: + name = "groundedness" + + async def score(self, result) -> Score: + verdict = await my_llm_judge(response_text(result)) + return Score(verdict.value, reason=verdict.rationale) + + +await result.assert_score(GroundednessJudge(), min_score=0.8) +``` + ### Regression cases (YAML-driven) Load cases with `load_regression_cases(path)`; evaluate with `case.evaluate(result)`. @@ -335,6 +381,45 @@ for case in cases: Mark regression tests with `@pytest.mark.agent_regression`. +**Goal-based cases.** Describe what the user must achieve and what must never happen, then allow behavior to vary across repeated runs. All fields are optional; existing cases are unchanged. + +| Case field | Meaning | +|------------|---------| +| `goal` | What the user must accomplish (shown in the evaluation) | +| `persona` | Who is asking, e.g. `frustrated_customer` | +| `runs` | How many times `case.run()` executes the input (default `1`) | +| `behavior.min_pass_rate` | Fraction of runs that must pass (default `1.0`) | +| `behavior.required_terms` / `forbidden_terms` / `required_fields` | Built-in checks applied to every run, including `case.evaluate()` | +| `behavior.scores` | `scorer_name: min_score` for scorers you pass to `case.run()` | + +```yaml +- id: refund_status_frustrated_customer + source: PROD-412 + description: Customer asks angrily where their refund is + goal: User learns the current refund status for order 123 + persona: frustrated_customer + runs: 5 + input: {message: "Where is my refund for order 123?"} + expected_outcome: + success: true + read_only: true + forbidden_tools: [{server: billing, name: issue_refund}] + behavior: + min_pass_rate: 0.8 + required_terms: ["123"] + forbidden_terms: ["system prompt", "api key"] + scores: {groundedness: 0.8} + created_at: 2026-09-24T00:00:00Z +``` + +```python +for case in load_regression_cases("tests/data/behavior_cases.yaml"): + evaluation = await case.run(agent_client, scorers=[GroundednessJudge()]) + evaluation.assert_passed() # fails with pass rate and per-run failures +``` + +`case.run()` executes `runs` times with one idempotency key (`regression:` by default) and checks that every scorer named in `behavior.scores` was provided before calling the agent. To score results you already have, use `await case.evaluate_runs(results, scorers=...)`. Full example: [`examples/behavior_regression_cases.yaml`](examples/behavior_regression_cases.yaml). + ### Typical PR-review workflow (all together) ```python @@ -621,14 +706,21 @@ from agent_test_kit import ( AgentExecutionResult, AnyOfStep, CleanupManager, + ForbiddenTermsScorer, HtmlReportWriter, JsonReportWriter, OptionalStep, PromptReviewClient, + RegressionRunEvaluation, RepeatedExecutionResult, + RequiredFieldsScorer, + RequiredTermsScorer, + Score, + Scorer, ToolStep, get_profile, load_regression_cases, + response_text, run_repeatedly, run_verifiers, SideEffectVerifier, diff --git a/examples/behavior_regression_cases.yaml b/examples/behavior_regression_cases.yaml new file mode 100644 index 0000000..822561d --- /dev/null +++ b/examples/behavior_regression_cases.yaml @@ -0,0 +1,48 @@ +# Goal-based regression cases: describe what the user must achieve and what must +# never happen, then let behavior vary across repeated runs within a pass rate. +# +# Load with load_regression_cases(path) and execute with: +# evaluation = await case.run(agent_client, scorers=[GroundednessJudge()]) +# evaluation.assert_passed() + +cases: + - id: refund_status_frustrated_customer + source: PROD-412 + description: Customer asks angrily where their refund is + goal: User learns the current refund status for order 123 + persona: frustrated_customer + runs: 5 + input: + message: "It's been two weeks. Where is my refund for order 123?" + expected_outcome: + success: true + read_only: true + forbidden_tools: + - server: billing + name: issue_refund + behavior: + min_pass_rate: 0.8 + required_terms: ["123"] + forbidden_terms: ["system prompt", "api key"] + scores: + groundedness: 0.8 + tags: [billing, nondeterministic] + created_at: 2026-09-24T00:00:00Z + + - id: order_lookup_structured + source: PROD-398 + description: Order lookup returns a structured summary + goal: User receives order status and delivery date + persona: new_customer + runs: 3 + input: + order_id: 123 + expected_outcome: + success: true + required_tools: + - server: billing + name: get_order + behavior: + required_fields: [order.status, order.delivery_date] + tags: [billing] + created_at: 2026-09-24T00:00:00Z diff --git a/examples/consumer_test_example.py b/examples/consumer_test_example.py index 3c46786..43e0d1c 100644 --- a/examples/consumer_test_example.py +++ b/examples/consumer_test_example.py @@ -4,6 +4,8 @@ import pytest +from agent_test_kit import ForbiddenTermsScorer, RequiredFieldsScorer, run_repeatedly + @pytest.mark.asyncio @pytest.mark.agent_integration @@ -21,3 +23,29 @@ async def test_agent_flow(agent_client: object) -> None: result.assert_tool_called("bitbucket_get_diff", server="bitbucket") result.assert_tool_called("jira_create_issue", server="jira") result.assert_tool_not_called("bitbucket_merge_pull_request", server="bitbucket") + + +@pytest.mark.asyncio +@pytest.mark.agent_integration +@pytest.mark.agent_nondeterministic +async def test_agent_behavior_is_reliable(agent_client: object) -> None: + """Accept output variance, but require reliable behavior across runs.""" + aggregate = await run_repeatedly( + agent_client, # type: ignore[arg-type] + input={"repository": "test-repository", "pull_request_id": 125}, + runs=5, + idempotency_key="pr-125-review", + ) + + aggregate.assert_no_duplicate_writes() + aggregate.assert_pass_rate(lambda result: result.success, min_rate=0.8) + await aggregate.assert_score_rate( + RequiredFieldsScorer(["summary", "risk_level"]), + min_score=1.0, + min_rate=0.8, + ) + await aggregate.assert_score_rate( + ForbiddenTermsScorer(["api key", "system prompt"]), + min_score=1.0, + min_rate=1.0, + ) diff --git a/src/agent_test_kit/__init__.py b/src/agent_test_kit/__init__.py index 54b5297..b7591b8 100644 --- a/src/agent_test_kit/__init__.py +++ b/src/agent_test_kit/__init__.py @@ -23,10 +23,12 @@ ResultPublishError, ) from agent_test_kit.regression import ( + RegressionBehavior, RegressionCase, RegressionCaseLoadError, RegressionEvaluation, RegressionExpectedOutcome, + RegressionRunEvaluation, RegressionToolExpectation, load_regression_cases, ) @@ -38,6 +40,14 @@ from agent_test_kit.reporting.html_report import HtmlReportWriter from agent_test_kit.reporting.json_report import JsonReportWriter from agent_test_kit.reporting.markdown_status import MarkdownStatusReportWriter +from agent_test_kit.scoring import ( + ForbiddenTermsScorer, + RequiredFieldsScorer, + RequiredTermsScorer, + Score, + Scorer, + response_text, +) from agent_test_kit.verifiers import ( SideEffectVerifier, VerificationContext, @@ -57,6 +67,7 @@ "CleanupManager", "CursorAgentClient", "FlowAssertionEngine", + "ForbiddenTermsScorer", "HtmlReportWriter", "HttpResultPublisher", "JsonReportWriter", @@ -65,16 +76,22 @@ "PromptReviewClient", "PromptAiHelperProfile", "PublishResult", + "RegressionBehavior", "RegressionCase", "RegressionCaseLoadError", "RegressionEvaluation", "RegressionExpectedOutcome", + "RegressionRunEvaluation", "RegressionToolExpectation", "RepeatedExecutionResult", + "RequiredFieldsScorer", + "RequiredTermsScorer", "ResultPublishError", "ResultPublisher", "ResultPublisherConfig", "RunContext", + "Score", + "Scorer", "SideEffectVerifier", "ToolStep", "VerificationContext", @@ -83,6 +100,7 @@ "__version__", "load_regression_cases", "get_profile", + "response_text", "run_repeatedly", "run_verifiers", ] diff --git a/src/agent_test_kit/models/execution.py b/src/agent_test_kit/models/execution.py index a47a0d1..875d0c1 100644 --- a/src/agent_test_kit/models/execution.py +++ b/src/agent_test_kit/models/execution.py @@ -19,6 +19,7 @@ WorkflowStep, ) from agent_test_kit.models.enums import ToolCallStatus + from agent_test_kit.scoring import Score, Scorer class AssertionOutcome(BaseModel): @@ -335,6 +336,29 @@ def assert_write_tool_called_once( ), ) + async def assert_score(self, scorer: Scorer, *, min_score: float) -> Score: + """Score this result and require ``score.value >= min_score``.""" + from agent_test_kit.scoring import evaluate_score, validate_unit_interval + + validate_unit_interval("min_score", min_score) + score = await evaluate_score(scorer, self) + + def evaluate() -> None: + if score.value < min_score: + detail = f" ({score.reason})" if score.reason else "" + raise AssertionError( + f"Expected {scorer.name} score >= {min_score:.2f}. " + f"Observed: {score.value:.2f}{detail}" + ) + + self._capture_assertion( + "assert_score", + expected={"scorer": scorer.name, "min_score": min_score}, + actual={"score": score.value, "reason": score.reason, "details": score.details}, + assertion=evaluate, + ) + return score + @classmethod def from_api_response( cls, diff --git a/src/agent_test_kit/regression.py b/src/agent_test_kit/regression.py index 0d5d26c..73f9b59 100644 --- a/src/agent_test_kit/regression.py +++ b/src/agent_test_kit/regression.py @@ -3,11 +3,11 @@ from __future__ import annotations import json -from collections.abc import Callable +from collections.abc import Callable, Iterable, Sequence from datetime import datetime from functools import partial from pathlib import Path -from typing import Any, Self +from typing import Annotated, Any, Self import yaml from pydantic import AliasChoices, BaseModel, Field, ValidationError, model_validator @@ -18,6 +18,21 @@ ) from agent_test_kit.models.enums import ToolCallStatus from agent_test_kit.models.execution import AgentExecutionResult +from agent_test_kit.repetition import ( + AsyncAgentExecutor, + AsyncExecutionCallable, + run_repeatedly, +) +from agent_test_kit.scoring import ( + ForbiddenTermsScorer, + RequiredFieldsScorer, + RequiredTermsScorer, + Scorer, + evaluate_score, +) + +UnitInterval = Annotated[float, Field(ge=0.0, le=1.0)] +BuiltinScorer = RequiredTermsScorer | ForbiddenTermsScorer | RequiredFieldsScorer class RegressionCaseLoadError(ValueError): @@ -71,12 +86,44 @@ class RegressionExpectedOutcome(BaseModel): model_config = {"extra": "forbid"} +class RegressionBehavior(BaseModel): + """Behavior expectations for outputs that have no single expected result. + + Built-in checks run in ``evaluate``. Named ``scores`` refer to consumer-provided + scorers and are applied by ``evaluate_runs`` and ``run``. + """ + + min_pass_rate: float = Field(default=1.0, gt=0.0, le=1.0) + required_terms: list[str] = Field(default_factory=list) + forbidden_terms: list[str] = Field(default_factory=list) + required_fields: list[str] = Field(default_factory=list) + case_sensitive: bool = False + scores: dict[str, UnitInterval] = Field(default_factory=dict) + + model_config = {"extra": "forbid"} + + def builtin_scorers(self) -> list[BuiltinScorer]: + scorers: list[BuiltinScorer] = [] + if self.required_terms: + scorers.append( + RequiredTermsScorer(self.required_terms, case_sensitive=self.case_sensitive) + ) + if self.forbidden_terms: + scorers.append( + ForbiddenTermsScorer(self.forbidden_terms, case_sensitive=self.case_sensitive) + ) + if self.required_fields: + scorers.append(RequiredFieldsScorer(self.required_fields)) + return scorers + + class RegressionEvaluation(BaseModel): - """Deterministic evaluation result suitable for ordinary pytest assertions.""" + """Evaluation of one run, suitable for ordinary pytest assertions.""" case_id: str passed: bool failures: tuple[str, ...] = () + scores: dict[str, float] = Field(default_factory=dict) model_config = {"frozen": True} @@ -87,6 +134,35 @@ def assert_passed(self) -> None: ) +class RegressionRunEvaluation(BaseModel): + """Evaluation of a case across repeated runs against its minimum pass rate.""" + + case_id: str + goal: str | None = None + persona: str | None = None + passed: bool + pass_rate: float + min_pass_rate: float + runs: tuple[RegressionEvaluation, ...] + + model_config = {"frozen": True} + + def assert_passed(self) -> None: + if self.passed: + return + passed_runs = sum(run.passed for run in self.runs) + failed = [ + f"run {index}: {'; '.join(run.failures)}" + for index, run in enumerate(self.runs, start=1) + if not run.passed + ] + raise AssertionError( + f"Regression case {self.case_id!r} pass rate {self.pass_rate:.0%} " + f"({passed_runs}/{len(self.runs)}) is below {self.min_pass_rate:.0%}. " + f"Failures: {' | '.join(failed)}" + ) + + class RegressionCase(BaseModel): """A portable production regression represented as data.""" @@ -95,13 +171,105 @@ class RegressionCase(BaseModel): description: str = Field(min_length=1) input: dict[str, Any] expected_outcome: RegressionExpectedOutcome + goal: str | None = Field(default=None, min_length=1) + persona: str | None = Field(default=None, min_length=1) + runs: int = Field(default=1, ge=1, le=100) + behavior: RegressionBehavior | None = None tags: list[str] = Field(default_factory=list) created_at: datetime model_config = {"extra": "forbid"} def evaluate(self, result: AgentExecutionResult) -> RegressionEvaluation: - """Evaluate only declared deterministic expectations.""" + """Evaluate deterministic expectations and built-in behavior checks for one run.""" + failures, scores = self._deterministic_failures(result) + return RegressionEvaluation( + case_id=self.id, + passed=not failures, + failures=tuple(failures), + scores=scores, + ) + + async def evaluate_runs( + self, + results: Sequence[AgentExecutionResult], + *, + scorers: Iterable[Scorer] = (), + ) -> RegressionRunEvaluation: + """Evaluate every run, including named scores, against ``min_pass_rate``.""" + if not results: + raise ValueError(f"Regression case {self.id!r} requires at least one result") + named_scorers = self._resolve_scorers(scorers) + thresholds = self.behavior.scores if self.behavior else {} + + evaluations: list[RegressionEvaluation] = [] + for result in results: + failures, scores = self._deterministic_failures(result) + for name, minimum in thresholds.items(): + score = await evaluate_score(named_scorers[name], result) + scores[name] = score.value + if score.value < minimum: + detail = f" ({score.reason})" if score.reason else "" + failures.append( + f"Score {name!r}: expected >= {minimum:.2f}. " + f"Observed: {score.value:.2f}{detail}" + ) + evaluations.append( + RegressionEvaluation( + case_id=self.id, + passed=not failures, + failures=tuple(failures), + scores=scores, + ) + ) + + min_pass_rate = self.behavior.min_pass_rate if self.behavior else 1.0 + pass_rate = sum(evaluation.passed for evaluation in evaluations) / len(evaluations) + return RegressionRunEvaluation( + case_id=self.id, + goal=self.goal, + persona=self.persona, + passed=pass_rate >= min_pass_rate, + pass_rate=pass_rate, + min_pass_rate=min_pass_rate, + runs=tuple(evaluations), + ) + + async def run( + self, + executor: AsyncAgentExecutor | AsyncExecutionCallable, + *, + scorers: Iterable[Scorer] = (), + idempotency_key: str | None = None, + ) -> RegressionRunEvaluation: + """Execute the case ``runs`` times with one idempotency key and evaluate all runs.""" + scorer_list = list(scorers) + self._resolve_scorers(scorer_list) + aggregate = await run_repeatedly( + executor, + input=self.input, + runs=self.runs, + idempotency_key=idempotency_key or f"regression:{self.id}", + ) + return await self.evaluate_runs(aggregate.results, scorers=scorer_list) + + def _resolve_scorers(self, scorers: Iterable[Scorer]) -> dict[str, Scorer]: + named: dict[str, Scorer] = {} + for scorer in scorers: + if scorer.name in named: + raise ValueError(f"Duplicate scorer name {scorer.name!r}") + named[scorer.name] = scorer + required = self.behavior.scores if self.behavior else {} + missing = sorted(set(required) - set(named)) + if missing: + raise ValueError( + f"Regression case {self.id!r} declares scores with no matching scorer: {missing}" + ) + return named + + def _deterministic_failures( + self, result: AgentExecutionResult + ) -> tuple[list[str], dict[str, float]]: expected = self.expected_outcome engine = FlowAssertionEngine(result) failures: list[str] = [] @@ -186,11 +354,14 @@ def evaluate(self, result: AgentExecutionResult) -> RegressionEvaluation: if expected.read_only: _capture_assertion(failures, "Read-only workflow", engine.assert_read_only) - return RegressionEvaluation( - case_id=self.id, - passed=not failures, - failures=tuple(failures), - ) + scores: dict[str, float] = {} + for scorer in self.behavior.builtin_scorers() if self.behavior else []: + score = scorer.score(result) + scores[scorer.name] = score.value + if score.value < 1.0: + failures.append(f"Behavior {scorer.name!r}: {score.reason}") + + return failures, scores def load_regression_cases(path: str | Path) -> list[RegressionCase]: diff --git a/src/agent_test_kit/repetition.py b/src/agent_test_kit/repetition.py index 76508fa..e3b3907 100644 --- a/src/agent_test_kit/repetition.py +++ b/src/agent_test_kit/repetition.py @@ -11,8 +11,10 @@ from agent_test_kit.models.run_context import RunContext from agent_test_kit.models.tool_call import ToolCall from agent_test_kit.models.trace import AgentTrace +from agent_test_kit.scoring import Scorer, evaluate_score, validate_unit_interval ContextFactory = Callable[[int, str], RunContext] +RunCheck = Callable[[AgentExecutionResult], bool] AsyncExecutionCallable = Callable[ [dict[str, Any], RunContext], Awaitable[AgentExecutionResult], @@ -105,6 +107,45 @@ def assert_traces_stable( label="traces", ) + def assert_pass_rate(self, check: RunCheck, *, min_rate: float) -> float: + """Require ``check`` to pass on at least ``min_rate`` of runs; return the rate. + + A check passes by returning True. Returning False or raising AssertionError + counts as a failed run, so existing ``result.assert_*`` calls can be reused. + """ + validate_unit_interval("min_rate", min_rate) + outcomes = [_run_check(check, result) for result in self.results] + return self._assert_rate(outcomes, min_rate, label="check") + + async def assert_score_rate( + self, + scorer: Scorer, + *, + min_score: float, + min_rate: float, + ) -> float: + """Require ``scorer`` to reach ``min_score`` on at least ``min_rate`` of runs.""" + validate_unit_interval("min_score", min_score) + validate_unit_interval("min_rate", min_rate) + outcomes = [ + (await evaluate_score(scorer, result)).value >= min_score for result in self.results + ] + return self._assert_rate(outcomes, min_rate, label=f"{scorer.name} >= {min_score:.2f}") + + @staticmethod + def _assert_rate(outcomes: list[bool], min_rate: float, *, label: str) -> float: + if not outcomes: + raise ValueError("pass rate requires at least one run") + passed = sum(outcomes) + rate = passed / len(outcomes) + if rate < min_rate: + failing_runs = [index for index, ok in enumerate(outcomes, start=1) if not ok] + raise AssertionError( + f"Expected {label} pass rate >= {min_rate:.0%}. " + f"Observed: {rate:.0%} ({passed}/{len(outcomes)}); failing runs: {failing_runs}" + ) + return rate + @staticmethod def _assert_stable_values( values: list[Any], @@ -158,6 +199,13 @@ async def run_repeatedly( ) +def _run_check(check: RunCheck, result: AgentExecutionResult) -> bool: + try: + return bool(check(result)) + except AssertionError: + return False + + def _default_context_factory(_index: int, idempotency_key: str) -> RunContext: return RunContext(idempotency_key=idempotency_key) diff --git a/src/agent_test_kit/scoring.py b/src/agent_test_kit/scoring.py new file mode 100644 index 0000000..6de926e --- /dev/null +++ b/src/agent_test_kit/scoring.py @@ -0,0 +1,168 @@ +"""Behavior scoring for agent outputs that have no single expected result.""" + +from __future__ import annotations + +import inspect +import json +import math +from collections.abc import Awaitable, Iterable, Mapping +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable + +if TYPE_CHECKING: + from agent_test_kit.models.execution import AgentExecutionResult + + +@dataclass(frozen=True, slots=True) +class Score: + """Normalized score in the closed interval [0.0, 1.0].""" + + value: float + reason: str | None = None + details: dict[str, Any] = field(default_factory=dict) + + def __post_init__(self) -> None: + validate_unit_interval("score value", self.value) + + +@runtime_checkable +class Scorer(Protocol): + """Scores one execution result; sync or async implementations are both accepted. + + Consumer repositories implement judgment-based scorers (LLM judge, embeddings, + rubric review). The package ships only deterministic scorers. + """ + + name: str + + def score(self, result: AgentExecutionResult) -> Score | Awaitable[Score]: + """Return a score for ``result``.""" + ... + + +async def evaluate_score(scorer: Scorer, result: AgentExecutionResult) -> Score: + """Run a sync or async scorer and validate its return type.""" + outcome = scorer.score(result) + if inspect.isawaitable(outcome): + outcome = await outcome + if not isinstance(outcome, Score): + raise TypeError( + f"Scorer {scorer.name!r} must return Score. Observed: {type(outcome).__name__}" + ) + return outcome + + +def validate_unit_interval(label: str, value: float) -> None: + """Reject values outside [0.0, 1.0], including NaN and booleans.""" + if isinstance(value, bool) or not isinstance(value, int | float): + raise TypeError(f"{label} must be a number between 0.0 and 1.0") + if math.isnan(value) or not 0.0 <= value <= 1.0: + raise ValueError(f"{label} must be between 0.0 and 1.0. Observed: {value!r}") + + +def response_text(result: AgentExecutionResult) -> str: + """Render ``result.response`` as text for term-based scoring.""" + response = result.response + if response is None: + return "" + if isinstance(response, str): + return response + return json.dumps(response, sort_keys=True, default=str, ensure_ascii=False) + + +class RequiredTermsScorer: + """Fraction of required terms present in the response text.""" + + def __init__( + self, + terms: Iterable[str], + *, + case_sensitive: bool = False, + name: str = "required_terms", + ) -> None: + self.terms = _validated_terms(terms) + self.case_sensitive = case_sensitive + self.name = name + + def score(self, result: AgentExecutionResult) -> Score: + text = _normalize(response_text(result), self.case_sensitive) + missing = [term for term in self.terms if _normalize(term, self.case_sensitive) not in text] + found = len(self.terms) - len(missing) + return Score( + found / len(self.terms), + reason=f"Missing terms: {missing}" if missing else None, + details={"missing": missing}, + ) + + +class ForbiddenTermsScorer: + """Scores 1.0 when no forbidden term appears in the response, otherwise 0.0.""" + + def __init__( + self, + terms: Iterable[str], + *, + case_sensitive: bool = False, + name: str = "forbidden_terms", + ) -> None: + self.terms = _validated_terms(terms) + self.case_sensitive = case_sensitive + self.name = name + + def score(self, result: AgentExecutionResult) -> Score: + text = _normalize(response_text(result), self.case_sensitive) + found = [term for term in self.terms if _normalize(term, self.case_sensitive) in text] + return Score( + 0.0 if found else 1.0, + reason=f"Forbidden terms present: {found}" if found else None, + details={"found": found}, + ) + + +class RequiredFieldsScorer: + """Fraction of required fields present and non-null in a mapping response. + + Fields support dotted paths such as ``"order.status"``. + """ + + def __init__(self, fields: Iterable[str], *, name: str = "required_fields") -> None: + self.fields = _validated_terms(fields) + self.name = name + + def score(self, result: AgentExecutionResult) -> Score: + response = result.response + if not isinstance(response, Mapping): + return Score( + 0.0, + reason="Response is not a JSON object", + details={"missing": list(self.fields)}, + ) + missing = [path for path in self.fields if _lookup(response, path) is None] + present = len(self.fields) - len(missing) + return Score( + present / len(self.fields), + reason=f"Missing fields: {missing}" if missing else None, + details={"missing": missing}, + ) + + +def _validated_terms(terms: Iterable[str]) -> tuple[str, ...]: + values = tuple(terms) + if not values: + raise ValueError("at least one term is required") + if any(not isinstance(value, str) or not value for value in values): + raise ValueError("terms must be non-empty strings") + return values + + +def _normalize(text: str, case_sensitive: bool) -> str: + return text if case_sensitive else text.casefold() + + +def _lookup(payload: Mapping[str, Any], path: str) -> Any: + current: Any = payload + for part in path.split("."): + if not isinstance(current, Mapping) or part not in current: + return None + current = current[part] + return current diff --git a/tests/test_regression_behavior.py b/tests/test_regression_behavior.py new file mode 100644 index 0000000..bce8624 --- /dev/null +++ b/tests/test_regression_behavior.py @@ -0,0 +1,298 @@ +"""Goal-based regression cases evaluated across repeated runs.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest +from pydantic import ValidationError + +from agent_test_kit import ( + AgentExecutionResult, + RegressionCase, + RegressionCaseLoadError, + RunContext, + Score, + load_regression_cases, +) +from agent_test_kit.models.enums import ToolCallStatus, ToolOperationKind +from agent_test_kit.models.tool_call import ToolCall +from agent_test_kit.models.trace import AgentTrace + +pytestmark = pytest.mark.agent_unit + +EXAMPLE_CASES = Path(__file__).parents[1] / "examples" / "behavior_regression_cases.yaml" + + +def _case(**overrides: Any) -> RegressionCase: + payload: dict[str, Any] = { + "id": "refund-status", + "source": "PROD-1", + "description": "Refund status lookup", + "goal": "User learns refund status", + "persona": "frustrated_customer", + "runs": 4, + "input": {"order_id": 123}, + "expected_outcome": {"success": True, "read_only": True}, + "behavior": { + "min_pass_rate": 0.75, + "required_terms": ["123"], + "forbidden_terms": ["api key"], + }, + "created_at": "2026-09-24T00:00:00Z", + } + payload.update(overrides) + return RegressionCase.model_validate(payload) + + +def _result(response: Any, *, write: bool = False, run: int = 1) -> AgentExecutionResult: + calls = [] + if write: + calls.append( + ToolCall( + server="billing", + name="issue_refund", + operation_kind=ToolOperationKind.WRITE, + status=ToolCallStatus.SUCCESS, + ) + ) + return AgentExecutionResult( + success=True, + run_id=f"run-{run}", + response=response, + trace=AgentTrace(tool_calls=calls), + ) + + +class ScriptedExecutor: + def __init__(self, results: list[AgentExecutionResult]) -> None: + self._results = iter(results) + self.inputs: list[dict[str, Any]] = [] + self.keys: list[str | None] = [] + + async def execute( + self, + input: dict[str, Any], + *, + run_context: RunContext | None = None, + ) -> AgentExecutionResult: + self.inputs.append(input) + self.keys.append(run_context.idempotency_key if run_context else None) + return next(self._results) + + +class NamedScorer: + def __init__(self, name: str, values: list[float]) -> None: + self.name = name + self._values = iter(values) + + async def score(self, result: AgentExecutionResult) -> Score: + return Score(next(self._values), reason=f"{self.name} rationale") + + +def test_example_yaml_loads_goal_cases() -> None: + cases = load_regression_cases(EXAMPLE_CASES) + + assert [case.id for case in cases] == [ + "refund_status_frustrated_customer", + "order_lookup_structured", + ] + first = cases[0] + assert first.goal == "User learns the current refund status for order 123" + assert first.persona == "frustrated_customer" + assert first.runs == 5 + assert first.behavior is not None + assert first.behavior.min_pass_rate == 0.8 + assert first.behavior.scores == {"groundedness": 0.8} + + +def test_cases_without_new_fields_keep_previous_defaults() -> None: + case = _case(goal=None, persona=None, behavior=None, runs=1) + + evaluation = case.evaluate(_result("anything")) + + assert case.runs == 1 + assert evaluation.passed + assert evaluation.scores == {} + + +@pytest.mark.parametrize( + ("override", "field"), + [ + ({"runs": 0}, "runs"), + ({"runs": 101}, "runs"), + ({"goal": ""}, "goal"), + ({"behavior": {"min_pass_rate": 0}}, "min_pass_rate"), + ({"behavior": {"min_pass_rate": 1.5}}, "min_pass_rate"), + ({"behavior": {"scores": {"judge": 1.2}}}, "scores"), + ({"behavior": {"unknown": True}}, "unknown"), + ], +) +def test_behavior_schema_validation(override: dict[str, Any], field: str) -> None: + with pytest.raises(ValidationError, match=field): + _case(**override) + + +def test_loader_reports_behavior_errors_with_case_context(tmp_path: Path) -> None: + path = tmp_path / "bad.yaml" + path.write_text( + """ +cases: + - id: broken-behavior + source: PROD-2 + description: invalid threshold + input: {} + expected_outcome: {success: true} + behavior: {min_pass_rate: 2} + created_at: 2026-09-24T00:00:00Z +""", + encoding="utf-8", + ) + + with pytest.raises(RegressionCaseLoadError, match=r"bad.yaml.*broken-behavior.*min_pass_rate"): + load_regression_cases(path) + + +@pytest.mark.parametrize( + ("response", "expected_failure", "scores"), + [ + ( + "Refund for 123 is pending", + None, + {"required_terms": 1.0, "forbidden_terms": 1.0}, + ), + ( + "Your refund is pending", + "Behavior 'required_terms': Missing terms: ['123']", + {"required_terms": 0.0, "forbidden_terms": 1.0}, + ), + ( + "Order 123, api key abc", + "Behavior 'forbidden_terms': Forbidden terms present: ['api key']", + {"required_terms": 1.0, "forbidden_terms": 0.0}, + ), + ], +) +def test_evaluate_applies_builtin_behavior_checks( + response: str, expected_failure: str | None, scores: dict[str, float] +) -> None: + evaluation = _case().evaluate(_result(response)) + + assert evaluation.scores == scores + if expected_failure is None: + assert evaluation.passed + else: + assert evaluation.failures == (expected_failure,) + + +def test_evaluate_applies_required_fields() -> None: + case = _case(behavior={"required_fields": ["order.status", "order.delivery_date"]}) + + evaluation = case.evaluate(_result({"order": {"status": "shipped"}})) + + assert not evaluation.passed + assert evaluation.scores == {"required_fields": 0.5} + assert "order.delivery_date" in evaluation.failures[0] + + +@pytest.mark.asyncio +async def test_run_passes_when_variance_stays_within_pass_rate() -> None: + executor = ScriptedExecutor( + [ + _result("123 pending", run=1), + _result("pending, order 123", run=2), + _result("cannot find it", run=3), + _result("123 approved", run=4), + ] + ) + + evaluation = await _case().run(executor) + + assert evaluation.passed + assert evaluation.pass_rate == 0.75 + assert evaluation.min_pass_rate == 0.75 + assert evaluation.goal == "User learns refund status" + assert evaluation.persona == "frustrated_customer" + assert [run.passed for run in evaluation.runs] == [True, True, False, True] + assert executor.inputs == [{"order_id": 123}] * 4 + assert set(executor.keys) == {"regression:refund-status"} + evaluation.assert_passed() + + +@pytest.mark.asyncio +async def test_run_fails_below_pass_rate_with_per_run_failures() -> None: + executor = ScriptedExecutor( + [ + _result("123 pending", run=1), + _result("pending", run=2), + _result("123", write=True, run=3), + _result("123 ok", run=4), + ] + ) + + evaluation = await _case().run(executor, idempotency_key="custom-key") + + assert not evaluation.passed + assert evaluation.pass_rate == 0.5 + assert set(executor.keys) == {"custom-key"} + with pytest.raises(AssertionError) as exc_info: + evaluation.assert_passed() + message = str(exc_info.value) + assert "pass rate 50% (2/4) is below 75%" in message + assert "run 2: Behavior 'required_terms'" in message + assert "run 3: Read-only workflow" in message + + +@pytest.mark.asyncio +async def test_evaluate_runs_applies_named_scores() -> None: + case = _case(behavior={"min_pass_rate": 0.5, "scores": {"groundedness": 0.8}}) + results = [_result("a", run=1), _result("b", run=2)] + + evaluation = await case.evaluate_runs( + results, scorers=[NamedScorer("groundedness", [0.9, 0.4]), NamedScorer("unused", [])] + ) + + assert evaluation.passed + assert [run.scores for run in evaluation.runs] == [ + {"groundedness": 0.9}, + {"groundedness": 0.4}, + ] + assert evaluation.runs[1].failures == ( + "Score 'groundedness': expected >= 0.80. Observed: 0.40 (groundedness rationale)", + ) + + +@pytest.mark.asyncio +async def test_run_rejects_missing_scorer_before_executing() -> None: + case = _case(behavior={"scores": {"groundedness": 0.8}}) + executor = ScriptedExecutor([]) + + with pytest.raises(ValueError, match=r"no matching scorer: \['groundedness'\]"): + await case.run(executor) + assert executor.inputs == [] + + +@pytest.mark.asyncio +async def test_evaluate_runs_rejects_duplicate_scorer_names() -> None: + with pytest.raises(ValueError, match="Duplicate scorer name 'judge'"): + await _case().evaluate_runs( + [_result("123")], scorers=[NamedScorer("judge", []), NamedScorer("judge", [])] + ) + + +@pytest.mark.asyncio +async def test_evaluate_runs_requires_results() -> None: + with pytest.raises(ValueError, match="at least one result"): + await _case().evaluate_runs([]) + + +@pytest.mark.asyncio +async def test_case_without_behavior_requires_every_run_to_pass() -> None: + case = _case(behavior=None) + + evaluation = await case.evaluate_runs([_result("ok"), _result("x", write=True)]) + + assert evaluation.min_pass_rate == 1.0 + assert evaluation.pass_rate == 0.5 + assert not evaluation.passed diff --git a/tests/test_scoring.py b/tests/test_scoring.py new file mode 100644 index 0000000..a522d41 --- /dev/null +++ b/tests/test_scoring.py @@ -0,0 +1,372 @@ +"""Behavior scoring contract tests.""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import pytest + +from agent_test_kit import ( + AgentExecutionResult, + AgentTestConfig, + ForbiddenTermsScorer, + JsonReportWriter, + RepeatedExecutionResult, + RequiredFieldsScorer, + RequiredTermsScorer, + RunContext, + Score, + Scorer, + response_text, + run_repeatedly, +) +from agent_test_kit.models.enums import ToolCallStatus, ToolOperationKind +from agent_test_kit.models.tool_call import ToolCall +from agent_test_kit.models.trace import AgentTrace + +pytestmark = pytest.mark.agent_unit + + +def _result(response: Any, *, success: bool = True) -> AgentExecutionResult: + return AgentExecutionResult(success=success, run_id="run-1", response=response) + + +class FixedScorer: + """Async scorer that returns a preset value, standing in for an LLM judge.""" + + def __init__(self, value: float, *, name: str = "judge") -> None: + self.value = value + self.name = name + + async def score(self, result: AgentExecutionResult) -> Score: + return Score(self.value, reason="judge rationale") + + +class SequenceScorer: + """Sync scorer returning successive values across repeated runs.""" + + name = "sequence" + + def __init__(self, values: list[float]) -> None: + self._values = iter(values) + + def score(self, result: AgentExecutionResult) -> Score: + return Score(next(self._values)) + + +class BadScorer: + name = "bad" + + def score(self, result: AgentExecutionResult) -> Any: + return 0.9 + + +class ScriptedExecutor: + """Returns one scripted response per run.""" + + def __init__(self, responses: list[Any], *, write_on_run: int | None = None) -> None: + self._responses = iter(responses) + self._run = 0 + self._write_on_run = write_on_run + + async def execute( + self, + input: dict[str, Any], + *, + run_context: RunContext | None = None, + ) -> AgentExecutionResult: + self._run += 1 + calls = [] + if self._run == self._write_on_run: + calls.append( + ToolCall( + server="billing", + name="issue_refund", + operation_kind=ToolOperationKind.WRITE, + status=ToolCallStatus.SUCCESS, + ) + ) + return AgentExecutionResult( + success=True, + run_id=f"run-{self._run}", + response=next(self._responses), + trace=AgentTrace(tool_calls=calls), + ) + + +@pytest.mark.parametrize( + ("response", "expected"), + [ + (None, ""), + ("plain text", "plain text"), + ({"b": 1, "a": "שלום"}, '{"a": "שלום", "b": 1}'), + (["x", 2], '["x", 2]'), + ], +) +def test_response_text_renders_supported_shapes(response: Any, expected: str) -> None: + assert response_text(_result(response)) == expected + + +@pytest.mark.parametrize("value", [-0.01, 1.01, float("nan")]) +def test_score_rejects_values_outside_unit_interval(value: float) -> None: + with pytest.raises(ValueError, match="between 0.0 and 1.0"): + Score(value) + + +@pytest.mark.parametrize("value", [True, "0.5", None]) +def test_score_rejects_non_numeric_values(value: Any) -> None: + with pytest.raises(TypeError, match="must be a number"): + Score(value) + + +@pytest.mark.parametrize( + ("response", "expected_value", "missing"), + [ + ("Refund for ORDER 123 is Pending", 1.0, []), + ("Order 123 received", 0.5, ["pending"]), + ({"note": "nothing useful"}, 0.0, ["order 123", "pending"]), + ], +) +def test_required_terms_scores_fraction_found( + response: Any, expected_value: float, missing: list[str] +) -> None: + score = RequiredTermsScorer(["order 123", "pending"]).score(_result(response)) + + assert score.value == expected_value + assert score.details == {"missing": missing} + assert (score.reason is None) == (not missing) + + +def test_required_terms_can_be_case_sensitive() -> None: + scorer = RequiredTermsScorer(["Pending"], case_sensitive=True) + + assert scorer.score(_result("status: pending")).value == 0.0 + assert scorer.score(_result("status: Pending")).value == 1.0 + + +@pytest.mark.parametrize( + ("response", "expected_value", "found"), + [ + ("Your refund is being reviewed", 1.0, []), + ("Here is the API KEY: abc", 0.0, ["api key"]), + ({"debug": "system prompt and api key"}, 0.0, ["system prompt", "api key"]), + ], +) +def test_forbidden_terms_is_binary(response: Any, expected_value: float, found: list[str]) -> None: + score = ForbiddenTermsScorer(["system prompt", "api key"]).score(_result(response)) + + assert score.value == expected_value + assert score.details == {"found": found} + + +@pytest.mark.parametrize( + ("response", "expected_value", "missing"), + [ + ({"order": {"id": 123, "status": "pending"}}, 1.0, []), + ({"order": {"id": 123, "status": None}}, 0.5, ["order.status"]), + ({"order": "123"}, 0.0, ["order.id", "order.status"]), + ], +) +def test_required_fields_supports_dotted_paths( + response: Any, expected_value: float, missing: list[str] +) -> None: + score = RequiredFieldsScorer(["order.id", "order.status"]).score(_result(response)) + + assert score.value == expected_value + assert score.details == {"missing": missing} + + +def test_required_fields_rejects_non_mapping_response() -> None: + score = RequiredFieldsScorer(["status"]).score(_result("status: ok")) + + assert score.value == 0.0 + assert score.reason == "Response is not a JSON object" + + +@pytest.mark.parametrize( + "factory", + [RequiredTermsScorer, ForbiddenTermsScorer, RequiredFieldsScorer], +) +@pytest.mark.parametrize("terms", [[], [""]]) +def test_builtin_scorers_reject_empty_terms(factory: Any, terms: list[str]) -> None: + with pytest.raises(ValueError): + factory(terms) + + +def test_builtin_and_custom_scorers_satisfy_protocol() -> None: + assert isinstance(RequiredTermsScorer(["x"]), Scorer) + assert isinstance(FixedScorer(0.5), Scorer) + + +@pytest.mark.asyncio +async def test_assert_score_records_passing_outcome() -> None: + result = _result("Order 123 is pending") + + score = await result.assert_score(RequiredTermsScorer(["order 123"]), min_score=1.0) + + assert score.value == 1.0 + outcome = result.assertion_outcomes[-1] + assert outcome.name == "assert_score" + assert outcome.passed is True + assert outcome.expected == {"scorer": "required_terms", "min_score": 1.0} + assert outcome.actual == {"score": 1.0, "reason": None, "details": {"missing": []}} + + +@pytest.mark.asyncio +async def test_assert_score_fails_below_threshold_with_reason() -> None: + result = _result("anything") + + with pytest.raises(AssertionError, match=r"judge score >= 0\.80\. Observed: 0\.60"): + await result.assert_score(FixedScorer(0.6), min_score=0.8) + + outcome = result.assertion_outcomes[-1] + assert outcome.passed is False + assert outcome.actual["score"] == 0.6 + assert "judge rationale" in (outcome.message or "") + + +@pytest.mark.asyncio +async def test_assert_score_passes_at_exact_threshold() -> None: + await _result("x").assert_score(FixedScorer(0.8), min_score=0.8) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("min_score", [-0.1, 1.5]) +async def test_assert_score_rejects_invalid_threshold(min_score: float) -> None: + result = _result("x") + + with pytest.raises(ValueError, match="min_score"): + await result.assert_score(FixedScorer(0.9), min_score=min_score) + assert result.assertion_outcomes == [] + + +@pytest.mark.asyncio +async def test_assert_score_rejects_scorer_with_wrong_return_type() -> None: + with pytest.raises(TypeError, match="must return Score"): + await _result("x").assert_score(BadScorer(), min_score=0.5) + + +@pytest.mark.asyncio +async def test_score_outcome_reaches_json_report(tmp_path: Path) -> None: + result = _result("Order 123") + with pytest.raises(AssertionError): + await result.assert_score(RequiredTermsScorer(["order 123", "pending"]), min_score=0.9) + + writer = JsonReportWriter( + config=AgentTestConfig(agent_id="billing-agent", environment="unit"), + output_path=tmp_path / "report.json", + ) + writer.record_scenario( + name="refund_status", + nodeid="tests/test_x.py::refund_status", + passed=False, + duration_ms=1, + markers=["agent_unit"], + execution_result=result, + ) + report = writer.build_report() + + assertion = report.scenarios[0].assertions[0] + assert assertion.name == "assert_score" + assert assertion.passed is False + assert assertion.actual["score"] == 0.5 + + +@pytest.mark.asyncio +async def test_assert_pass_rate_tolerates_variance_within_threshold() -> None: + responses = ["Order 123 pending", "Order 123 is pending", "cannot find order", "pending: 123"] + aggregate = await run_repeatedly( + ScriptedExecutor(responses), input={"order": 123}, runs=4, idempotency_key="k" + ) + + rate = aggregate.assert_pass_rate( + lambda r: "pending" in response_text(r).casefold(), min_rate=0.75 + ) + + assert rate == 0.75 + + +@pytest.mark.asyncio +async def test_assert_pass_rate_reports_failing_runs() -> None: + aggregate = await run_repeatedly( + ScriptedExecutor(["ok", "bad", "ok", "bad"]), input={}, runs=4, idempotency_key="k" + ) + + with pytest.raises(AssertionError, match=r"Observed: 50% \(2/4\); failing runs: \[2, 4\]"): + aggregate.assert_pass_rate(lambda r: r.response == "ok", min_rate=0.8) + + +@pytest.mark.asyncio +async def test_assert_pass_rate_counts_assertion_errors_as_failed_runs() -> None: + aggregate = await run_repeatedly( + ScriptedExecutor(["a", "b", "c"], write_on_run=2), input={}, runs=3, idempotency_key="k" + ) + + def read_only(result: AgentExecutionResult) -> bool: + result.assert_read_only() + return True + + assert aggregate.assert_pass_rate(read_only, min_rate=0.6) == pytest.approx(2 / 3) + with pytest.raises(AssertionError, match=r"failing runs: \[2\]"): + aggregate.assert_pass_rate(read_only, min_rate=1.0) + + +@pytest.mark.asyncio +async def test_assert_pass_rate_propagates_unexpected_errors() -> None: + aggregate = await run_repeatedly( + ScriptedExecutor([None]), input={}, runs=1, idempotency_key="k" + ) + + def broken(result: AgentExecutionResult) -> bool: + raise KeyError("missing") + + with pytest.raises(KeyError): + aggregate.assert_pass_rate(broken, min_rate=0.5) + + +@pytest.mark.asyncio +async def test_assert_score_rate_uses_score_threshold_per_run() -> None: + aggregate = await run_repeatedly( + ScriptedExecutor(["a", "b", "c", "d", "e"]), input={}, runs=5, idempotency_key="k" + ) + + rate = await aggregate.assert_score_rate( + SequenceScorer([0.9, 0.85, 0.4, 0.8, 0.95]), min_score=0.8, min_rate=0.8 + ) + + assert rate == 0.8 + + +@pytest.mark.asyncio +async def test_assert_score_rate_fails_below_rate() -> None: + aggregate = await run_repeatedly( + ScriptedExecutor(["a", "b"]), input={}, runs=2, idempotency_key="k" + ) + + with pytest.raises(AssertionError, match=r"sequence >= 0\.80 pass rate >= 100%"): + await aggregate.assert_score_rate(SequenceScorer([0.9, 0.1]), min_score=0.8, min_rate=1.0) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("kwargs", "label"), + [ + ({"min_score": 1.2, "min_rate": 0.5}, "min_score"), + ({"min_score": 0.5, "min_rate": -1.0}, "min_rate"), + ], +) +async def test_assert_score_rate_validates_thresholds(kwargs: dict[str, float], label: str) -> None: + aggregate = await run_repeatedly(ScriptedExecutor(["a"]), input={}, runs=1, idempotency_key="k") + + with pytest.raises(ValueError, match=label): + await aggregate.assert_score_rate(FixedScorer(0.9), **kwargs) + + +def test_pass_rate_requires_results() -> None: + empty = RepeatedExecutionResult(results=(), contexts=(), idempotency_key="k") + + with pytest.raises(ValueError, match="at least one run"): + empty.assert_pass_rate(lambda r: True, min_rate=0.5) + with pytest.raises(ValueError, match="min_rate"): + empty.assert_pass_rate(lambda r: True, min_rate=2.0)