pydantic_ai.messages

ModelMessage 的结构可以显示为图表

graph RL
    SystemPromptPart(SystemPromptPart) --- ModelRequestPart
    UserPromptPart(UserPromptPart) --- ModelRequestPart
    ToolReturnPart(ToolReturnPart) --- ModelRequestPart
    RetryPromptPart(RetryPromptPart) --- ModelRequestPart
    TextPart(TextPart) --- ModelResponsePart
    ToolCallPart(ToolCallPart) --- 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
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."""

    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) -> Event:
        return Event('gen_ai.system.message', body={'content': self.content, 'role': 'system'})

content instance-attribute

content: str

提示的内容。

timestamp class-attribute instance-attribute

timestamp: datetime = field(default_factory=now_utc)

提示的时间戳。

dynamic_ref class-attribute instance-attribute

dynamic_ref: str | None = None

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

仅当系统提示是动态的时设置，请参阅 system_prompt 了解更多信息。

part_kind class-attribute instance-attribute

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

部分类型标识符，这在所有部分上都可用作鉴别器。

AudioUrl dataclass

音频文件的 URL。

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

@dataclass
class AudioUrl:
    """A URL to an audio file."""

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

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

    @property
    def media_type(self) -> AudioMediaType:
        """Return the media type of the audio file, based on the url."""
        if self.url.endswith('.mp3'):
            return 'audio/mpeg'
        elif self.url.endswith('.wav'):
            return 'audio/wav'
        else:
            raise ValueError(f'Unknown audio file extension: {self.url}')

url instance-attribute

url: str

音频文件的 URL。

kind class-attribute instance-attribute

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

类型标识符，这在所有部分上都可用作鉴别器。

media_type property

media_type: AudioMediaType

根据 URL 返回音频文件的媒体类型。

ImageUrl dataclass

图像的 URL。

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

@dataclass
class ImageUrl:
    """A URL to an image."""

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

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

    @property
    def 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'Unknown image file extension: {self.url}')

    @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(self.media_type)

url instance-attribute

url: str

图像的 URL。

kind class-attribute instance-attribute

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

类型标识符，这在所有部分上都可用作鉴别器。

media_type property

media_type: ImageMediaType

根据 URL 返回图像的媒体类型。

format property

format: ImageFormat

图像的文件格式。

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

DocumentUrl dataclass

文档的 URL。

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

@dataclass
class DocumentUrl:
    """The URL of the document."""

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

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

    @property
    def media_type(self) -> str:
        """Return the media type of the document, based on the url."""
        type_, _ = guess_type(self.url)
        if type_ is None:
            raise RuntimeError(f'Unknown document file extension: {self.url}')
        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.
        """
        return _document_format(self.media_type)

url instance-attribute

url: str

文档的 URL。

kind class-attribute instance-attribute

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

类型标识符，这在所有部分上都可用作鉴别器。

media_type property

media_type: str

根据 URL 返回文档的媒体类型。

format property

format: DocumentFormat

文档的文件格式。

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

BinaryContent dataclass

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

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

@dataclass
class BinaryContent:
    """Binary content, e.g. an audio or image file."""

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

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

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

    @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_document(self) -> bool:
        """Return `True` if the media type is a document type."""
        return self.media_type in {
            'application/pdf',
            'text/plain',
            'text/csv',
            'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
            'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
            'text/html',
            'text/markdown',
            'application/vnd.ms-excel',
        }

    @property
    def format(self) -> str:
        """The file format of the binary content."""
        if self.is_audio:
            if self.media_type == 'audio/mpeg':
                return 'mp3'
            elif self.media_type == 'audio/wav':
                return 'wav'
        elif self.is_image:
            return _image_format(self.media_type)
        elif self.is_document:
            return _document_format(self.media_type)
        raise ValueError(f'Unknown media type: {self.media_type}')

data instance-attribute

data: bytes

二进制数据。

media_type instance-attribute

media_type: (
    AudioMediaType
    | ImageMediaType
    | DocumentMediaType
    | str
)

二进制数据的媒体类型。

kind class-attribute instance-attribute

kind: Literal['binary'] = 'binary'

类型标识符，这在所有部分上都可用作鉴别器。

is_audio property

is_audio: bool

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

is_image property

is_image: bool

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

is_document property

is_document: bool

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

format property

format: str

二进制内容的文件格式。

UserPromptPart dataclass

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

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

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

@dataclass
class UserPromptPart:
    """A user prompt, generally written by the end user.

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

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

    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) -> Event:
        if isinstance(self.content, str):
            content = self.content
        else:
            # TODO figure out what to record for images and audio
            content = [part if isinstance(part, str) else {'kind': part.kind} for part in self.content]
        return Event('gen_ai.user.message', body={'content': content, 'role': 'user'})

content instance-attribute

content: str | Sequence[UserContent]

提示的内容。

timestamp class-attribute instance-attribute

timestamp: datetime = field(default_factory=now_utc)

提示的时间戳。

part_kind class-attribute instance-attribute

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

部分类型标识符，这在所有部分上都可用作鉴别器。

ToolReturnPart dataclass

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

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

@dataclass
class ToolReturnPart:
    """A tool return message, this encodes the result of running a tool."""

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

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

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

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

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

    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) -> Event:
        return Event(
            'gen_ai.tool.message',
            body={'content': self.content, 'role': 'tool', 'id': self.tool_call_id, 'name': self.tool_name},
        )

tool_name instance-attribute

tool_name: str

调用的“工具”的名称。

content instance-attribute

content: Any

返回值。

tool_call_id class-attribute instance-attribute

tool_call_id: str | None = None

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

timestamp class-attribute instance-attribute

timestamp: datetime = field(default_factory=now_utc)

工具返回时的时间戳。

part_kind class-attribute instance-attribute

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

部分类型标识符，这在所有部分上都可用作鉴别器。

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')}

RetryPromptPart dataclass

发回给模型的消息，要求它重试。

发送此消息的原因有很多

• 工具参数的 Pydantic 验证失败，此处内容来自 Pydantic ValidationError
• 工具引发了 ModelRetry 异常
• 找不到工具名称对应的工具
• 当需要结构化响应时，模型返回了纯文本
• 结构化响应的 Pydantic 验证失败，此处内容来自 Pydantic ValidationError
• 结果验证器引发了 ModelRetry 异常

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

@dataclass
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]
    * a result 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.
    """

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

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

    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):
            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) -> 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(),
                    'role': 'tool',
                    'id': self.tool_call_id,
                    'name': self.tool_name,
                },
            )

content instance-attribute

content: list[ErrorDetails] | str

模型应该重试的原因和方式的详细信息。

如果重试是由 ValidationError 触发的，这将是错误详细信息列表。

tool_name class-attribute instance-attribute

tool_name: str | None = None

调用的工具的名称（如果有）。

tool_call_id class-attribute instance-attribute

tool_call_id: str | None = None

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

timestamp class-attribute instance-attribute

timestamp: datetime = field(default_factory=now_utc)

触发重试时的时间戳。

part_kind class-attribute instance-attribute

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):
        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 module-attribute

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

PydanticAI 发送到模型的消息部分。

ModelRequest dataclass

由 PydanticAI 生成并发送到模型的请求，例如从 PydanticAI 应用发送到模型的消息。

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

@dataclass
class ModelRequest:
    """A request generated by PydanticAI and sent to a model, e.g. a message from the PydanticAI app to the model."""

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

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

parts instance-attribute

parts: list[ModelRequestPart]

用户消息的各个部分。

kind class-attribute instance-attribute

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

消息类型标识符，这在所有部分上都可用作鉴别器。

TextPart dataclass

来自模型的纯文本响应。

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

@dataclass
class TextPart:
    """A plain text response from a model."""

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

    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)

content instance-attribute

content: str

响应的文本内容。

part_kind class-attribute instance-attribute

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)

ToolCallPart dataclass

来自模型的工具调用。

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

@dataclass
class ToolCallPart:
    """A tool call from a model."""

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

    args: str | dict[str, Any]
    """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 | None = None
    """Optional tool call identifier, this is used by some models including OpenAI."""

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

    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 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 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)

tool_name instance-attribute

tool_name: str

要调用的工具的名称。

args instance-attribute

args: str | dict[str, Any]

传递给工具的参数。

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

tool_call_id class-attribute instance-attribute

tool_call_id: str | None = None

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

part_kind class-attribute instance-attribute

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

部分类型标识符，这在所有部分上都可用作鉴别器。

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 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 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)

ModelResponsePart module-attribute

ModelResponsePart = Annotated[
    Union[TextPart, ToolCallPart],
    Discriminator("part_kind"),
]

模型返回的消息部分。

ModelResponse dataclass

来自模型的响应，例如从模型发送到 PydanticAI 应用的消息。

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

@dataclass
class ModelResponse:
    """A response from a model, e.g. a message from the model to the PydanticAI app."""

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

    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."""

    def otel_events(self) -> 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',  # TODO https://github.com/pydantic/pydantic-ai/issues/888
                        'function': {
                            'name': part.tool_name,
                            'arguments': part.args,
                        },
                    }
                )
            elif isinstance(part, TextPart):
                if body.get('content'):
                    body = new_event_body()
                body['content'] = part.content

        return result

parts instance-attribute

parts: list[ModelResponsePart]

模型消息的各个部分。

model_name class-attribute instance-attribute

model_name: str | None = None

生成响应的模型的名称。

timestamp class-attribute instance-attribute

timestamp: datetime = field(default_factory=now_utc)

响应的时间戳。

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

kind class-attribute instance-attribute

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

消息类型标识符，这在所有部分上都可用作鉴别器。

otel_events

otel_events() -> list[Event]

返回响应的 OpenTelemetry 事件。

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

def otel_events(self) -> 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',  # TODO https://github.com/pydantic/pydantic-ai/issues/888
                    'function': {
                        'name': part.tool_name,
                        'arguments': part.args,
                    },
                }
            )
        elif isinstance(part, TextPart):
            if body.get('content'):
                body = new_event_body()
            body['content'] = part.content

    return result

ModelMessage module-attribute

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

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

ModelMessagesTypeAdapter module-attribute

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

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

TextPartDelta dataclass

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

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

@dataclass
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."""

    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')
        return replace(part, content=part.content + self.content_delta)

content_delta instance-attribute

content_delta: str

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

part_delta_kind class-attribute instance-attribute

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')
    return replace(part, content=part.content + self.content_delta)

ToolCallPartDelta dataclass

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

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

@dataclass
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 both `tool_name_delta` and `args_delta` are set, otherwise `None`.
        """
        if self.tool_name_delta is None or self.args_delta is None:
            return None

        return ToolCallPart(
            self.tool_name_delta,
            self.args_delta,
            self.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(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:
            # Set the tool_call_id if it wasn't present, otherwise error if it has changed
            if delta.tool_call_id is not None and delta.tool_call_id != self.tool_call_id:
                raise UnexpectedModelBehavior(
                    f'Cannot apply a new tool_call_id to a ToolCallPartDelta that already has one ({delta=}, {self=})'
                )
            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 and delta.args_delta is not None:
            return ToolCallPart(
                delta.tool_name_delta,
                delta.args_delta,
                delta.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 not isinstance(part.args, str):
                raise UnexpectedModelBehavior(f'Cannot apply JSON deltas to non-JSON tool arguments ({part=}, {self=})')
            updated_json = part.args + self.args_delta
            part = replace(part, args=updated_json)
        elif isinstance(self.args_delta, dict):
            if not isinstance(part.args, dict):
                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:
            # Replace the tool_call_id entirely if given
            if part.tool_call_id is not None and part.tool_call_id != self.tool_call_id:
                raise UnexpectedModelBehavior(
                    f'Cannot apply a new tool_call_id to a ToolCallPartDelta that already has one ({part=}, {self=})'
                )
            part = replace(part, tool_call_id=self.tool_call_id)
        return part

tool_name_delta class-attribute instance-attribute

tool_name_delta: str | None = None

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

args_delta class-attribute instance-attribute

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

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

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

tool_call_id class-attribute instance-attribute

tool_call_id: str | None = None

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

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

part_delta_kind class-attribute instance-attribute

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

部分增量类型标识符，用作鉴别器。

as_part

as_part() -> ToolCallPart | None

如果可能，将此增量转换为完全形成的 ToolCallPart，否则返回 None。

返回值

类型	描述
ToolCallPart | None	如果 tool_name_delta 和 args_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 both `tool_name_delta` and `args_delta` are set, otherwise `None`.
    """
    if self.tool_name_delta is None or self.args_delta is None:
        return None

    return ToolCallPart(
        self.tool_name_delta,
        self.args_delta,
        self.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(f'Can only apply ToolCallPartDeltas to ToolCallParts or ToolCallPartDeltas, not {part}')

ModelResponsePartDelta module-attribute

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

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

PartStartEvent dataclass

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

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

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

@dataclass
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."""

index instance-attribute

index: int

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

part instance-attribute

part: ModelResponsePart

新开始的 ModelResponsePart。

event_kind class-attribute instance-attribute

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

事件类型标识符，用作鉴别器。

PartDeltaEvent dataclass

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

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

@dataclass
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."""

index instance-attribute

index: int

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

delta instance-attribute

delta: ModelResponsePartDelta

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

event_kind class-attribute instance-attribute

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

事件类型标识符，用作鉴别器。

FinalResultEvent dataclass

指示对当前模型请求的响应与结果模式匹配的事件。

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

@dataclass
class FinalResultEvent:
    """An event indicating the response to the current model request matches the result schema."""

    tool_name: str | None
    """The name of the result 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."""

tool_name instance-attribute

tool_name: str | None

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

tool_call_id instance-attribute

tool_call_id: str | None

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

event_kind class-attribute instance-attribute

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

事件类型标识符，用作鉴别器。

ModelResponseStreamEvent module-attribute

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

模型响应流中的事件，可以是开始新部分或将增量应用于现有部分。

AgentStreamEvent module-attribute

AgentStreamEvent = Annotated[
    Union[PartStartEvent, PartDeltaEvent, FinalResultEvent],
    Discriminator("event_kind"),
]

代理流中的事件。

FunctionToolCallEvent dataclass

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

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

@dataclass
class FunctionToolCallEvent:
    """An event indicating the start to a call to a function tool."""

    part: ToolCallPart
    """The (function) tool call to make."""
    call_id: str = field(init=False)
    """An ID used for matching details about the call to its result. If present, defaults to the part's tool_call_id."""
    event_kind: Literal['function_tool_call'] = 'function_tool_call'
    """Event type identifier, used as a discriminator."""

    def __post_init__(self):
        self.call_id = self.part.tool_call_id or str(uuid.uuid4())

part instance-attribute

part: ToolCallPart

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

call_id class-attribute instance-attribute

call_id: str = field(init=False)

用于将有关调用的详细信息与其结果匹配的 ID。如果存在，则默认为部分的 tool_call_id。

event_kind class-attribute instance-attribute

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

事件类型标识符，用作鉴别器。

FunctionToolResultEvent dataclass

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

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

@dataclass
class FunctionToolResultEvent:
    """An event indicating the result of a function tool call."""

    result: ToolReturnPart | RetryPromptPart
    """The result of the call to the function tool."""
    tool_call_id: str
    """An ID used to match the result to its original call."""
    event_kind: Literal['function_tool_result'] = 'function_tool_result'
    """Event type identifier, used as a discriminator."""

result instance-attribute

result: ToolReturnPart | RetryPromptPart

调用函数工具的结果。

tool_call_id instance-attribute

tool_call_id: str

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

event_kind class-attribute instance-attribute

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

事件类型标识符，用作鉴别器。