From 6e357c547d115336b29b60277960299a927b7777 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Thu, 16 Jul 2026 15:47:45 -0700 Subject: [PATCH 1/7] feat(ai): add TestJudgeAgent trajectory-match evaluator + ModelBuilder fake registry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a deterministic evals harness for testing AI agents, modeled on LangChain's agentevals create_trajectory_match_evaluator. TestJudgeAgent(trajectory_match_mode=...).record() yields a callable evaluator that compares an actual message trajectory against a reference trajectory with no live LLM call — pure structural comparison, typically driven by Agent.fake() output: with TestJudgeAgent(trajectory_match_mode="strict").record() as evaluator: evaluation = evaluator(outputs=response["messages"], reference_outputs=reference) assert evaluation["score"] is True Supported modes mirror agentevals semantics: - strict: same messages in order, same role and tool calls per position (message content is not compared) - unordered: same set of tool calls, any order - subset: output tool calls all appear in the reference - superset: output tool calls cover all reference tool calls (extras ok) The evaluator returns a subscriptable result with key/score/comment, following the LangSmith evaluator-result convention. Also add a model-level fake registry to ModelBuilder: ModelBuilder.fake() registers a GenericFakeChatModel for an agent (by class name or instance), and get_model_for() returns it when present, otherwise builds a real provider model as before. Agent._build_model() now routes through get_model_for(), so a faked agent runs the same message-building / pipeline / tool-execution path as a real one with only the underlying model swapped. The existing Agent.fake()/FakeAgent/AgentBinding path is unchanged. --- .../src/fastapi_startkit/ai/__init__.py | 4 + .../src/fastapi_startkit/ai/agent.py | 2 +- .../src/fastapi_startkit/ai/evals.py | 195 ++++++++++++ .../src/fastapi_startkit/ai/model_builder.py | 49 +++ fastapi_startkit/tests/ai/test_evals.py | 281 ++++++++++++++++++ .../tests/ai/test_model_builder_fake.py | 128 ++++++++ 6 files changed, 658 insertions(+), 1 deletion(-) create mode 100644 fastapi_startkit/src/fastapi_startkit/ai/evals.py create mode 100644 fastapi_startkit/tests/ai/test_evals.py create mode 100644 fastapi_startkit/tests/ai/test_model_builder_fake.py diff --git a/fastapi_startkit/src/fastapi_startkit/ai/__init__.py b/fastapi_startkit/src/fastapi_startkit/ai/__init__.py index dbf4c85e..7b105325 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/__init__.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/__init__.py @@ -6,6 +6,7 @@ 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 @@ -36,6 +37,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..ac88e1ec 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/agent.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/agent.py @@ -225,7 +225,7 @@ def _build_messages( def _build_model(self, model: str | None = None, provider_options: dict | None = None) -> Any: from .model_builder import ModelBuilder # noqa: PLC0415 - return ModelBuilder(agent=self).build(model, provider_options) + return ModelBuilder(agent=self).get_model_for(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..079c5993 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/model_builder.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/model_builder.py @@ -9,9 +9,58 @@ class ModelBuilder: + # 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_models: dict[str, Any] = {} + # Reserved for response-level fakes (mirroring fake_models, but for + # cached final replies rather than whole chat models). Not yet wired up. + fake_responses: dict[str, Any] = {} + def __init__(self, agent: "Agent") -> None: self._agent = agent + @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_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_models + + @classmethod + def get_fake_model_for(cls, agent: "Agent | str") -> Any: + return cls.fake_models[cls._key(agent)] + + @classmethod + def reset_fakes(cls) -> None: + cls.fake_models.clear() + cls.fake_responses.clear() + + def get_model_for(self, model: str | None = None, provider_options: dict | None = None) -> Any: + """Resolve the model to run: a registered fake if one exists for + this builder's agent, otherwise a freshly-built provider model.""" + if self.has_fake_model_for(self._agent): + return self.get_fake_model_for(self._agent) + return self.build(model, provider_options) + def build(self, model: str | None = None, provider_options: dict | None = None) -> Any: from langchain.chat_models import init_chat_model # noqa: PLC0415 diff --git a/fastapi_startkit/tests/ai/test_evals.py b/fastapi_startkit/tests/ai/test_evals.py new file mode 100644 index 00000000..f0afe04c --- /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({"*weather*": AgentResponse(content="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/tests/ai/test_model_builder_fake.py b/fastapi_startkit/tests/ai/test_model_builder_fake.py new file mode 100644 index 00000000..6201527a --- /dev/null +++ b/fastapi_startkit/tests/ai/test_model_builder_fake.py @@ -0,0 +1,128 @@ +"""Tests for ModelBuilder's fake-model registry. + +ModelBuilder.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. +ModelBuilder(agent=...).get_model_for() is what Agent._build_model() calls: +it returns the registered fake when one exists, otherwise it builds a real +provider model exactly as ModelBuilder.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 ModelBuilder +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 TestModelBuilderFakeBase(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(ModelBuilder.reset_fakes) + + +class TestModelBuilderFakeRegistration(TestModelBuilderFakeBase): + def test_fake_registers_a_model_for_an_agent_class_name(self): + ModelBuilder.fake("SimpleAgent", [AIMessage(content="hi")]) + + self.assertTrue(ModelBuilder.has_fake_model_for("SimpleAgent")) + + def test_fake_accepts_an_agent_instance_keyed_by_its_class_name(self): + ModelBuilder.fake(SimpleAgent(), [AIMessage(content="hi")]) + + self.assertTrue(ModelBuilder.has_fake_model_for(SimpleAgent())) + self.assertTrue(ModelBuilder.has_fake_model_for("SimpleAgent")) + + def test_has_fake_model_for_is_false_when_nothing_registered(self): + self.assertFalse(ModelBuilder.has_fake_model_for("SimpleAgent")) + + def test_fake_coerces_plain_strings_into_ai_messages(self): + model = ModelBuilder.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 = ModelBuilder.fake("SimpleAgent", [AIMessage(content="hi")]) + + self.assertIs(ModelBuilder.get_fake_model_for("SimpleAgent"), model) + + +class TestModelBuilderGetModelFor(TestModelBuilderFakeBase): + def test_returns_registered_fake_without_building_a_real_model(self): + fake_model = ModelBuilder.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 = ModelBuilder(agent=SimpleAgent()).get_model_for() + + 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 = ModelBuilder(agent=SimpleAgent()).get_model_for() + + self.assertIs(resolved, sentinel) + + +class TestAgentPromptUsesFakeModelEndToEnd(TestModelBuilderFakeBase): + async def test_prompt_replays_the_registered_fake_model_reply(self): + ModelBuilder.fake("SimpleAgent", [AIMessage(content="faked via model builder")]) + + result = await SimpleAgent().prompt("hi there") + + self.assertEqual(result.content, "faked via model builder") + + async def test_prompt_runs_a_faked_tool_call_end_to_end(self): + ModelBuilder.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): + ModelBuilder.fake("SimpleAgent", [AIMessage(content="only for SimpleAgent")]) + + self.assertFalse(ModelBuilder.has_fake_model_for("JobAssistant")) From e29efc9d7cd51d19512dfed50271a98c9df09bc1 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Thu, 16 Jul 2026 16:20:39 -0700 Subject: [PATCH 2/7] refactor(ai): rename ModelBuilder to Ai, rebuild fake()/record() around model-level fakes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ai (formerly ModelBuilder) drops the per-instance agent binding: fake_agent_models/ fake_agent_responses are class-level registries keyed by agent class name, and get_model_for()/build() take the agent as an argument instead of storing it in the constructor. Agent.fake() now registers a fixed, ordered list of replies as a deterministic chat model (via Ai.fake()) instead of binding a whole FakeAgent stand-in into the container. Replies flow through the real message-building/pipeline/tool- execution path, so faked tool calls actually execute — only the model at the bottom is swapped. Agent.record()/RecordingAgent/AgentBinding are unchanged. Drops the dead _match_fake()/self._fakes machinery on Agent (never populated by any code path) along with the now-unreachable FakeAgent/NoFakeResponse testing helpers. --- .../tests/features/test_chat_controller.py | 6 +- .../src/fastapi_startkit/ai/__init__.py | 7 +- .../src/fastapi_startkit/ai/agent.py | 46 ++----- .../src/fastapi_startkit/ai/model_builder.py | 68 +++++----- .../src/fastapi_startkit/ai/testing.py | 68 +++++----- fastapi_startkit/tests/ai/test_agent.py | 8 +- fastapi_startkit/tests/ai/test_agent_fake.py | 122 ++++++++---------- .../tests/ai/test_agent_schema.py | 10 +- ..._model_builder_fake.py => test_ai_fake.py} | 76 ++++++----- fastapi_startkit/tests/ai/test_evals.py | 2 +- fastapi_startkit/uv.lock | 2 +- 11 files changed, 199 insertions(+), 216 deletions(-) rename fastapi_startkit/tests/ai/{test_model_builder_fake.py => test_ai_fake.py} (53%) 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 7b105325..790ef065 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/__init__.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/__init__.py @@ -10,21 +10,22 @@ 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", diff --git a/fastapi_startkit/src/fastapi_startkit/ai/agent.py b/fastapi_startkit/src/fastapi_startkit/ai/agent.py index ac88e1ec..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).get_model_for(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/model_builder.py b/fastapi_startkit/src/fastapi_startkit/ai/model_builder.py index 079c5993..5b65a237 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/model_builder.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/model_builder.py @@ -8,19 +8,19 @@ from .agent import Agent -class ModelBuilder: +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_models: dict[str, Any] = {} - # Reserved for response-level fakes (mirroring fake_models, but for + 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_responses: dict[str, Any] = {} + fake_agent_responses: dict[str, Any] = {} - def __init__(self, agent: "Agent") -> None: - self._agent = agent + def __init__(self) -> None: + pass @staticmethod def _key(agent: "Agent | str") -> str: @@ -38,59 +38,63 @@ def fake(cls, agent: "Agent | str", messages: list) -> Any: turns = [message if hasattr(message, "content") else AIMessage(content=str(message)) for message in messages] model = GenericFakeChatModel(messages=iter(turns)) - cls.fake_models[cls._key(agent)] = model + 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_models + return cls._key(agent) in cls.fake_agent_models @classmethod def get_fake_model_for(cls, agent: "Agent | str") -> Any: - return cls.fake_models[cls._key(agent)] + 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_models.clear() - cls.fake_responses.clear() + cls.fake_agent_models.clear() + cls.fake_agent_responses.clear() - def get_model_for(self, model: str | None = None, provider_options: dict | None = None) -> Any: + 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 - this builder's agent, otherwise a freshly-built provider model.""" - if self.has_fake_model_for(self._agent): - return self.get_fake_model_for(self._agent) - return self.build(model, provider_options) + ``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, model: str | None = None, provider_options: dict | None = None) -> Any: + 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_model_builder_fake.py b/fastapi_startkit/tests/ai/test_ai_fake.py similarity index 53% rename from fastapi_startkit/tests/ai/test_model_builder_fake.py rename to fastapi_startkit/tests/ai/test_ai_fake.py index 6201527a..30b2e39e 100644 --- a/fastapi_startkit/tests/ai/test_model_builder_fake.py +++ b/fastapi_startkit/tests/ai/test_ai_fake.py @@ -1,11 +1,11 @@ -"""Tests for ModelBuilder's fake-model registry. - -ModelBuilder.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. -ModelBuilder(agent=...).get_model_for() is what Agent._build_model() calls: -it returns the registered fake when one exists, otherwise it builds a real -provider model exactly as ModelBuilder.build() always has. +"""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 @@ -16,7 +16,7 @@ from langchain_core.tools import tool 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.application import app @@ -35,47 +35,61 @@ class SimpleAgent(Agent): pass -class TestModelBuilderFakeBase(unittest.IsolatedAsyncioTestCase): +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(ModelBuilder.reset_fakes) + self.addCleanup(Ai.reset_fakes) -class TestModelBuilderFakeRegistration(TestModelBuilderFakeBase): +class TestAiFakeRegistration(TestAiFakeBase): def test_fake_registers_a_model_for_an_agent_class_name(self): - ModelBuilder.fake("SimpleAgent", [AIMessage(content="hi")]) + Ai.fake("SimpleAgent", [AIMessage(content="hi")]) - self.assertTrue(ModelBuilder.has_fake_model_for("SimpleAgent")) + self.assertTrue(Ai.has_fake_model_for("SimpleAgent")) def test_fake_accepts_an_agent_instance_keyed_by_its_class_name(self): - ModelBuilder.fake(SimpleAgent(), [AIMessage(content="hi")]) + Ai.fake(SimpleAgent(), [AIMessage(content="hi")]) - self.assertTrue(ModelBuilder.has_fake_model_for(SimpleAgent())) - self.assertTrue(ModelBuilder.has_fake_model_for("SimpleAgent")) + 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(ModelBuilder.has_fake_model_for("SimpleAgent")) + self.assertFalse(Ai.has_fake_model_for("SimpleAgent")) def test_fake_coerces_plain_strings_into_ai_messages(self): - model = ModelBuilder.fake("SimpleAgent", ["plain text reply"]) + 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 = ModelBuilder.fake("SimpleAgent", [AIMessage(content="hi")]) + model = Ai.fake("SimpleAgent", [AIMessage(content="hi")]) - self.assertIs(ModelBuilder.get_fake_model_for("SimpleAgent"), model) + 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")]) -class TestModelBuilderGetModelFor(TestModelBuilderFakeBase): + 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 = ModelBuilder.fake("SimpleAgent", [AIMessage(content="faked")]) + 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") @@ -84,7 +98,7 @@ def fail_if_called(*args, **kwargs): patcher.start() self.addCleanup(patcher.stop) - resolved = ModelBuilder(agent=SimpleAgent()).get_model_for() + resolved = Ai().get_model_for(SimpleAgent()) self.assertIs(resolved, fake_model) @@ -94,21 +108,21 @@ def test_falls_back_to_build_when_no_fake_is_registered(self): patcher.start() self.addCleanup(patcher.stop) - resolved = ModelBuilder(agent=SimpleAgent()).get_model_for() + resolved = Ai().get_model_for(SimpleAgent()) self.assertIs(resolved, sentinel) -class TestAgentPromptUsesFakeModelEndToEnd(TestModelBuilderFakeBase): +class TestAgentPromptUsesFakeModelEndToEnd(TestAiFakeBase): async def test_prompt_replays_the_registered_fake_model_reply(self): - ModelBuilder.fake("SimpleAgent", [AIMessage(content="faked via model builder")]) + Ai.fake("SimpleAgent", [AIMessage(content="faked via ai")]) result = await SimpleAgent().prompt("hi there") - self.assertEqual(result.content, "faked via model builder") + self.assertEqual(result.content, "faked via ai") async def test_prompt_runs_a_faked_tool_call_end_to_end(self): - ModelBuilder.fake( + Ai.fake( "JobAssistant", [ AIMessage( @@ -123,6 +137,6 @@ async def test_prompt_runs_a_faked_tool_call_end_to_end(self): self.assertEqual(result.content, "Python Developer at Shopify") async def test_registering_a_fake_does_not_affect_other_agent_classes(self): - ModelBuilder.fake("SimpleAgent", [AIMessage(content="only for SimpleAgent")]) + Ai.fake("SimpleAgent", [AIMessage(content="only for SimpleAgent")]) - self.assertFalse(ModelBuilder.has_fake_model_for("JobAssistant")) + 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 index f0afe04c..5e7e2ccc 100644 --- a/fastapi_startkit/tests/ai/test_evals.py +++ b/fastapi_startkit/tests/ai/test_evals.py @@ -233,7 +233,7 @@ class TestJudgeAgentWithAgentFakeOutput(unittest.TestCase): async def _fake_response(self) -> AgentResponse: agent = SimpleAgent() - with SimpleAgent.fake({"*weather*": AgentResponse(content="It is sunny in San Francisco.")}): + 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): 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" }, From 767a9adc1ea046cbb0d51e406b46b6dd9aea479e Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Thu, 16 Jul 2026 17:32:47 -0700 Subject: [PATCH 3/7] refactor(ai): rename model_builder.py to ai.py --- fastapi_startkit/src/fastapi_startkit/ai/__init__.py | 2 +- fastapi_startkit/src/fastapi_startkit/ai/agent.py | 2 +- .../src/fastapi_startkit/ai/{model_builder.py => ai.py} | 0 fastapi_startkit/src/fastapi_startkit/ai/testing.py | 4 ++-- fastapi_startkit/tests/ai/test_agent.py | 2 +- fastapi_startkit/tests/ai/test_agent_fake.py | 2 +- fastapi_startkit/tests/ai/test_agent_schema.py | 2 +- fastapi_startkit/tests/ai/test_ai_fake.py | 2 +- 8 files changed, 8 insertions(+), 8 deletions(-) rename fastapi_startkit/src/fastapi_startkit/ai/{model_builder.py => ai.py} (100%) diff --git a/fastapi_startkit/src/fastapi_startkit/ai/__init__.py b/fastapi_startkit/src/fastapi_startkit/ai/__init__.py index 790ef065..ab0aedac 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/__init__.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/__init__.py @@ -10,7 +10,7 @@ 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 diff --git a/fastapi_startkit/src/fastapi_startkit/ai/agent.py b/fastapi_startkit/src/fastapi_startkit/ai/agent.py index e67a89ac..15a2fb37 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/agent.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/agent.py @@ -193,7 +193,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 100% rename from fastapi_startkit/src/fastapi_startkit/ai/model_builder.py rename to fastapi_startkit/src/fastapi_startkit/ai/ai.py diff --git a/fastapi_startkit/src/fastapi_startkit/ai/testing.py b/fastapi_startkit/src/fastapi_startkit/ai/testing.py index d456234c..2c5ecdee 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/testing.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/testing.py @@ -68,12 +68,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 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..0b7eab7c 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 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 From 54fa7ac8c09c1ede2483ea9c52b90361d231961c Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Thu, 16 Jul 2026 17:52:47 -0700 Subject: [PATCH 4/7] feat(ai): fluent Agent.record() testing DSL Bind record() with `as agent` to get a synchronous prompt() plus assert_text_response(), assert_tool_called()/assert_tool_not_called(), assert_response_time_lt(), and assert_response_judged() (LLM-as-judge, verdict cached in the cassette) that all judge the most recent turn. record(cassette, messages=...) seeds a session's prior history; cassette keys now fold in the conversation so far (not just the literal message text) so two sessions with different histories but the same follow-up text don't collide. The pre-existing bare context-manager usage from task #327 is unchanged. Removes TestJudgeAgent/evals.py: its deterministic structural trajectory matching is unused anywhere outside its own tests, and this DSL now covers response judging via assert_response_judged. --- .../tests/units/agents/test_router_agent.py | 30 ++ .../src/fastapi_startkit/ai/__init__.py | 7 +- .../src/fastapi_startkit/ai/agent.py | 6 +- .../src/fastapi_startkit/ai/evals.py | 195 ---------- .../src/fastapi_startkit/ai/testing.py | 165 +++++++- fastapi_startkit/tests/ai/test_agent_fake.py | 3 +- .../tests/ai/test_agent_record_fluent.py | 354 ++++++++++++++++++ fastapi_startkit/tests/ai/test_evals.py | 281 -------------- 8 files changed, 549 insertions(+), 492 deletions(-) create mode 100644 example/agents/tests/units/agents/test_router_agent.py delete mode 100644 fastapi_startkit/src/fastapi_startkit/ai/evals.py create mode 100644 fastapi_startkit/tests/ai/test_agent_record_fluent.py delete mode 100644 fastapi_startkit/tests/ai/test_evals.py 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..1564e90a --- /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: + def test_the_router_agent(self): + with RouterAgent.record("record_stream.json") as agent: + 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) + + agent.prompt("suggest python developer jobs") + agent.assert_tool_called("job_search_tool", lambda tool: tool.name == "job_search_tool") + + 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: + 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 ab0aedac..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 .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 15a2fb37..44c6868e 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/agent.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/agent.py @@ -51,7 +51,7 @@ async def prompt( ) -> AgentResponse: stand_in = self._faked() if stand_in is not None: - response = await stand_in.prompt(message, attachments=attachments) + response = await stand_in._prompt_async(message, attachments=attachments) self._log_call("prompt", message) return self._apply_schema(response) @@ -91,10 +91,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: 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 2c5ecdee..bdca65ae 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/testing.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/testing.py @@ -1,11 +1,14 @@ from __future__ import annotations +import asyncio import fnmatch import functools 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 @@ -96,16 +99,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 synchronous (wraps the real, async agent call) 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 +169,42 @@ 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_async(self, message: str, attachments: list[Document] | None = None) -> AgentResponse: self._record_call(message, attachments) cassette, store = self._load() key = self._key(message, attachments) 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._remember_turn(message, response) + return response + + def prompt(self, message: str, *, attachments: list[Document] | None = None) -> AgentResponse: + """Synchronous fluent entry point: run (or replay) one turn and make + it the "current" response that assert_*() methods judge.""" + start = time.monotonic() + response = asyncio.run(self._prompt_async(message, attachments)) + self.last_elapsed = time.monotonic() - start + self.last_response = response return response async def stream(self, message: str) -> AsyncIterator[str]: @@ -142,6 +221,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_fake.py b/fastapi_startkit/tests/ai/test_agent_fake.py index 0b7eab7c..42ff7adb 100644 --- a/fastapi_startkit/tests/ai/test_agent_fake.py +++ b/fastapi_startkit/tests/ai/test_agent_fake.py @@ -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..3683d376 --- /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 +with a synchronous ``prompt()`` and assertion methods that 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: + agent.prompt("hello") + agent.assert_text_response() + agent.assert_tool_not_called(["job_search_tool"]) + agent.assert_response_time_lt(5) + + 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.TestCase): + 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) + + def test_prompt_is_synchronous_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 = agent.prompt("hi") + + self.assertIsInstance(response, AgentResponse) + self.assertEqual(response.content, "Hello there!") + + 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: + agent.prompt("hello") + agent.assert_text_response() + + agent.prompt("suggest python developer jobs") + agent.assert_tool_called("job_search_tool") + + 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: + agent.prompt("find jobs") + + # No queued responses left — this must replay from cassette, not call _run again. + with SimpleAgent.record(cassette) as agent: + agent.prompt("find jobs") + agent.assert_tool_called("job_search_tool") + + +class TestAssertTextResponse(unittest.TestCase): + 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) + + 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: + agent.prompt("hi") + agent.assert_text_response() + + 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: + agent.prompt("hi") + with self.assertRaises(AssertionError): + agent.assert_text_response() + + 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.TestCase): + 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) + + 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: + agent.prompt("find jobs") + agent.assert_tool_called("job_search_tool") + + 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: + agent.prompt("hi") + with self.assertRaises(AssertionError): + agent.assert_tool_called("job_search_tool") + + 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: + agent.prompt("find jobs") + agent.assert_tool_called("job_search_tool", lambda tool: tool.name == "job_search_tool") + + 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: + 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.TestCase): + 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) + + 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: + agent.prompt("hi") + agent.assert_tool_not_called(["job_search_tool"]) + + 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: + agent.prompt("find jobs") + with self.assertRaises(AssertionError): + agent.assert_tool_not_called(["job_search_tool"]) + + +class TestAssertResponseTimeLt(unittest.TestCase): + 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) + + 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: + agent.prompt("hi") + agent.assert_response_time_lt(5) + + 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: + agent.prompt("hi") + with self.assertRaises(AssertionError): + agent.assert_response_time_lt(0) + + 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.TestCase): + 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"}) + + 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 = 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 = 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.TestCase): + 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) + + 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: + agent.prompt("hello") + agent.assert_response_judged( + model="gpt-3.5-turbo", expectation="The llm should respond with greetings" + ) + + 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: + agent.prompt("hello") + with self.assertRaises(AssertionError): + agent.assert_response_judged( + model="gpt-3.5-turbo", expectation="The llm should respond with greetings" + ) + + 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: + 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() + + 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: + 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: + agent.prompt("hello") + agent.assert_response_judged(model="gpt-3.5-turbo", expectation="greet") + + judge.assert_not_called() + + 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_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) From 6195956f005195ab999fab53e4d4f93c318d875d Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Thu, 16 Jul 2026 18:32:17 -0700 Subject: [PATCH 5/7] fix(ai): fluent record() prompt() is async, not sync MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RecordingAgent.prompt() no longer wraps the real async call with asyncio.run() — it's now the same async method the bare context-manager path already awaited internally. Wrapping every call in a fresh event loop was unnecessary and would break inside any already-running loop. Fluent tests now use IsolatedAsyncioTestCase + await agent.prompt(...). --- .../tests/units/agents/test_router_agent.py | 10 +- .../src/fastapi_startkit/ai/agent.py | 2 +- .../src/fastapi_startkit/ai/testing.py | 23 ++-- .../tests/ai/test_agent_record_fluent.py | 112 +++++++++--------- 4 files changed, 71 insertions(+), 76 deletions(-) diff --git a/example/agents/tests/units/agents/test_router_agent.py b/example/agents/tests/units/agents/test_router_agent.py index 1564e90a..26ddfae7 100644 --- a/example/agents/tests/units/agents/test_router_agent.py +++ b/example/agents/tests/units/agents/test_router_agent.py @@ -4,9 +4,9 @@ class TestRouterAgent: - def test_the_router_agent(self): + async def test_the_router_agent(self): with RouterAgent.record("record_stream.json") as agent: - agent.prompt("hello") + await agent.prompt("hello") agent.assert_text_response() agent.assert_tool_not_called(["job_search_tool"]) agent.assert_response_judged( @@ -15,10 +15,10 @@ def test_the_router_agent(self): ) agent.assert_response_time_lt(5) - agent.prompt("suggest python developer jobs") + await agent.prompt("suggest python developer jobs") agent.assert_tool_called("job_search_tool", lambda tool: tool.name == "job_search_tool") - def test_the_router_with_initial_messages(self): + async def test_the_router_with_initial_messages(self): with RouterAgent.record( "record_stream.json", messages=[ @@ -26,5 +26,5 @@ def test_the_router_with_initial_messages(self): AIMessage(content="Hello, How can I help you?"), ], ) as agent: - agent.prompt("suggest python developer jobs") + 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/agent.py b/fastapi_startkit/src/fastapi_startkit/ai/agent.py index 44c6868e..b68297ac 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/agent.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/agent.py @@ -51,7 +51,7 @@ async def prompt( ) -> AgentResponse: stand_in = self._faked() if stand_in is not None: - response = await stand_in._prompt_async(message, attachments=attachments) + response = await stand_in.prompt(message, attachments=attachments) self._log_call("prompt", message) return self._apply_schema(response) diff --git a/fastapi_startkit/src/fastapi_startkit/ai/testing.py b/fastapi_startkit/src/fastapi_startkit/ai/testing.py index bdca65ae..639fda71 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/testing.py +++ b/fastapi_startkit/src/fastapi_startkit/ai/testing.py @@ -1,6 +1,5 @@ from __future__ import annotations -import asyncio import fnmatch import functools import hashlib @@ -117,10 +116,10 @@ class RecordingAgent(_Recorder): """Bound as ``agent`` by ``with Agent.record(cassette) as agent:``. Fluent testing handle around a record-and-replay session: ``prompt()`` - is synchronous (wraps the real, async agent call) 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. + 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 @@ -186,25 +185,21 @@ def _remember_turn(self, message: str, response: AgentResponse) -> None: turn["tool_calls"] = response.tool_calls self._transcript.append(turn) - async def _prompt_async(self, message: str, attachments: list[Document] | None = None) -> AgentResponse: + 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: 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._remember_turn(message, response) - return response - - def prompt(self, message: str, *, attachments: list[Document] | None = None) -> AgentResponse: - """Synchronous fluent entry point: run (or replay) one turn and make - it the "current" response that assert_*() methods judge.""" - start = time.monotonic() - response = asyncio.run(self._prompt_async(message, attachments)) 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]: diff --git a/fastapi_startkit/tests/ai/test_agent_record_fluent.py b/fastapi_startkit/tests/ai/test_agent_record_fluent.py index 3683d376..30d12989 100644 --- a/fastapi_startkit/tests/ai/test_agent_record_fluent.py +++ b/fastapi_startkit/tests/ai/test_agent_record_fluent.py @@ -1,17 +1,17 @@ """Tests for the fluent Agent.record() testing DSL. ``with Agent.record(cassette) as agent:`` binds a ``RecordingAgent`` handle -with a synchronous ``prompt()`` and assertion methods that judge the most -recent turn — mirroring how a browser-testing ``page`` object exposes -assertions against current page state: +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: - agent.prompt("hello") + await agent.prompt("hello") agent.assert_text_response() agent.assert_tool_not_called(["job_search_tool"]) agent.assert_response_time_lt(5) - agent.prompt("suggest python developer jobs") + await agent.prompt("suggest python developer jobs") agent.assert_tool_called("job_search_tool", lambda tool: tool.name == "job_search_tool") """ @@ -36,7 +36,7 @@ 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.TestCase): +class TestFluentPromptMechanics(unittest.IsolatedAsyncioTestCase): def setup_agent(self, responses: list): """responses: list of (content, tool_calls) tuples, consumed in call order.""" queue = list(responses) @@ -49,39 +49,39 @@ async def fake_run(agent_self, message, **kwargs): patcher.start() self.addCleanup(patcher.stop) - def test_prompt_is_synchronous_and_returns_agent_response(self): + 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 = agent.prompt("hi") + response = await agent.prompt("hi") self.assertIsInstance(response, AgentResponse) self.assertEqual(response.content, "Hello there!") - def test_second_prompt_continues_the_same_session(self): + 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: - agent.prompt("hello") + await agent.prompt("hello") agent.assert_text_response() - agent.prompt("suggest python developer jobs") + await agent.prompt("suggest python developer jobs") agent.assert_tool_called("job_search_tool") - def test_replaying_from_cassette_preserves_tool_calls(self): + 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: - agent.prompt("find jobs") + await agent.prompt("find jobs") # No queued responses left — this must replay from cassette, not call _run again. with SimpleAgent.record(cassette) as agent: - agent.prompt("find jobs") + await agent.prompt("find jobs") agent.assert_tool_called("job_search_tool") -class TestAssertTextResponse(unittest.TestCase): +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 []) @@ -90,29 +90,29 @@ async def fake_run(agent_self, message, **kwargs): patcher.start() self.addCleanup(patcher.stop) - def test_passes_when_content_present(self): + 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: - agent.prompt("hi") + await agent.prompt("hi") agent.assert_text_response() - def test_fails_on_empty_content(self): + 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: - agent.prompt("hi") + await agent.prompt("hi") with self.assertRaises(AssertionError): agent.assert_text_response() - def test_fails_when_no_prompt_has_been_made(self): + 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.TestCase): +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) @@ -121,38 +121,38 @@ async def fake_run(agent_self, message, **kwargs): patcher.start() self.addCleanup(patcher.stop) - def test_passes_when_tool_present(self): + 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: - agent.prompt("find jobs") + await agent.prompt("find jobs") agent.assert_tool_called("job_search_tool") - def test_fails_when_tool_absent(self): + 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: - agent.prompt("hi") + await agent.prompt("hi") with self.assertRaises(AssertionError): agent.assert_tool_called("job_search_tool") - def test_predicate_can_accept_via_attribute_access(self): + 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: - agent.prompt("find jobs") + await agent.prompt("find jobs") agent.assert_tool_called("job_search_tool", lambda tool: tool.name == "job_search_tool") - def test_predicate_can_reject(self): + 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: - agent.prompt("find jobs") + 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.TestCase): +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) @@ -161,23 +161,23 @@ async def fake_run(agent_self, message, **kwargs): patcher.start() self.addCleanup(patcher.stop) - def test_passes_when_absent(self): + 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: - agent.prompt("hi") + await agent.prompt("hi") agent.assert_tool_not_called(["job_search_tool"]) - def test_fails_when_present(self): + 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: - agent.prompt("find jobs") + await agent.prompt("find jobs") with self.assertRaises(AssertionError): agent.assert_tool_not_called(["job_search_tool"]) -class TestAssertResponseTimeLt(unittest.TestCase): +class TestAssertResponseTimeLt(unittest.IsolatedAsyncioTestCase): def setup_agent(self): async def fake_run(agent_self, message, **kwargs): return AgentResponse(content="Hello!") @@ -186,30 +186,30 @@ async def fake_run(agent_self, message, **kwargs): patcher.start() self.addCleanup(patcher.stop) - def test_passes_for_a_fast_call(self): + 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: - agent.prompt("hi") + await agent.prompt("hi") agent.assert_response_time_lt(5) - def test_fails_when_exceeded(self): + 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: - agent.prompt("hi") + await agent.prompt("hi") with self.assertRaises(AssertionError): agent.assert_response_time_lt(0) - def test_fails_when_no_prompt_has_been_made(self): + 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.TestCase): - def test_seed_messages_are_included_when_building_the_real_agents_messages(self): +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: @@ -219,7 +219,7 @@ def test_seed_messages_are_included_when_building_the_real_agents_messages(self) self.assertEqual(built[1], seed[1]) self.assertEqual(built[-1], {"role": "user", "content": "suggest python developer jobs"}) - def test_same_followup_text_with_different_seed_history_does_not_collide(self): + 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") @@ -228,7 +228,7 @@ async def run_a(agent_self, message, **kwargs): with mock.patch.object(SimpleAgent, "_run", run_a): with SimpleAgent.record(cassette) as agent: - response_a = agent.prompt("suggest python developer jobs") + response_a = await agent.prompt("suggest python developer jobs") async def run_b(agent_self, message, **kwargs): return AgentResponse(content="job list B") @@ -236,13 +236,13 @@ async def run_b(agent_self, message, **kwargs): 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 = agent.prompt("suggest python developer jobs") + 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.TestCase): +class TestAssertResponseJudged(unittest.IsolatedAsyncioTestCase): def setup_agent(self, content): async def fake_run(agent_self, message, **kwargs): return AgentResponse(content=content) @@ -251,62 +251,62 @@ async def fake_run(agent_self, message, **kwargs): patcher.start() self.addCleanup(patcher.stop) - def test_passes_when_judge_approves(self): + 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: - agent.prompt("hello") + await agent.prompt("hello") agent.assert_response_judged( model="gpt-3.5-turbo", expectation="The llm should respond with greetings" ) - def test_fails_when_judge_rejects(self): + 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: - agent.prompt("hello") + await agent.prompt("hello") with self.assertRaises(AssertionError): agent.assert_response_judged( model="gpt-3.5-turbo", expectation="The llm should respond with greetings" ) - def test_verdict_is_cached_in_the_cassette_and_not_re_judged(self): + 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: - agent.prompt("hello") + 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() - def test_verdict_persists_to_disk_for_a_later_replay(self): + 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: - agent.prompt("hello") + 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: - agent.prompt("hello") + await agent.prompt("hello") agent.assert_response_judged(model="gpt-3.5-turbo", expectation="greet") judge.assert_not_called() - def test_fails_when_no_prompt_has_been_made(self): + 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): From 7510b6421c95e8b4cd13dc146f99e30a4c0c662f Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Thu, 16 Jul 2026 19:05:07 -0700 Subject: [PATCH 6/7] refactor(ai): drop dead non-streaming fallback in Agent.stream() MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AgentBinding only ever wraps a RecordingAgent (from record()), which always implements stream() — the hasattr(swapped, "stream") check and its buffered-response fallback could never actually run. --- fastapi_startkit/src/fastapi_startkit/ai/agent.py | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/fastapi_startkit/src/fastapi_startkit/ai/agent.py b/fastapi_startkit/src/fastapi_startkit/ai/agent.py index b68297ac..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): From 9647d4b4d53e4aa6c8a72ef39004cddd5a672300 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Thu, 16 Jul 2026 19:08:07 -0700 Subject: [PATCH 7/7] style(ai): remove stale comments on Ai's fake registries --- fastapi_startkit/src/fastapi_startkit/ai/ai.py | 7 ------- 1 file changed, 7 deletions(-) diff --git a/fastapi_startkit/src/fastapi_startkit/ai/ai.py b/fastapi_startkit/src/fastapi_startkit/ai/ai.py index 5b65a237..da1dc599 100644 --- a/fastapi_startkit/src/fastapi_startkit/ai/ai.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: