pydantic_ai.models

与 LLM 发送请求相关的逻辑。

这里的目的是为不同的 LLM 创建一个通用接口，以便代码的其余部分可以忽略正在使用的特定 LLM。

KnownModelName module-attribute

KnownModelName = Literal[
    "anthropic:claude-3-7-sonnet-latest",
    "anthropic:claude-3-5-haiku-latest",
    "anthropic:claude-3-5-sonnet-latest",
    "anthropic:claude-3-opus-latest",
    "claude-3-7-sonnet-latest",
    "claude-3-5-haiku-latest",
    "bedrock:amazon.titan-tg1-large",
    "bedrock:amazon.titan-text-lite-v1",
    "bedrock:amazon.titan-text-express-v1",
    "bedrock:us.amazon.nova-pro-v1:0",
    "bedrock:us.amazon.nova-lite-v1:0",
    "bedrock:us.amazon.nova-micro-v1:0",
    "bedrock:anthropic.claude-3-5-sonnet-20241022-v2:0",
    "bedrock:us.anthropic.claude-3-5-sonnet-20241022-v2:0",
    "bedrock:anthropic.claude-3-5-haiku-20241022-v1:0",
    "bedrock:us.anthropic.claude-3-5-haiku-20241022-v1:0",
    "bedrock:anthropic.claude-instant-v1",
    "bedrock:anthropic.claude-v2:1",
    "bedrock:anthropic.claude-v2",
    "bedrock:anthropic.claude-3-sonnet-20240229-v1:0",
    "bedrock:us.anthropic.claude-3-sonnet-20240229-v1:0",
    "bedrock:anthropic.claude-3-haiku-20240307-v1:0",
    "bedrock:us.anthropic.claude-3-haiku-20240307-v1:0",
    "bedrock:anthropic.claude-3-opus-20240229-v1:0",
    "bedrock:us.anthropic.claude-3-opus-20240229-v1:0",
    "bedrock:anthropic.claude-3-5-sonnet-20240620-v1:0",
    "bedrock:us.anthropic.claude-3-5-sonnet-20240620-v1:0",
    "bedrock:anthropic.claude-3-7-sonnet-20250219-v1:0",
    "bedrock:us.anthropic.claude-3-7-sonnet-20250219-v1:0",
    "bedrock:cohere.command-text-v14",
    "bedrock:cohere.command-r-v1:0",
    "bedrock:cohere.command-r-plus-v1:0",
    "bedrock:cohere.command-light-text-v14",
    "bedrock:meta.llama3-8b-instruct-v1:0",
    "bedrock:meta.llama3-70b-instruct-v1:0",
    "bedrock:meta.llama3-1-8b-instruct-v1:0",
    "bedrock:us.meta.llama3-1-8b-instruct-v1:0",
    "bedrock:meta.llama3-1-70b-instruct-v1:0",
    "bedrock:us.meta.llama3-1-70b-instruct-v1:0",
    "bedrock:meta.llama3-1-405b-instruct-v1:0",
    "bedrock:us.meta.llama3-2-11b-instruct-v1:0",
    "bedrock:us.meta.llama3-2-90b-instruct-v1:0",
    "bedrock:us.meta.llama3-2-1b-instruct-v1:0",
    "bedrock:us.meta.llama3-2-3b-instruct-v1:0",
    "bedrock:us.meta.llama3-3-70b-instruct-v1:0",
    "bedrock:mistral.mistral-7b-instruct-v0:2",
    "bedrock:mistral.mixtral-8x7b-instruct-v0:1",
    "bedrock:mistral.mistral-large-2402-v1:0",
    "bedrock:mistral.mistral-large-2407-v1:0",
    "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",
    "deepseek:deepseek-chat",
    "deepseek:deepseek-reasoner",
    "google-gla:gemini-1.0-pro",
    "google-gla:gemini-1.5-flash",
    "google-gla:gemini-1.5-flash-8b",
    "google-gla:gemini-1.5-pro",
    "google-gla:gemini-2.0-flash-exp",
    "google-gla:gemini-2.0-flash-thinking-exp-01-21",
    "google-gla:gemini-exp-1206",
    "google-gla:gemini-2.0-flash",
    "google-gla:gemini-2.0-flash-lite-preview-02-05",
    "google-gla:gemini-2.0-pro-exp-02-05",
    "google-vertex:gemini-1.0-pro",
    "google-vertex:gemini-1.5-flash",
    "google-vertex:gemini-1.5-flash-8b",
    "google-vertex:gemini-1.5-pro",
    "google-vertex:gemini-2.0-flash-exp",
    "google-vertex:gemini-2.0-flash-thinking-exp-01-21",
    "google-vertex:gemini-exp-1206",
    "google-vertex:gemini-2.0-flash",
    "google-vertex:gemini-2.0-flash-lite-preview-02-05",
    "google-vertex:gemini-2.0-pro-exp-02-05",
    "gpt-3.5-turbo",
    "gpt-3.5-turbo-0125",
    "gpt-3.5-turbo-0301",
    "gpt-3.5-turbo-0613",
    "gpt-3.5-turbo-1106",
    "gpt-3.5-turbo-16k",
    "gpt-3.5-turbo-16k-0613",
    "gpt-4",
    "gpt-4-0125-preview",
    "gpt-4-0314",
    "gpt-4-0613",
    "gpt-4-1106-preview",
    "gpt-4-32k",
    "gpt-4-32k-0314",
    "gpt-4-32k-0613",
    "gpt-4-turbo",
    "gpt-4-turbo-2024-04-09",
    "gpt-4-turbo-preview",
    "gpt-4-vision-preview",
    "gpt-4.5-preview",
    "gpt-4.5-preview-2025-02-27",
    "gpt-4o",
    "gpt-4o-2024-05-13",
    "gpt-4o-2024-08-06",
    "gpt-4o-2024-11-20",
    "gpt-4o-audio-preview",
    "gpt-4o-audio-preview-2024-10-01",
    "gpt-4o-audio-preview-2024-12-17",
    "gpt-4o-mini",
    "gpt-4o-mini-2024-07-18",
    "gpt-4o-mini-audio-preview",
    "gpt-4o-mini-audio-preview-2024-12-17",
    "groq:gemma2-9b-it",
    "groq:llama-3.1-8b-instant",
    "groq:llama-3.2-11b-vision-preview",
    "groq:llama-3.2-1b-preview",
    "groq:llama-3.2-3b-preview",
    "groq:llama-3.2-90b-vision-preview",
    "groq:llama-3.3-70b-specdec",
    "groq:llama-3.3-70b-versatile",
    "groq:llama3-70b-8192",
    "groq:llama3-8b-8192",
    "groq:mixtral-8x7b-32768",
    "mistral:codestral-latest",
    "mistral:mistral-large-latest",
    "mistral:mistral-moderation-latest",
    "mistral:mistral-small-latest",
    "o1",
    "o1-2024-12-17",
    "o1-mini",
    "o1-mini-2024-09-12",
    "o1-preview",
    "o1-preview-2024-09-12",
    "o3-mini",
    "o3-mini-2025-01-31",
    "openai:chatgpt-4o-latest",
    "openai:gpt-3.5-turbo",
    "openai:gpt-3.5-turbo-0125",
    "openai:gpt-3.5-turbo-0301",
    "openai:gpt-3.5-turbo-0613",
    "openai:gpt-3.5-turbo-1106",
    "openai:gpt-3.5-turbo-16k",
    "openai:gpt-3.5-turbo-16k-0613",
    "openai:gpt-4",
    "openai:gpt-4-0125-preview",
    "openai:gpt-4-0314",
    "openai:gpt-4-0613",
    "openai:gpt-4-1106-preview",
    "openai:gpt-4-32k",
    "openai:gpt-4-32k-0314",
    "openai:gpt-4-32k-0613",
    "openai:gpt-4-turbo",
    "openai:gpt-4-turbo-2024-04-09",
    "openai:gpt-4-turbo-preview",
    "openai:gpt-4-vision-preview",
    "openai:gpt-4.5-preview",
    "openai:gpt-4.5-preview-2025-02-27",
    "openai:gpt-4o",
    "openai:gpt-4o-2024-05-13",
    "openai:gpt-4o-2024-08-06",
    "openai:gpt-4o-2024-11-20",
    "openai:gpt-4o-audio-preview",
    "openai:gpt-4o-audio-preview-2024-10-01",
    "openai:gpt-4o-audio-preview-2024-12-17",
    "openai:gpt-4o-mini",
    "openai:gpt-4o-mini-2024-07-18",
    "openai:gpt-4o-mini-audio-preview",
    "openai:gpt-4o-mini-audio-preview-2024-12-17",
    "openai:o1",
    "openai:o1-2024-12-17",
    "openai:o1-mini",
    "openai:o1-mini-2024-09-12",
    "openai:o1-preview",
    "openai:o1-preview-2024-09-12",
    "openai:o3-mini",
    "openai:o3-mini-2025-01-31",
    "test",
]

可与 Agent 的 model 参数一起使用的已知模型名称。

提供 KnownModelName 作为指定模型的简洁方式。

ModelRequestParameters dataclass

代理向模型发出请求的配置，特别是与工具和结果处理相关的配置。

源代码位于 pydantic_ai_slim/pydantic_ai/models/__init__.py 中

@dataclass
class ModelRequestParameters:
    """Configuration for an agent's request to a model, specifically related to tools and result handling."""

    function_tools: list[ToolDefinition]
    allow_text_result: bool
    result_tools: list[ToolDefinition]

模型

基类：ABC

模型的抽象类。

源代码位于 pydantic_ai_slim/pydantic_ai/models/__init__.py 中

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

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

    @asynccontextmanager
    async def request_stream(
        self,
        messages: list[ModelMessage],
        model_settings: ModelSettings | None,
        model_request_parameters: ModelRequestParameters,
    ) -> 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

    @property
    @abstractmethod
    def model_name(self) -> str:
        """The model name."""
        raise NotImplementedError()

    @property
    @abstractmethod
    def system(self) -> str:
        """The system / model provider, ex: openai.

        Use to populate the `gen_ai.system` OpenTelemetry semantic convention attribute,
        so should use well-known values listed in
        https://opentelemetry.io/docs/specs/semconv/attributes-registry/gen-ai/#gen-ai-system
        when applicable.
        """
        raise NotImplementedError()

    @property
    def base_url(self) -> str | None:
        """The base URL for the provider API, if available."""
        return None

request abstractmethod async

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

向模型发出请求。

源代码位于 pydantic_ai_slim/pydantic_ai/models/__init__.py 中

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

request_stream async

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

向模型发出请求并返回流式响应。

源代码位于 pydantic_ai_slim/pydantic_ai/models/__init__.py 中

@asynccontextmanager
async def request_stream(
    self,
    messages: list[ModelMessage],
    model_settings: ModelSettings | None,
    model_request_parameters: ModelRequestParameters,
) -> 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

model_name abstractmethod property

model_name: str

模型名称。

system abstractmethod property

system: str

系统/模型提供商，例如：openai。

用于填充 gen_ai.system OpenTelemetry 语义约定属性，因此应使用 https://opentelemetry.io/docs/specs/semconv/attributes-registry/gen-ai/#gen-ai-system 中列出的众所周知的值（如果适用）。

base_url property

base_url: str | None

提供商 API 的基本 URL（如果可用）。

StreamedResponse dataclass

基类：ABC

调用工具时来自 LLM 的流式响应。

源代码位于 pydantic_ai_slim/pydantic_ai/models/__init__.py 中

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

    _parts_manager: ModelResponsePartsManager = field(default_factory=ModelResponsePartsManager, init=False)
    _event_iterator: AsyncIterator[ModelResponseStreamEvent] | None = field(default=None, init=False)
    _usage: Usage = field(default_factory=Usage, 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.

        It should use the `_parts_manager` to handle deltas, and should update the `_usage` attributes as it goes.
        """
        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 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

    @property
    @abstractmethod
    def model_name(self) -> str:
        """Get the model name of the response."""
        raise NotImplementedError()

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

__aiter__

__aiter__() -> AsyncIterator[ModelResponseStreamEvent]

将响应作为 ModelResponseStreamEvent 的异步可迭代对象进行流式传输。

源代码位于 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

从到目前为止从流中接收到的数据构建 ModelResponse。

源代码位于 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
    )

usage

usage() -> Usage

获取到目前为止的响应使用情况。在流耗尽之前，这不会是最终的使用情况。

源代码位于 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

model_name abstractmethod property

model_name: str

获取响应的模型名称。

timestamp abstractmethod property

timestamp: datetime

获取响应的时间戳。

ALLOW_MODEL_REQUESTS module-attribute

ALLOW_MODEL_REQUESTS = True

是否允许向模型发出请求。

此全局设置允许您禁用向大多数模型发出请求，例如，确保您不会在测试期间意外地向模型发出代价高昂的请求。

测试模型 TestModel 和 FunctionModel 不受此设置的影响。

check_allow_model_requests

check_allow_model_requests() -> None

检查是否允许模型请求。

如果您正在定义自己的模型，这些模型具有与其使用相关的成本或延迟，则应在 Model.request 和 Model.request_stream 中调用此方法。

引发

类型	描述
RuntimeError	如果模型请求不被允许。

源代码位于 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.request`][pydantic_ai.models.Model.request] and [`Model.request_stream`][pydantic_ai.models.Model.request_stream].

    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]

临时覆盖 ALLOW_MODEL_REQUESTS 的上下文管理器。

参数

名称	类型	描述	默认值
allow_model_requests	bool	是否允许在此上下文中进行模型请求。	必需

源代码位于 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]