pydantic_ai.messages

ModelMessage 的结构可以显示为一张图表

graph RL
    SystemPromptPart(SystemPromptPart) --- ModelRequestPart
    UserPromptPart(UserPromptPart) --- ModelRequestPart
    ToolReturnPart(ToolReturnPart) --- ModelRequestPart
    RetryPromptPart(RetryPromptPart) --- ModelRequestPart
    TextPart(TextPart) --- ModelResponsePart
    ToolCallPart(ToolCallPart) --- ModelResponsePart
    ThinkingPart(ThinkingPart) --- ModelResponsePart
    ModelRequestPart("ModelRequestPart<br>(Union)") --- ModelRequest
    ModelRequest("ModelRequest(parts=list[...])") --- ModelMessage
    ModelResponsePart("ModelResponsePart<br>(Union)") --- ModelResponse
    ModelResponse("ModelResponse(parts=list[...])") --- ModelMessage("ModelMessage<br>(Union)")

SystemPromptPart dataclass

一个系统提示，通常由应用程序开发人员编写。

这为模型提供了上下文和如何响应的指导。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class SystemPromptPart:
    """A system prompt, generally written by the application developer.

    This gives the model context and guidance on how to respond.
    """

    content: str
    """The content of the prompt."""

    _: KW_ONLY

    timestamp: datetime = field(default_factory=_now_utc)
    """The timestamp of the prompt."""

    dynamic_ref: str | None = None
    """The ref of the dynamic system prompt function that generated this part.

    Only set if system prompt is dynamic, see [`system_prompt`][pydantic_ai.Agent.system_prompt] for more information.
    """

    part_kind: Literal['system-prompt'] = 'system-prompt'
    """Part type identifier, this is available on all parts as a discriminator."""

    def otel_event(self, settings: InstrumentationSettings) -> Event:
        return Event(
            'gen_ai.system.message',
            body={'role': 'system', **({'content': self.content} if settings.include_content else {})},
        )

    def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_messages.MessagePart]:
        return [_otel_messages.TextPart(type='text', **{'content': self.content} if settings.include_content else {})]

    __repr__ = _utils.dataclasses_no_defaults_repr

内容 实例属性

content: str

提示的内容。

timestamp 类属性 实例属性

timestamp: datetime = field(default_factory=now_utc)

提示的时间戳。

dynamic_ref 类属性 实例属性

dynamic_ref: str | None = None

生成此部分的动态系统提示函数的引用。

仅在系统提示为动态时设置，更多信息请参见 system_prompt。

part_kind 类属性 实例属性

part_kind: Literal['system-prompt'] = 'system-prompt'

部分类型标识符，作为区分符在所有部分上都可用。

FileUrl dataclass

基类：ABC

任何基于 URL 的文件的抽象基类。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(init=False, repr=False)
class FileUrl(ABC):
    """Abstract base class for any URL-based file."""

    url: str
    """The URL of the file."""

    _: KW_ONLY

    force_download: bool = False
    """If the model supports it:

    * If True, the file is downloaded and the data is sent to the model as bytes.
    * If False, the URL is sent directly to the model and no download is performed.
    """

    vendor_metadata: dict[str, Any] | None = None
    """Vendor-specific metadata for the file.

    Supported by:
    - `GoogleModel`: `VideoUrl.vendor_metadata` is used as `video_metadata`: https://ai.google.dev/gemini-api/docs/video-understanding#customize-video-processing
    """

    _media_type: Annotated[str | None, pydantic.Field(alias='media_type', default=None, exclude=True)] = field(
        compare=False, default=None
    )

    identifier: str | None = None
    """The identifier of the file, such as a unique ID. generating one from the url if not explicitly set

    This identifier can be provided to the model in a message to allow it to refer to this file in a tool call argument,
    and the tool can look up the file in question by iterating over the message history and finding the matching `FileUrl`.

    This identifier is only automatically passed to the model when the `FileUrl` is returned by a tool.
    If you're passing the `FileUrl` as a user message, it's up to you to include a separate text part with the identifier,
    e.g. "This is file <identifier>:" preceding the `FileUrl`.
    """

    def __init__(
        self,
        url: str,
        *,
        force_download: bool = False,
        vendor_metadata: dict[str, Any] | None = None,
        media_type: str | None = None,
        identifier: str | None = None,
    ) -> None:
        self.url = url
        self.force_download = force_download
        self.vendor_metadata = vendor_metadata
        self._media_type = media_type
        self.identifier = identifier or _multi_modal_content_identifier(url)

    @pydantic.computed_field
    @property
    def media_type(self) -> str:
        """Return the media type of the file, based on the URL or the provided `media_type`."""
        return self._media_type or self._infer_media_type()

    @abstractmethod
    def _infer_media_type(self) -> str:
        """Infer the media type of the file based on the URL."""
        raise NotImplementedError

    @property
    @abstractmethod
    def format(self) -> str:
        """The file format."""
        raise NotImplementedError

    __repr__ = _utils.dataclasses_no_defaults_repr

url 实例属性

url: str = url

文件的 URL。

force_download 类属性 实例属性

force_download: bool = force_download

如果模型支持

• 如果为 True，则文件将被下载，数据作为字节发送给模型。
• 如果为 False，URL 将直接发送到模型，不执行下载。

vendor_metadata 类属性 实例属性

vendor_metadata: dict[str, Any] | None = vendor_metadata

文件的特定于供应商的元数据。

支持者：- GoogleModel：VideoUrl.vendor_metadata 被用作 video_metadata：https://ai.google.dev/gemini-api/docs/video-understanding#customize-video-processing

identifier 类属性 实例属性

identifier: str | None = (
    identifier or _multi_modal_content_identifier(url)
)

文件的标识符，例如唯一ID。如果未明确设置，则从 url 生成一个。

此标识符可以在消息中提供给模型，以允许它在工具调用参数中引用此文件，并且工具可以通过迭代消息历史记录并找到匹配的 FileUrl 来查找所讨论的文件。

此标识符仅在工具返回 FileUrl 时自动传递给模型。如果您将 FileUrl 作为用户消息传递，则需要您自己包含一个带有标识符的单独文本部分，例如“这是文件:” 位于 FileUrl 之前。

media_type 属性

media_type: str

根据 URL 或提供的 media_type 返回文件的媒体类型。

format 抽象方法 属性

format: str

文件格式。

VideoUrl dataclass

基类：FileUrl

视频的 URL。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(init=False, repr=False)
class VideoUrl(FileUrl):
    """A URL to a video."""

    url: str
    """The URL of the video."""

    _: KW_ONLY

    kind: Literal['video-url'] = 'video-url'
    """Type identifier, this is available on all parts as a discriminator."""

    def __init__(
        self,
        url: str,
        *,
        force_download: bool = False,
        vendor_metadata: dict[str, Any] | None = None,
        media_type: str | None = None,
        kind: Literal['video-url'] = 'video-url',
        identifier: str | None = None,
        # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs.
        _media_type: str | None = None,
    ) -> None:
        super().__init__(
            url=url,
            force_download=force_download,
            vendor_metadata=vendor_metadata,
            media_type=media_type or _media_type,
            identifier=identifier,
        )
        self.kind = kind

    def _infer_media_type(self) -> VideoMediaType:
        """Return the media type of the video, based on the url."""
        if self.url.endswith('.mkv'):
            return 'video/x-matroska'
        elif self.url.endswith('.mov'):
            return 'video/quicktime'
        elif self.url.endswith('.mp4'):
            return 'video/mp4'
        elif self.url.endswith('.webm'):
            return 'video/webm'
        elif self.url.endswith('.flv'):
            return 'video/x-flv'
        elif self.url.endswith(('.mpeg', '.mpg')):
            return 'video/mpeg'
        elif self.url.endswith('.wmv'):
            return 'video/x-ms-wmv'
        elif self.url.endswith('.three_gp'):
            return 'video/3gpp'
        # Assume that YouTube videos are mp4 because there would be no extension
        # to infer from. This should not be a problem, as Gemini disregards media
        # type for YouTube URLs.
        elif self.is_youtube:
            return 'video/mp4'
        else:
            raise ValueError(
                f'Could not infer media type from video URL: {self.url}. Explicitly provide a `media_type` instead.'
            )

    @property
    def is_youtube(self) -> bool:
        """True if the URL has a YouTube domain."""
        return self.url.startswith(('https://youtu.be/', 'https://youtube.com/', 'https://www.youtube.com/'))

    @property
    def format(self) -> VideoFormat:
        """The file format of the video.

        The choice of supported formats were based on the Bedrock Converse API. Other APIs don't require to use a format.
        """
        return _video_format_lookup[self.media_type]

url 实例属性

url: str

视频的 URL。

kind 类属性 实例属性

kind: Literal['video-url'] = kind

类型标识符，作为区分符在所有部分上都可用。

is_youtube 属性

is_youtube: bool

如果 URL 具有 YouTube 域名，则为 True。

format 属性

format: VideoFormat

视频的文件格式。

支持的格式选择基于 Bedrock Converse API。其他 API 不要求使用格式。

AudioUrl dataclass

基类：FileUrl

音频文件的 URL。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(init=False, repr=False)
class AudioUrl(FileUrl):
    """A URL to an audio file."""

    url: str
    """The URL of the audio file."""

    _: KW_ONLY

    kind: Literal['audio-url'] = 'audio-url'
    """Type identifier, this is available on all parts as a discriminator."""

    def __init__(
        self,
        url: str,
        *,
        force_download: bool = False,
        vendor_metadata: dict[str, Any] | None = None,
        media_type: str | None = None,
        kind: Literal['audio-url'] = 'audio-url',
        identifier: str | None = None,
        # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs.
        _media_type: str | None = None,
    ) -> None:
        super().__init__(
            url=url,
            force_download=force_download,
            vendor_metadata=vendor_metadata,
            media_type=media_type or _media_type,
            identifier=identifier,
        )
        self.kind = kind

    def _infer_media_type(self) -> AudioMediaType:
        """Return the media type of the audio file, based on the url.

        References:
        - Gemini: https://ai.google.dev/gemini-api/docs/audio#supported-formats
        """
        if self.url.endswith('.mp3'):
            return 'audio/mpeg'
        if self.url.endswith('.wav'):
            return 'audio/wav'
        if self.url.endswith('.flac'):
            return 'audio/flac'
        if self.url.endswith('.oga'):
            return 'audio/ogg'
        if self.url.endswith('.aiff'):
            return 'audio/aiff'
        if self.url.endswith('.aac'):
            return 'audio/aac'

        raise ValueError(
            f'Could not infer media type from audio URL: {self.url}. Explicitly provide a `media_type` instead.'
        )

    @property
    def format(self) -> AudioFormat:
        """The file format of the audio file."""
        return _audio_format_lookup[self.media_type]

url 实例属性

url: str

音频文件的 URL。

kind 类属性 实例属性

kind: Literal['audio-url'] = kind

类型标识符，作为区分符在所有部分上都可用。

format 属性

format: AudioFormat

音频文件的文件格式。

ImageUrl dataclass

基类：FileUrl

图像的 URL。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(init=False, repr=False)
class ImageUrl(FileUrl):
    """A URL to an image."""

    url: str
    """The URL of the image."""

    _: KW_ONLY

    kind: Literal['image-url'] = 'image-url'
    """Type identifier, this is available on all parts as a discriminator."""

    def __init__(
        self,
        url: str,
        *,
        force_download: bool = False,
        vendor_metadata: dict[str, Any] | None = None,
        media_type: str | None = None,
        kind: Literal['image-url'] = 'image-url',
        identifier: str | None = None,
        # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs.
        _media_type: str | None = None,
    ) -> None:
        super().__init__(
            url=url,
            force_download=force_download,
            vendor_metadata=vendor_metadata,
            media_type=media_type or _media_type,
            identifier=identifier,
        )
        self.kind = kind

    def _infer_media_type(self) -> ImageMediaType:
        """Return the media type of the image, based on the url."""
        if self.url.endswith(('.jpg', '.jpeg')):
            return 'image/jpeg'
        elif self.url.endswith('.png'):
            return 'image/png'
        elif self.url.endswith('.gif'):
            return 'image/gif'
        elif self.url.endswith('.webp'):
            return 'image/webp'
        else:
            raise ValueError(
                f'Could not infer media type from image URL: {self.url}. Explicitly provide a `media_type` instead.'
            )

    @property
    def format(self) -> ImageFormat:
        """The file format of the image.

        The choice of supported formats were based on the Bedrock Converse API. Other APIs don't require to use a format.
        """
        return _image_format_lookup[self.media_type]

url 实例属性

url: str

图像的 URL。

kind 类属性 实例属性

kind: Literal['image-url'] = kind

类型标识符，作为区分符在所有部分上都可用。

format 属性

format: ImageFormat

图像的文件格式。

支持的格式选择基于 Bedrock Converse API。其他 API 不要求使用格式。

DocumentUrl dataclass

基类：FileUrl

文档的 URL。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(init=False, repr=False)
class DocumentUrl(FileUrl):
    """The URL of the document."""

    url: str
    """The URL of the document."""

    _: KW_ONLY

    kind: Literal['document-url'] = 'document-url'
    """Type identifier, this is available on all parts as a discriminator."""

    def __init__(
        self,
        url: str,
        *,
        force_download: bool = False,
        vendor_metadata: dict[str, Any] | None = None,
        media_type: str | None = None,
        kind: Literal['document-url'] = 'document-url',
        identifier: str | None = None,
        # Required for inline-snapshot which expects all dataclass `__init__` methods to take all field names as kwargs.
        _media_type: str | None = None,
    ) -> None:
        super().__init__(
            url=url,
            force_download=force_download,
            vendor_metadata=vendor_metadata,
            media_type=media_type or _media_type,
            identifier=identifier,
        )
        self.kind = kind

    def _infer_media_type(self) -> str:
        """Return the media type of the document, based on the url."""
        # Common document types are hardcoded here as mime-type support for these
        # extensions varies across operating systems.
        if self.url.endswith(('.md', '.mdx', '.markdown')):
            return 'text/markdown'
        elif self.url.endswith('.asciidoc'):
            return 'text/x-asciidoc'
        elif self.url.endswith('.txt'):
            return 'text/plain'
        elif self.url.endswith('.pdf'):
            return 'application/pdf'
        elif self.url.endswith('.rtf'):
            return 'application/rtf'
        elif self.url.endswith('.docx'):
            return 'application/vnd.openxmlformats-officedocument.wordprocessingml.document'
        elif self.url.endswith('.xlsx'):
            return 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'

        type_, _ = guess_type(self.url)
        if type_ is None:
            raise ValueError(
                f'Could not infer media type from document URL: {self.url}. Explicitly provide a `media_type` instead.'
            )
        return type_

    @property
    def format(self) -> DocumentFormat:
        """The file format of the document.

        The choice of supported formats were based on the Bedrock Converse API. Other APIs don't require to use a format.
        """
        media_type = self.media_type
        try:
            return _document_format_lookup[media_type]
        except KeyError as e:
            raise ValueError(f'Unknown document media type: {media_type}') from e

url 实例属性

url: str

文档的 URL。

kind 类属性 实例属性

kind: Literal['document-url'] = kind

类型标识符，作为区分符在所有部分上都可用。

format 属性

format: DocumentFormat

文档的文件格式。

支持的格式选择基于 Bedrock Converse API。其他 API 不要求使用格式。

BinaryContent dataclass

二进制内容，例如音频或图像文件。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(init=False, repr=False)
class BinaryContent:
    """Binary content, e.g. an audio or image file."""

    data: bytes
    """The binary data."""

    _: KW_ONLY

    media_type: AudioMediaType | ImageMediaType | DocumentMediaType | str
    """The media type of the binary data."""

    identifier: str
    """Identifier for the binary content, such as a unique ID. generating one from the data if not explicitly set
    This identifier can be provided to the model in a message to allow it to refer to this file in a tool call argument,
    and the tool can look up the file in question by iterating over the message history and finding the matching `BinaryContent`.

    This identifier is only automatically passed to the model when the `BinaryContent` is returned by a tool.
    If you're passing the `BinaryContent` as a user message, it's up to you to include a separate text part with the identifier,
    e.g. "This is file <identifier>:" preceding the `BinaryContent`.
    """

    vendor_metadata: dict[str, Any] | None = None
    """Vendor-specific metadata for the file.

    Supported by:
    - `GoogleModel`: `BinaryContent.vendor_metadata` is used as `video_metadata`: https://ai.google.dev/gemini-api/docs/video-understanding#customize-video-processing
    """

    kind: Literal['binary'] = 'binary'
    """Type identifier, this is available on all parts as a discriminator."""

    def __init__(
        self,
        data: bytes,
        *,
        media_type: AudioMediaType | ImageMediaType | DocumentMediaType | str,
        identifier: str | None = None,
        vendor_metadata: dict[str, Any] | None = None,
        kind: Literal['binary'] = 'binary',
    ) -> None:
        self.data = data
        self.media_type = media_type
        self.identifier = identifier or _multi_modal_content_identifier(data)
        self.vendor_metadata = vendor_metadata
        self.kind = kind

    @property
    def is_audio(self) -> bool:
        """Return `True` if the media type is an audio type."""
        return self.media_type.startswith('audio/')

    @property
    def is_image(self) -> bool:
        """Return `True` if the media type is an image type."""
        return self.media_type.startswith('image/')

    @property
    def is_video(self) -> bool:
        """Return `True` if the media type is a video type."""
        return self.media_type.startswith('video/')

    @property
    def is_document(self) -> bool:
        """Return `True` if the media type is a document type."""
        return self.media_type in _document_format_lookup

    @property
    def format(self) -> str:
        """The file format of the binary content."""
        try:
            if self.is_audio:
                return _audio_format_lookup[self.media_type]
            elif self.is_image:
                return _image_format_lookup[self.media_type]
            elif self.is_video:
                return _video_format_lookup[self.media_type]
            else:
                return _document_format_lookup[self.media_type]
        except KeyError as e:
            raise ValueError(f'Unknown media type: {self.media_type}') from e

    __repr__ = _utils.dataclasses_no_defaults_repr

data 实例属性

data: bytes = data

二进制数据。

media_type 实例属性

media_type: (
    AudioMediaType
    | ImageMediaType
    | DocumentMediaType
    | str
) = media_type

二进制数据的媒体类型。

identifier 实例属性

identifier: str = (
    identifier or _multi_modal_content_identifier(data)
)

二进制内容的标识符，例如唯一ID。如果未明确设置，则从数据生成一个。此标识符可以在消息中提供给模型，以允许它在工具调用参数中引用此文件，并且工具可以通过迭代消息历史记录并找到匹配的 BinaryContent 来查找所讨论的文件。

此标识符仅在工具返回 BinaryContent 时自动传递给模型。如果您将 BinaryContent 作为用户消息传递，则需要您自己包含一个带有标识符的单独文本部分，例如“这是文件:” 位于 BinaryContent 之前。

vendor_metadata 类属性 实例属性

vendor_metadata: dict[str, Any] | None = vendor_metadata

文件的特定于供应商的元数据。

支持者：- GoogleModel：BinaryContent.vendor_metadata 被用作 video_metadata：https://ai.google.dev/gemini-api/docs/video-understanding#customize-video-processing

kind 类属性 实例属性

kind: Literal['binary'] = kind

类型标识符，作为区分符在所有部分上都可用。

is_audio 属性

is_audio: bool

如果媒体类型是音频类型，则返回 True。

is_image 属性

is_image: bool

如果媒体类型是图像类型，则返回 True。

is_video 属性

is_video: bool

如果媒体类型是视频类型，则返回 True。

is_document 属性

is_document: bool

如果媒体类型是文档类型，则返回 True。

format 属性

format: str

二进制内容的文件格式。

ToolReturn dataclass

为需要同时提供返回值和自定义内容给模型的工具提供的结构化返回值。

此类允许工具返回复杂的响应，包括： - 用于实际工具返回的返回值 - 作为 UserPromptPart 发送给模型的自定义内容（包括多模态内容） - 用于应用程序的可选元数据

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class ToolReturn:
    """A structured return value for tools that need to provide both a return value and custom content to the model.

    This class allows tools to return complex responses that include:
    - A return value for actual tool return
    - Custom content (including multi-modal content) to be sent to the model as a UserPromptPart
    - Optional metadata for application use
    """

    return_value: Any
    """The return value to be used in the tool response."""

    _: KW_ONLY

    content: str | Sequence[UserContent] | None = None
    """The content to be sent to the model as a UserPromptPart."""

    metadata: Any = None
    """Additional data that can be accessed programmatically by the application but is not sent to the LLM."""

    kind: Literal['tool-return'] = 'tool-return'

    __repr__ = _utils.dataclasses_no_defaults_repr

return_value 实例属性

return_value: Any

要在工具响应中使用的返回值。

内容 类属性 实例属性

content: str | Sequence[UserContent] | None = None

要作为 UserPromptPart 发送给模型的内容。

metadata 类属性 实例属性

metadata: Any = None

可由应用程序以编程方式访问但不会发送给 LLM 的附加数据。

UserPromptPart dataclass

用户提示，通常由最终用户编写。

内容来自 Agent.run、Agent.run_sync 和 Agent.run_stream 的 user_prompt 参数。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class UserPromptPart:
    """A user prompt, generally written by the end user.

    Content comes from the `user_prompt` parameter of [`Agent.run`][pydantic_ai.agent.AbstractAgent.run],
    [`Agent.run_sync`][pydantic_ai.agent.AbstractAgent.run_sync], and [`Agent.run_stream`][pydantic_ai.agent.AbstractAgent.run_stream].
    """

    content: str | Sequence[UserContent]
    """The content of the prompt."""

    _: KW_ONLY

    timestamp: datetime = field(default_factory=_now_utc)
    """The timestamp of the prompt."""

    part_kind: Literal['user-prompt'] = 'user-prompt'
    """Part type identifier, this is available on all parts as a discriminator."""

    def otel_event(self, settings: InstrumentationSettings) -> Event:
        content = [{'kind': part.pop('type'), **part} for part in self.otel_message_parts(settings)]
        for part in content:
            if part['kind'] == 'binary' and 'content' in part:
                part['binary_content'] = part.pop('content')
        content = [
            part['content'] if part == {'kind': 'text', 'content': part.get('content')} else part for part in content
        ]
        if content in ([{'kind': 'text'}], [self.content]):
            content = content[0]
        return Event('gen_ai.user.message', body={'content': content, 'role': 'user'})

    def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_messages.MessagePart]:
        parts: list[_otel_messages.MessagePart] = []
        content: Sequence[UserContent] = [self.content] if isinstance(self.content, str) else self.content
        for part in content:
            if isinstance(part, str):
                parts.append(
                    _otel_messages.TextPart(type='text', **({'content': part} if settings.include_content else {}))
                )
            elif isinstance(part, ImageUrl | AudioUrl | DocumentUrl | VideoUrl):
                parts.append(
                    _otel_messages.MediaUrlPart(
                        type=part.kind,
                        **{'url': part.url} if settings.include_content else {},
                    )
                )
            elif isinstance(part, BinaryContent):
                converted_part = _otel_messages.BinaryDataPart(type='binary', media_type=part.media_type)
                if settings.include_content and settings.include_binary_content:
                    converted_part['content'] = base64.b64encode(part.data).decode()
                parts.append(converted_part)
            else:
                parts.append({'type': part.kind})  # pragma: no cover
        return parts

    __repr__ = _utils.dataclasses_no_defaults_repr

内容 实例属性

content: str | Sequence[UserContent]

提示的内容。

timestamp 类属性 实例属性

timestamp: datetime = field(default_factory=now_utc)

提示的时间戳。

part_kind 类属性 实例属性

part_kind: Literal['user-prompt'] = 'user-prompt'

部分类型标识符，作为区分符在所有部分上都可用。

BaseToolReturnPart dataclass

工具返回部分的基类。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class BaseToolReturnPart:
    """Base class for tool return parts."""

    tool_name: str
    """The name of the "tool" was called."""

    content: Any
    """The return value."""

    tool_call_id: str
    """The tool call identifier, this is used by some models including OpenAI."""

    _: KW_ONLY

    metadata: Any = None
    """Additional data that can be accessed programmatically by the application but is not sent to the LLM."""

    timestamp: datetime = field(default_factory=_now_utc)
    """The timestamp, when the tool returned."""

    def model_response_str(self) -> str:
        """Return a string representation of the content for the model."""
        if isinstance(self.content, str):
            return self.content
        else:
            return tool_return_ta.dump_json(self.content).decode()

    def model_response_object(self) -> dict[str, Any]:
        """Return a dictionary representation of the content, wrapping non-dict types appropriately."""
        # gemini supports JSON dict return values, but no other JSON types, hence we wrap anything else in a dict
        if isinstance(self.content, dict):
            return tool_return_ta.dump_python(self.content, mode='json')  # pyright: ignore[reportUnknownMemberType]
        else:
            return {'return_value': tool_return_ta.dump_python(self.content, mode='json')}

    def otel_event(self, settings: InstrumentationSettings) -> Event:
        return Event(
            'gen_ai.tool.message',
            body={
                **({'content': self.content} if settings.include_content else {}),
                'role': 'tool',
                'id': self.tool_call_id,
                'name': self.tool_name,
            },
        )

    def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_messages.MessagePart]:
        from .models.instrumented import InstrumentedModel

        return [
            _otel_messages.ToolCallResponsePart(
                type='tool_call_response',
                id=self.tool_call_id,
                name=self.tool_name,
                **({'result': InstrumentedModel.serialize_any(self.content)} if settings.include_content else {}),
            )
        ]

    def has_content(self) -> bool:
        """Return `True` if the tool return has content."""
        return self.content is not None  # pragma: no cover

    __repr__ = _utils.dataclasses_no_defaults_repr

tool_name 实例属性

tool_name: str

被调用的“工具”的名称。

内容 实例属性

content: Any

返回值。

tool_call_id 实例属性

tool_call_id: str

工具调用标识符，包括 OpenAI 在内的一些模型会使用它。

metadata 类属性 实例属性

metadata: Any = None

可由应用程序以编程方式访问但不会发送给 LLM 的附加数据。

timestamp 类属性 实例属性

timestamp: datetime = field(default_factory=now_utc)

时间戳，工具返回时的时间。

model_response_str

model_response_str() -> str

返回内容的字符串表示形式供模型使用。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def model_response_str(self) -> str:
    """Return a string representation of the content for the model."""
    if isinstance(self.content, str):
        return self.content
    else:
        return tool_return_ta.dump_json(self.content).decode()

model_response_object

model_response_object() -> dict[str, Any]

返回内容的字典表示形式，适当地包装非字典类型。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def model_response_object(self) -> dict[str, Any]:
    """Return a dictionary representation of the content, wrapping non-dict types appropriately."""
    # gemini supports JSON dict return values, but no other JSON types, hence we wrap anything else in a dict
    if isinstance(self.content, dict):
        return tool_return_ta.dump_python(self.content, mode='json')  # pyright: ignore[reportUnknownMemberType]
    else:
        return {'return_value': tool_return_ta.dump_python(self.content, mode='json')}

has_content

has_content() -> bool

如果工具返回有内容，则返回 True。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def has_content(self) -> bool:
    """Return `True` if the tool return has content."""
    return self.content is not None  # pragma: no cover

ToolReturnPart dataclass

基类：BaseToolReturnPart

工具返回消息，它编码了运行工具的结果。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class ToolReturnPart(BaseToolReturnPart):
    """A tool return message, this encodes the result of running a tool."""

    _: KW_ONLY

    part_kind: Literal['tool-return'] = 'tool-return'
    """Part type identifier, this is available on all parts as a discriminator."""

part_kind 类属性 实例属性

part_kind: Literal['tool-return'] = 'tool-return'

部分类型标识符，作为区分符在所有部分上都可用。

BuiltinToolReturnPart dataclass

基类：BaseToolReturnPart

来自内置工具的工具返回消息。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class BuiltinToolReturnPart(BaseToolReturnPart):
    """A tool return message from a built-in tool."""

    _: KW_ONLY

    provider_name: str | None = None
    """The name of the provider that generated the response."""

    part_kind: Literal['builtin-tool-return'] = 'builtin-tool-return'
    """Part type identifier, this is available on all parts as a discriminator."""

provider_name 类属性 实例属性

provider_name: str | None = None

生成响应的提供商的名称。

part_kind 类属性 实例属性

part_kind: Literal["builtin-tool-return"] = (
    "builtin-tool-return"
)

部分类型标识符，作为区分符在所有部分上都可用。

RetryPromptPart dataclass

一条返回给模型，要求其重试的消息。

这可能因多种原因被发送

• 工具参数的 Pydantic 验证失败，此时内容派生自 Pydantic 的 ValidationError
• 一个工具引发了 ModelRetry 异常
• 找不到工具名称对应的工具
• 当期望结构化响应时，模型返回了纯文本
• 结构化响应的 Pydantic 验证失败，此时内容派生自 Pydantic 的 ValidationError
• 一个输出验证器引发了 ModelRetry 异常

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class RetryPromptPart:
    """A message back to a model asking it to try again.

    This can be sent for a number of reasons:

    * Pydantic validation of tool arguments failed, here content is derived from a Pydantic
      [`ValidationError`][pydantic_core.ValidationError]
    * a tool raised a [`ModelRetry`][pydantic_ai.exceptions.ModelRetry] exception
    * no tool was found for the tool name
    * the model returned plain text when a structured response was expected
    * Pydantic validation of a structured response failed, here content is derived from a Pydantic
      [`ValidationError`][pydantic_core.ValidationError]
    * an output validator raised a [`ModelRetry`][pydantic_ai.exceptions.ModelRetry] exception
    """

    content: list[pydantic_core.ErrorDetails] | str
    """Details of why and how the model should retry.

    If the retry was triggered by a [`ValidationError`][pydantic_core.ValidationError], this will be a list of
    error details.
    """

    _: KW_ONLY

    tool_name: str | None = None
    """The name of the tool that was called, if any."""

    tool_call_id: str = field(default_factory=_generate_tool_call_id)
    """The tool call identifier, this is used by some models including OpenAI.

    In case the tool call id is not provided by the model, Pydantic AI will generate a random one.
    """

    timestamp: datetime = field(default_factory=_now_utc)
    """The timestamp, when the retry was triggered."""

    part_kind: Literal['retry-prompt'] = 'retry-prompt'
    """Part type identifier, this is available on all parts as a discriminator."""

    def model_response(self) -> str:
        """Return a string message describing why the retry is requested."""
        if isinstance(self.content, str):
            if self.tool_name is None:
                description = f'Validation feedback:\n{self.content}'
            else:
                description = self.content
        else:
            json_errors = error_details_ta.dump_json(self.content, exclude={'__all__': {'ctx'}}, indent=2)
            description = f'{len(self.content)} validation errors: {json_errors.decode()}'
        return f'{description}\n\nFix the errors and try again.'

    def otel_event(self, settings: InstrumentationSettings) -> Event:
        if self.tool_name is None:
            return Event('gen_ai.user.message', body={'content': self.model_response(), 'role': 'user'})
        else:
            return Event(
                'gen_ai.tool.message',
                body={
                    **({'content': self.model_response()} if settings.include_content else {}),
                    'role': 'tool',
                    'id': self.tool_call_id,
                    'name': self.tool_name,
                },
            )

    def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_messages.MessagePart]:
        if self.tool_name is None:
            return [_otel_messages.TextPart(type='text', content=self.model_response())]
        else:
            return [
                _otel_messages.ToolCallResponsePart(
                    type='tool_call_response',
                    id=self.tool_call_id,
                    name=self.tool_name,
                    **({'result': self.model_response()} if settings.include_content else {}),
                )
            ]

    __repr__ = _utils.dataclasses_no_defaults_repr

内容 实例属性

content: list[ErrorDetails] | str

关于模型为何以及如何重试的详细信息。

如果重试是由 ValidationError 触发的，这将是一个错误详情列表。

tool_name 类属性 实例属性

tool_name: str | None = None

被调用的工具的名称，如果有的话。

tool_call_id 类属性 实例属性

tool_call_id: str = field(
    default_factory=generate_tool_call_id
)

工具调用标识符，包括 OpenAI 在内的一些模型会使用它。

如果模型未提供工具调用 ID，Pydantic AI 将生成一个随机的。

timestamp 类属性 实例属性

timestamp: datetime = field(default_factory=now_utc)

时间戳，重试被触发时的时间。

part_kind 类属性 实例属性

part_kind: Literal['retry-prompt'] = 'retry-prompt'

部分类型标识符，作为区分符在所有部分上都可用。

model_response

model_response() -> str

返回一条描述请求重试原因的字符串消息。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def model_response(self) -> str:
    """Return a string message describing why the retry is requested."""
    if isinstance(self.content, str):
        if self.tool_name is None:
            description = f'Validation feedback:\n{self.content}'
        else:
            description = self.content
    else:
        json_errors = error_details_ta.dump_json(self.content, exclude={'__all__': {'ctx'}}, indent=2)
        description = f'{len(self.content)} validation errors: {json_errors.decode()}'
    return f'{description}\n\nFix the errors and try again.'

ModelRequestPart 模块属性

ModelRequestPart = Annotated[
    SystemPromptPart
    | UserPromptPart
    | ToolReturnPart
    | RetryPromptPart,
    Discriminator("part_kind"),
]

由 Pydantic AI 发送给模型的消息部分。

ModelRequest dataclass

由 Pydantic AI 生成并发送给模型的请求，例如从 Pydantic AI 应用程序到模型的消息。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class ModelRequest:
    """A request generated by Pydantic AI and sent to a model, e.g. a message from the Pydantic AI app to the model."""

    parts: Sequence[ModelRequestPart]
    """The parts of the user message."""

    _: KW_ONLY

    instructions: str | None = None
    """The instructions for the model."""

    kind: Literal['request'] = 'request'
    """Message type identifier, this is available on all parts as a discriminator."""

    @classmethod
    def user_text_prompt(cls, user_prompt: str, *, instructions: str | None = None) -> ModelRequest:
        """Create a `ModelRequest` with a single user prompt as text."""
        return cls(parts=[UserPromptPart(user_prompt)], instructions=instructions)

    __repr__ = _utils.dataclasses_no_defaults_repr

parts 实例属性

parts: Sequence[ModelRequestPart]

用户消息的各个部分。

instructions 类属性 实例属性

instructions: str | None = None

给模型的指令。

kind 类属性 实例属性

kind: Literal['request'] = 'request'

消息类型标识符，作为区分符在所有部分上都可用。

user_text_prompt 类方法

user_text_prompt(
    user_prompt: str, *, instructions: str | None = None
) -> ModelRequest

创建一个带单个文本用户提示的 ModelRequest。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@classmethod
def user_text_prompt(cls, user_prompt: str, *, instructions: str | None = None) -> ModelRequest:
    """Create a `ModelRequest` with a single user prompt as text."""
    return cls(parts=[UserPromptPart(user_prompt)], instructions=instructions)

TextPart dataclass

来自模型的纯文本响应。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class TextPart:
    """A plain text response from a model."""

    content: str
    """The text content of the response."""

    _: KW_ONLY

    part_kind: Literal['text'] = 'text'
    """Part type identifier, this is available on all parts as a discriminator."""

    def has_content(self) -> bool:
        """Return `True` if the text content is non-empty."""
        return bool(self.content)

    __repr__ = _utils.dataclasses_no_defaults_repr

内容 实例属性

content: str

响应的文本内容。

part_kind 类属性 实例属性

part_kind: Literal['text'] = 'text'

部分类型标识符，作为区分符在所有部分上都可用。

has_content

has_content() -> bool

如果文本内容非空，则返回 True。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def has_content(self) -> bool:
    """Return `True` if the text content is non-empty."""
    return bool(self.content)

ThinkingPart dataclass

来自模型的思考响应。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class ThinkingPart:
    """A thinking response from a model."""

    content: str
    """The thinking content of the response."""

    _: KW_ONLY

    id: str | None = None
    """The identifier of the thinking part."""

    signature: str | None = None
    """The signature of the thinking.

    The signature is only available on the Anthropic models.
    """

    part_kind: Literal['thinking'] = 'thinking'
    """Part type identifier, this is available on all parts as a discriminator."""

    def has_content(self) -> bool:
        """Return `True` if the thinking content is non-empty."""
        return bool(self.content)

    __repr__ = _utils.dataclasses_no_defaults_repr

内容 实例属性

content: str

响应的思考内容。

id 类属性 实例属性

id: str | None = None

思考部分的标识符。

signature 类属性 实例属性

signature: str | None = None

思考的签名。

签名仅在 Anthropic 模型上可用。

part_kind 类属性 实例属性

part_kind: Literal['thinking'] = 'thinking'

部分类型标识符，作为区分符在所有部分上都可用。

has_content

has_content() -> bool

如果思考内容非空，则返回 True。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def has_content(self) -> bool:
    """Return `True` if the thinking content is non-empty."""
    return bool(self.content)

BaseToolCallPart dataclass

来自模型的工具调用。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class BaseToolCallPart:
    """A tool call from a model."""

    tool_name: str
    """The name of the tool to call."""

    args: str | dict[str, Any] | None = None
    """The arguments to pass to the tool.

    This is stored either as a JSON string or a Python dictionary depending on how data was received.
    """

    tool_call_id: str = field(default_factory=_generate_tool_call_id)
    """The tool call identifier, this is used by some models including OpenAI.

    In case the tool call id is not provided by the model, Pydantic AI will generate a random one.
    """

    def args_as_dict(self) -> dict[str, Any]:
        """Return the arguments as a Python dictionary.

        This is just for convenience with models that require dicts as input.
        """
        if not self.args:
            return {}
        if isinstance(self.args, dict):
            return self.args
        args = pydantic_core.from_json(self.args)
        assert isinstance(args, dict), 'args should be a dict'
        return cast(dict[str, Any], args)

    def args_as_json_str(self) -> str:
        """Return the arguments as a JSON string.

        This is just for convenience with models that require JSON strings as input.
        """
        if not self.args:
            return '{}'
        if isinstance(self.args, str):
            return self.args
        return pydantic_core.to_json(self.args).decode()

    def has_content(self) -> bool:
        """Return `True` if the arguments contain any data."""
        if isinstance(self.args, dict):
            # TODO: This should probably return True if you have the value False, or 0, etc.
            #   It makes sense to me to ignore empty strings, but not sure about empty lists or dicts
            return any(self.args.values())
        else:
            return bool(self.args)

    __repr__ = _utils.dataclasses_no_defaults_repr

tool_name 实例属性

tool_name: str

要调用的工具的名称。

args 类属性 实例属性

args: str | dict[str, Any] | None = None

传递给工具的参数。

这可以存储为 JSON 字符串或 Python 字典，具体取决于接收数据的方式。

tool_call_id 类属性 实例属性

tool_call_id: str = field(
    default_factory=generate_tool_call_id
)

工具调用标识符，包括 OpenAI 在内的一些模型会使用它。

如果模型未提供工具调用 ID，Pydantic AI 将生成一个随机的。

args_as_dict

args_as_dict() -> dict[str, Any]

将参数作为 Python 字典返回。

这只是为了方便那些需要字典作为输入的模型。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def args_as_dict(self) -> dict[str, Any]:
    """Return the arguments as a Python dictionary.

    This is just for convenience with models that require dicts as input.
    """
    if not self.args:
        return {}
    if isinstance(self.args, dict):
        return self.args
    args = pydantic_core.from_json(self.args)
    assert isinstance(args, dict), 'args should be a dict'
    return cast(dict[str, Any], args)

args_as_json_str

args_as_json_str() -> str

将参数作为 JSON 字符串返回。

这只是为了方便那些需要 JSON 字符串作为输入的模型。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def args_as_json_str(self) -> str:
    """Return the arguments as a JSON string.

    This is just for convenience with models that require JSON strings as input.
    """
    if not self.args:
        return '{}'
    if isinstance(self.args, str):
        return self.args
    return pydantic_core.to_json(self.args).decode()

has_content

has_content() -> bool

如果参数包含任何数据，则返回 True。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def has_content(self) -> bool:
    """Return `True` if the arguments contain any data."""
    if isinstance(self.args, dict):
        # TODO: This should probably return True if you have the value False, or 0, etc.
        #   It makes sense to me to ignore empty strings, but not sure about empty lists or dicts
        return any(self.args.values())
    else:
        return bool(self.args)

ToolCallPart dataclass

基类：BaseToolCallPart

来自模型的工具调用。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class ToolCallPart(BaseToolCallPart):
    """A tool call from a model."""

    _: KW_ONLY

    part_kind: Literal['tool-call'] = 'tool-call'
    """Part type identifier, this is available on all parts as a discriminator."""

part_kind 类属性 实例属性

part_kind: Literal['tool-call'] = 'tool-call'

部分类型标识符，作为区分符在所有部分上都可用。

BuiltinToolCallPart dataclass

基类：BaseToolCallPart

对内置工具的工具调用。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class BuiltinToolCallPart(BaseToolCallPart):
    """A tool call to a built-in tool."""

    _: KW_ONLY

    provider_name: str | None = None
    """The name of the provider that generated the response."""

    part_kind: Literal['builtin-tool-call'] = 'builtin-tool-call'
    """Part type identifier, this is available on all parts as a discriminator."""

provider_name 类属性 实例属性

provider_name: str | None = None

生成响应的提供商的名称。

part_kind 类属性 实例属性

part_kind: Literal["builtin-tool-call"] = (
    "builtin-tool-call"
)

部分类型标识符，作为区分符在所有部分上都可用。

ModelResponsePart 模块属性

ModelResponsePart = Annotated[
    TextPart
    | ToolCallPart
    | BuiltinToolCallPart
    | BuiltinToolReturnPart
    | ThinkingPart,
    Discriminator("part_kind"),
]

由模型返回的消息部分。

ModelResponse dataclass

来自模型的响应，例如从模型到 Pydantic AI 应用程序的消息。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class ModelResponse:
    """A response from a model, e.g. a message from the model to the Pydantic AI app."""

    parts: Sequence[ModelResponsePart]
    """The parts of the model message."""

    _: KW_ONLY

    usage: RequestUsage = field(default_factory=RequestUsage)
    """Usage information for the request.

    This has a default to make tests easier, and to support loading old messages where usage will be missing.
    """

    model_name: str | None = None
    """The name of the model that generated the response."""

    timestamp: datetime = field(default_factory=_now_utc)
    """The timestamp of the response.

    If the model provides a timestamp in the response (as OpenAI does) that will be used.
    """

    kind: Literal['response'] = 'response'
    """Message type identifier, this is available on all parts as a discriminator."""

    provider_name: str | None = None
    """The name of the LLM provider that generated the response."""

    provider_details: Annotated[
        dict[str, Any] | None,
        # `vendor_details` is deprecated, but we still want to support deserializing model responses stored in a DB before the name was changed
        pydantic.Field(validation_alias=pydantic.AliasChoices('provider_details', 'vendor_details')),
    ] = None
    """Additional provider-specific details in a serializable format.

    This allows storing selected vendor-specific data that isn't mapped to standard ModelResponse fields.
    For OpenAI models, this may include 'logprobs', 'finish_reason', etc.
    """

    provider_response_id: Annotated[
        str | None,
        # `vendor_id` is deprecated, but we still want to support deserializing model responses stored in a DB before the name was changed
        pydantic.Field(validation_alias=pydantic.AliasChoices('provider_response_id', 'vendor_id')),
    ] = None
    """request ID as specified by the model provider. This can be used to track the specific request to the model."""

    @deprecated('`price` is deprecated, use `cost` instead')
    def price(self) -> genai_types.PriceCalculation:  # pragma: no cover
        return self.cost()

    def cost(self) -> genai_types.PriceCalculation:
        """Calculate the cost of the usage.

        Uses [`genai-prices`](https://github.com/pydantic/genai-prices).
        """
        assert self.model_name, 'Model name is required to calculate price'
        return calc_price(
            self.usage,
            self.model_name,
            provider_id=self.provider_name,
            genai_request_timestamp=self.timestamp,
        )

    def otel_events(self, settings: InstrumentationSettings) -> list[Event]:
        """Return OpenTelemetry events for the response."""
        result: list[Event] = []

        def new_event_body():
            new_body: dict[str, Any] = {'role': 'assistant'}
            ev = Event('gen_ai.assistant.message', body=new_body)
            result.append(ev)
            return new_body

        body = new_event_body()
        for part in self.parts:
            if isinstance(part, ToolCallPart):
                body.setdefault('tool_calls', []).append(
                    {
                        'id': part.tool_call_id,
                        'type': 'function',
                        'function': {
                            'name': part.tool_name,
                            **({'arguments': part.args} if settings.include_content else {}),
                        },
                    }
                )
            elif isinstance(part, TextPart | ThinkingPart):
                kind = part.part_kind
                body.setdefault('content', []).append(
                    {'kind': kind, **({'text': part.content} if settings.include_content else {})}
                )

        if content := body.get('content'):
            text_content = content[0].get('text')
            if content == [{'kind': 'text', 'text': text_content}]:
                body['content'] = text_content

        return result

    def otel_message_parts(self, settings: InstrumentationSettings) -> list[_otel_messages.MessagePart]:
        parts: list[_otel_messages.MessagePart] = []
        for part in self.parts:
            if isinstance(part, TextPart):
                parts.append(
                    _otel_messages.TextPart(
                        type='text',
                        **({'content': part.content} if settings.include_content else {}),
                    )
                )
            elif isinstance(part, ThinkingPart):
                parts.append(
                    _otel_messages.ThinkingPart(
                        type='thinking',
                        **({'content': part.content} if settings.include_content else {}),
                    )
                )
            elif isinstance(part, ToolCallPart):
                call_part = _otel_messages.ToolCallPart(type='tool_call', id=part.tool_call_id, name=part.tool_name)
                if settings.include_content and part.args is not None:
                    from .models.instrumented import InstrumentedModel

                    if isinstance(part.args, str):
                        call_part['arguments'] = part.args
                    else:
                        call_part['arguments'] = {k: InstrumentedModel.serialize_any(v) for k, v in part.args.items()}

                parts.append(call_part)
        return parts

    @property
    @deprecated('`vendor_details` is deprecated, use `provider_details` instead')
    def vendor_details(self) -> dict[str, Any] | None:
        return self.provider_details

    @property
    @deprecated('`vendor_id` is deprecated, use `provider_response_id` instead')
    def vendor_id(self) -> str | None:
        return self.provider_response_id

    @property
    @deprecated('`provider_request_id` is deprecated, use `provider_response_id` instead')
    def provider_request_id(self) -> str | None:
        return self.provider_response_id

    __repr__ = _utils.dataclasses_no_defaults_repr

parts 实例属性

parts: Sequence[ModelResponsePart]

模型消息的各个部分。

usage 类属性 实例属性

usage: RequestUsage = field(default_factory=RequestUsage)

请求的使用信息。

这有一个默认值，以方便测试，并支持加载缺少使用信息的旧消息。

model_name 类属性 实例属性

model_name: str | None = None

生成响应的模型的名称。

timestamp 类属性 实例属性

timestamp: datetime = field(default_factory=now_utc)

响应的时间戳。

如果模型在响应中提供了时间戳（如 OpenAI 所做），则将使用该时间戳。

kind 类属性 实例属性

kind: Literal['response'] = 'response'

消息类型标识符，作为区分符在所有部分上都可用。

provider_name 类属性 实例属性

provider_name: str | None = None

生成响应的 LLM 提供商的名称。

provider_details 类属性 实例属性

provider_details: Annotated[
    dict[str, Any] | None,
    Field(
        validation_alias=AliasChoices(
            provider_details, vendor_details
        )
    ),
] = None

以可序列化格式提供的额外特定于提供商的详细信息。

这允许存储未映射到标准 ModelResponse 字段的选定特定于供应商的数据。对于 OpenAI 模型，这可能包括 'logprobs'、'finish_reason' 等。

provider_response_id 类属性 实例属性

provider_response_id: Annotated[
    str | None,
    Field(
        validation_alias=AliasChoices(
            provider_response_id, vendor_id
        )
    ),
] = None

由模型提供商指定的请求 ID。这可用于跟踪对模型的特定请求。

price 已弃用

price() -> PriceCalculation

已弃用

price 已被弃用，请改用 cost

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@deprecated('`price` is deprecated, use `cost` instead')
def price(self) -> genai_types.PriceCalculation:  # pragma: no cover
    return self.cost()

cost

cost() -> PriceCalculation

计算使用成本。

使用 genai-prices。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def cost(self) -> genai_types.PriceCalculation:
    """Calculate the cost of the usage.

    Uses [`genai-prices`](https://github.com/pydantic/genai-prices).
    """
    assert self.model_name, 'Model name is required to calculate price'
    return calc_price(
        self.usage,
        self.model_name,
        provider_id=self.provider_name,
        genai_request_timestamp=self.timestamp,
    )

otel_events

otel_events(
    settings: InstrumentationSettings,
) -> list[Event]

返回响应的 OpenTelemetry 事件。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def otel_events(self, settings: InstrumentationSettings) -> list[Event]:
    """Return OpenTelemetry events for the response."""
    result: list[Event] = []

    def new_event_body():
        new_body: dict[str, Any] = {'role': 'assistant'}
        ev = Event('gen_ai.assistant.message', body=new_body)
        result.append(ev)
        return new_body

    body = new_event_body()
    for part in self.parts:
        if isinstance(part, ToolCallPart):
            body.setdefault('tool_calls', []).append(
                {
                    'id': part.tool_call_id,
                    'type': 'function',
                    'function': {
                        'name': part.tool_name,
                        **({'arguments': part.args} if settings.include_content else {}),
                    },
                }
            )
        elif isinstance(part, TextPart | ThinkingPart):
            kind = part.part_kind
            body.setdefault('content', []).append(
                {'kind': kind, **({'text': part.content} if settings.include_content else {})}
            )

    if content := body.get('content'):
        text_content = content[0].get('text')
        if content == [{'kind': 'text', 'text': text_content}]:
            body['content'] = text_content

    return result

ModelMessage 模块属性

ModelMessage = Annotated[
    ModelRequest | ModelResponse, Discriminator("kind")
]

发送到模型或由模型返回的任何消息。

ModelMessagesTypeAdapter 模块属性

ModelMessagesTypeAdapter = TypeAdapter(
    list[ModelMessage],
    config=ConfigDict(
        defer_build=True,
        ser_json_bytes="base64",
        val_json_bytes="base64",
    ),
)

用于（反）序列化消息的 Pydantic TypeAdapter。

TextPartDelta dataclass

对 TextPart 的部分更新（增量），以附加新的文本内容。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class TextPartDelta:
    """A partial update (delta) for a `TextPart` to append new text content."""

    content_delta: str
    """The incremental text content to add to the existing `TextPart` content."""

    _: KW_ONLY

    part_delta_kind: Literal['text'] = 'text'
    """Part delta type identifier, used as a discriminator."""

    def apply(self, part: ModelResponsePart) -> TextPart:
        """Apply this text delta to an existing `TextPart`.

        Args:
            part: The existing model response part, which must be a `TextPart`.

        Returns:
            A new `TextPart` with updated text content.

        Raises:
            ValueError: If `part` is not a `TextPart`.
        """
        if not isinstance(part, TextPart):
            raise ValueError('Cannot apply TextPartDeltas to non-TextParts')  # pragma: no cover
        return replace(part, content=part.content + self.content_delta)

    __repr__ = _utils.dataclasses_no_defaults_repr

content_delta 实例属性

content_delta: str

要添加到现有 TextPart 内容的增量文本内容。

part_delta_kind 类属性 实例属性

part_delta_kind: Literal['text'] = 'text'

部分增量类型标识符，用作区分符。

apply

apply(part: ModelResponsePart) -> TextPart

将此文本增量应用于现有的 TextPart。

参数

名称	类型	描述	默认值
part	ModelResponsePart	现有的模型响应部分，必须是 TextPart。	必需

返回

类型	描述
TextPart	一个带有更新文本内容的新的 TextPart。

引发

类型	描述
ValueError	如果 part 不是 TextPart。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def apply(self, part: ModelResponsePart) -> TextPart:
    """Apply this text delta to an existing `TextPart`.

    Args:
        part: The existing model response part, which must be a `TextPart`.

    Returns:
        A new `TextPart` with updated text content.

    Raises:
        ValueError: If `part` is not a `TextPart`.
    """
    if not isinstance(part, TextPart):
        raise ValueError('Cannot apply TextPartDeltas to non-TextParts')  # pragma: no cover
    return replace(part, content=part.content + self.content_delta)

ThinkingPartDelta dataclass

对 ThinkingPart 的部分更新（增量），以附加新的思考内容。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False, kw_only=True)
class ThinkingPartDelta:
    """A partial update (delta) for a `ThinkingPart` to append new thinking content."""

    content_delta: str | None = None
    """The incremental thinking content to add to the existing `ThinkingPart` content."""

    signature_delta: str | None = None
    """Optional signature delta.

    Note this is never treated as a delta — it can replace None.
    """

    part_delta_kind: Literal['thinking'] = 'thinking'
    """Part delta type identifier, used as a discriminator."""

    @overload
    def apply(self, part: ModelResponsePart) -> ThinkingPart: ...

    @overload
    def apply(self, part: ModelResponsePart | ThinkingPartDelta) -> ThinkingPart | ThinkingPartDelta: ...

    def apply(self, part: ModelResponsePart | ThinkingPartDelta) -> ThinkingPart | ThinkingPartDelta:
        """Apply this thinking delta to an existing `ThinkingPart`.

        Args:
            part: The existing model response part, which must be a `ThinkingPart`.

        Returns:
            A new `ThinkingPart` with updated thinking content.

        Raises:
            ValueError: If `part` is not a `ThinkingPart`.
        """
        if isinstance(part, ThinkingPart):
            new_content = part.content + self.content_delta if self.content_delta else part.content
            new_signature = self.signature_delta if self.signature_delta is not None else part.signature
            return replace(part, content=new_content, signature=new_signature)
        elif isinstance(part, ThinkingPartDelta):
            if self.content_delta is None and self.signature_delta is None:
                raise ValueError('Cannot apply ThinkingPartDelta with no content or signature')
            if self.signature_delta is not None:
                return replace(part, signature_delta=self.signature_delta)
            if self.content_delta is not None:
                return replace(part, content_delta=self.content_delta)
        raise ValueError(  # pragma: no cover
            f'Cannot apply ThinkingPartDeltas to non-ThinkingParts or non-ThinkingPartDeltas ({part=}, {self=})'
        )

    __repr__ = _utils.dataclasses_no_defaults_repr

content_delta 类属性 实例属性

content_delta: str | None = None

要添加到现有 ThinkingPart 内容的增量思考内容。

signature_delta 类属性 实例属性

signature_delta: str | None = None

可选的签名增量。

请注意，这永远不会被视为增量——它可以替换 None。

part_delta_kind 类属性 实例属性

part_delta_kind: Literal['thinking'] = 'thinking'

部分增量类型标识符，用作区分符。

apply

apply(part: ModelResponsePart) -> ThinkingPart

apply(
    part: ModelResponsePart | ThinkingPartDelta,
) -> ThinkingPart | ThinkingPartDelta

apply(
    part: ModelResponsePart | ThinkingPartDelta,
) -> ThinkingPart | ThinkingPartDelta

将此思考增量应用于现有的 ThinkingPart。

参数

名称	类型	描述	默认值
part	ModelResponsePart | ThinkingPartDelta	现有的模型响应部分，必须是 ThinkingPart。	必需

返回

类型	描述
ThinkingPart | ThinkingPartDelta	一个带有更新思考内容的新的 ThinkingPart。

引发

类型	描述
ValueError	如果 part 不是 ThinkingPart。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def apply(self, part: ModelResponsePart | ThinkingPartDelta) -> ThinkingPart | ThinkingPartDelta:
    """Apply this thinking delta to an existing `ThinkingPart`.

    Args:
        part: The existing model response part, which must be a `ThinkingPart`.

    Returns:
        A new `ThinkingPart` with updated thinking content.

    Raises:
        ValueError: If `part` is not a `ThinkingPart`.
    """
    if isinstance(part, ThinkingPart):
        new_content = part.content + self.content_delta if self.content_delta else part.content
        new_signature = self.signature_delta if self.signature_delta is not None else part.signature
        return replace(part, content=new_content, signature=new_signature)
    elif isinstance(part, ThinkingPartDelta):
        if self.content_delta is None and self.signature_delta is None:
            raise ValueError('Cannot apply ThinkingPartDelta with no content or signature')
        if self.signature_delta is not None:
            return replace(part, signature_delta=self.signature_delta)
        if self.content_delta is not None:
            return replace(part, content_delta=self.content_delta)
    raise ValueError(  # pragma: no cover
        f'Cannot apply ThinkingPartDeltas to non-ThinkingParts or non-ThinkingPartDeltas ({part=}, {self=})'
    )

ToolCallPartDelta dataclass

对 ToolCallPart 的部分更新（增量），以修改工具名称、参数或工具调用 ID。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False, kw_only=True)
class ToolCallPartDelta:
    """A partial update (delta) for a `ToolCallPart` to modify tool name, arguments, or tool call ID."""

    tool_name_delta: str | None = None
    """Incremental text to add to the existing tool name, if any."""

    args_delta: str | dict[str, Any] | None = None
    """Incremental data to add to the tool arguments.

    If this is a string, it will be appended to existing JSON arguments.
    If this is a dict, it will be merged with existing dict arguments.
    """

    tool_call_id: str | None = None
    """Optional tool call identifier, this is used by some models including OpenAI.

    Note this is never treated as a delta — it can replace None, but otherwise if a
    non-matching value is provided an error will be raised."""

    part_delta_kind: Literal['tool_call'] = 'tool_call'
    """Part delta type identifier, used as a discriminator."""

    def as_part(self) -> ToolCallPart | None:
        """Convert this delta to a fully formed `ToolCallPart` if possible, otherwise return `None`.

        Returns:
            A `ToolCallPart` if `tool_name_delta` is set, otherwise `None`.
        """
        if self.tool_name_delta is None:
            return None

        return ToolCallPart(self.tool_name_delta, self.args_delta, self.tool_call_id or _generate_tool_call_id())

    @overload
    def apply(self, part: ModelResponsePart) -> ToolCallPart: ...

    @overload
    def apply(self, part: ModelResponsePart | ToolCallPartDelta) -> ToolCallPart | ToolCallPartDelta: ...

    def apply(self, part: ModelResponsePart | ToolCallPartDelta) -> ToolCallPart | ToolCallPartDelta:
        """Apply this delta to a part or delta, returning a new part or delta with the changes applied.

        Args:
            part: The existing model response part or delta to update.

        Returns:
            Either a new `ToolCallPart` or an updated `ToolCallPartDelta`.

        Raises:
            ValueError: If `part` is neither a `ToolCallPart` nor a `ToolCallPartDelta`.
            UnexpectedModelBehavior: If applying JSON deltas to dict arguments or vice versa.
        """
        if isinstance(part, ToolCallPart):
            return self._apply_to_part(part)

        if isinstance(part, ToolCallPartDelta):
            return self._apply_to_delta(part)

        raise ValueError(  # pragma: no cover
            f'Can only apply ToolCallPartDeltas to ToolCallParts or ToolCallPartDeltas, not {part}'
        )

    def _apply_to_delta(self, delta: ToolCallPartDelta) -> ToolCallPart | ToolCallPartDelta:
        """Internal helper to apply this delta to another delta."""
        if self.tool_name_delta:
            # Append incremental text to the existing tool_name_delta
            updated_tool_name_delta = (delta.tool_name_delta or '') + self.tool_name_delta
            delta = replace(delta, tool_name_delta=updated_tool_name_delta)

        if isinstance(self.args_delta, str):
            if isinstance(delta.args_delta, dict):
                raise UnexpectedModelBehavior(
                    f'Cannot apply JSON deltas to non-JSON tool arguments ({delta=}, {self=})'
                )
            updated_args_delta = (delta.args_delta or '') + self.args_delta
            delta = replace(delta, args_delta=updated_args_delta)
        elif isinstance(self.args_delta, dict):
            if isinstance(delta.args_delta, str):
                raise UnexpectedModelBehavior(
                    f'Cannot apply dict deltas to non-dict tool arguments ({delta=}, {self=})'
                )
            updated_args_delta = {**(delta.args_delta or {}), **self.args_delta}
            delta = replace(delta, args_delta=updated_args_delta)

        if self.tool_call_id:
            delta = replace(delta, tool_call_id=self.tool_call_id)

        # If we now have enough data to create a full ToolCallPart, do so
        if delta.tool_name_delta is not None:
            return ToolCallPart(delta.tool_name_delta, delta.args_delta, delta.tool_call_id or _generate_tool_call_id())

        return delta

    def _apply_to_part(self, part: ToolCallPart) -> ToolCallPart:
        """Internal helper to apply this delta directly to a `ToolCallPart`."""
        if self.tool_name_delta:
            # Append incremental text to the existing tool_name
            tool_name = part.tool_name + self.tool_name_delta
            part = replace(part, tool_name=tool_name)

        if isinstance(self.args_delta, str):
            if isinstance(part.args, dict):
                raise UnexpectedModelBehavior(f'Cannot apply JSON deltas to non-JSON tool arguments ({part=}, {self=})')
            updated_json = (part.args or '') + self.args_delta
            part = replace(part, args=updated_json)
        elif isinstance(self.args_delta, dict):
            if isinstance(part.args, str):
                raise UnexpectedModelBehavior(f'Cannot apply dict deltas to non-dict tool arguments ({part=}, {self=})')
            updated_dict = {**(part.args or {}), **self.args_delta}
            part = replace(part, args=updated_dict)

        if self.tool_call_id:
            part = replace(part, tool_call_id=self.tool_call_id)
        return part

    __repr__ = _utils.dataclasses_no_defaults_repr

tool_name_delta 类属性 实例属性

tool_name_delta: str | None = None

要添加到现有工具名称的增量文本，如果有的话。

args_delta 类属性 实例属性

args_delta: str | dict[str, Any] | None = None

要添加到工具参数的增量数据。

如果这是一个字符串，它将被附加到现有的 JSON 参数。如果这是一个字典，它将与现有的字典参数合并。

tool_call_id 类属性 实例属性

tool_call_id: str | None = None

可选的工具调用标识符，包括 OpenAI 在内的一些模型会使用它。

请注意，这永远不会被视为增量——它可以替换 None，但否则如果提供了不匹配的值，将引发错误。

part_delta_kind 类属性 实例属性

part_delta_kind: Literal['tool_call'] = 'tool_call'

部分增量类型标识符，用作区分符。

as_part

as_part() -> ToolCallPart | None

如果可能，将此增量转换为一个完整的 ToolCallPart，否则返回 None。

返回

类型	描述
ToolCallPart | None	如果设置了 tool_name_delta，则为一个 ToolCallPart，否则为 None。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def as_part(self) -> ToolCallPart | None:
    """Convert this delta to a fully formed `ToolCallPart` if possible, otherwise return `None`.

    Returns:
        A `ToolCallPart` if `tool_name_delta` is set, otherwise `None`.
    """
    if self.tool_name_delta is None:
        return None

    return ToolCallPart(self.tool_name_delta, self.args_delta, self.tool_call_id or _generate_tool_call_id())

apply

apply(part: ModelResponsePart) -> ToolCallPart

apply(
    part: ModelResponsePart | ToolCallPartDelta,
) -> ToolCallPart | ToolCallPartDelta

apply(
    part: ModelResponsePart | ToolCallPartDelta,
) -> ToolCallPart | ToolCallPartDelta

将此增量应用于一个部分或增量，返回一个应用了更改的新部分或增量。

参数

名称	类型	描述	默认值
part	ModelResponsePart | ToolCallPartDelta	要更新的现有模型响应部分或增量。	必需

返回

类型	描述
ToolCallPart | ToolCallPartDelta	一个新的 ToolCallPart 或一个更新的 ToolCallPartDelta。

引发

类型	描述
ValueError	如果 part 既不是 ToolCallPart 也不是 ToolCallPartDelta。
UnexpectedModelBehavior	如果将 JSON 增量应用于字典参数，反之亦然。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

def apply(self, part: ModelResponsePart | ToolCallPartDelta) -> ToolCallPart | ToolCallPartDelta:
    """Apply this delta to a part or delta, returning a new part or delta with the changes applied.

    Args:
        part: The existing model response part or delta to update.

    Returns:
        Either a new `ToolCallPart` or an updated `ToolCallPartDelta`.

    Raises:
        ValueError: If `part` is neither a `ToolCallPart` nor a `ToolCallPartDelta`.
        UnexpectedModelBehavior: If applying JSON deltas to dict arguments or vice versa.
    """
    if isinstance(part, ToolCallPart):
        return self._apply_to_part(part)

    if isinstance(part, ToolCallPartDelta):
        return self._apply_to_delta(part)

    raise ValueError(  # pragma: no cover
        f'Can only apply ToolCallPartDeltas to ToolCallParts or ToolCallPartDeltas, not {part}'
    )

ModelResponsePartDelta 模块属性

ModelResponsePartDelta = Annotated[
    TextPartDelta | ThinkingPartDelta | ToolCallPartDelta,
    Discriminator("part_delta_kind"),
]

对任何模型响应部分的部分更新（增量）。

PartStartEvent dataclass

表示新部分已开始的事件。

如果收到多个具有相同索引的 PartStartEvent，新的应该完全替换旧的。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False, kw_only=True)
class PartStartEvent:
    """An event indicating that a new part has started.

    If multiple `PartStartEvent`s are received with the same index,
    the new one should fully replace the old one.
    """

    index: int
    """The index of the part within the overall response parts list."""

    part: ModelResponsePart
    """The newly started `ModelResponsePart`."""

    event_kind: Literal['part_start'] = 'part_start'
    """Event type identifier, used as a discriminator."""

    __repr__ = _utils.dataclasses_no_defaults_repr

index 实例属性

index: int

部分在整个响应部分列表中的索引。

part 实例属性

part: ModelResponsePart

新开始的 ModelResponsePart。

event_kind 类属性 实例属性

event_kind: Literal['part_start'] = 'part_start'

事件类型标识符，用作区分符。

PartDeltaEvent dataclass

表示对现有部分的增量更新的事件。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False, kw_only=True)
class PartDeltaEvent:
    """An event indicating a delta update for an existing part."""

    index: int
    """The index of the part within the overall response parts list."""

    delta: ModelResponsePartDelta
    """The delta to apply to the specified part."""

    event_kind: Literal['part_delta'] = 'part_delta'
    """Event type identifier, used as a discriminator."""

    __repr__ = _utils.dataclasses_no_defaults_repr

index 实例属性

index: int

部分在整个响应部分列表中的索引。

delta 实例属性

delta: ModelResponsePartDelta

要应用于指定部分的增量。

event_kind 类属性 实例属性

event_kind: Literal['part_delta'] = 'part_delta'

事件类型标识符，用作区分符。

FinalResultEvent dataclass

表示对当前模型请求的响应与输出模式匹配并将产生结果的事件。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False, kw_only=True)
class FinalResultEvent:
    """An event indicating the response to the current model request matches the output schema and will produce a result."""

    tool_name: str | None
    """The name of the output tool that was called. `None` if the result is from text content and not from a tool."""
    tool_call_id: str | None
    """The tool call ID, if any, that this result is associated with."""
    event_kind: Literal['final_result'] = 'final_result'
    """Event type identifier, used as a discriminator."""

    __repr__ = _utils.dataclasses_no_defaults_repr

tool_name 实例属性

tool_name: str | None

被调用的输出工具的名称。如果结果来自文本内容而不是工具，则为 None。

tool_call_id 实例属性

tool_call_id: str | None

与此结果关联的工具调用 ID（如果有）。

event_kind 类属性 实例属性

event_kind: Literal['final_result'] = 'final_result'

事件类型标识符，用作区分符。

ModelResponseStreamEvent 模块属性

ModelResponseStreamEvent = Annotated[
    PartStartEvent | PartDeltaEvent | FinalResultEvent,
    Discriminator("event_kind"),
]

模型响应流中的一个事件，开始一个新部分、对现有部分应用增量或指示最终结果。

FunctionToolCallEvent dataclass

表示对函数工具调用开始的事件。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class FunctionToolCallEvent:
    """An event indicating the start to a call to a function tool."""

    part: ToolCallPart
    """The (function) tool call to make."""

    _: KW_ONLY

    event_kind: Literal['function_tool_call'] = 'function_tool_call'
    """Event type identifier, used as a discriminator."""

    @property
    def tool_call_id(self) -> str:
        """An ID used for matching details about the call to its result."""
        return self.part.tool_call_id

    @property
    @deprecated('`call_id` is deprecated, use `tool_call_id` instead.')
    def call_id(self) -> str:
        """An ID used for matching details about the call to its result."""
        return self.part.tool_call_id  # pragma: no cover

    __repr__ = _utils.dataclasses_no_defaults_repr

part 实例属性

part: ToolCallPart

要进行的（函数）工具调用。

event_kind 类属性 实例属性

event_kind: Literal["function_tool_call"] = (
    "function_tool_call"
)

事件类型标识符，用作区分符。

tool_call_id 属性

tool_call_id: str

用于将调用的详细信息与其结果匹配的 ID。

call_id 属性

call_id: str

用于将调用的详细信息与其结果匹配的 ID。

FunctionToolResultEvent dataclass

表示函数工具调用结果的事件。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class FunctionToolResultEvent:
    """An event indicating the result of a function tool call."""

    result: ToolReturnPart | RetryPromptPart
    """The result of the call to the function tool."""

    _: KW_ONLY

    event_kind: Literal['function_tool_result'] = 'function_tool_result'
    """Event type identifier, used as a discriminator."""

    @property
    def tool_call_id(self) -> str:
        """An ID used to match the result to its original call."""
        return self.result.tool_call_id

    __repr__ = _utils.dataclasses_no_defaults_repr

result 实例属性

result: ToolReturnPart | RetryPromptPart

函数工具调用的结果。

event_kind 类属性 实例属性

event_kind: Literal["function_tool_result"] = (
    "function_tool_result"
)

事件类型标识符，用作区分符。

tool_call_id 属性

tool_call_id: str

用于将结果与其原始调用匹配的 ID。

BuiltinToolCallEvent dataclass

表示对内置工具调用开始的事件。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class BuiltinToolCallEvent:
    """An event indicating the start to a call to a built-in tool."""

    part: BuiltinToolCallPart
    """The built-in tool call to make."""

    _: KW_ONLY

    event_kind: Literal['builtin_tool_call'] = 'builtin_tool_call'
    """Event type identifier, used as a discriminator."""

part 实例属性

part: BuiltinToolCallPart

要进行的内置工具调用。

event_kind 类属性 实例属性

event_kind: Literal["builtin_tool_call"] = (
    "builtin_tool_call"
)

事件类型标识符，用作区分符。

BuiltinToolResultEvent dataclass

表示内置工具调用结果的事件。

源代码位于 pydantic_ai_slim/pydantic_ai/messages.py

@dataclass(repr=False)
class BuiltinToolResultEvent:
    """An event indicating the result of a built-in tool call."""

    result: BuiltinToolReturnPart
    """The result of the call to the built-in tool."""

    _: KW_ONLY

    event_kind: Literal['builtin_tool_result'] = 'builtin_tool_result'
    """Event type identifier, used as a discriminator."""

result 实例属性

result: BuiltinToolReturnPart

内置工具调用的结果。

event_kind 类属性 实例属性

event_kind: Literal["builtin_tool_result"] = (
    "builtin_tool_result"
)

事件类型标识符，用作区分符。

HandleResponseEvent 模块属性

HandleResponseEvent = Annotated[
    FunctionToolCallEvent
    | FunctionToolResultEvent
    | BuiltinToolCallEvent
    | BuiltinToolResultEvent,
    Discriminator("event_kind"),
]

处理模型响应时产生的事件，指示工具调用和结果。

AgentStreamEvent 模块属性

AgentStreamEvent = Annotated[
    ModelResponseStreamEvent | HandleResponseEvent,
    Discriminator("event_kind"),
]

代理流中的一个事件：模型响应流事件和响应处理事件。