diff --git a/example/agents/tests/units/agents/test_router_agent.py b/example/agents/tests/units/agents/test_router_agent.py new file mode 100644 index 00000000..26ddfae7 --- /dev/null +++ b/example/agents/tests/units/agents/test_router_agent.py @@ -0,0 +1,30 @@ +from langchain_core.messages import AIMessage, HumanMessage + +from app.agents.chat import RouterAgent + + +class TestRouterAgent: + async def test_the_router_agent(self): + with RouterAgent.record("record_stream.json") as agent: + await agent.prompt("hello") + agent.assert_text_response() + agent.assert_tool_not_called(["job_search_tool"]) + agent.assert_response_judged( + model="gpt-3.5-turbo", + expectation="The llm should respond with greetings", + ) + agent.assert_response_time_lt(5) + + await agent.prompt("suggest python developer jobs") + agent.assert_tool_called("job_search_tool", lambda tool: tool.name == "job_search_tool") + + async def test_the_router_with_initial_messages(self): + with RouterAgent.record( + "record_stream.json", + messages=[ + HumanMessage(content="Hi"), + AIMessage(content="Hello, How can I help you?"), + ], + ) as agent: + await agent.prompt("suggest python developer jobs") + agent.assert_tool_called("job_search_tool", lambda tool: tool.name == "job_search_tool") diff --git a/fastapi_startkit/src/fastapi_startkit/ai/__init__.py b/fastapi_startkit/src/fastapi_startkit/ai/__init__.py index 790ef065..1e8f8b8c 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/__init__.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/__init__.py @@ -6,14 +6,13 @@ from .config.ai import AIConfig from .decorators import max_steps, max_tokens, model, provider, timeout, top_p from .document import Document -from .evals import TestJudgeAgent, TrajectoryEvaluator, TrajectoryMatchMode from .fakes import fake_chat_model from .image import Image, ImageResponse from .image_factory import ImageFactory -from .model_builder import Ai +from .ai import Ai from .providers.ai_provider import AIProvider from .response import AgentResponse, AgentSnapshot -from .testing import AgentBinding, AgentModelFake, RecordingAgent +from .testing import AgentBinding, AgentModelFake, RecordingAgent, ToolCallView __all__ = [ "Agent", @@ -27,6 +26,7 @@ "AIProvider", "AnthropicConfig", "RecordingAgent", + "ToolCallView", "Audio", "AudioResponse", "AudioFactory", @@ -38,9 +38,6 @@ "ImageFactory", "ImageResponse", "OpenAIConfig", - "TestJudgeAgent", - "TrajectoryEvaluator", - "TrajectoryMatchMode", "max_steps", "max_tokens", "model", diff --git a/fastapi_startkit/src/fastapi_startkit/ai/agent.py b/fastapi_startkit/src/fastapi_startkit/ai/agent.py index e67a89ac..96926185 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/agent.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/agent.py @@ -73,12 +73,8 @@ async def stream( swapped = self._faked() if swapped is not None: - if hasattr(swapped, "stream"): - async for chunk in swapped.stream(message): - yield chunk - else: - response = await swapped.prompt(message) - yield response.content + async for chunk in swapped.stream(message): + yield chunk return async for chunk in self._stream(message, model=model, provider_options=provider_options): @@ -91,10 +87,10 @@ def fake(cls, responses: list) -> "AgentModelFake": return AgentModelFake(cls, responses) @classmethod - def record(cls, cassette: str | None = None) -> "AgentBinding": + def record(cls, cassette: str | None = None, messages: list | None = None) -> "AgentBinding": from .testing import AgentBinding, RecordingAgent - return AgentBinding(cls, RecordingAgent(cls(), cassette)) + return AgentBinding(cls, RecordingAgent(cls(), cassette, messages)) @classmethod def _binding(cls) -> Any: @@ -193,7 +189,7 @@ def _build_messages( return messages def _build_model(self, model: str | None = None, provider_options: dict | None = None) -> Any: - from .model_builder import Ai # noqa: PLC0415 + from .ai import Ai # noqa: PLC0415 return Ai().get_model_for(self, model, provider_options) diff --git a/fastapi_startkit/src/fastapi_startkit/ai/model_builder.py b/fastapi_startkit/src/fastapi_startkit/ai/ai.py similarity index 87% rename from fastapi_startkit/src/fastapi_startkit/ai/model_builder.py rename to fastapi_startkit/src/fastapi_startkit/ai/ai.py index 5b65a237..da1dc599 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/model_builder.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/ai.py @@ -9,14 +9,7 @@ class Ai: - # Keyed by agent class name (see _key()) so a fake can be registered - # before any instance of that agent exists. get_model_for() consults - # this registry, so a faked agent runs through the same message-building - # / pipeline / tool-execution path as a real one — only the model at the - # bottom is swapped for a deterministic stand-in. fake_agent_models: dict[str, Any] = {} - # Reserved for response-level fakes (mirroring fake_agent_models, but for - # cached final replies rather than whole chat models). Not yet wired up. fake_agent_responses: dict[str, Any] = {} def __init__(self) -> None: diff --git a/fastapi_startkit/src/fastapi_startkit/ai/evals.py b/fastapi_startkit/src/fastapi_startkit/ai/evals.py deleted file mode 100644 index de59f3d9..00000000 --- a/fastapi_startkit/src/fastapi_startkit/ai/evals.py +++ /dev/null @@ -1,195 +0,0 @@ -"""Deterministic trajectory-match evaluators for testing AI agents. - -``TestJudgeAgent`` compares an agent's actual message trajectory against a -reference trajectory using matching semantics equivalent to LangChain's -``agentevals.trajectory.match.create_trajectory_match_evaluator``. There is -no live LLM call involved — it is pure, deterministic structural comparison, -meant to be driven by data the test already has on hand (typically the -output of ``Agent.fake()``). - -Mirrors the ``Agent.fake()``/``Agent.record()`` testing DSL: ``.record()`` -returns a context manager yielding a callable evaluator:: - - with TestJudgeAgent(trajectory_match_mode="strict").record() as evaluator: - evaluation = evaluator(outputs=actual_messages, reference_outputs=reference_trajectory) - - assert evaluation["score"] is True -""" - -from __future__ import annotations - -from typing import Any, Callable, Literal, TypedDict, get_args - -TrajectoryMatchMode = Literal["strict", "unordered", "subset", "superset"] - -_TRAJECTORY_MATCH_MODES: tuple[str, ...] = get_args(TrajectoryMatchMode) - -_ROLE_BY_MESSAGE_TYPE = { - "human": "user", - "ai": "assistant", - "system": "system", - "tool": "tool", - "function": "tool", -} - - -class TrajectoryToolCall(TypedDict): - name: str - args: dict[str, Any] - - -class NormalizedMessage(TypedDict): - role: str - tool_calls: list[TrajectoryToolCall] - - -class EvaluatorResult(TypedDict): - key: str - score: bool - comment: str | None - - -def _normalize_tool_call(tool_call: dict) -> TrajectoryToolCall: - return {"name": tool_call.get("name", ""), "args": tool_call.get("args") or {}} - - -def _normalize_message(message: Any) -> NormalizedMessage: - if isinstance(message, dict): - role = message.get("role", "") - raw_tool_calls = message.get("tool_calls") or [] - else: - message_type = getattr(message, "type", "") - role = _ROLE_BY_MESSAGE_TYPE.get(message_type, message_type) - raw_tool_calls = getattr(message, "tool_calls", None) or [] - - return {"role": role, "tool_calls": [_normalize_tool_call(tc) for tc in raw_tool_calls]} - - -def _normalize_trajectory(trajectory: Any) -> list[NormalizedMessage]: - if isinstance(trajectory, dict): - trajectory = trajectory.get("messages", []) - return [_normalize_message(message) for message in trajectory] - - -def _extract_tool_calls(messages: list[NormalizedMessage]) -> list[TrajectoryToolCall]: - tool_calls: list[TrajectoryToolCall] = [] - for message in messages: - tool_calls.extend(message["tool_calls"]) - return tool_calls - - -def _is_tool_call_superset(calls: list[TrajectoryToolCall], wanted: list[TrajectoryToolCall]) -> bool: - """True if every tool call in ``wanted`` has a matching, unused tool call in ``calls``.""" - used = [False] * len(calls) - for want in wanted: - matched = False - for index, candidate in enumerate(calls): - if not used[index] and candidate == want: - used[index] = True - matched = True - break - if not matched: - return False - return True - - -def _tool_calls_equal(a: list[TrajectoryToolCall], b: list[TrajectoryToolCall]) -> bool: - return len(a) == len(b) and _is_tool_call_superset(a, b) and _is_tool_call_superset(b, a) - - -def _strict_match(outputs: list[NormalizedMessage], reference_outputs: list[NormalizedMessage]) -> bool: - if len(outputs) != len(reference_outputs): - return False - return all( - output["role"] == reference["role"] and _tool_calls_equal(output["tool_calls"], reference["tool_calls"]) - for output, reference in zip(outputs, reference_outputs) - ) - - -def _unordered_match(outputs: list[NormalizedMessage], reference_outputs: list[NormalizedMessage]) -> bool: - return _tool_calls_equal(_extract_tool_calls(outputs), _extract_tool_calls(reference_outputs)) - - -def _subset_match(outputs: list[NormalizedMessage], reference_outputs: list[NormalizedMessage]) -> bool: - """``outputs``' tool calls must all appear in ``reference_outputs``.""" - return _is_tool_call_superset(_extract_tool_calls(reference_outputs), _extract_tool_calls(outputs)) - - -def _superset_match(outputs: list[NormalizedMessage], reference_outputs: list[NormalizedMessage]) -> bool: - """``outputs``' tool calls must cover all of ``reference_outputs``'s.""" - return _is_tool_call_superset(_extract_tool_calls(outputs), _extract_tool_calls(reference_outputs)) - - -_SCORERS: dict[str, Callable[[list[NormalizedMessage], list[NormalizedMessage]], bool]] = { - "strict": _strict_match, - "unordered": _unordered_match, - "subset": _subset_match, - "superset": _superset_match, -} - - -class TrajectoryEvaluator: - """Callable returned by ``TestJudgeAgent.record()``; compares two trajectories.""" - - __test__ = False - - def __init__(self, trajectory_match_mode: TrajectoryMatchMode) -> None: - self.trajectory_match_mode: TrajectoryMatchMode = trajectory_match_mode - - def __call__(self, *, outputs: Any, reference_outputs: Any) -> EvaluatorResult: - normalized_outputs = _normalize_trajectory(outputs) - normalized_reference = _normalize_trajectory(reference_outputs) - scorer = _SCORERS[self.trajectory_match_mode] - score = scorer(normalized_outputs, normalized_reference) - return { - "key": f"trajectory_{self.trajectory_match_mode}_match", - "score": score, - "comment": None, - } - - -class TrajectoryJudgeBinding: - """Context manager returned by ``TestJudgeAgent.record()``.""" - - def __init__(self, trajectory_match_mode: TrajectoryMatchMode) -> None: - self._trajectory_match_mode: TrajectoryMatchMode = trajectory_match_mode - - def __enter__(self) -> TrajectoryEvaluator: - return TrajectoryEvaluator(self._trajectory_match_mode) - - def __exit__(self, *exc: Any) -> bool: - return False - - -class TestJudgeAgent: - """Deterministic judge for comparing agent trajectories in tests. - - Mirrors the ``Agent.fake()``/``Agent.record()`` testing DSL: ``.record()`` - returns a context manager yielding a callable evaluator, so trajectory - assertions read the same way as the rest of the agent testing toolkit. - - ``trajectory_match_mode`` controls how ``outputs`` is compared against - ``reference_outputs`` (each a list of LangChain messages, a list of - role/content/tool_calls dicts, or a dict with a ``messages`` key): - - - ``"strict"``: same number of messages, same role and same tool calls - at each position, in order. Message content is not compared. - - ``"unordered"``: the same set of tool calls were made, in any order - or position. - - ``"subset"``: every tool call in ``outputs`` also appears in - ``reference_outputs`` (no unexpected tool calls). - - ``"superset"``: every tool call in ``reference_outputs`` also appears - in ``outputs`` (no missing tool calls; extras are allowed). - """ - - __test__ = False - - def __init__(self, trajectory_match_mode: TrajectoryMatchMode = "strict") -> None: - if trajectory_match_mode not in _TRAJECTORY_MATCH_MODES: - raise ValueError( - f"Invalid trajectory_match_mode: {trajectory_match_mode!r}. Must be one of {_TRAJECTORY_MATCH_MODES!r}." - ) - self.trajectory_match_mode: TrajectoryMatchMode = trajectory_match_mode - - def record(self) -> TrajectoryJudgeBinding: - return TrajectoryJudgeBinding(self.trajectory_match_mode) diff --git a/fastapi_startkit/src/fastapi_startkit/ai/testing.py b/fastapi_startkit/src/fastapi_startkit/ai/testing.py index d456234c..639fda71 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/testing.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/testing.py @@ -5,7 +5,9 @@ import hashlib import inspect import json +import re import sys +import time from collections.abc import AsyncIterator from pathlib import Path from typing import TYPE_CHECKING, Any, Callable @@ -68,12 +70,12 @@ def __init__(self, agent_cls: type[Agent], responses: list) -> None: self._responses = responses def __enter__(self) -> None: - from .model_builder import Ai + from .ai import Ai Ai.fake(self._agent_cls.__name__, self._responses) def __exit__(self, *_exc: Any) -> bool: - from .model_builder import Ai + from .ai import Ai Ai.forget(self._agent_cls.__name__) return False @@ -96,16 +98,64 @@ def wrapper(*args: Any, **kwargs: Any) -> Any: return wrapper +class ToolCallView: + """Ergonomic, attribute-style view of a raw ``tool_calls`` dict, passed + to ``assert_tool_called``'s predicate.""" + + def __init__(self, data: dict) -> None: + self.name = data.get("name", "") + self.args = data.get("args") or {} + self.id = data.get("id") + self._data = data + + def __repr__(self) -> str: + return f"ToolCallView(name={self.name!r}, args={self.args!r})" + + class RecordingAgent(_Recorder): - def __init__(self, real: Agent, cassette: str | None = None) -> None: + """Bound as ``agent`` by ``with Agent.record(cassette) as agent:``. + + Fluent testing handle around a record-and-replay session: ``prompt()`` + is async (it's the same real agent call underneath, just cached) and + each call mutates the handle's "current turn" state, which the + ``assert_*`` methods judge against — mirroring how a browser-testing + ``page`` object exposes assertions against the current page state. + + On a cassette miss, the real agent is called once and the response is + cached to disk (keyed by the conversation history so far, plus the new + message, so two sessions with different histories but the same latest + message text don't collide). On a hit, it's replayed with no live call. + """ + + def __init__(self, real: Agent, cassette: str | None = None, messages: list | None = None) -> None: super().__init__() self._real = real self.cassette: Path | None = Path(cassette) if cassette else None + self._seed_messages: list = list(messages or []) + self._transcript: list[dict] = [] + self._real.messages = self._history # type: ignore[method-assign] + self.last_response: AgentResponse | None = None + self.last_elapsed: float | None = None + + def _history(self) -> list: + return self._seed_messages + self._transcript @staticmethod - def _key(message: str, attachments: list[Document] | None) -> str: + def _serialize(value: Any) -> Any: + if isinstance(value, dict): + return value + return {"type": type(value).__name__, "content": getattr(value, "content", str(value))} + + def _key(self, message: str, attachments: list[Document] | None) -> str: names = [getattr(doc, "name", "") for doc in (attachments or [])] - payload = json.dumps({"message": message, "attachments": names}, sort_keys=True) + payload = json.dumps( + { + "history": [self._serialize(m) for m in self._history()], + "message": message, + "attachments": names, + }, + sort_keys=True, + ) return hashlib.sha256(payload.encode()).hexdigest() def _load(self) -> tuple[Path, dict]: @@ -118,14 +168,38 @@ def _save(self, cassette: Path, store: dict, key: str, value: Any) -> None: cassette.parent.mkdir(parents=True, exist_ok=True) cassette.write_text(json.dumps(store, indent=2, sort_keys=True)) - async def prompt(self, message: str, attachments: list[Document] | None = None) -> AgentResponse: + @staticmethod + def _cache_prompt_value(response: AgentResponse) -> dict: + return {"content": response.content, "tool_calls": response.tool_calls} + + @staticmethod + def _response_from_cache(value: Any) -> AgentResponse: + if isinstance(value, dict) and "content" in value: + return AgentResponse(content=_joined(value.get("content", "")), tool_calls=value.get("tool_calls") or []) + return AgentResponse(content=_joined(value)) + + def _remember_turn(self, message: str, response: AgentResponse) -> None: + self._transcript.append({"role": "user", "content": message}) + turn: dict[str, Any] = {"role": "assistant", "content": response.content} + if response.tool_calls: + turn["tool_calls"] = response.tool_calls + self._transcript.append(turn) + + async def prompt(self, message: str, *, attachments: list[Document] | None = None) -> AgentResponse: + """Run (or replay) one turn and make it the "current" response that + assert_*() methods judge.""" self._record_call(message, attachments) cassette, store = self._load() key = self._key(message, attachments) + start = time.monotonic() if key in store: - return AgentResponse(content=_joined(store[key])) - response = await self._real._run(message, attachments=attachments) - self._save(cassette, store, key, response.content) + response = self._response_from_cache(store[key]) + else: + response = await self._real._run(message, attachments=attachments) + self._save(cassette, store, key, self._cache_prompt_value(response)) + self.last_elapsed = time.monotonic() - start + self.last_response = response + self._remember_turn(message, response) return response async def stream(self, message: str) -> AsyncIterator[str]: @@ -142,6 +216,78 @@ async def stream(self, message: str) -> AsyncIterator[str]: for chunk in chunks: yield chunk + def _require_response(self) -> AgentResponse: + assert self.last_response is not None, "No prompt() call has been made yet." + return self.last_response + + def _tool_call_names(self) -> list[str]: + return [tc.get("name", "") for tc in self._require_response().tool_calls] + + def assert_text_response(self) -> None: + response = self._require_response() + assert response.content, "Expected a non-empty text response, but content was empty." + + def assert_tool_called(self, name: str, predicate: Callable[[ToolCallView], bool] | None = None) -> None: + response = self._require_response() + matches = [tc for tc in response.tool_calls if tc.get("name") == name] + assert matches, f"Expected tool {name!r} to be called, but it wasn't. Called: {self._tool_call_names()}" + if predicate is not None: + assert any(predicate(ToolCallView(tc)) for tc in matches), ( + f"Tool {name!r} was called, but no call satisfied the given predicate." + ) + + def assert_tool_not_called(self, names: list[str]) -> None: + unexpected = set(self._tool_call_names()) & set(names) + assert not unexpected, f"Expected tools {sorted(names)} not to be called, but got: {sorted(unexpected)}" + + def assert_response_time_lt(self, seconds: float) -> None: + assert self.last_elapsed is not None, "No prompt() call has been made yet." + assert self.last_elapsed < seconds, f"Expected response time < {seconds}s, took {self.last_elapsed:.3f}s" + + def assert_response_judged(self, *, model: str, expectation: str) -> None: + response = self._require_response() + verdict = self._judge(model, expectation, response.content) + assert verdict.get("passed"), ( + f"Judge ({model}) rejected the response for expectation {expectation!r}: " + f"{verdict.get('reasoning', '')!r} — response was {response.content!r}" + ) + + def _judge(self, model: str, expectation: str, content: str) -> dict: + cassette, store = self._load() + key = self._judge_key(model, expectation, content) + if key in store: + return store[key] + verdict = self._judge_live(model, expectation, content) + self._save(cassette, store, key, verdict) + return verdict + + @staticmethod + def _judge_key(model: str, expectation: str, content: str) -> str: + payload = json.dumps( + {"judge_model": model, "expectation": expectation, "content": content}, + sort_keys=True, + ) + return "judge:" + hashlib.sha256(payload.encode()).hexdigest() + + def _judge_live(self, model: str, expectation: str, content: str) -> dict: + from langchain.chat_models import init_chat_model # noqa: PLC0415 + + prompt = ( + "You are grading whether an AI agent's response satisfies an expectation.\n" + f"Expectation: {expectation}\n" + f"Response: {content}\n\n" + 'Reply with strict JSON only, no prose: {"passed": true|false, "reasoning": ""}' + ) + chat_model = init_chat_model(model) + result = chat_model.invoke(prompt) + return self._parse_verdict(result.content) + + @staticmethod + def _parse_verdict(raw: str) -> dict: + match = re.search(r"\{.*\}", raw, re.DOTALL) + data = json.loads(match.group(0) if match else raw) + return {"passed": bool(data.get("passed")), "reasoning": data.get("reasoning", "")} + class AgentBinding: def __init__(self, agent_cls: type[Agent], stand_in: Any) -> None: diff --git a/fastapi_startkit/tests/ai/test_agent.py b/fastapi_startkit/tests/ai/test_agent.py index 2d63867a..145a1cfb 100644 --- a/fastapi_startkit/tests/ai/test_agent.py +++ b/fastapi_startkit/tests/ai/test_agent.py @@ -7,7 +7,7 @@ from fastapi_startkit.ai import AIConfig, Document, fake_chat_model from fastapi_startkit.ai.agent import Agent -from fastapi_startkit.ai.model_builder import Ai +from fastapi_startkit.ai.ai import Ai from fastapi_startkit.ai.response import AgentResponse from fastapi_startkit.application import app diff --git a/fastapi_startkit/tests/ai/test_agent_fake.py b/fastapi_startkit/tests/ai/test_agent_fake.py index 7b560d74..42ff7adb 100644 --- a/fastapi_startkit/tests/ai/test_agent_fake.py +++ b/fastapi_startkit/tests/ai/test_agent_fake.py @@ -16,7 +16,7 @@ from unittest import mock from fastapi_startkit.ai.agent import Agent -from fastapi_startkit.ai.model_builder import Ai +from fastapi_startkit.ai.ai import Ai from fastapi_startkit.ai.response import AgentResponse @@ -204,7 +204,8 @@ async def test_first_run_records_response_to_cassette(self): self.assertEqual(calls, ["hello"]) self.assertTrue(os.path.exists(cassette)) with open(cassette) as f: - self.assertIn("recorded reply", json.load(f).values()) + store = json.load(f) + self.assertTrue(any(v.get("content") == "recorded reply" for v in store.values())) async def test_second_run_replays_without_calling_run(self): calls = self.setup_agent("recorded reply") diff --git a/fastapi_startkit/tests/ai/test_agent_record_fluent.py b/fastapi_startkit/tests/ai/test_agent_record_fluent.py new file mode 100644 index 00000000..30d12989 --- /dev/null +++ b/fastapi_startkit/tests/ai/test_agent_record_fluent.py @@ -0,0 +1,354 @@ +"""Tests for the fluent Agent.record() testing DSL. + +``with Agent.record(cassette) as agent:`` binds a ``RecordingAgent`` handle +whose async ``prompt()`` and assertion methods judge the most recent turn — +mirroring how a browser-testing ``page`` object exposes assertions against +current page state: + + with RouterAgent.record("cassette.json") as agent: + await agent.prompt("hello") + agent.assert_text_response() + agent.assert_tool_not_called(["job_search_tool"]) + agent.assert_response_time_lt(5) + + await agent.prompt("suggest python developer jobs") + agent.assert_tool_called("job_search_tool", lambda tool: tool.name == "job_search_tool") +""" + +import os +import tempfile +import unittest +from unittest import mock + +import langchain.chat_models as chat_models +from langchain_core.messages import AIMessage, HumanMessage + +from fastapi_startkit.ai.agent import Agent +from fastapi_startkit.ai.response import AgentResponse +from fastapi_startkit.ai.testing import RecordingAgent + + +class SimpleAgent(Agent): + pass + + +def _tool_call(name: str, args: dict | None = None, call_id: str = "c1") -> dict: + return {"name": name, "args": args or {}, "id": call_id} + + +class TestFluentPromptMechanics(unittest.IsolatedAsyncioTestCase): + def setup_agent(self, responses: list): + """responses: list of (content, tool_calls) tuples, consumed in call order.""" + queue = list(responses) + + async def fake_run(agent_self, message, **kwargs): + content, tool_calls = queue.pop(0) + return AgentResponse(content=content, tool_calls=tool_calls or []) + + patcher = mock.patch.object(SimpleAgent, "_run", fake_run) + patcher.start() + self.addCleanup(patcher.stop) + + async def test_prompt_is_async_and_returns_agent_response(self): + self.setup_agent([("Hello there!", [])]) + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + response = await agent.prompt("hi") + + self.assertIsInstance(response, AgentResponse) + self.assertEqual(response.content, "Hello there!") + + async def test_second_prompt_continues_the_same_session(self): + self.setup_agent([("Hi!", []), ("here are some jobs", [_tool_call("job_search_tool")])]) + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("hello") + agent.assert_text_response() + + await agent.prompt("suggest python developer jobs") + agent.assert_tool_called("job_search_tool") + + async def test_replaying_from_cassette_preserves_tool_calls(self): + with tempfile.TemporaryDirectory() as tmp: + cassette = os.path.join(tmp, "c.json") + self.setup_agent([("", [_tool_call("job_search_tool", {"q": "python"})])]) + with SimpleAgent.record(cassette) as agent: + await agent.prompt("find jobs") + + # No queued responses left — this must replay from cassette, not call _run again. + with SimpleAgent.record(cassette) as agent: + await agent.prompt("find jobs") + agent.assert_tool_called("job_search_tool") + + +class TestAssertTextResponse(unittest.IsolatedAsyncioTestCase): + def setup_agent(self, content, tool_calls=None): + async def fake_run(agent_self, message, **kwargs): + return AgentResponse(content=content, tool_calls=tool_calls or []) + + patcher = mock.patch.object(SimpleAgent, "_run", fake_run) + patcher.start() + self.addCleanup(patcher.stop) + + async def test_passes_when_content_present(self): + self.setup_agent("Hi!") + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("hi") + agent.assert_text_response() + + async def test_fails_on_empty_content(self): + self.setup_agent("", tool_calls=[_tool_call("search")]) + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("hi") + with self.assertRaises(AssertionError): + agent.assert_text_response() + + async def test_fails_when_no_prompt_has_been_made(self): + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + with self.assertRaises(AssertionError): + agent.assert_text_response() + + +class TestAssertToolCalled(unittest.IsolatedAsyncioTestCase): + def setup_agent(self, tool_calls): + async def fake_run(agent_self, message, **kwargs): + return AgentResponse(content="", tool_calls=tool_calls) + + patcher = mock.patch.object(SimpleAgent, "_run", fake_run) + patcher.start() + self.addCleanup(patcher.stop) + + async def test_passes_when_tool_present(self): + self.setup_agent([_tool_call("job_search_tool", {"q": "python"})]) + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("find jobs") + agent.assert_tool_called("job_search_tool") + + async def test_fails_when_tool_absent(self): + self.setup_agent([]) + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("hi") + with self.assertRaises(AssertionError): + agent.assert_tool_called("job_search_tool") + + async def test_predicate_can_accept_via_attribute_access(self): + self.setup_agent([_tool_call("job_search_tool", {"q": "python"})]) + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("find jobs") + agent.assert_tool_called("job_search_tool", lambda tool: tool.name == "job_search_tool") + + async def test_predicate_can_reject(self): + self.setup_agent([_tool_call("job_search_tool", {"q": "python"})]) + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("find jobs") + with self.assertRaises(AssertionError): + agent.assert_tool_called("job_search_tool", lambda tool: tool.args.get("q") == "java") + + +class TestAssertToolNotCalled(unittest.IsolatedAsyncioTestCase): + def setup_agent(self, tool_calls): + async def fake_run(agent_self, message, **kwargs): + return AgentResponse(content="Hello!", tool_calls=tool_calls) + + patcher = mock.patch.object(SimpleAgent, "_run", fake_run) + patcher.start() + self.addCleanup(patcher.stop) + + async def test_passes_when_absent(self): + self.setup_agent([]) + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("hi") + agent.assert_tool_not_called(["job_search_tool"]) + + async def test_fails_when_present(self): + self.setup_agent([_tool_call("job_search_tool")]) + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("find jobs") + with self.assertRaises(AssertionError): + agent.assert_tool_not_called(["job_search_tool"]) + + +class TestAssertResponseTimeLt(unittest.IsolatedAsyncioTestCase): + def setup_agent(self): + async def fake_run(agent_self, message, **kwargs): + return AgentResponse(content="Hello!") + + patcher = mock.patch.object(SimpleAgent, "_run", fake_run) + patcher.start() + self.addCleanup(patcher.stop) + + async def test_passes_for_a_fast_call(self): + self.setup_agent() + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("hi") + agent.assert_response_time_lt(5) + + async def test_fails_when_exceeded(self): + self.setup_agent() + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("hi") + with self.assertRaises(AssertionError): + agent.assert_response_time_lt(0) + + async def test_fails_when_no_prompt_has_been_made(self): + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + with self.assertRaises(AssertionError): + agent.assert_response_time_lt(5) + + +class TestRecordMessagesSeed(unittest.IsolatedAsyncioTestCase): + async def test_seed_messages_are_included_when_building_the_real_agents_messages(self): + seed = [HumanMessage(content="Hi"), AIMessage(content="Hello, how can I help?")] + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json"), messages=seed) as agent: + built = agent._real._build_messages("suggest python developer jobs") + + self.assertEqual(built[0], seed[0]) + self.assertEqual(built[1], seed[1]) + self.assertEqual(built[-1], {"role": "user", "content": "suggest python developer jobs"}) + + async def test_same_followup_text_with_different_seed_history_does_not_collide(self): + with tempfile.TemporaryDirectory() as tmp: + cassette = os.path.join(tmp, "shared.json") + + async def run_a(agent_self, message, **kwargs): + return AgentResponse(content="job list A") + + with mock.patch.object(SimpleAgent, "_run", run_a): + with SimpleAgent.record(cassette) as agent: + response_a = await agent.prompt("suggest python developer jobs") + + async def run_b(agent_self, message, **kwargs): + return AgentResponse(content="job list B") + + seed = [HumanMessage(content="Hi"), AIMessage(content="Hello, how can I help?")] + with mock.patch.object(SimpleAgent, "_run", run_b): + with SimpleAgent.record(cassette, messages=seed) as agent: + response_b = await agent.prompt("suggest python developer jobs") + + self.assertEqual(response_a.content, "job list A") + self.assertEqual(response_b.content, "job list B") + + +class TestAssertResponseJudged(unittest.IsolatedAsyncioTestCase): + def setup_agent(self, content): + async def fake_run(agent_self, message, **kwargs): + return AgentResponse(content=content) + + patcher = mock.patch.object(SimpleAgent, "_run", fake_run) + patcher.start() + self.addCleanup(patcher.stop) + + async def test_passes_when_judge_approves(self): + self.setup_agent("Hello there, welcome!") + with tempfile.TemporaryDirectory() as tmp: + with mock.patch.object( + RecordingAgent, "_judge_live", return_value={"passed": True, "reasoning": "greets the user"} + ): + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("hello") + agent.assert_response_judged( + model="gpt-3.5-turbo", expectation="The llm should respond with greetings" + ) + + async def test_fails_when_judge_rejects(self): + self.setup_agent("Completely unrelated content") + with tempfile.TemporaryDirectory() as tmp: + with mock.patch.object( + RecordingAgent, "_judge_live", return_value={"passed": False, "reasoning": "not a greeting"} + ): + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + await agent.prompt("hello") + with self.assertRaises(AssertionError): + agent.assert_response_judged( + model="gpt-3.5-turbo", expectation="The llm should respond with greetings" + ) + + async def test_verdict_is_cached_in_the_cassette_and_not_re_judged(self): + self.setup_agent("Hello there!") + judge = mock.Mock(return_value={"passed": True, "reasoning": "ok"}) + with tempfile.TemporaryDirectory() as tmp: + cassette = os.path.join(tmp, "c.json") + with mock.patch.object(RecordingAgent, "_judge_live", judge): + with SimpleAgent.record(cassette) as agent: + await agent.prompt("hello") + agent.assert_response_judged(model="gpt-3.5-turbo", expectation="greet") + agent.assert_response_judged(model="gpt-3.5-turbo", expectation="greet") + + judge.assert_called_once() + + async def test_verdict_persists_to_disk_for_a_later_replay(self): + self.setup_agent("Hello there!") + with tempfile.TemporaryDirectory() as tmp: + cassette = os.path.join(tmp, "c.json") + with mock.patch.object(RecordingAgent, "_judge_live", return_value={"passed": True, "reasoning": "ok"}): + with SimpleAgent.record(cassette) as agent: + await agent.prompt("hello") + agent.assert_response_judged(model="gpt-3.5-turbo", expectation="greet") + + judge = mock.Mock(side_effect=AssertionError("must not be called on replay")) + with mock.patch.object(RecordingAgent, "_judge_live", judge): + with SimpleAgent.record(cassette) as agent: + await agent.prompt("hello") + agent.assert_response_judged(model="gpt-3.5-turbo", expectation="greet") + + judge.assert_not_called() + + async def test_fails_when_no_prompt_has_been_made(self): + with tempfile.TemporaryDirectory() as tmp: + with SimpleAgent.record(os.path.join(tmp, "c.json")) as agent: + with self.assertRaises(AssertionError): + agent.assert_response_judged(model="gpt-3.5-turbo", expectation="greet") + + +class TestJudgeLiveModelCall(unittest.TestCase): + def test_calls_init_chat_model_and_parses_json_verdict(self): + captured = {} + + class FakeResult: + content = '{"passed": true, "reasoning": "Greets the user politely."}' + + class FakeModel: + def invoke(self, prompt): + captured["prompt"] = prompt + return FakeResult() + + patcher = mock.patch.object(chat_models, "init_chat_model", lambda *a, **k: FakeModel()) + patcher.start() + self.addCleanup(patcher.stop) + + agent = RecordingAgent(SimpleAgent()) + verdict = agent._judge_live("gpt-3.5-turbo", "The llm should respond with greetings", "Hello there!") + + self.assertTrue(verdict["passed"]) + self.assertIn("Greets", verdict["reasoning"]) + self.assertIn("Hello there!", captured["prompt"]) + + +class TestExistingRecordApiIsUnaffected(unittest.IsolatedAsyncioTestCase): + """The pre-existing bare-context-manager Agent.record() usage (task #327) + must keep working unchanged alongside the new fluent handle.""" + + async def test_bare_context_manager_prompt_still_works(self): + async def fake_run(agent_self, message, **kwargs): + return AgentResponse(content="recorded reply") + + with tempfile.TemporaryDirectory() as tmp: + cassette = os.path.join(tmp, "c.json") + with mock.patch.object(SimpleAgent, "_run", fake_run): + with SimpleAgent.record(cassette): + result = await SimpleAgent().prompt("hello") + + self.assertEqual(result.content, "recorded reply") diff --git a/fastapi_startkit/tests/ai/test_agent_schema.py b/fastapi_startkit/tests/ai/test_agent_schema.py index 1a30772c..40306ea3 100644 --- a/fastapi_startkit/tests/ai/test_agent_schema.py +++ b/fastapi_startkit/tests/ai/test_agent_schema.py @@ -6,7 +6,7 @@ from pydantic import BaseModel from fastapi_startkit.ai.agent import Agent -from fastapi_startkit.ai.model_builder import Ai +from fastapi_startkit.ai.ai import Ai from fastapi_startkit.ai.response import AgentResponse diff --git a/fastapi_startkit/tests/ai/test_ai_fake.py b/fastapi_startkit/tests/ai/test_ai_fake.py index 30b2e39e..33a29247 100644 --- a/fastapi_startkit/tests/ai/test_ai_fake.py +++ b/fastapi_startkit/tests/ai/test_ai_fake.py @@ -16,7 +16,7 @@ from langchain_core.tools import tool from fastapi_startkit.ai.agent import Agent -from fastapi_startkit.ai.model_builder import Ai +from fastapi_startkit.ai.ai import Ai from fastapi_startkit.application import app diff --git a/fastapi_startkit/tests/ai/test_evals.py b/fastapi_startkit/tests/ai/test_evals.py deleted file mode 100644 index 5e7e2ccc..00000000 --- a/fastapi_startkit/tests/ai/test_evals.py +++ /dev/null @@ -1,281 +0,0 @@ -"""Tests for TestJudgeAgent — the deterministic trajectory-match evaluator. - -TestJudgeAgent compares an actual agent message trajectory against a reference -trajectory with no live LLM calls: it is pure, deterministic structural -comparison, driven entirely by data the tests construct themselves (typically -the output of Agent.fake()). Mirrors the Agent.fake()/Agent.record() testing -DSL: ``.record()`` returns a context manager yielding a callable evaluator. -""" - -import unittest - -from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage - -from fastapi_startkit.ai.agent import Agent -from fastapi_startkit.ai.evals import TestJudgeAgent -from fastapi_startkit.ai.response import AgentResponse - - -def _weather_trajectory(city: str = "San Francisco") -> list: - return [ - HumanMessage(content=f"What is the weather in {city}?"), - AIMessage( - content="", - tool_calls=[{"id": "call_1", "name": "get_weather", "args": {"city": city}}], - ), - ToolMessage(content="It is 75 degrees and sunny.", tool_call_id="call_1"), - AIMessage(content=f"The weather in {city} is 75 degrees and sunny."), - ] - - -class TestJudgeAgentConstruction(unittest.TestCase): - def test_defaults_to_strict_mode(self): - judge = TestJudgeAgent() - self.assertEqual(judge.trajectory_match_mode, "strict") - - def test_accepts_each_supported_mode(self): - for mode in ("strict", "unordered", "subset", "superset"): - judge = TestJudgeAgent(trajectory_match_mode=mode) - self.assertEqual(judge.trajectory_match_mode, mode) - - def test_rejects_unknown_mode(self): - with self.assertRaises(ValueError): - TestJudgeAgent(trajectory_match_mode="fuzzy") - - -class TestJudgeAgentRecordContextManager(unittest.TestCase): - def test_record_yields_callable_evaluator(self): - with TestJudgeAgent().record() as evaluator: - self.assertTrue(callable(evaluator)) - - def test_evaluation_result_is_subscriptable_with_score_key(self): - trajectory = _weather_trajectory() - with TestJudgeAgent(trajectory_match_mode="strict").record() as evaluator: - evaluation = evaluator(outputs=trajectory, reference_outputs=trajectory) - - self.assertIs(evaluation["score"], True) - self.assertEqual(evaluation["key"], "trajectory_strict_match") - self.assertIn("comment", evaluation) - - -class TestStrictTrajectoryMatch(unittest.TestCase): - def setUp(self): - self.judge = TestJudgeAgent(trajectory_match_mode="strict") - - def _evaluate(self, outputs, reference_outputs): - with self.judge.record() as evaluator: - return evaluator(outputs=outputs, reference_outputs=reference_outputs) - - def test_identical_trajectory_matches(self): - trajectory = _weather_trajectory() - evaluation = self._evaluate(trajectory, trajectory) - self.assertIs(evaluation["score"], True) - - def test_matches_even_when_final_message_content_differs(self): - reference = _weather_trajectory() - outputs = _weather_trajectory() - outputs[-1] = AIMessage(content="Completely different wording, still sunny.") - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], True) - - def test_fails_when_tool_call_args_differ(self): - reference = _weather_trajectory(city="San Francisco") - outputs = _weather_trajectory(city="Oakland") - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], False) - - def test_fails_on_extra_trailing_message(self): - reference = _weather_trajectory() - outputs = _weather_trajectory() + [AIMessage(content="One more thing...")] - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], False) - - def test_fails_on_missing_message(self): - reference = _weather_trajectory() - outputs = _weather_trajectory()[:-1] - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], False) - - def test_fails_when_role_order_differs(self): - reference = _weather_trajectory() - outputs = list(reversed(_weather_trajectory())) - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], False) - - def test_fails_when_one_side_has_tool_calls_and_other_does_not(self): - reference = _weather_trajectory() - outputs = _weather_trajectory() - outputs[1] = AIMessage(content="") - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], False) - - def test_works_with_plain_dict_messages(self): - reference = _weather_trajectory() - outputs = [ - {"role": "user", "content": "What is the weather in San Francisco?"}, - { - "role": "assistant", - "content": "", - "tool_calls": [{"id": "call_1", "name": "get_weather", "args": {"city": "San Francisco"}}], - }, - {"role": "tool", "content": "It is 75 degrees and sunny.", "tool_call_id": "call_1"}, - {"role": "assistant", "content": "The weather in San Francisco is 75 degrees and sunny."}, - ] - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], True) - - -class TestUnorderedTrajectoryMatch(unittest.TestCase): - def setUp(self): - self.judge = TestJudgeAgent(trajectory_match_mode="unordered") - - def _evaluate(self, outputs, reference_outputs): - with self.judge.record() as evaluator: - return evaluator(outputs=outputs, reference_outputs=reference_outputs) - - def test_same_tool_calls_in_different_message_order_matches(self): - weather_call = AIMessage(content="", tool_calls=[{"id": "1", "name": "get_weather", "args": {"city": "SF"}}]) - time_call = AIMessage(content="", tool_calls=[{"id": "2", "name": "get_time", "args": {"tz": "PT"}}]) - - reference = [HumanMessage(content="hi"), weather_call, time_call] - outputs = [HumanMessage(content="hi"), time_call, weather_call] - - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], True) - - def test_missing_tool_call_fails(self): - weather_call = AIMessage(content="", tool_calls=[{"id": "1", "name": "get_weather", "args": {"city": "SF"}}]) - time_call = AIMessage(content="", tool_calls=[{"id": "2", "name": "get_time", "args": {"tz": "PT"}}]) - - reference = [weather_call, time_call] - outputs = [weather_call] - - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], False) - - def test_extra_tool_call_fails(self): - weather_call = AIMessage(content="", tool_calls=[{"id": "1", "name": "get_weather", "args": {"city": "SF"}}]) - time_call = AIMessage(content="", tool_calls=[{"id": "2", "name": "get_time", "args": {"tz": "PT"}}]) - - reference = [weather_call] - outputs = [weather_call, time_call] - - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], False) - - -class TestSubsetTrajectoryMatch(unittest.TestCase): - def setUp(self): - self.judge = TestJudgeAgent(trajectory_match_mode="subset") - - def _evaluate(self, outputs, reference_outputs): - with self.judge.record() as evaluator: - return evaluator(outputs=outputs, reference_outputs=reference_outputs) - - def test_output_tool_calls_within_reference_matches(self): - weather_call = AIMessage(content="", tool_calls=[{"id": "1", "name": "get_weather", "args": {"city": "SF"}}]) - time_call = AIMessage(content="", tool_calls=[{"id": "2", "name": "get_time", "args": {"tz": "PT"}}]) - - reference = [weather_call, time_call] - outputs = [weather_call] - - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], True) - - def test_output_tool_call_not_in_reference_fails(self): - weather_call = AIMessage(content="", tool_calls=[{"id": "1", "name": "get_weather", "args": {"city": "SF"}}]) - time_call = AIMessage(content="", tool_calls=[{"id": "2", "name": "get_time", "args": {"tz": "PT"}}]) - - reference = [weather_call] - outputs = [weather_call, time_call] - - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], False) - - -class TestSupersetTrajectoryMatch(unittest.TestCase): - def setUp(self): - self.judge = TestJudgeAgent(trajectory_match_mode="superset") - - def _evaluate(self, outputs, reference_outputs): - with self.judge.record() as evaluator: - return evaluator(outputs=outputs, reference_outputs=reference_outputs) - - def test_output_contains_all_reference_tool_calls_plus_extra_matches(self): - weather_call = AIMessage(content="", tool_calls=[{"id": "1", "name": "get_weather", "args": {"city": "SF"}}]) - time_call = AIMessage(content="", tool_calls=[{"id": "2", "name": "get_time", "args": {"tz": "PT"}}]) - - reference = [weather_call] - outputs = [weather_call, time_call] - - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], True) - - def test_output_missing_a_required_reference_tool_call_fails(self): - weather_call = AIMessage(content="", tool_calls=[{"id": "1", "name": "get_weather", "args": {"city": "SF"}}]) - time_call = AIMessage(content="", tool_calls=[{"id": "2", "name": "get_time", "args": {"tz": "PT"}}]) - - reference = [weather_call, time_call] - outputs = [weather_call] - - evaluation = self._evaluate(outputs, reference) - self.assertIs(evaluation["score"], False) - - -class SimpleAgent(Agent): - pass - - -class TestJudgeAgentWithAgentFakeOutput(unittest.TestCase): - """Demonstrates the intended workflow: Agent.fake() drives deterministic - output with no live LLM call, and TestJudgeAgent judges the resulting - trajectory against a reference.""" - - async def _fake_response(self) -> AgentResponse: - agent = SimpleAgent() - with SimpleAgent.fake(["It is sunny in San Francisco."]): - return await agent.prompt("What is the weather in San Francisco?") - - def test_fake_agent_reply_matches_reference_trajectory_in_strict_mode(self): - import asyncio - - response = asyncio.run(self._fake_response()) - - outputs = [ - HumanMessage(content="What is the weather in San Francisco?"), - AIMessage(content=response.content), - ] - reference_trajectory = [ - HumanMessage(content="What is the weather in San Francisco?"), - AIMessage(content="It is sunny in San Francisco."), - ] - - with TestJudgeAgent(trajectory_match_mode="strict").record() as evaluator: - evaluation = evaluator(outputs=outputs, reference_outputs=reference_trajectory) - - self.assertIs(evaluation["score"], True) - - -class TestJudgeAgentMessagesDictInput(unittest.TestCase): - def test_accepts_outputs_dict_with_messages_key(self): - trajectory = _weather_trajectory() - with TestJudgeAgent(trajectory_match_mode="strict").record() as evaluator: - evaluation = evaluator(outputs={"messages": trajectory}, reference_outputs=trajectory) - - self.assertIs(evaluation["score"], True) - - -class TestJudgeAgentIsNotCollectedByPytest(unittest.TestCase): - def test_has_test_dunder_set_to_false(self): - self.assertFalse(getattr(TestJudgeAgent, "__test__")) - - -class TestSystemMessagesAreComparedByRole(unittest.TestCase): - def test_strict_mode_compares_system_message_role(self): - reference = [SystemMessage(content="Be concise."), HumanMessage(content="hi")] - outputs = [SystemMessage(content="Be very concise."), HumanMessage(content="hi")] - - with TestJudgeAgent(trajectory_match_mode="strict").record() as evaluator: - evaluation = evaluator(outputs=outputs, reference_outputs=reference) - - self.assertIs(evaluation["score"], True)