字数 4772，阅读大约需 24 分钟
Python：Agent开发为什么使用Pydantic AI，核心概念中篇（函数工具、输出、消息和历史）
什么是pydantic ai
超级好用的AI Agent开发框架：Pydantic AI。构建人工智能驱动的应用程序往往会导致非结构化输出、类型不匹配和生产可靠性问题。将大型语言模型集成到Python应用程序中的传统方法缺乏生产系统所需的结构和验证。Pydantic AI通过将Pydantic的数据验证与用于大型语言模型交互的智能体框架相结合，解决了这一问题。
1. Pydantic AI 是由 Pydantic 团队开发的 Python Agent 框架,旨在帮助开发者快速、自信且轻松地构建生产级的生成式 AI 应用程序和工作流 PydanticAI（
GenAI智能体框架，Pydantic风格）。2. 这个框架的设计初衷是将 FastAPI 的开发体验带到生成式 AI 应用和 Agent 开发中。Pydantic AI 的设计理念是将生产级工程最佳实践带入 AI Agent 开发领域,让构建可靠的 AI 应用变得更加简单和安全。
3. Pydantic AI 是一个 Python 智能体框架，为大语言模型（LLM）应用带来结构化和类型安全性。与仅返回原始文本的基础 LLM 集成不同，Pydantic AI 将 AI 交互视为具有经过验证的输入和输出的结构化对话，从而更易于构建可靠的应用程序。
包含：结构化输出验证、函数工具、类型安全、系统提示指令、依赖注入、多执行模式、智能体复用
1. 结构化输出验证——人工智能响应会自动符合Pydantic模型，从而防止解析错误并确保数据一致性
2. 函数工具——大型语言模型可以在对话过程中调用你的Python函数，从而使它们能够获取真实数据并进行计算。
3. 类型安全——全面的类型支持意味着在构建复杂的AI工作流时，运行时错误更少，IDE辅助更出色。
4. 系统提示-为AI代理定义清晰的指令，在所有交互中保持一致
5. 依赖注入——清晰地在所有智能体组件之间共享上下文、数据库连接和用户偏好
6. 多种执行模式——可根据需求同步、异步运行智能体，或实时流式传输响应
7. 智能体可复用性——创建一次智能体，即可在整个应用程序中复用，类似于FastAPI路由。
Pydantic AI 框架的文档地址[1]
核心概念（函数工具、输出、消息和历史）
函数工具
执行操作和检索额外信息，函数工具基本上是 RAG（检索增强生成）中的“R”——它们通过让模型请求额外信息来增强模型的能力。
1. 通过 @agent.tool 装饰器 — 适用于需要访问代理上下文的工具
2. 通过 @agent.tool_plain 装饰器 — 适用于不需要访问代理上下文的工具
3. 通过 Agent 的 tools 关键字参数，它可以接受普通函数或 Tool 的实例
1. @agent.tool 装饰器
• 用途：适用于需要访问代理上下文（RunContext）的工具
• 特点：
• 工具函数可以接收
RunContext参数• 可以访问依赖注入（dependencies）
• 可以访问当前的对话状态和历史
• 适合需要上下文信息的复杂工具
示例：
from pydantic_ai import Agent, RunContext
agent = Agent('openai:gpt-4')
@agent.tool
async def get_user_data(ctx: RunContext[UserDeps], user_id: int) -> str:
# 可以访问 ctx.deps 获取依赖
# 可以访问 ctx.messages 获取对话历史
user = await ctx.deps.database.get_user(user_id)
return f"User: {user.name}"2. @agent.tool_plain 装饰器
• 用途：适用于不需要访问代理上下文的简单工具
• 特点：
• 纯函数，不接收
RunContext• 更简洁，适合独立的工具函数
• 性能稍好（无需传递上下文）
• 适合无状态的计算型工具
示例：
@agent.tool_plain
def calculate_sum(a: int, b: int) -> int:
"""简单的加法计算"""
return a + b3. 通过 Agent 的 tools 参数
• 用途：在创建 Agent 时直接注册工具
• 特点：
• 可以传递普通函数
• 可以传递
Tool实例• 适合复用工具或从外部导入工具
• 更灵活，可以动态配置
示例：
def search_web(query: str) -> str:
"""搜索网络"""
return f"Results for: {query}"
# 方式1：直接传递函数
agent = Agent(
'openai:gpt-4',
tools=[search_web]
)
# 方式2：传递 Tool 实例
from pydantic_ai import Tool
custom_tool = Tool(search_web, name="web_search")
agent = Agent(
'openai:gpt-4',
tools=[custom_tool]
)三种方式的对比
使用建议
• 使用
@agent.tool：当工具需要访问数据库、用户信息、会话状态等上下文数据• 使用
@agent.tool_plain：当工具是纯计算函数，如数学运算、字符串处理等• 使用
tools参数：当需要在多个 agent 之间共享工具，或需要动态添加工具时
输出
1. 纯文本输出 - 最基本的文本返回
2. 结构化数据输出 - 使用 Pydantic Model、TypedDict 和联合类型
3. 函数输出 - 使用函数处理输出，支持额外的业务逻辑
4. 文本输出函数 - 将纯文本传递给函数处理
5. 三种输出模式:
• 工具输出（Tool Output）- 默认模式
• 原生输出（Native Output）- 使用模型原生功能
• 提示输出（Prompted Output）- 通过指令提示
6. 输出验证器 - 使用
@agent.output_validator进行自定义验证7. 流式文本输出 - 实时获取文本响应
8. 流式结构化输出 - 实时获取结构化数据
9. 自定义 JSON Schema - 使用
StructuredDict定义灵活的输出格式
示例
"""
Pydantic AI 输出（Output）概念完整示例
演示三种输出类型、三种输出模式、验证和流式输出
"""
import asyncio
from datetime import date
from typing import List
from pydantic import BaseModel, ValidationError
from pydantic_ai import Agent, RunContext, ModelRetry, ToolOutput, NativeOutput, PromptedOutput, TextOutput
from typing_extensions import TypedDict, NotRequired
# ============================================================
# 1. 纯文本输出 (Text Output)
# ============================================================
print("=" * 60)
print("1. 纯文本输出示例")
print("=" * 60)
# 默认情况下，Agent 返回纯文本
text_agent = Agent('openai:gpt-4o-mini')
result = text_agent.run_sync('用一句话介绍 Python')
print(f"纯文本输出: {result.output}")
print(f"使用情况: {result.usage()}\n")
# ============================================================
# 2. 结构化数据输出 (Structured Output)
# ============================================================
print("=" * 60)
print("2. 结构化数据输出示例")
print("=" * 60)
# 2.1 使用 Pydantic Model
class CityInfo(BaseModel):
"""城市信息模型"""
city: str
country: str
population: int
city_agent = Agent('openai:gpt-4o-mini', output_type=CityInfo)
result = city_agent.run_sync('告诉我关于东京的信息')
print(f"城市信息: {result.output}")
print(f"类型: {type(result.output)}\n")
# 2.2 使用 TypedDict
class PersonProfile(TypedDict):
"""个人档案"""
name: str
age: int
occupation: str
person_agent = Agent('openai:gpt-4o-mini', output_type=PersonProfile)
result = person_agent.run_sync('创建一个软件工程师的档案')
print(f"个人档案: {result.output}")
print(f"类型: {type(result.output)}\n")
# 2.3 联合类型输出
class Box(BaseModel):
"""盒子尺寸"""
width: int
height: int
depth: int
units: str
# 可以返回结构化数据或文本
union_agent = Agent(
'openai:gpt-4o-mini',
output_type=[Box, str],
system_prompt="提取盒子尺寸，如果无法提取所有数据，请要求用户重试。"
)
result = union_agent.run_sync('盒子是 10x20x30')
print(f"缺少单位时: {result.output}")
result = union_agent.run_sync('盒子是 10x20x30 厘米')
print(f"完整信息时: {result.output}\n")
# ============================================================
# 3. 函数输出 (Output Functions)
# ============================================================
print("=" * 60)
print("3. 函数输出示例")
print("=" * 60)
class ProductInfo(BaseModel):
"""产品信息"""
name: str
price: float
category: str
def process_product(name: str, price: float, category: str) -> ProductInfo:
"""处理产品信息并返回 ProductInfo 对象"""
# 可以在这里进行额外的处理、验证或业务逻辑
print(f" [处理中] 产品名称: {name}, 价格: {price}, 类别: {category}")
return ProductInfo(name=name, price=price, category=category)
# 使用函数作为输出
function_agent = Agent[None, ProductInfo](
'openai:gpt-4o-mini',
output_type=process_product,
)
result = function_agent.run_sync('创建一个笔记本电脑产品，价格 5000 元')
print(f"函数输出结果: {result.output}")
print(f"类型: {type(result.output)}\n")
# 3.2 带 RunContext 的输出函数
class OrderSummary(BaseModel):
"""订单摘要"""
items: List[str]
total: float
async def create_order(ctx: RunContext, items: List[str], total: float) -> OrderSummary:
"""创建订单，可以访问上下文"""
print(f" [上下文信息] 消息数: {len(ctx.messages)}")
# 可以进行验证，如果不满足条件可以要求模型重试
if total < 0:
raise ModelRetry("订单总额不能为负数")
return OrderSummary(items=items, total=total)
order_agent = Agent[None, OrderSummary](
'openai:gpt-4o-mini',
output_type=create_order,
)
result = await order_agent.run('创建一个包含苹果和香蕉的订单，总计 50 元')
print(f"订单输出: {result.output}\n")
# ============================================================
# 4. 文本输出函数 (Text Output Function)
# ============================================================
print("=" * 60)
print("4. 文本输出函数示例")
print("=" * 60)
def split_into_words(text: str) -> List[str]:
"""将文本分割成单词列表"""
return text.split()
text_func_agent = Agent(
'openai:gpt-4o-mini',
output_type=TextOutput(split_into_words),
)
result = text_func_agent.run_sync('Albert Einstein was a physicist')
print(f"文本分词结果: {result.output}\n")
# ============================================================
# 5. 三种输出模式 (Output Modes)
# ============================================================
print("=" * 60)
print("5. 三种输出模式示例")
print("=" * 60)
class Fruit(BaseModel):
"""水果"""
name: str
color: str
class Vehicle(BaseModel):
"""交通工具"""
name: str
wheels: int
# 5.1 工具输出 (Tool Output) - 默认模式
print("5.1 工具输出模式 (默认)")
tool_agent = Agent(
'openai:gpt-4o-mini',
output_type=[
ToolOutput(Fruit, name='return_fruit'),
ToolOutput(Vehicle, name='return_vehicle'),
],
)
result = tool_agent.run_sync('什么是香蕉？')
print(f"工具输出: {result.output}")
result = tool_agent.run_sync('什么是自行车？')
print(f"工具输出: {result.output}\n")
# 5.2 原生输出 (Native Output)
print("5.2 原生输出模式")
native_agent = Agent(
'openai:gpt-4o-mini',
output_type=NativeOutput(
[Fruit, Vehicle],
name='Fruit_or_vehicle',
description='返回一个水果或交通工具'
),
)
result = native_agent.run_sync('什么是汽车？')
print(f"原生输出: {result.output}\n")
# 5.3 提示输出 (Prompted Output)
print("5.3 提示输出模式")
class Device(BaseModel):
"""设备"""
name: str
kind: str
prompted_agent = Agent(
'openai:gpt-4o-mini',
output_type=PromptedOutput(
[Vehicle, Device],
name='Vehicle or device',
description='返回一个交通工具或设备'
),
)
result = prompted_agent.run_sync('什么是 iPhone？')
print(f"提示输出: {result.output}\n")
# ============================================================
# 6. 输出验证器 (Output Validator)
# ============================================================
print("=" * 60)
print("6. 输出验证器示例")
print("=" * 60)
class EmailData(BaseModel):
"""邮箱数据"""
email: str
domain: str
email_agent = Agent('openai:gpt-4o-mini', output_type=EmailData)
@email_agent.output_validator
async def validate_email(ctx: RunContext, output: EmailData) -> EmailData:
"""验证邮箱格式"""
print(f" [验证中] 邮箱: {output.email}, 域名: {output.domain}")
# 检查域名是否在邮箱中
if output.domain not in output.email:
raise ModelRetry(f'域名 {output.domain} 必须在邮箱 {output.email} 中')
# 检查是否包含 @
if '@' not in output.email:
raise ModelRetry('邮箱必须包含 @ 符号')
return output
try:
result = await email_agent.run('提取邮箱: [email protected]')
print(f"验证通过: {result.output}\n")
except Exception as e:
print(f"验证失败: {e}\n")
# ============================================================
# 7. 流式文本输出 (Streaming Text)
# ============================================================
print("=" * 60)
print("7. 流式文本输出示例")
print("=" * 60)
async def stream_text_example():
stream_agent = Agent('openai:gpt-4o-mini')
print("7.1 完整文本流式输出:")
async with stream_agent.run_stream('用三句话介绍人工智能') as result:
async for text in result.stream_text():
print(f" 收到: {text[:50]}..." if len(text) > 50 else f" 收到: {text}")
print("\n7.2 增量文本流式输出:")
async with stream_agent.run_stream('列举三个编程语言') as result:
async for delta in result.stream_text(delta=True):
print(f" 增量: {delta}")
await stream_text_example()
print()
# ============================================================
# 8. 流式结构化输出 (Streaming Structured Output)
# ============================================================
print("=" * 60)
print("8. 流式结构化输出示例")
print("=" * 60)
class UserProfile(TypedDict):
"""用户档案"""
name: str
dob: NotRequired[date]
bio: NotRequired[str]
async def stream_structured_example():
profile_agent = Agent(
'openai:gpt-4o-mini',
output_type=UserProfile,
system_prompt='从输入中提取用户档案'
)
user_input = '我的名字是张三，1990年1月28日出生，我喜欢编程和阅读。'
print("流式结构化输出:")
async with profile_agent.run_stream(user_input) as result:
async for profile in result.stream_output():
print(f" 当前档案: {profile}")
await stream_structured_example()
print()
# ============================================================
# 9. 自定义 JSON Schema (StructuredDict)
# ============================================================
print("=" * 60)
print("9. 自定义 JSON Schema 示例")
print("=" * 60)
from pydantic_ai import StructuredDict
HumanDict = StructuredDict(
{
'type': 'object',
'properties': {
'name': {'type': 'string'},
'age': {'type': 'integer'}
},
'required': ['name', 'age']
},
name='Human',
description='具有姓名和年龄的人类'
)
dict_agent = Agent('openai:gpt-4o-mini', output_type=HumanDict)
result = dict_agent.run_sync('创建一个 25 岁的人，名字叫李四')
print(f"自定义 Schema 输出: {result.output}")
print(f"类型: {type(result.output)}")
# ============================================================
# 运行主函数
# ============================================================
if __name__ == '__main__':
print("\n" + "=" * 60)
print("所有示例执行完成！")
print("=" * 60)消息和聊天历史
访问智能体运行期间交换的消息的功能。这些消息既可以用来继续连贯的对话，也可以用来了解智能体的执行情况
消息类型
ModelRequest：用户的请求消息
ModelResponse：模型的响应消息，包含使用情况、模型名称和时间戳
访问消息的方法
RunResult（由 Agent.run、Agent.run_sync 返回）和 StreamedRunResult（由 Agent.run_stream 返回）都具有以下方法：
• all_messages() - 获取所有消息
• new_messages() - 仅获取当前运行的新消息
• all_messages_json() - 序列化所有消息为JSON
• new_messages_json() - 序列化新消息为JSON
消息和聊天记录 - Pydantic AI 框架[2]
示例
"""
Pydantic AI 消息和历史消息（Messages and Chat History）完整示例
演示消息访问、对话上下文维护、序列化和历史处理
"""
import asyncio
import json
from datetime import datetime
from pydantic_ai import Agent, RunContext
from pydantic_ai.messages import (
ModelMessage,
ModelRequest,
ModelResponse,
UserPromptPart,
TextPart,
ModelMessagesTypeAdapter
)
from pydantic_core import to_jsonable_python
# ============================================================
# 1. 基本消息访问
# ============================================================
print("=" * 60)
print("1. 从结果访问消息")
print("=" * 60)
agent = Agent('openai:gpt-4o-mini', system_prompt='你是一个有用的助手。')
# 第一次运行
result1 = agent.run_sync('给我讲个笑话。')
print(f"第一次响应: {result1.output}\n")
# 访问所有消息
all_messages = result1.all_messages()
print(f"消息总数: {len(all_messages)}")
for i, msg in enumerate(all_messages):
if isinstance(msg, ModelRequest):
print(f" [{i}] 请求: {msg.parts[0].content if msg.parts else 'N/A'}")
elif isinstance(msg, ModelResponse):
print(f" [{i}] 响应: {msg.parts[0].content[:50] if msg.parts else 'N/A'}...")
# 仅访问新消息
new_messages = result1.new_messages()
print(f"\n新消息数: {len(new_messages)}")
# ============================================================
# 2. 维护对话上下文
# ============================================================
print("\n" + "=" * 60)
print("2. 维护多轮对话上下文")
print("=" * 60)
# 第一轮对话
result1 = agent.run_sync('我的名字是张三，我喜欢编程。')
print(f"第一轮: {result1.output}")
# 第二轮对话 - 使用历史消息
result2 = agent.run_sync(
'我叫什么名字？',
message_history=result1.new_messages() # 传递上一轮的消息
)
print(f"第二轮: {result2.output}")
# 第三轮对话 - 累积历史
result3 = agent.run_sync(
'我的爱好是什么？',
message_history=result2.all_messages() # 使用全部历史
)
print(f"第三轮: {result3.output}")
print(f"\n总消息数: {len(result3.all_messages())}")
# ============================================================
# 3. 消息序列化和持久化
# ============================================================
print("\n" + "=" * 60)
print("3. 消息序列化和持久化")
print("=" * 60)
# 3.1 序列化为 Python 对象
history_messages = result1.all_messages()
python_objects = to_jsonable_python(history_messages)
print(f"序列化为 Python 对象: {type(python_objects)}")
print(f"对象数量: {len(python_objects)}")
# 3.2 序列化为 JSON 字符串
json_str = json.dumps(python_objects, indent=2, ensure_ascii=False)
print(f"\nJSON 字符串长度: {len(json_str)}")
print(f"JSON 预览:\n{json_str[:200]}...\n")
# 3.3 反序列化回消息对象
restored_messages = ModelMessagesTypeAdapter.validate_python(python_objects)
print(f"恢复的消息数: {len(restored_messages)}")
# 3.4 使用恢复的消息继续对话
result_restored = agent.run_sync(
'再讲一个笑话',
message_history=restored_messages
)
print(f"使用恢复消息的响应: {result_restored.output}")
# 3.5 直接使用 JSON 方法
json_bytes = result1.all_messages_json()
print(f"\n使用 all_messages_json(): {len(json_bytes)} 字节")
new_json_bytes = result1.new_messages_json()
print(f"使用 new_messages_json(): {len(new_json_bytes)} 字节")
# ============================================================
# 4. 历史处理器 - 过滤消息
# ============================================================
print("\n" + "=" * 60)
print("4. 历史处理器 - 过滤响应消息")
print("=" * 60)
def filter_responses(messages: list[ModelMessage]) -> list[ModelMessage]:
"""移除所有 ModelResponse 消息，只保留 ModelRequest"""
filtered = [msg for msg in messages if isinstance(msg, ModelRequest)]
print(f" [过滤器] 原始消息数: {len(messages)}, 过滤后: {len(filtered)}")
return filtered
# 创建带有历史处理器的代理
filter_agent = Agent(
'openai:gpt-4o-mini',
history_processors=[filter_responses]
)
# 创建一些对话历史
message_history = [
ModelRequest(parts=[UserPromptPart(content='2+2等于多少？')]),
ModelResponse(parts=[TextPart(content='2+2等于4')]),
ModelRequest(parts=[UserPromptPart(content='3+3等于多少？')]),
ModelResponse(parts=[TextPart(content='3+3等于6')]),
]
result = filter_agent.run_sync(
'5+5等于多少？',
message_history=message_history
)
print(f"过滤后的结果: {result.output}")
# ============================================================
# 5. 历史处理器 - 限制消息数量
# ============================================================
print("\n" + "=" * 60)
print("5. 历史处理器 - 限制消息数量")
print("=" * 60)
def limit_messages(messages: list[ModelMessage]) -> list[ModelMessage]:
"""只保留最近的5条消息"""
max_messages = 5
if len(messages) > max_messages:
print(f" [限制器] 截断消息: {len(messages)} -> {max_messages}")
return messages[-max_messages:]
return messages
limit_agent = Agent(
'openai:gpt-4o-mini',
history_processors=[limit_messages]
)
# 创建大量历史消息
large_history = []
for i in range(20):
large_history.append(
ModelRequest(parts=[UserPromptPart(content=f'问题 {i+1}')])
)
large_history.append(
ModelResponse(parts=[TextPart(content=f'答案 {i+1}')])
)
result = limit_agent.run_sync(
'最新的问题',
message_history=large_history
)
print(f"限制后的结果: {result.output[:100]}...")
# ============================================================
# 6. 历史处理器 - 异步处理和总结
# ============================================================
print("\n" + "=" * 60)
print("6. 历史处理器 - 异步总结旧消息")
print("=" * 60)
# 创建用于总结的代理
summarize_agent = Agent(
'openai:gpt-4o-mini',
system_prompt='总结对话内容，省略闲聊和无关话题。'
)
async def summarize_old_messages(
ctx: RunContext,
messages: list[ModelMessage]
) -> list[ModelMessage]:
"""总结最旧的10条消息"""
if len(messages) > 10:
print(f" [总结器] 正在总结 {len(messages)} 条消息的前10条")
oldest_messages = messages[:10]
summary = await summarize_agent.run(message_history=oldest_messages)
# 返回总结和最后一条消息
return summary.new_messages() + messages[-1:]
return messages
async def async_summary_example():
summary_agent = Agent(
'openai:gpt-4o-mini',
history_processors=[summarize_old_messages]
)
# 创建长对话历史
history = []
for i in range(12):
history.append(ModelRequest(parts=[UserPromptPart(content=f'话题 {i+1}')]))
history.append(ModelResponse(parts=[TextPart(content=f'关于话题 {i+1} 的详细回答')]))
result = await summary_agent.run('最终问题', message_history=history)
print(f"总结后的结果: {result.output[:100]}...")
await async_summary_example()
# ============================================================
# 7. 历史处理器 - 多个处理器链式调用
# ============================================================
print("\n" + "=" * 60)
print("7. 多个历史处理器链式调用")
print("=" * 60)
def remove_system_prompts(messages: list[ModelMessage]) -> list[ModelMessage]:
"""移除系统提示消息"""
print(f" [处理器1] 移除系统提示")
# 这里简化处理，实际可能需要更复杂的逻辑
return messages
def truncate_long_messages(messages: list[ModelMessage]) -> list[ModelMessage]:
"""截断过长的消息"""
print(f" [处理器2] 截断长消息")
max_length = 100
processed = []
for msg in messages:
if isinstance(msg, ModelRequest) and msg.parts:
if len(msg.parts[0].content) > max_length:
truncated_content = msg.parts[0].content[:max_length] + "..."
processed.append(
ModelRequest(parts=[UserPromptPart(content=truncated_content)])
)
else:
processed.append(msg)
else:
processed.append(msg)
return processed
chain_agent = Agent(
'openai:gpt-4o-mini',
history_processors=[
remove_system_prompts,
truncate_long_messages,
limit_messages
]
)
result = chain_agent.run_sync('测试多处理器', message_history=large_history)
print(f"链式处理后的结果: {result.output[:100]}...")
# ============================================================
# 8. 实际应用 - 保存到文件
# ============================================================
print("\n" + "=" * 60)
print("8. 实际应用 - 消息持久化到文件")
print("=" * 60)
def save_conversation_to_file(messages: list[ModelMessage], filename: str):
"""保存对话到JSON文件"""
python_objects = to_jsonable_python(messages)
with open(filename, 'w', encoding='utf-8') as f:
json.dump(python_objects, f, indent=2, ensure_ascii=False)
print(f" 对话已保存到: {filename}")
def load_conversation_from_file(filename: str) -> list[ModelMessage]:
"""从JSON文件加载对话"""
with open(filename, 'r', encoding='utf-8') as f:
python_objects = json.load(f)
messages = ModelMessagesTypeAdapter.validate_python(python_objects)
print(f" 已从 {filename} 加载 {len(messages)} 条消息")
return messages
# 保存对话
save_conversation_to_file(result3.all_messages(), 'conversation.json')
# 加载对话
loaded_messages = load_conversation_from_file('conversation.json')
# 使用加载的消息继续对话
result_from_file = agent.run_sync(
'继续我们之前的对话',
message_history=loaded_messages
)
print(f"从文件恢复后的响应: {result_from_file.output[:100]}...")
# ============================================================
# 9. 检查消息结构
# ============================================================
print("\n" + "=" * 60)
print("9. 检查消息结构详情")
print("=" * 60)
sample_messages = result1.all_messages()
for i, msg in enumerate(sample_messages[:3]): # 只显示前3条
print(f"\n消息 {i+1}:")
print(f" 类型: {type(msg).__name__}")
if isinstance(msg, ModelRequest):
print(f" 部分数: {len(msg.parts)}")
if msg.parts:
part = msg.parts[0]
print(f" 内容: {part.content[:50]}...")
if hasattr(part, 'timestamp'):
print(f" 时间戳: {part.timestamp}")
elif isinstance(msg, ModelResponse):
print(f" 部分数: {len(msg.parts)}")
if msg.parts:
part = msg.parts[0]
if isinstance(part, TextPart):
print(f" 内容: {part.content[:50]}...")
if hasattr(msg, 'usage') and msg.usage:
print(f" 使用情况: 输入={msg.usage.input_tokens}, 输出={msg.usage.output_tokens}")
if hasattr(msg, 'model_name'):
print(f" 模型: {msg.model_name}")
if hasattr(msg, 'timestamp'):
print(f" 时间戳: {msg.timestamp}")
# ============================================================
# 清理
# ============================================================
import os
if os.path.exists('conversation.json'):
os.remove('conversation.json')
print("\n已清理临时文件")
print("\n" + "=" * 60)
print("所有示例执行完成！")
print("=" * 60)彩蛋结尾
嘿，别滑了！手指停一停，听我说句悄悄话👇
🌟 关注我：下次更新，系统会自动弹窗提醒你，就像外卖到了那样准时！再也不怕错过我的脑洞和干货啦~
📌 收藏本文：这篇宝藏文章，现在不码住，以后想找只能捶胸顿足！点个收藏，让它成为你的私人知识库，随时回来挖宝~
❤️ 点赞在看：如果逗笑你了或者对你有用，麻烦高抬贵手点个赞！你的每个赞都是我熬夜写文的“鸡血”，让我更有动力产出更多有趣内容~
轻量云主机
专属优惠地址[3]：https://my.racknerd.com/aff.php?aff=14942
在线文档
邀请码：Zeus邀请你加入语雀，注册后在会员信息页填写邀请码 RD2QGU 即可领取 30 天语雀会员。前往注册：https://www.yuque.com/about
引用链接
[1] Pydantic AI 框架的文档地址: https://ai.pydantic.org.cn/[2] 消息和聊天记录 - Pydantic AI 框架: https://ai.pydantic.org.cn/message-history/#accessing-messages-from-results[3] 专属优惠地址: https://my.racknerd.com/aff.php?aff=14942
评论区