diff --git a/README.md b/README.md index 116ab2c..8b442fe 100644 --- a/README.md +++ b/README.md @@ -279,6 +279,45 @@ result.assert_no_failed_tool_calls() result.assert_no_failed_tool_calls("search", server="jira") ``` +### Semantic meaning (Jev) + +String checks break when the model rewords a reply. `assert_means` sends the response text to [pytest-jev](https://github.com/allebee/pytest-jev), which asks TypeSafe's Jev whether each claim holds. One request covers every claim. The probabilities are stored on `assertion_outcomes` and show up in the HTML report. + +```bash +pip install "agent-test-kit[jev]" +export OPENROUTER_API_KEY=... # or TYPESAFE_API_KEY from https://console.typesafe.ai +``` + +Request the `jev` fixture pytest-jev provides. Tests that use it are marked `jev`, so `pytest -m "not jev"` stays offline. Without an API key those tests are skipped. In CI, pass `--jev-require` so a missing key fails the job. + +```python +@pytest.mark.asyncio +async def test_refund_reply(agent_scenario, jev): + result = await agent_client.execute({"ticket": "I was charged twice"}) + agent_scenario.attach_execution_result(result) + result.assert_success() + result.assert_means( + jev, + holds=[ + "apologizes to the customer", + "says the duplicate payment was refunded", + ], + lacks=["asks for a password or a full card number"], + # context={"policy": REFUND_POLICY}, # claims can name `policy` + # threshold=0.9, + ) +``` + +`response` strings are judged as-is. Dict responses use the first non-empty field among `text`, `summary`, `message`, `content`, `reply`, and `suggested_prompt`. Read that string yourself with `result.response_text` when you call `jev.choice` or `jev.score` directly. + +Pin the model when a run must be reproducible: + +```ini +[pytest] +jev_model = jev-1.13 +jev_threshold = 0.8 +``` + ### Repeated execution (idempotency) Use `run_repeatedly()` when the same input + idempotency key must produce stable behavior across N runs. diff --git a/examples/jev_semantic_example.py b/examples/jev_semantic_example.py new file mode 100644 index 0000000..b593dd6 --- /dev/null +++ b/examples/jev_semantic_example.py @@ -0,0 +1,29 @@ +"""Semantic check of an agent reply with pytest-jev. + +Copy into an agent repo's tests/ after: + + pip install "agent-test-kit[jev]" + +Set OPENROUTER_API_KEY or TYPESAFE_API_KEY. Without a key, pytest-jev skips +the test. ``pytest -m "not jev"`` runs everything else offline. +""" + +from __future__ import annotations + +import pytest + + +@pytest.mark.asyncio +async def test_refund_reply_means(agent_client: object, jev: object) -> None: + result = await agent_client.execute( # type: ignore[attr-defined] + input={"ticket": "I was charged twice for order #1042."}, + ) + result.assert_success() + result.assert_means( + jev, + holds=[ + "apologizes to the customer", + "says the duplicate payment was refunded", + ], + lacks=["asks for a password or a full card number"], + ) diff --git a/pyproject.toml b/pyproject.toml index 87d2161..8031876 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,9 @@ dependencies = [ ] [project.optional-dependencies] +jev = [ + "pytest-jev>=0.1.0", +] dev = [ "build>=1.2", "mypy>=1.8", diff --git a/src/agent_test_kit/assertions/semantic.py b/src/agent_test_kit/assertions/semantic.py new file mode 100644 index 0000000..89c7641 --- /dev/null +++ b/src/agent_test_kit/assertions/semantic.py @@ -0,0 +1,80 @@ +"""Semantic assertions judged by TypeSafe's Jev (pytest-jev).""" + +from __future__ import annotations + +import json +from collections.abc import Iterable, Mapping +from typing import Any, Protocol + +# Keys tried, in order, when an agent response is a mapping. +_TEXT_KEYS = ( + "text", + "summary", + "message", + "content", + "reply", + "suggested_prompt", +) + + +class SemanticJudge(Protocol): + """The pytest-jev ``jev`` fixture: ``expect`` checks many claims in one request.""" + + def expect( + self, + text: Any, + *, + holds: str | Iterable[str] = (), + lacks: str | Iterable[str] = (), + context: Mapping[str, Any] | None = None, + threshold: float | None = None, + ) -> Any: + """Return the claims when every one passes, or raise AssertionError.""" + + +def claim_list(claims: str | Iterable[str]) -> list[str]: + """Normalize a single claim or a sequence of claims into a list of strings.""" + if isinstance(claims, str): + return [claims] + return [str(claim) for claim in claims] + + +def response_text(response: Any) -> str: + """Text Jev should judge from an agent ``response`` payload. + + Strings are used as-is. Mappings prefer a known text field + (``text``, ``summary``, ``message``, ``content``, ``reply``, + ``suggested_prompt``). Anything else is serialized to JSON so a + structured payload can still be judged. + """ + if isinstance(response, str): + if not response.strip(): + raise ValueError("agent response text is empty") + return response + if isinstance(response, Mapping): + for key in _TEXT_KEYS: + value = response.get(key) + if isinstance(value, str) and value.strip(): + return value + return json.dumps(response, sort_keys=True, ensure_ascii=False, default=str) + if response is None: + raise ValueError("agent response is empty; assert_means needs text to judge") + rendered = str(response).strip() + if not rendered: + raise ValueError("agent response text is empty") + return str(response) + + +def summarize_claims(claims: Any) -> list[dict[str, Any]]: + """JSON-ready rows for the HTML report from a pytest-jev ``Claims`` result.""" + rows: list[dict[str, Any]] = [] + for claim in claims: + rows.append( + { + "claim": getattr(claim, "claim", str(claim)), + "expect": getattr(claim, "expect", None), + "p": getattr(claim, "p", None), + "passed": bool(getattr(claim, "passed", claim)), + } + ) + return rows diff --git a/src/agent_test_kit/models/execution.py b/src/agent_test_kit/models/execution.py index a47a0d1..b67ae93 100644 --- a/src/agent_test_kit/models/execution.py +++ b/src/agent_test_kit/models/execution.py @@ -2,12 +2,20 @@ from __future__ import annotations -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Iterable, Mapping, Sequence from datetime import UTC, datetime from typing import TYPE_CHECKING, Any from pydantic import BaseModel, Field +from agent_test_kit.assertions.semantic import ( + SemanticJudge, + claim_list, + summarize_claims, +) +from agent_test_kit.assertions.semantic import ( + response_text as extract_response_text, +) 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 @@ -53,6 +61,11 @@ class AgentExecutionResult(BaseModel): def tool_calls(self) -> list[ToolCall]: return self.trace.tool_calls + @property + def response_text(self) -> str: + """Plain text of ``response`` for semantic checks.""" + return extract_response_text(self.response) + def _engine(self) -> FlowAssertionEngine: from agent_test_kit.assertions.flow import FlowAssertionEngine @@ -88,6 +101,55 @@ def _capture_assertion( ) ) + def assert_means( + self, + jev: SemanticJudge, + *, + holds: str | Iterable[str] = (), + lacks: str | Iterable[str] = (), + context: Mapping[str, Any] | None = None, + threshold: float | None = None, + ) -> Any: + """Judge what ``response`` means using the pytest-jev ``jev`` fixture. + + Install the optional extra (``pip install 'agent-test-kit[jev]'``) and + request the ``jev`` fixture. Every claim is one Jev request. A passing + run records each claim's probability on ``assertion_outcomes``. + + ``holds`` claims must be true of the reply. ``lacks`` claims must be + false. Pass ``context`` for a policy or retrieved documents the claim + names in backticks. + """ + held = claim_list(holds) + absent = claim_list(lacks) + if not held and not absent: + raise ValueError("assert_means needs at least one claim in holds= or lacks=") + text = self.response_text + judged: list[Any] = [] + + def evaluate() -> None: + judged.append( + jev.expect( + text, + holds=held, + lacks=absent, + context=context, + threshold=threshold, + ) + ) + + self._capture_assertion( + "assert_means", + expected={"holds": held, "lacks": absent, "threshold": threshold}, + actual=text, + assertion=evaluate, + ) + claims = judged[0] + summary = summarize_claims(claims) + if summary: + self.assertion_outcomes[-1].actual = summary + return claims + def assert_success(self) -> None: def evaluate() -> None: if not self.success: diff --git a/tests/test_semantic_assertions.py b/tests/test_semantic_assertions.py new file mode 100644 index 0000000..f4e1df2 --- /dev/null +++ b/tests/test_semantic_assertions.py @@ -0,0 +1,144 @@ +"""Semantic assertion tests using a stand-in for the pytest-jev fixture.""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from typing import Any + +import pytest + +from agent_test_kit.assertions.semantic import response_text +from agent_test_kit.models.execution import AgentExecutionResult + + +@dataclass +class _Claim: + claim: str + expect: str + p: float + + @property + def passed(self) -> bool: + if self.expect == "holds": + return self.p >= 0.8 + return self.p <= 0.2 + + +class _FakeJev: + def __init__(self, *, fail: bool = False) -> None: + self.fail = fail + self.calls: list[dict[str, Any]] = [] + + def expect( + self, + text: Any, + *, + holds: str | Iterable[str] = (), + lacks: str | Iterable[str] = (), + context: Mapping[str, Any] | None = None, + threshold: float | None = None, + ) -> list[_Claim]: + self.calls.append( + { + "text": text, + "holds": list(holds) if not isinstance(holds, str) else [holds], + "lacks": list(lacks) if not isinstance(lacks, str) else [lacks], + "context": context, + "threshold": threshold, + } + ) + if self.fail: + raise AssertionError( + "jev: 1 of 1 claims failed\n" + " ✗ holds p=0.04 apologizes to the customer (needs >= 0.80)" + ) + return [ + _Claim(claim=claim, expect="holds", p=0.91) + for claim in self.calls[-1]["holds"] + ] + [ + _Claim(claim=claim, expect="lacks", p=0.02) + for claim in self.calls[-1]["lacks"] + ] + + +@pytest.mark.agent_unit +def test_assert_means_records_claim_probabilities() -> None: + jev = _FakeJev() + result = AgentExecutionResult( + success=True, + run_id="run_1", + response={"summary": "Sorry — the duplicate charge was refunded."}, + ) + claims = result.assert_means( + jev, + holds="apologizes to the customer", + lacks=["asks for a password"], + threshold=0.9, + ) + assert jev.calls[0]["text"] == "Sorry — the duplicate charge was refunded." + assert jev.calls[0]["threshold"] == 0.9 + assert len(claims) == 2 + outcome = result.assertion_outcomes[-1] + assert outcome.name == "assert_means" + assert outcome.passed is True + assert outcome.actual == [ + { + "claim": "apologizes to the customer", + "expect": "holds", + "p": 0.91, + "passed": True, + }, + { + "claim": "asks for a password", + "expect": "lacks", + "p": 0.02, + "passed": True, + }, + ] + + +@pytest.mark.agent_unit +def test_assert_means_records_failure_then_raises() -> None: + result = AgentExecutionResult( + success=True, + run_id="run_1", + response="Double charges happen when you click twice.", + ) + with pytest.raises(AssertionError, match="apologizes to the customer"): + result.assert_means(_FakeJev(fail=True), holds=["apologizes to the customer"]) + outcome = result.assertion_outcomes[-1] + assert outcome.passed is False + assert outcome.actual == "Double charges happen when you click twice." + assert "p=0.04" in (outcome.message or "") + + +@pytest.mark.agent_unit +def test_assert_means_requires_a_claim() -> None: + result = AgentExecutionResult(success=True, run_id="run_1", response="ok") + with pytest.raises(ValueError, match="at least one claim"): + result.assert_means(_FakeJev(), holds=[], lacks=[]) + + +@pytest.mark.agent_unit +@pytest.mark.parametrize( + ("response", "expected"), + [ + ("plain reply", "plain reply"), + ({"summary": "from summary"}, "from summary"), + ({"suggested_prompt": "rewrite this"}, "rewrite this"), + ({"ticket": 12}, '{"ticket": 12}'), + ], +) +def test_response_text_prefers_known_fields(response: object, expected: str) -> None: + assert response_text(response) == expected + + +@pytest.mark.agent_unit +def test_response_text_rejects_empty_payload() -> None: + result = AgentExecutionResult(success=True, run_id="run_1", response=None) + with pytest.raises(ValueError, match="empty"): + result.assert_means(_FakeJev(), holds=["says hello"]) + blank = AgentExecutionResult(success=True, run_id="run_1", response=" ") + with pytest.raises(ValueError, match="empty"): + _ = blank.response_text