diff --git a/example/agents/tests/features/test_chat_controller.py b/example/agents/tests/features/test_chat_controller.py index c0228d72..41eb02f7 100644 --- a/example/agents/tests/features/test_chat_controller.py +++ b/example/agents/tests/features/test_chat_controller.py @@ -4,14 +4,14 @@ class TestChatController(TestCase): - @RouterAgent.fake({"*hello*": "Hello there, This is no stream chat, Hope you are doing well."}) + @RouterAgent.fake(["Hello there, This is no stream chat, Hope you are doing well."]) async def test_it_responds_without_stream(self): response = await self.post("/chat", json={"message": "hello"}) response.assert_ok() response.assert_contents("Hello there, This is no stream chat, Hope you are doing well.") - @RouterAgent.fake({"*hello*": "Hello there, This is no stream chat, Hope you are doing well."}) + @RouterAgent.fake(["Hello there, This is no stream chat, Hope you are doing well."]) async def test_stream_assertions_are_rejected_on_a_buffered_response(self): response = await self.post("/chat", json={"message": "hello"}) @@ -21,7 +21,7 @@ async def test_stream_assertions_are_rejected_on_a_buffered_response(self): with self.assertRaises(AssertionError): response.assert_stream("Hello there, This is no stream chat, Hope you are doing well.") - @RouterAgent.fake({"*hello*": "Hello there, This is stream chat, Hope you are doing well."}) + @RouterAgent.fake(["Hello there, This is stream chat, Hope you are doing well."]) async def test_it_responds_with_stream(self): response = await self.post("/chat/stream", json={"message": "hello"}) diff --git a/fastapi_startkit/src/fastapi_startkit/ai/__init__.py b/fastapi_startkit/src/fastapi_startkit/ai/__init__.py index dbf4c85e..790ef065 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/__init__.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/__init__.py @@ -6,24 +6,26 @@ 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 .providers.ai_provider import AIProvider from .response import AgentResponse, AgentSnapshot -from .testing import AgentBinding, FakeAgent, NoFakeResponse, RecordingAgent +from .testing import AgentBinding, AgentModelFake, RecordingAgent __all__ = [ "Agent", + "Ai", "Middleware", "AgentBinding", + "AgentModelFake", "AgentResponse", "AgentSnapshot", "AIConfig", "AIProvider", "AnthropicConfig", - "FakeAgent", - "NoFakeResponse", "RecordingAgent", "Audio", "AudioResponse", @@ -36,6 +38,9 @@ "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 17c947f3..e67a89ac 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/agent.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/agent.py @@ -1,15 +1,16 @@ from __future__ import annotations -import fnmatch from typing import TYPE_CHECKING, Any, AsyncIterator, Callable, Optional, Type from .document import Document -from .response import AgentResponse, AgentSnapshot +from .response import AgentResponse from .testing import AgentBinding if TYPE_CHECKING: from langchain_core.tools import BaseTool + from .testing import AgentModelFake + class Agent: provider: str | None = None @@ -20,7 +21,6 @@ class Agent: top_p: float = 1.0 def __init__(self): - self._fakes: dict[str, AgentResponse | AgentSnapshot] = {} self._call_log: list[dict] = [] def messages(self) -> list[dict]: @@ -55,21 +55,6 @@ async def prompt( self._log_call("prompt", message) return self._apply_schema(response) - _run_kwargs = dict( - model=model, - attachments=attachments, - provider_options=provider_options, - ) - - match = self._match_fake(message) - if match is not None: - if isinstance(match, AgentSnapshot): - response = await match.resolve(self, message, **_run_kwargs) - else: - response = match - self._log_call("prompt", message) - return self._apply_schema(response) - messages = self._build_messages(message, attachments) chat_model = self._build_model(model, provider_options) @@ -96,22 +81,14 @@ async def stream( yield response.content return - fake = self._match_fake(message) - if fake is not None: - if isinstance(fake, AgentSnapshot): - response = await fake.resolve(self, message) - else: - response = fake - yield response.content - return async for chunk in self._stream(message, model=model, provider_options=provider_options): yield chunk @classmethod - def fake(cls, responses: dict | None = None) -> "AgentBinding": - from .testing import AgentBinding, FakeAgent + def fake(cls, responses: list) -> "AgentModelFake": + from .testing import AgentModelFake - return AgentBinding(cls, FakeAgent(responses)) + return AgentModelFake(cls, responses) @classmethod def record(cls, cassette: str | None = None) -> "AgentBinding": @@ -146,16 +123,9 @@ def assert_not_prompted(self) -> None: self.assert_prompted(times=0) def reset(self) -> "Agent": - self._fakes.clear() self._call_log.clear() return self - def _match_fake(self, message: str) -> Optional[AgentResponse | AgentSnapshot]: - for pattern, value in self._fakes.items(): - if fnmatch.fnmatch(message.lower(), pattern.lower()): - return value - return None - def _log_call(self, method: str, message: str) -> None: self._call_log.append({"method": method, "message": message}) @@ -223,9 +193,9 @@ def _build_messages( return messages def _build_model(self, model: str | None = None, provider_options: dict | None = None) -> Any: - from .model_builder import ModelBuilder # noqa: PLC0415 + from .model_builder import Ai # noqa: PLC0415 - return ModelBuilder(agent=self).build(model, provider_options) + return Ai().get_model_for(self, model, provider_options) def _to_agent_response(self, result: Any) -> AgentResponse: messages = result.get("messages", []) if isinstance(result, dict) else [] diff --git a/fastapi_startkit/src/fastapi_startkit/ai/evals.py b/fastapi_startkit/src/fastapi_startkit/ai/evals.py new file mode 100644 index 00000000..de59f3d9 --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/ai/evals.py @@ -0,0 +1,195 @@ +"""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/model_builder.py b/fastapi_startkit/src/fastapi_startkit/ai/model_builder.py index 7f82254f..5b65a237 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/model_builder.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/model_builder.py @@ -8,40 +8,93 @@ from .agent import Agent -class ModelBuilder: - def __init__(self, agent: "Agent") -> None: - self._agent = agent +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 build(self, model: str | None = None, provider_options: dict | None = None) -> Any: + def __init__(self) -> None: + pass + + @staticmethod + def _key(agent: "Agent | str") -> str: + return agent if isinstance(agent, str) else type(agent).__name__ + + @classmethod + def fake(cls, agent: "Agent | str", messages: list) -> Any: + """Register a deterministic stand-in chat model for ``agent``. + + Replays ``messages`` in order via a GenericFakeChatModel — no live + LLM call. Plain strings are coerced into ``AIMessage(content=...)``. + """ + from langchain_core.language_models.fake_chat_models import GenericFakeChatModel + from langchain_core.messages import AIMessage + + turns = [message if hasattr(message, "content") else AIMessage(content=str(message)) for message in messages] + model = GenericFakeChatModel(messages=iter(turns)) + cls.fake_agent_models[cls._key(agent)] = model + return model + + @classmethod + def has_fake_model_for(cls, agent: "Agent | str") -> bool: + return cls._key(agent) in cls.fake_agent_models + + @classmethod + def get_fake_model_for(cls, agent: "Agent | str") -> Any: + return cls.fake_agent_models[cls._key(agent)] + + @classmethod + def forget(cls, agent: "Agent | str") -> None: + cls.fake_agent_models.pop(cls._key(agent), None) + + @classmethod + def reset_fakes(cls) -> None: + cls.fake_agent_models.clear() + cls.fake_agent_responses.clear() + + def get_model_for(self, agent: "Agent", model: str | None = None, provider_options: dict | None = None) -> Any: + """Resolve the model to run: a registered fake if one exists for + ``agent``, otherwise a freshly-built provider model.""" + if self.has_fake_model_for(agent): + return self.get_fake_model_for(agent) + return self.build(agent, model, provider_options) + + def build(self, agent: "Agent", model: str | None = None, provider_options: dict | None = None) -> Any: from langchain.chat_models import init_chat_model # noqa: PLC0415 - lab = Lab.get_provider(self._agent.provider) + lab = Lab.get_provider(agent.provider) kwargs: dict[str, Any] = {"model_provider": lab.get_provider_key()} api_key = lab.get_api_key() if api_key: kwargs["api_key"] = api_key - if self._agent.max_tokens: - kwargs["max_tokens"] = self._agent.max_tokens - if self._agent.top_p != 1.0: - kwargs["top_p"] = self._agent.top_p - if self._agent.timeout: - kwargs["timeout"] = self._agent.timeout + if agent.max_tokens: + kwargs["max_tokens"] = agent.max_tokens + if agent.top_p != 1.0: + kwargs["top_p"] = agent.top_p + if agent.timeout: + kwargs["timeout"] = agent.timeout - kwargs.update(self._resolve_provider_options(provider_options)) + kwargs.update(self._resolve_provider_options(agent, provider_options)) - chat_model = init_chat_model(self._resolve_model(model), **kwargs) + chat_model = init_chat_model(self._resolve_model(agent, model), **kwargs) - tools = list(self._agent.tools()) + tools = list(agent.tools()) return chat_model.bind_tools(tools) if tools else chat_model - def _resolve_model(self, override: str | None = None) -> str: - return Lab.get_provider(self._agent.provider).get_model(override or self._agent.model or None) + def _resolve_model(self, agent: "Agent", override: str | None = None) -> str: + return Lab.get_provider(agent.provider).get_model(override or agent.model or None) - def _resolve_provider_options(self, override: dict | None = None) -> dict: - options = dict(self._agent.provider_options().get(self._agent.provider, {})) + def _resolve_provider_options(self, agent: "Agent", override: dict | None = None) -> dict: + options = dict(agent.provider_options().get(agent.provider, {})) if override: - provider_specific = override.get(self._agent.provider, override) + provider_specific = override.get(agent.provider, override) if isinstance(provider_specific, dict): options.update(provider_specific) return options diff --git a/fastapi_startkit/src/fastapi_startkit/ai/testing.py b/fastapi_startkit/src/fastapi_startkit/ai/testing.py index e3e8533b..d456234c 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/testing.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/testing.py @@ -5,7 +5,6 @@ import hashlib import inspect import json -import re import sys from collections.abc import AsyncIterator from pathlib import Path @@ -18,10 +17,6 @@ from .document import Document -class NoFakeResponse(LookupError): - pass - - def _matches(pattern: str, message: str) -> bool: pattern, message = pattern.lower(), message.lower() if any(ch in pattern for ch in "*?["): @@ -29,12 +24,6 @@ def _matches(pattern: str, message: str) -> bool: return pattern in message -def _reply_text(reply: Any) -> str: - if isinstance(reply, AgentResponse): - return reply.content - return getattr(reply, "content", None) or str(reply) - - class _Recorder: def __init__(self) -> None: self.calls: list[str] = [] @@ -65,33 +54,46 @@ def _joined(value: Any) -> str: return "".join(value) if isinstance(value, list) else value -def _word_chunks(text: str) -> list[str]: - """Split text into word chunks (word + trailing whitespace) so a fake can - mimic a token stream. Loss-less: ``"".join(_word_chunks(t)) == t``.""" - return re.findall(r"\S+\s*", text) or [text] +class AgentModelFake: + """Registers a fixed, ordered list of replies as ``agent_cls``'s chat + model for the duration of a ``with`` block (or a decorated function). + Unlike the old pattern-matching stand-in, this swaps only the model — + ``prompt()``/``stream()`` still run the real message-building, pipeline, + and tool-execution path; see ``Ai.fake()``. + """ -class FakeAgent(_Recorder): - def __init__(self, responses: dict[str, Any] | None = None) -> None: - super().__init__() - self.responses = responses or {} + def __init__(self, agent_cls: type[Agent], responses: list) -> None: + self._agent_cls = agent_cls + self._responses = responses - def _resolve(self, message: str) -> str: - if not self.responses: - return "" - for pattern, reply in self.responses.items(): - if _matches(pattern, message): - return _reply_text(reply) - raise NoFakeResponse(f"No fake response matched message: {message!r}") + def __enter__(self) -> None: + from .model_builder import Ai - async def prompt(self, message: str, attachments: list[Document] | None = None) -> AgentResponse: - self._record_call(message, attachments) - return AgentResponse(content=self._resolve(message)) + Ai.fake(self._agent_cls.__name__, self._responses) - async def stream(self, message: str) -> AsyncIterator[str]: - self._record_call(message, None) - for chunk in _word_chunks(self._resolve(message)): - yield chunk + def __exit__(self, *_exc: Any) -> bool: + from .model_builder import Ai + + Ai.forget(self._agent_cls.__name__) + return False + + def __call__(self, func: Callable) -> Callable: + if inspect.iscoroutinefunction(func): + + @functools.wraps(func) + async def async_wrapper(*args: Any, **kwargs: Any) -> Any: + with self: + return await func(*args, **kwargs) + + return async_wrapper + + @functools.wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Any: + with self: + return func(*args, **kwargs) + + return wrapper class RecordingAgent(_Recorder): diff --git a/fastapi_startkit/tests/ai/test_agent.py b/fastapi_startkit/tests/ai/test_agent.py index 91a2203a..2d63867a 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 ModelBuilder +from fastapi_startkit.ai.model_builder import Ai from fastapi_startkit.ai.response import AgentResponse from fastapi_startkit.application import app @@ -156,15 +156,15 @@ async def test_stream_yields_tool_result_without_calling_model_again(self): self.assertEqual(chunks, ["Python Developer at Shopify"]) def test_resolve_model_falls_back_to_lab_default(self): - self.assertEqual(ModelBuilder(Agent())._resolve_model(), "gemini-2.5-flash-lite") + self.assertEqual(Ai()._resolve_model(Agent()), "gemini-2.5-flash-lite") class AnthropicAgent(Agent): provider = "anthropic" - self.assertEqual(ModelBuilder(AnthropicAgent())._resolve_model(), "claude-sonnet-4-6") + self.assertEqual(Ai()._resolve_model(AnthropicAgent()), "claude-sonnet-4-6") def test_resolve_model_prefers_explicit_override(self): - self.assertEqual(ModelBuilder(Agent())._resolve_model("my-model"), "my-model") + self.assertEqual(Ai()._resolve_model(Agent(), "my-model"), "my-model") def test_instructions_lead_the_message_list(self): messages = JobAssistant()._build_messages("find me a job") diff --git a/fastapi_startkit/tests/ai/test_agent_fake.py b/fastapi_startkit/tests/ai/test_agent_fake.py index 3191a8b3..7b560d74 100644 --- a/fastapi_startkit/tests/ai/test_agent_fake.py +++ b/fastapi_startkit/tests/ai/test_agent_fake.py @@ -1,9 +1,12 @@ """Tests for Agent.fake() / Agent.record() and the assert_prompted/reset helpers. -``Agent.fake()`` binds a canned stand-in into the container for the duration of a -``with`` block. ``Agent.record()`` binds a record-and-replay stand-in: on a cassette -miss it calls the real agent once and caches the response to disk; on a hit it -replays from the cassette without calling the agent again. +``Agent.fake()`` registers a fixed, ordered list of replies as a deterministic +stand-in chat model (see ``Ai.fake()``) for the duration of a ``with`` block — +each call to ``prompt()``/``stream()`` replays the next reply, going through +the real message-building / pipeline / tool-execution path, only the model at +the bottom is swapped. ``Agent.record()`` binds a record-and-replay stand-in: +on a cassette miss it calls the real agent once and caches the response to +disk; on a hit it replays from the cassette without calling the agent again. """ import json @@ -13,6 +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.response import AgentResponse @@ -21,92 +25,78 @@ class SimpleAgent(Agent): class TestAgentFake(unittest.IsolatedAsyncioTestCase): - async def test_fake_with_agent_response_returns_it(self): + def tearDown(self): + Ai.reset_fakes() + + async def test_fake_replays_the_only_response(self): agent = SimpleAgent() - with SimpleAgent.fake({"*": AgentResponse(content="Hello world!")}): + with SimpleAgent.fake(["Hello world!"]): result = await agent.prompt("anything") self.assertEqual(result.content, "Hello world!") - async def test_fake_does_not_call_provider_run(self): - agent = SimpleAgent() - called = [] - - original_run = agent._run - - async def patched_run(*args, **kwargs): - called.append(True) - return await original_run(*args, **kwargs) + async def test_fake_does_not_call_provider_build(self): + import langchain.chat_models as chat_models - agent._run = patched_run + def fail_if_called(*args, **kwargs): + raise AssertionError("init_chat_model must not be called when a fake is registered") - with SimpleAgent.fake({"*": AgentResponse(content="faked")}): - await agent.prompt("hello") - - self.assertEqual(called, [], "_run() must not be called when a fake matches") + patcher = mock.patch.object(chat_models, "init_chat_model", fail_if_called) + patcher.start() + self.addCleanup(patcher.stop) - async def test_fake_with_exact_pattern(self): agent = SimpleAgent() - with SimpleAgent.fake({"hello": AgentResponse(content="matched hello")}): - result = await agent.prompt("hello") - - self.assertEqual(result.content, "matched hello") + with SimpleAgent.fake(["faked"]): + await agent.prompt("hello") - async def test_fake_glob_hello_wildcard(self): + async def test_fake_replays_responses_in_order(self): agent = SimpleAgent() - with SimpleAgent.fake({"*hello*": AgentResponse(content="hi there")}): - result = await agent.prompt("say hello to me") + with SimpleAgent.fake(["first reply", "second reply"]): + first = await agent.prompt("call one") + second = await agent.prompt("call two") - self.assertEqual(result.content, "hi there") + self.assertEqual(first.content, "first reply") + self.assertEqual(second.content, "second reply") - async def test_fake_glob_analyze_wildcard(self): + async def test_fake_raises_once_responses_are_exhausted(self): agent = SimpleAgent() - with SimpleAgent.fake({"*analyze*": AgentResponse(content="analysis done")}): - result = await agent.prompt("please analyze this report") + with SimpleAgent.fake(["only reply"]): + await agent.prompt("call one") - self.assertEqual(result.content, "analysis done") + with self.assertRaises(RuntimeError): + await agent.prompt("call two") - async def test_fake_no_match_raises(self): - agent = SimpleAgent() - with SimpleAgent.fake({"*hello*": AgentResponse(content="hi")}): - with self.assertRaises(Exception): - await agent.prompt("goodbye") + async def test_fake_unregisters_after_the_block_exits(self): + with SimpleAgent.fake(["faked"]): + self.assertTrue(Ai.has_fake_model_for("SimpleAgent")) - async def test_fake_glob_case_insensitive(self): - agent = SimpleAgent() - with SimpleAgent.fake({"*HELLO*": AgentResponse(content="case insensitive")}): - result = await agent.prompt("say hello please") + self.assertFalse(Ai.has_fake_model_for("SimpleAgent")) - self.assertEqual(result.content, "case insensitive") + async def test_fake_as_decorator(self): + @SimpleAgent.fake(["decorated reply"]) + async def run(): + return await SimpleAgent().prompt("call") - async def test_fake_first_matching_pattern_wins(self): - agent = SimpleAgent() - with SimpleAgent.fake( - { - "*hello*": AgentResponse(content="first match"), - "*hello world*": AgentResponse(content="second match"), - } - ): - result = await agent.prompt("hello world") + result = await run() - self.assertEqual(result.content, "first match") + self.assertEqual(result.content, "decorated reply") async def test_assert_prompted_passes_after_one_call(self): agent = SimpleAgent() - with SimpleAgent.fake({"*": AgentResponse(content="ok")}): + with SimpleAgent.fake(["ok"]): await agent.prompt("first") agent.assert_prompted() async def test_assert_prompted_times_2_passes_after_exactly_2_calls(self): agent = SimpleAgent() - with SimpleAgent.fake({"*": AgentResponse(content="ok")}): + with SimpleAgent.fake(["ok", "ok"]): await agent.prompt("first") await agent.prompt("second") agent.assert_prompted(times=2) async def test_assert_prompted_times_fails_when_count_mismatch(self): agent = SimpleAgent() - with SimpleAgent.fake({"*": AgentResponse(content="ok")}): + with SimpleAgent.fake(["ok"]): await agent.prompt("only once") with self.assertRaises(AssertionError): @@ -128,7 +118,7 @@ def test_assert_not_prompted_passes_when_no_calls_made(self): async def test_assert_not_prompted_fails_after_one_call(self): agent = SimpleAgent() - with SimpleAgent.fake({"*": AgentResponse(content="ok")}): + with SimpleAgent.fake(["ok"]): await agent.prompt("a prompt") with self.assertRaises(AssertionError): @@ -136,7 +126,7 @@ async def test_assert_not_prompted_fails_after_one_call(self): async def test_reset_clears_call_log(self): agent = SimpleAgent() - with SimpleAgent.fake({"*": AgentResponse(content="ok")}): + with SimpleAgent.fake(["ok"]): await agent.prompt("first") self.assertEqual(len(agent._call_log), 1) @@ -150,7 +140,7 @@ def test_reset_returns_agent_for_chaining(self): async def test_assert_not_prompted_passes_after_reset(self): agent = SimpleAgent() - with SimpleAgent.fake({"*": AgentResponse(content="ok")}): + with SimpleAgent.fake(["ok"]): await agent.prompt("call before reset") agent.reset() @@ -159,33 +149,31 @@ async def test_assert_not_prompted_passes_after_reset(self): async def test_fake_rebinding_overrides_previous(self): agent = SimpleAgent() - with SimpleAgent.fake({"*": AgentResponse(content="first fake")}): + with SimpleAgent.fake(["first fake"]): self.assertEqual((await agent.prompt("call")).content, "first fake") - with SimpleAgent.fake({"*": AgentResponse(content="second fake")}): + with SimpleAgent.fake(["second fake"]): self.assertEqual((await agent.prompt("call again")).content, "second fake") async def test_stream_returns_fake_response(self): agent = SimpleAgent() - with SimpleAgent.fake({"*hello*": AgentResponse(content="Faked stream!")}): + with SimpleAgent.fake(["Faked stream!"]): chunks = [chunk async for chunk in agent.stream("hello world")] - # A faked stream is split into word chunks but rejoins to the value. self.assertEqual("".join(chunks), "Faked stream!") self.assertGreater(len(chunks), 1) agent.assert_prompted(times=1) - async def test_fake_stream_splits_value_into_word_chunks(self): + async def test_stream_replays_the_registered_text_exactly(self): agent = SimpleAgent() - with SimpleAgent.fake({"*": "Hello there, friend"}): + with SimpleAgent.fake(["Hello there, friend"]): chunks = [chunk async for chunk in agent.stream("hi")] - self.assertEqual(chunks, ["Hello ", "there, ", "friend"]) self.assertEqual("".join(chunks), "Hello there, friend") async def test_stream_records_one_call_not_two(self): agent = SimpleAgent() - with SimpleAgent.fake({"*": AgentResponse(content="x")}): + with SimpleAgent.fake(["x"]): [chunk async for chunk in agent.stream("once")] # Streaming must log exactly one prompt — not one for stream + one for prompt. diff --git a/fastapi_startkit/tests/ai/test_agent_schema.py b/fastapi_startkit/tests/ai/test_agent_schema.py index 0e7d705b..1a30772c 100644 --- a/fastapi_startkit/tests/ai/test_agent_schema.py +++ b/fastapi_startkit/tests/ai/test_agent_schema.py @@ -6,6 +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.response import AgentResponse @@ -20,9 +21,12 @@ def schema(self): class TestAgentSchema(unittest.IsolatedAsyncioTestCase): + def tearDown(self): + Ai.reset_fakes() + async def test_fake_json_is_built_into_the_schema(self): agent = UserAgent() - with UserAgent.fake({"*": '{"id": "u-1", "name": "Alex"}'}): + with UserAgent.fake(['{"id": "u-1", "name": "Alex"}']): response = await agent.prompt("get the user") self.assertIsInstance(response.parsed, User) @@ -32,14 +36,14 @@ async def test_fake_json_is_built_into_the_schema(self): async def test_no_schema_leaves_parsed_none(self): agent = Agent() - with Agent.fake({"*": '{"id": "u-1"}'}): + with Agent.fake(['{"id": "u-1"}']): response = await agent.prompt("anything") self.assertIsNone(response.parsed) async def test_invalid_json_for_schema_raises(self): agent = UserAgent() - with UserAgent.fake({"*": '{"name": "no id here"}'}): + with UserAgent.fake(['{"name": "no id here"}']): with self.assertRaises(Exception): await agent.prompt("get the user") diff --git a/fastapi_startkit/tests/ai/test_ai_fake.py b/fastapi_startkit/tests/ai/test_ai_fake.py new file mode 100644 index 00000000..30b2e39e --- /dev/null +++ b/fastapi_startkit/tests/ai/test_ai_fake.py @@ -0,0 +1,142 @@ +"""Tests for Ai's fake-model registry. + +Ai.fake() swaps the chat model a given agent (by class name or instance) +resolves to for a deterministic GenericFakeChatModel that replays a fixed +list of message turns — no live LLM call, no network access. +Ai().get_model_for(agent) is what Agent._build_model() calls: it returns +the registered fake when one exists, otherwise it builds a real provider +model exactly as Ai.build() always has. +""" + +import unittest +from unittest import mock + +import langchain.chat_models as chat_models +from langchain_core.messages import AIMessage, ToolCall +from langchain_core.tools import tool + +from fastapi_startkit.ai.agent import Agent +from fastapi_startkit.ai.model_builder import Ai +from fastapi_startkit.application import app + + +@tool +def search_jobs(query: str) -> str: + """Search the job board for roles matching the query.""" + return "Python Developer at Shopify" + + +class JobAssistant(Agent): + def tools(self): + return [search_jobs] + + +class SimpleAgent(Agent): + pass + + +class TestAiFakeBase(unittest.IsolatedAsyncioTestCase): + def setUp(self): + from fastapi_startkit.ai import AIConfig + + container = app() + container.bind("ai", AIConfig()) + container.make("config").set("ai", AIConfig()) + self.addCleanup(Ai.reset_fakes) + + +class TestAiFakeRegistration(TestAiFakeBase): + def test_fake_registers_a_model_for_an_agent_class_name(self): + Ai.fake("SimpleAgent", [AIMessage(content="hi")]) + + self.assertTrue(Ai.has_fake_model_for("SimpleAgent")) + + def test_fake_accepts_an_agent_instance_keyed_by_its_class_name(self): + Ai.fake(SimpleAgent(), [AIMessage(content="hi")]) + + self.assertTrue(Ai.has_fake_model_for(SimpleAgent())) + self.assertTrue(Ai.has_fake_model_for("SimpleAgent")) + + def test_has_fake_model_for_is_false_when_nothing_registered(self): + self.assertFalse(Ai.has_fake_model_for("SimpleAgent")) + + def test_fake_coerces_plain_strings_into_ai_messages(self): + model = Ai.fake("SimpleAgent", ["plain text reply"]) + + result = model.invoke([]) + + self.assertEqual(result.content, "plain text reply") + + def test_fake_returns_the_registered_chat_model(self): + model = Ai.fake("SimpleAgent", [AIMessage(content="hi")]) + + self.assertIs(Ai.get_fake_model_for("SimpleAgent"), model) + + def test_forget_removes_a_single_registration(self): + Ai.fake("SimpleAgent", [AIMessage(content="hi")]) + Ai.fake("JobAssistant", [AIMessage(content="hi")]) + + Ai.forget("SimpleAgent") + + self.assertFalse(Ai.has_fake_model_for("SimpleAgent")) + self.assertTrue(Ai.has_fake_model_for("JobAssistant")) + + def test_forget_is_a_no_op_when_nothing_registered(self): + Ai.forget("SimpleAgent") + + self.assertFalse(Ai.has_fake_model_for("SimpleAgent")) + + +class TestAiGetModelFor(TestAiFakeBase): + def test_returns_registered_fake_without_building_a_real_model(self): + fake_model = Ai.fake("SimpleAgent", [AIMessage(content="faked")]) + + def fail_if_called(*args, **kwargs): + raise AssertionError("init_chat_model must not be called when a fake is registered") + + patcher = mock.patch.object(chat_models, "init_chat_model", fail_if_called) + patcher.start() + self.addCleanup(patcher.stop) + + resolved = Ai().get_model_for(SimpleAgent()) + + self.assertIs(resolved, fake_model) + + def test_falls_back_to_build_when_no_fake_is_registered(self): + sentinel = object() + patcher = mock.patch.object(chat_models, "init_chat_model", lambda *a, **k: sentinel) + patcher.start() + self.addCleanup(patcher.stop) + + resolved = Ai().get_model_for(SimpleAgent()) + + self.assertIs(resolved, sentinel) + + +class TestAgentPromptUsesFakeModelEndToEnd(TestAiFakeBase): + async def test_prompt_replays_the_registered_fake_model_reply(self): + Ai.fake("SimpleAgent", [AIMessage(content="faked via ai")]) + + result = await SimpleAgent().prompt("hi there") + + self.assertEqual(result.content, "faked via ai") + + async def test_prompt_runs_a_faked_tool_call_end_to_end(self): + Ai.fake( + "JobAssistant", + [ + AIMessage( + content="", + tool_calls=[ToolCall(name="search_jobs", args={"query": "python"}, id="c1", type="tool_call")], + ) + ], + ) + + result = await JobAssistant().prompt("find me a python job") + + self.assertEqual(result.content, "Python Developer at Shopify") + + async def test_registering_a_fake_does_not_affect_other_agent_classes(self): + Ai.fake("SimpleAgent", [AIMessage(content="only for SimpleAgent")]) + + self.assertFalse(Ai.has_fake_model_for("JobAssistant")) diff --git a/fastapi_startkit/tests/ai/test_evals.py b/fastapi_startkit/tests/ai/test_evals.py new file mode 100644 index 00000000..5e7e2ccc --- /dev/null +++ b/fastapi_startkit/tests/ai/test_evals.py @@ -0,0 +1,281 @@ +"""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) diff --git a/fastapi_startkit/uv.lock b/fastapi_startkit/uv.lock index 5808c35b..c9ea4cba 100644 --- a/fastapi_startkit/uv.lock +++ b/fastapi_startkit/uv.lock @@ -527,7 +527,7 @@ wheels = [ [[package]] name = "fastapi-startkit" -version = "0.48.0" +version = "0.50.0" source = { editable = "." } dependencies = [ { name = "cleo" },