Skip to content
Draft
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
39 changes: 39 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
29 changes: 29 additions & 0 deletions examples/jev_semantic_example.py
Original file line number Diff line number Diff line change
@@ -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"],
)
3 changes: 3 additions & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,9 @@ dependencies = [
]

[project.optional-dependencies]
jev = [
"pytest-jev>=0.1.0",
]
dev = [
"build>=1.2",
"mypy>=1.8",
Expand Down
80 changes: 80 additions & 0 deletions src/agent_test_kit/assertions/semantic.py
Original file line number Diff line number Diff line change
@@ -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
64 changes: 63 additions & 1 deletion src/agent_test_kit/models/execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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:
Expand Down
Loading
Loading