pydantic_ai.models

Logic related to making requests to an LLM.

The aim here is to make a common interface for different LLMs, so that the rest of the code can be agnostic to the specific LLM being used.

KnownModelName module-attribute

KnownModelName = Literal[
    "openai:gpt-4o",
    "openai:gpt-4o-mini",
    "openai:gpt-4-turbo",
    "openai:gpt-4",
    "openai:o1-preview",
    "openai:o1-mini",
    "openai:o1",
    "openai:gpt-3.5-turbo",
    "groq:llama-3.3-70b-versatile",
    "groq:llama-3.1-70b-versatile",
    "groq:llama3-groq-70b-8192-tool-use-preview",
    "groq:llama3-groq-8b-8192-tool-use-preview",
    "groq:llama-3.1-70b-specdec",
    "groq:llama-3.1-8b-instant",
    "groq:llama-3.2-1b-preview",
    "groq:llama-3.2-3b-preview",
    "groq:llama-3.2-11b-vision-preview",
    "groq:llama-3.2-90b-vision-preview",
    "groq:llama3-70b-8192",
    "groq:llama3-8b-8192",
    "groq:mixtral-8x7b-32768",
    "groq:gemma2-9b-it",
    "groq:gemma-7b-it",
    "google-gla:gemini-1.5-flash",
    "google-gla:gemini-1.5-pro",
    "google-gla:gemini-2.0-flash-exp",
    "google-vertex:gemini-1.5-flash",
    "google-vertex:gemini-1.5-pro",
    "google-vertex:gemini-2.0-flash-exp",
    "mistral:mistral-small-latest",
    "mistral:mistral-large-latest",
    "mistral:codestral-latest",
    "mistral:mistral-moderation-latest",
    "ollama:codellama",
    "ollama:deepseek-r1",
    "ollama:gemma",
    "ollama:gemma2",
    "ollama:llama3",
    "ollama:llama3.1",
    "ollama:llama3.2",
    "ollama:llama3.2-vision",
    "ollama:llama3.3",
    "ollama:mistral",
    "ollama:mistral-nemo",
    "ollama:mixtral",
    "ollama:phi3",
    "ollama:phi4",
    "ollama:qwq",
    "ollama:qwen",
    "ollama:qwen2",
    "ollama:qwen2.5",
    "ollama:starcoder2",
    "anthropic:claude-3-5-haiku-latest",
    "anthropic:claude-3-5-sonnet-latest",
    "anthropic:claude-3-opus-latest",
    "claude-3-5-haiku-latest",
    "claude-3-5-sonnet-latest",
    "claude-3-opus-latest",
    "cohere:c4ai-aya-expanse-32b",
    "cohere:c4ai-aya-expanse-8b",
    "cohere:command",
    "cohere:command-light",
    "cohere:command-light-nightly",
    "cohere:command-nightly",
    "cohere:command-r",
    "cohere:command-r-03-2024",
    "cohere:command-r-08-2024",
    "cohere:command-r-plus",
    "cohere:command-r-plus-04-2024",
    "cohere:command-r-plus-08-2024",
    "cohere:command-r7b-12-2024",
    "test",
]

Known model names that can be used with the model parameter of Agent.

KnownModelName is provided as a concise way to specify a model.

Model

Bases: ABC

Abstract class for a model.

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

class Model(ABC):
    """Abstract class for a model."""

    @abstractmethod
    async def agent_model(
        self,
        *,
        function_tools: list[ToolDefinition],
        allow_text_result: bool,
        result_tools: list[ToolDefinition],
    ) -> AgentModel:
        """Create an agent model, this is called for each step of an agent run.

        This is async in case slow/async config checks need to be performed that can't be done in `__init__`.

        Args:
            function_tools: The tools available to the agent.
            allow_text_result: Whether a plain text final response/result is permitted.
            result_tools: Tool definitions for the final result tool(s), if any.

        Returns:
            An agent model.
        """
        raise NotImplementedError()

    @abstractmethod
    def name(self) -> str:
        raise NotImplementedError()

agent_model abstractmethod async

agent_model(
    *,
    function_tools: list[ToolDefinition],
    allow_text_result: bool,
    result_tools: list[ToolDefinition]
) -> AgentModel

Create an agent model, this is called for each step of an agent run.

This is async in case slow/async config checks need to be performed that can't be done in __init__.

Parameters:

Name	Type	Description	Default
function_tools	list[ToolDefinition]	The tools available to the agent.	required
allow_text_result	bool	Whether a plain text final response/result is permitted.	required
result_tools	list[ToolDefinition]	Tool definitions for the final result tool(s), if any.	required

Returns:

Type	Description
AgentModel	An agent model.

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

@abstractmethod
async def agent_model(
    self,
    *,
    function_tools: list[ToolDefinition],
    allow_text_result: bool,
    result_tools: list[ToolDefinition],
) -> AgentModel:
    """Create an agent model, this is called for each step of an agent run.

    This is async in case slow/async config checks need to be performed that can't be done in `__init__`.

    Args:
        function_tools: The tools available to the agent.
        allow_text_result: Whether a plain text final response/result is permitted.
        result_tools: Tool definitions for the final result tool(s), if any.

    Returns:
        An agent model.
    """
    raise NotImplementedError()

AgentModel

Bases: ABC

Model configured for each step of an Agent run.

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

class AgentModel(ABC):
    """Model configured for each step of an Agent run."""

    @abstractmethod
    async def request(
        self, messages: list[ModelMessage], model_settings: ModelSettings | None
    ) -> tuple[ModelResponse, Usage]:
        """Make a request to the model."""
        raise NotImplementedError()

    @asynccontextmanager
    async def request_stream(
        self, messages: list[ModelMessage], model_settings: ModelSettings | None
    ) -> AsyncIterator[StreamedResponse]:
        """Make a request to the model and return a streaming response."""
        # This method is not required, but you need to implement it if you want to support streamed responses
        raise NotImplementedError(f'Streamed requests not supported by this {self.__class__.__name__}')
        # yield is required to make this a generator for type checking
        # noinspection PyUnreachableCode
        yield  # pragma: no cover

request abstractmethod async

request(
    messages: list[ModelMessage],
    model_settings: ModelSettings | None,
) -> tuple[ModelResponse, Usage]

Make a request to the model.

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

@abstractmethod
async def request(
    self, messages: list[ModelMessage], model_settings: ModelSettings | None
) -> tuple[ModelResponse, Usage]:
    """Make a request to the model."""
    raise NotImplementedError()

request_stream async

request_stream(
    messages: list[ModelMessage],
    model_settings: ModelSettings | None,
) -> AsyncIterator[StreamedResponse]

Make a request to the model and return a streaming response.

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

@asynccontextmanager
async def request_stream(
    self, messages: list[ModelMessage], model_settings: ModelSettings | None
) -> AsyncIterator[StreamedResponse]:
    """Make a request to the model and return a streaming response."""
    # This method is not required, but you need to implement it if you want to support streamed responses
    raise NotImplementedError(f'Streamed requests not supported by this {self.__class__.__name__}')
    # yield is required to make this a generator for type checking
    # noinspection PyUnreachableCode
    yield  # pragma: no cover

StreamedResponse dataclass

Bases: ABC

Streamed response from an LLM when calling a tool.

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

@dataclass
class StreamedResponse(ABC):
    """Streamed response from an LLM when calling a tool."""

    _model_name: str
    _usage: Usage = field(default_factory=Usage, init=False)
    _parts_manager: ModelResponsePartsManager = field(default_factory=ModelResponsePartsManager, init=False)
    _event_iterator: AsyncIterator[ModelResponseStreamEvent] | None = field(default=None, init=False)

    def __aiter__(self) -> AsyncIterator[ModelResponseStreamEvent]:
        """Stream the response as an async iterable of [`ModelResponseStreamEvent`][pydantic_ai.messages.ModelResponseStreamEvent]s."""
        if self._event_iterator is None:
            self._event_iterator = self._get_event_iterator()
        return self._event_iterator

    @abstractmethod
    async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
        """Return an async iterator of [`ModelResponseStreamEvent`][pydantic_ai.messages.ModelResponseStreamEvent]s.

        This method should be implemented by subclasses to translate the vendor-specific stream of events into
        pydantic_ai-format events.
        """
        raise NotImplementedError()
        # noinspection PyUnreachableCode
        yield

    def get(self) -> ModelResponse:
        """Build a [`ModelResponse`][pydantic_ai.messages.ModelResponse] from the data received from the stream so far."""
        return ModelResponse(
            parts=self._parts_manager.get_parts(), model_name=self._model_name, timestamp=self.timestamp()
        )

    def model_name(self) -> str:
        """Get the model name of the response."""
        return self._model_name

    def usage(self) -> Usage:
        """Get the usage of the response so far. This will not be the final usage until the stream is exhausted."""
        return self._usage

    @abstractmethod
    def timestamp(self) -> datetime:
        """Get the timestamp of the response."""
        raise NotImplementedError()

__aiter__

__aiter__() -> AsyncIterator[ModelResponseStreamEvent]

Stream the response as an async iterable of ModelResponseStreamEvents.

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

def __aiter__(self) -> AsyncIterator[ModelResponseStreamEvent]:
    """Stream the response as an async iterable of [`ModelResponseStreamEvent`][pydantic_ai.messages.ModelResponseStreamEvent]s."""
    if self._event_iterator is None:
        self._event_iterator = self._get_event_iterator()
    return self._event_iterator

get

get() -> ModelResponse

Build a ModelResponse from the data received from the stream so far.

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

def get(self) -> ModelResponse:
    """Build a [`ModelResponse`][pydantic_ai.messages.ModelResponse] from the data received from the stream so far."""
    return ModelResponse(
        parts=self._parts_manager.get_parts(), model_name=self._model_name, timestamp=self.timestamp()
    )

model_name

model_name() -> str

Get the model name of the response.

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

def model_name(self) -> str:
    """Get the model name of the response."""
    return self._model_name

usage

usage() -> Usage

Get the usage of the response so far. This will not be the final usage until the stream is exhausted.

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

def usage(self) -> Usage:
    """Get the usage of the response so far. This will not be the final usage until the stream is exhausted."""
    return self._usage

timestamp abstractmethod

timestamp() -> datetime

Get the timestamp of the response.

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

@abstractmethod
def timestamp(self) -> datetime:
    """Get the timestamp of the response."""
    raise NotImplementedError()

ALLOW_MODEL_REQUESTS module-attribute

ALLOW_MODEL_REQUESTS = True

Whether to allow requests to models.

This global setting allows you to disable request to most models, e.g. to make sure you don't accidentally make costly requests to a model during tests.

The testing models TestModel and FunctionModel are no affected by this setting.

check_allow_model_requests

check_allow_model_requests() -> None

Check if model requests are allowed.

If you're defining your own models that have costs or latency associated with their use, you should call this in Model.agent_model.

Raises:

Type	Description
RuntimeError	If model requests are not allowed.

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

def check_allow_model_requests() -> None:
    """Check if model requests are allowed.

    If you're defining your own models that have costs or latency associated with their use, you should call this in
    [`Model.agent_model`][pydantic_ai.models.Model.agent_model].

    Raises:
        RuntimeError: If model requests are not allowed.
    """
    if not ALLOW_MODEL_REQUESTS:
        raise RuntimeError('Model requests are not allowed, since ALLOW_MODEL_REQUESTS is False')

override_allow_model_requests

override_allow_model_requests(
    allow_model_requests: bool,
) -> Iterator[None]

Context manager to temporarily override ALLOW_MODEL_REQUESTS.

Parameters:

Name	Type	Description	Default
allow_model_requests	bool	Whether to allow model requests within the context.	required

Source code in pydantic_ai_slim/pydantic_ai/models/__init__.py

@contextmanager
def override_allow_model_requests(allow_model_requests: bool) -> Iterator[None]:
    """Context manager to temporarily override [`ALLOW_MODEL_REQUESTS`][pydantic_ai.models.ALLOW_MODEL_REQUESTS].

    Args:
        allow_model_requests: Whether to allow model requests within the context.
    """
    global ALLOW_MODEL_REQUESTS
    old_value = ALLOW_MODEL_REQUESTS
    ALLOW_MODEL_REQUESTS = allow_model_requests  # pyright: ignore[reportConstantRedefinition]
    try:
        yield
    finally:
        ALLOW_MODEL_REQUESTS = old_value  # pyright: ignore[reportConstantRedefinition]