Evaluators API Reference¶

@dataclass(slots=True)
class ContainsEvaluator:
    r"""Check that the reply contains expected substrings or matches regex patterns.

    Args:
        any_of: Reply must contain at least one of these strings.
        all_of: Reply must contain every one of these strings.
        matches_any: Reply must match at least one of these regex patterns (``re.search``).
        matches_all: Reply must match every one of these regex patterns (``re.search``).
        case_sensitive: When False (the default), substring and regex checks ignore case.

    Raises:
        ValueError: If a regex pattern in matches_any/matches_all does not compile.

    Example:
        ```python
        ContainsEvaluator(any_of=["confirmed", "booked"])
        ContainsEvaluator(all_of=["booking", "reference number"])
        ContainsEvaluator(matches_any=[r"ref(erence)? number[:# ]*[A-Z]{2}-\\d+"])
        ContainsEvaluator(all_of=["Booking"], case_sensitive=True)
        ```
    """

    any_of: list[str] = field(default_factory=list)
    all_of: list[str] = field(default_factory=list)
    matches_any: list[str] = field(default_factory=list)
    matches_all: list[str] = field(default_factory=list)
    case_sensitive: bool = False

    def __post_init__(self) -> None:
        """Validate every pattern at construction time."""
        # Compile and discard: a bad pattern is an authoring error and must fail at
        # construction time, not surface as a per-turn evaluation failure. The compiled
        # objects are not stored, so this stays a plain dataclass with no hidden
        # attributes; re.compile is memoised by the re module cache, so recompiling in
        # evaluate() costs a dict lookup.
        self._compile()

    def _compile(self) -> list[list[re.Pattern[str]]]:
        """Compile both pattern lists, raising a didactic ValueError on a bad pattern."""
        flags = 0 if self.case_sensitive else re.IGNORECASE
        try:
            return [[re.compile(p, flags) for p in patterns] for patterns in (self.matches_any, self.matches_all)]
        except re.error as exc:
            raise ValueError(f"Invalid regex pattern {exc.pattern!r}: {exc}") from exc

    async def evaluate(self, ctx: TurnContext) -> EvalResult:
        """Evaluate substring and regex checks against the reply."""
        # Bound once instead of through a one-line _norm() method: the substring checks
        # compare a folded needle against a folded reply, and this is the only place that
        # decision is made. The regex checks below use re.IGNORECASE on the raw reply
        # instead, so a pattern's own anchors and character classes still mean what they say.
        fold = str if self.case_sensitive else str.lower
        reply = fold(ctx.reply)
        matches_any_compiled, matches_all_compiled = self._compile()

        if self.any_of and not any(fold(s) in reply for s in self.any_of):
            return EvalResult(
                passed=False,
                reasoning=f"Reply did not contain any of {self.any_of!r}",
            )

        missing = [s for s in self.all_of if fold(s) not in reply]
        if missing:
            return EvalResult(
                passed=False,
                reasoning=f"Reply missing required strings: {missing!r}",
            )

        if matches_any_compiled and not any(p.search(ctx.reply) for p in matches_any_compiled):
            return EvalResult(
                passed=False,
                reasoning=f"Reply did not match any of {self.matches_any!r}",
            )

        unmatched = [p.pattern for p in matches_all_compiled if not p.search(ctx.reply)]
        if unmatched:
            return EvalResult(
                passed=False,
                reasoning=f"Reply missing required patterns: {unmatched!r}",
            )

        return EvalResult(passed=True, reasoning="All substring and pattern checks passed")

@dataclass
class ToolCallEvaluator:
    """Validate that specific tools were (or were not) called.

    Args:
        must_include: Tool names that must appear in tool_calls.
        must_exclude: Tool names that must NOT appear in tool_calls.
        ordered: If True, must_include tools must appear in the given order.

    Example:
        ```python
        ToolCallEvaluator(must_include=["book_slot"], must_exclude=["cancel_slot"])
        ToolCallEvaluator(must_include=["auth", "fetch", "respond"], ordered=True)
        ```
    """

    must_include: list[str] = field(default_factory=list)
    must_exclude: list[str] = field(default_factory=list)
    ordered: bool = False

    async def evaluate(self, ctx: TurnContext) -> EvalResult:
        """Evaluate tool call presence and ordering."""
        failures: list[str] = []
        if not self.ordered:
            failures += [
                f"Expected tool {tool!r} not in {ctx.tool_calls!r}"
                for tool in self.must_include
                if tool not in ctx.tool_calls
            ]
        failures += [f"Forbidden tool {tool!r} was called" for tool in self.must_exclude if tool in ctx.tool_calls]
        if self.ordered and self.must_include and not _is_ordered_subsequence(self.must_include, ctx.tool_calls):
            failures.append(f"Tools {self.must_include!r} not called in order in {ctx.tool_calls!r}")

        if failures:
            return EvalResult(passed=False, reasoning="\n".join(failures))
        return EvalResult(passed=True, reasoning="All tool call checks passed")

@dataclass
class ToolCallArgsEvaluator:
    """Assert the arguments a tool was called with.

    When the tool was called more than once in a turn, the check passes if ANY
    of those calls matches the expected arguments.

    Args:
        tool: Name of the tool to check.
        args: Expected arguments.
        mode: ``"subset"`` (every expected top-level key/value must appear in the
            observed args; extra observed keys are fine, but nested values are
            compared exactly) or ``"exact"`` (observed args must equal the
            expected dict exactly).

    Example:
        ```python
        ToolCallArgsEvaluator(tool="book_slot", args={"time": "10am"})
        ToolCallArgsEvaluator(tool="book_slot", args={"time": "10am", "date": "tomorrow"}, mode="exact")
        ```
    """

    tool: str
    args: JsonMapping
    mode: ToolCallArgsMode = "subset"

    def __post_init__(self) -> None:
        """Reject an unknown comparison mode at construction time."""
        if self.mode not in ("subset", "exact"):
            raise ValueError(f"ToolCallArgsEvaluator mode must be 'subset' or 'exact', got {self.mode!r}")

    def _matches(self, observed: ToolArgs) -> bool:
        """Compare one call's captured arguments against the expected ones.

        ``observed`` is ``ToolArgs`` while ``self.args`` stays ``JsonMapping``: the
        expected side is written by hand in a transcript and really is JSON, whereas the
        observed side is whatever an SDK captured.
        """
        if self.mode == "exact":
            return observed == self.args
        return all(k in observed and observed[k] == v for k, v in self.args.items())

    async def evaluate(self, ctx: TurnContext) -> EvalResult:
        """Evaluate the expected arguments against every call of the tool this turn."""
        found = capture_tool_args(self.tool, ctx.tool_calls)
        if found.failure is not None:
            return found.failure
        captured = found.args

        if any(self._matches(observed) for observed in captured):
            return EvalResult(passed=True, reasoning=f"Tool {self.tool!r} called with expected args ({self.mode})")

        return EvalResult(
            passed=False,
            reasoning=(
                f"Tool {self.tool!r} argument mismatch ({self.mode} mode): "
                f"expected {self.args!r}, observed {captured!r}"
            ),
        )

@dataclass(kw_only=True)
class ToolCallArgsJudgeEvaluator(_JudgeEvaluatorBase):
    """Use an LLM to evaluate a tool's call arguments against a rubric.

    Deterministic short-circuits run before any LLM call: if the tool was never
    called, or was called but no arguments were captured, the evaluator fails
    with a precise message and no judge tokens are spent. Otherwise the judge
    receives the tool name and the JSON arguments of every call to it this
    turn, and passes if any call satisfies the rubric.

    Args:
        tool: Name of the tool whose arguments to judge.
        rubric: Natural language rubric describing acceptable arguments.
        model: pydantic-ai model string (e.g. ``"openai:gpt-4o"``). Falls back to
            ``[tool.agent_eval] model`` in pyproject.toml if None.
        retries: Number of retry attempts on API failure before returning a FAIL verdict.
        timeout: Seconds before the judge call times out.

    Example:
        ```python
        ToolCallArgsJudgeEvaluator(
            tool="book_slot",
            rubric="The booking time must be within business hours (9am-5pm).",
        )
        ```
    """

    tool: str
    rubric: str

    _system_prompt: ClassVar[str] = _ARGS_SYSTEM_PROMPT

    async def evaluate(self, ctx: TurnContext) -> EvalResult:
        """Judge the tool's captured arguments, short-circuiting when there is nothing to judge."""
        found = capture_tool_args(self.tool, ctx.tool_calls)
        if found.failure is not None:
            return found.failure
        captured = found.args

        calls_text = "\n\n".join(
            f"CALL {i + 1} ARGUMENTS:\n{json.dumps(args, indent=2, default=str)}" for i, args in enumerate(captured)
        )
        user_msg = f"RUBRIC:\n{self.rubric}\n\nTOOL: {self.tool}\n\n{calls_text}"
        return await _run_judge(self._agent, user_msg, self.retries, self.timeout)

@dataclass(kw_only=True)
class JudgeEvaluator(_JudgeEvaluatorBase):
    """Use an LLM to evaluate the reply against a rubric.

    Uses pydantic-ai under the hood; supports any pydantic-ai compatible model.

    Args:
        rubric: Natural language rubric describing what a passing reply looks like.
        model: pydantic-ai model string (e.g. ``"openai:gpt-4o"``). Falls back to
            ``[tool.agent_eval] model`` in pyproject.toml if None.
        retries: Number of retry attempts on API failure before returning a FAIL verdict.
        timeout: Seconds before the judge call times out.

    Example:
        ```python
        JudgeEvaluator(
            rubric="Reply must confirm booking with date and time",
            model="anthropic:claude-3-5-sonnet-latest",
        )
        ```
    """

    rubric: str

    _system_prompt: ClassVar[str] = _SYSTEM_PROMPT

    async def evaluate(self, ctx: TurnContext) -> EvalResult:
        """Run the LLM judge against the turn and return its verdict."""
        user_msg = _format_judge_prompt(self.rubric, ctx)
        return await _run_judge(self._agent, user_msg, self.retries, self.timeout)