跳过内容

消息和聊天记录

PydanticAI 提供了在代理运行期间交换的消息的访问。这些消息可以用于继续连贯的对话，并理解代理的表现。

从结果中访问消息

在运行代理后，您可以从 result 对象访问在该运行期间交换的消息。

两个 RunResult （由 Agent.run、Agent.run_sync 返回） 和 StreamedRunResult （由 Agent.run_stream 返回）具有以下方法：

流式运行结果和完整消息

在 StreamedRunResult 中，这些方法返回的消息将仅在流结束后包含最终结果消息。

例如，您已等待以下协程之一：

注意： 如果您使用 .stream_text(delta=True)，最终结果消息将不会添加到结果消息中，因为在这种情况下结果内容永远不会构建为一个字符串。

访问RunResult的方法的示例：

run_result_messages.py
from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', system_prompt='做一个有帮助的助手.')

result = agent.run_sync('告诉我一个笑话.')
print(result.data)
#> 你听说过牙膏丑闻吗？他们把它称为高露洁。

# 从运行中获取的所有消息
print(result.all_messages())
"""
[
    ModelRequest(
        parts=[
            SystemPromptPart(
                content='做一个有帮助的助手.',
                dynamic_ref=None,
                part_kind='system-prompt',
            ),
            UserPromptPart(
                content='告诉我一个笑话.',
                timestamp=datetime.datetime(...),
                part_kind='user-prompt',
            ),
        ],
        kind='request',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='你听说过牙膏丑闻吗？他们把它称为高露洁.',
                part_kind='text',
            )
        ],
        model_name='function:model_logic',
        timestamp=datetime.datetime(...),
        kind='response',
    ),
]
"""
(这个例子是完整的，可以“照原样”运行)

访问StreamedRunResult的方法示例：

streamed_run_result_messages.py
from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', system_prompt='Be a helpful assistant.')


async def main():
    async with agent.run_stream('Tell me a joke.') as result:
        # incomplete messages before the stream finishes
        print(result.all_messages())
        """
        [
            ModelRequest(
                parts=[
                    SystemPromptPart(
                        content='Be a helpful assistant.',
                        dynamic_ref=None,
                        part_kind='system-prompt',
                    ),
                    UserPromptPart(
                        content='Tell me a joke.',
                        timestamp=datetime.datetime(...),
                        part_kind='user-prompt',
                    ),
                ],
                kind='request',
            )
        ]
        """

        async for text in result.stream_text():
            print(text)
            #> Did you hear
            #> Did you hear about the toothpaste
            #> Did you hear about the toothpaste scandal? They called
            #> Did you hear about the toothpaste scandal? They called it Colgate.

        # complete messages once the stream finishes
        print(result.all_messages())
        """
        [
            ModelRequest(
                parts=[
                    SystemPromptPart(
                        content='Be a helpful assistant.',
                        dynamic_ref=None,
                        part_kind='system-prompt',
                    ),
                    UserPromptPart(
                        content='Tell me a joke.',
                        timestamp=datetime.datetime(...),
                        part_kind='user-prompt',
                    ),
                ],
                kind='request',
            ),
            ModelResponse(
                parts=[
                    TextPart(
                        content='Did you hear about the toothpaste scandal? They called it Colgate.',
                        part_kind='text',
                    )
                ],
                model_name='function:stream_model_logic',
                timestamp=datetime.datetime(...),
                kind='response',
            ),
        ]
        """
(This example is complete, it can be run "as is" — you'll need to add asyncio.run(main()) to run main)

使用消息作为进一步代理运行的输入

PydanticAI中消息历史的主要用途是保持多个代理运行之间的上下文。

要在运行中使用现有消息，请将它们传递给Agent.run、Agent.run_sync或Agent.run_stream的message_history参数。

如果 message_history 被设置且不为空，则不会生成新的系统提示 —— 我们假设现有的消息历史包含一个系统提示。

在对话中重用消息
from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', system_prompt='做一个有帮助的助手.')

result1 = agent.run_sync('给我讲个笑话.')
print(result1.data)
#> 你听说过牙膏丑闻吗？他们称之为高露洁。

result2 = agent.run_sync('解释一下？', message_history=result1.new_messages())
print(result2.data)
#> 这是一个由塞缪尔·科尔文发明的优秀笑话，不需要解释。

print(result2.all_messages())
"""
[
    ModelRequest(
        parts=[
            SystemPromptPart(
                content='做一个有帮助的助手.',
                dynamic_ref=None,
                part_kind='system-prompt',
            ),
            UserPromptPart(
                content='给我讲个笑话.',
                timestamp=datetime.datetime(...),
                part_kind='user-prompt',
            ),
        ],
        kind='request',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='你听说过牙膏丑闻吗？他们称之为高露洁.',
                part_kind='text',
            )
        ],
        model_name='function:model_logic',
        timestamp=datetime.datetime(...),
        kind='response',
    ),
    ModelRequest(
        parts=[
            UserPromptPart(
                content='解释一下？',
                timestamp=datetime.datetime(...),
                part_kind='user-prompt',
            )
        ],
        kind='request',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='这是一个由塞缪尔·科尔文发明的优秀笑话，不需要解释.',
                part_kind='text',
            )
        ],
        model_name='function:model_logic',
        timestamp=datetime.datetime(...),
        kind='response',
    ),
]
"""
(这个例子是完整的，可以“按原样”运行)

使用消息的其他方式

由于消息是由简单的数据类定义的，您可以手动创建和操作，例如用于测试。

消息格式与所使用的模型无关，因此您可以在不同的代理中使用消息，或者在同一代理中使用不同的模型。

在下面的示例中，我们在第二个代理运行中重用了第一个代理运行的消息，该运行使用了 openai:gpt-4o 模型，而第二个代理运行使用了 google-gla:gemini-1.5-pro 模型。

Reusing messages with a different model
from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', system_prompt='Be a helpful assistant.')

result1 = agent.run_sync('Tell me a joke.')
print(result1.data)
#> Did you hear about the toothpaste scandal? They called it Colgate.

result2 = agent.run_sync(
    'Explain?',
    model='google-gla:gemini-1.5-pro',
    message_history=result1.new_messages(),
)
print(result2.data)
#> This is an excellent joke invented by Samuel Colvin, it needs no explanation.

print(result2.all_messages())
"""
[
    ModelRequest(
        parts=[
            SystemPromptPart(
                content='Be a helpful assistant.',
                dynamic_ref=None,
                part_kind='system-prompt',
            ),
            UserPromptPart(
                content='Tell me a joke.',
                timestamp=datetime.datetime(...),
                part_kind='user-prompt',
            ),
        ],
        kind='request',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='Did you hear about the toothpaste scandal? They called it Colgate.',
                part_kind='text',
            )
        ],
        model_name='function:model_logic',
        timestamp=datetime.datetime(...),
        kind='response',
    ),
    ModelRequest(
        parts=[
            UserPromptPart(
                content='Explain?',
                timestamp=datetime.datetime(...),
                part_kind='user-prompt',
            )
        ],
        kind='request',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='This is an excellent joke invented by Samuel Colvin, it needs no explanation.',
                part_kind='text',
            )
        ],
        model_name='function:model_logic',
        timestamp=datetime.datetime(...),
        kind='response',
    ),
]
"""

示例

有关在对话中使用消息的更完整示例，请参见聊天应用示例。