TestModel

Utilities for testing apps built with PydanticAI.

TestModel dataclass

Bases: Model

A model specifically for testing purposes.

This will (by default) call all retrievers in the agent model, then return a tool response if possible, otherwise a plain response.

How useful this function will be is unknown, it may be useless, it may require significant changes to be useful.

Apart from __init__ derived by the dataclass decorator, all methods are private or match those of the base class.

Source code in pydantic_ai/models/test.py

@dataclass
class TestModel(Model):
    """A model specifically for testing purposes.

    This will (by default) call all retrievers in the agent model, then return a tool response if possible,
    otherwise a plain response.

    How useful this function will be is unknown, it may be useless, it may require significant changes to be useful.

    Apart from `__init__` derived by the `dataclass` decorator, all methods are private or match those
    of the base class.
    """

    # NOTE: Avoid test discovery by pytest.
    __test__ = False

    call_retrievers: list[str] | Literal['all'] = 'all'
    """List of retrievers to call. If `'all'`, all retrievers will be called."""
    custom_result_text: str | None = None
    """If set, this text is return as teh final result."""
    custom_result_args: Any | None = None
    """If set, these args will be passed to the result tool."""
    seed: int = 0
    """Seed for generating random data."""
    # these fields are set when the model is called by the agent
    agent_model_retrievers: Mapping[str, AbstractToolDefinition] | None = field(default=None, init=False)
    agent_model_allow_text_result: bool | None = field(default=None, init=False)
    agent_model_result_tools: list[AbstractToolDefinition] | None = field(default=None, init=False)

    def agent_model(
        self,
        retrievers: Mapping[str, AbstractToolDefinition],
        allow_text_result: bool,
        result_tools: Sequence[AbstractToolDefinition] | None,
    ) -> AgentModel:
        self.agent_model_retrievers = retrievers
        self.agent_model_allow_text_result = allow_text_result
        self.agent_model_result_tools = list(result_tools) if result_tools is not None else None

        if self.call_retrievers == 'all':
            retriever_calls = [(r.name, r) for r in retrievers.values()]
        else:
            retrievers_to_call = (retrievers[name] for name in self.call_retrievers)
            retriever_calls = [(r.name, r) for r in retrievers_to_call]

        if self.custom_result_text is not None:
            assert allow_text_result, 'Plain response not allowed, but `custom_result_text` is set.'
            assert self.custom_result_args is None, 'Cannot set both `custom_result_text` and `custom_result_args`.'
            result: _utils.Either[str | None, Any | None] = _utils.Either(left=self.custom_result_text)
        elif self.custom_result_args is not None:
            assert result_tools is not None, 'No result tools provided, but `custom_result_args` is set.'
            result_tool = result_tools[0]

            if k := result_tool.outer_typed_dict_key:
                result = _utils.Either(right={k: self.custom_result_args})
            else:
                result = _utils.Either(right=self.custom_result_args)
        elif allow_text_result:
            result = _utils.Either(left=None)
        elif result_tools is not None:
            result = _utils.Either(right=None)
        else:
            result = _utils.Either(left=None)
        return TestAgentModel(retriever_calls, result, self.agent_model_result_tools, self.seed)

    def name(self) -> str:
        return 'test-model'

call_retrievers class-attribute instance-attribute

call_retrievers: list[str] | Literal['all'] = 'all'

List of retrievers to call. If 'all', all retrievers will be called.

custom_result_text class-attribute instance-attribute

custom_result_text: str | None = None

If set, this text is return as teh final result.

custom_result_args class-attribute instance-attribute

custom_result_args: Any | None = None

If set, these args will be passed to the result tool.

seed class-attribute instance-attribute

seed: int = 0

Seed for generating random data.

TestAgentModel dataclass

Bases: AgentModel

Implementation of AgentModel for testing purposes.

Source code in pydantic_ai/models/test.py

@dataclass
class TestAgentModel(AgentModel):
    """Implementation of `AgentModel` for testing purposes."""

    # NOTE: Avoid test discovery by pytest.
    __test__ = False

    retriever_calls: list[tuple[str, AbstractToolDefinition]]
    # left means the text is plain text; right means it's a function call
    result: _utils.Either[str | None, Any | None]
    result_tools: list[AbstractToolDefinition] | None
    seed: int
    step: int = 0
    last_message_count: int = 0

    async def request(self, messages: list[Message]) -> tuple[ModelAnyResponse, Cost]:
        return self._request(messages), Cost()

    @asynccontextmanager
    async def request_stream(self, messages: list[Message]) -> AsyncIterator[EitherStreamedResponse]:
        msg = self._request(messages)
        cost = Cost()
        if isinstance(msg, ModelTextResponse):
            yield TestStreamTextResponse(msg.content, cost)
        else:
            yield TestStreamStructuredResponse(msg, cost)

    def gen_retriever_args(self, tool_def: AbstractToolDefinition) -> Any:
        return _JsonSchemaTestData(tool_def.json_schema, self.seed).generate()

    def _request(self, messages: list[Message]) -> ModelAnyResponse:
        if self.step == 0 and self.retriever_calls:
            calls = [ToolCall.from_object(name, self.gen_retriever_args(args)) for name, args in self.retriever_calls]
            self.step += 1
            self.last_message_count = len(messages)
            return ModelStructuredResponse(calls=calls)

        new_messages = messages[self.last_message_count :]
        self.last_message_count = len(messages)
        new_retry_names = {m.tool_name for m in new_messages if isinstance(m, RetryPrompt)}
        if new_retry_names:
            calls = [
                ToolCall.from_object(name, self.gen_retriever_args(args))
                for name, args in self.retriever_calls
                if name in new_retry_names
            ]
            self.step += 1
            return ModelStructuredResponse(calls=calls)
        else:
            if response_text := self.result.left:
                self.step += 1
                if response_text.value is None:
                    # build up details of retriever responses
                    output: dict[str, Any] = {}
                    for message in messages:
                        if isinstance(message, ToolReturn):
                            output[message.tool_name] = message.content
                    return ModelTextResponse(content=pydantic_core.to_json(output).decode())
                else:
                    return ModelTextResponse(content=response_text.value)
            else:
                assert self.result_tools is not None, 'No result tools provided'
                custom_result_args = self.result.right
                result_tool = self.result_tools[self.seed % len(self.result_tools)]
                if custom_result_args is not None:
                    self.step += 1
                    return ModelStructuredResponse(calls=[ToolCall.from_object(result_tool.name, custom_result_args)])
                else:
                    response_args = self.gen_retriever_args(result_tool)
                    self.step += 1
                    return ModelStructuredResponse(calls=[ToolCall.from_object(result_tool.name, response_args)])