当前位置：首页 > news >正文

Pydantic AI：用类型安全与依赖注入构建生产级AI Agent

news 2026/4/26 11:06:20

1. 项目概述：当Pydantic遇见AI Agent

如果你和我一样，在过去一两年里折腾过各种AI Agent框架，从LangChain到LlamaIndex，再到CrewAI，那你大概率经历过这样的场景：为了接入一个模型，你得写一堆胶水代码；想给Agent加个工具，得研究半天复杂的装饰器；最头疼的是，代码跑起来看着没问题，但一上线就各种类型错误、运行时异常，调试起来像在迷宫里找出口。我们总在期待一个框架，能像FastAPI定义Web服务那样，用清晰、直观、符合直觉的方式来定义AI Agent。现在，这个期待可能被填上了——Pydantic AI来了。

简单说，Pydantic AI是一个用Python构建生产级GenAI应用和智能体工作流的框架。它的核心口号是“The Pydantic way”，这可不是随便说说的。Pydantic团队在Python生态里是什么地位？数据验证的绝对标杆。OpenAI SDK、Google AI SDK、Anthropic SDK、LangChain……这些你耳熟能详的库，底层的数据验证层都是Pydantic。所以，当这个团队决定亲自下场做AI Agent框架时，他们带来的不是又一个“跟风者”，而是一个从基因里就带着类型安全、开发者体验至上、生产就绪理念的解决方案。

我花了一周时间深度试用了Pydantic AI，从最简单的“Hello World”到构建一个具备工具调用、依赖注入、结构化输出的复杂客服Agent。我的整体感受是：它确实把那种“FastAPI式”的愉悦感带到了AI开发中。你不用再和复杂的抽象层搏斗，而是可以专注于定义“你的Agent是什么”、“它需要什么”、“它应该输出什么”。接下来，我就结合自己的实操，带你拆解这个框架的核心设计、上手步骤，以及那些官方文档里不会明说，但实际开发中能让你少踩80%坑的细节。

2. 核心设计哲学：为什么是“Pydantic Way”？

在深入代码之前，我们必须先理解Pydantic AI的设计哲学。这决定了你用它时的思维模式，也解释了它为什么在很多地方显得“不一样”。

2.1 类型安全不是可选项，是基础设施

大多数AI框架把类型提示（Type Hints）当作“锦上添花”的文档。但在Pydantic AI里，类型是框架运转的基石。你的Agent、它的依赖（Dependencies）、工具（Tools）的参数和返回值、以及最终的输出（Output），全部通过Pydantic模型和Python类型注解来定义。这意味着什么？

首先，错误从运行时提前到了编写时。如果你的工具函数期望一个int类型的customer_id，但你传了个字符串，你的IDE（比如VSCode或PyCharm）会在你写代码的时候就用红色波浪线提醒你，而不是等到程序运行到一半才抛出一个晦涩的ValidationError。这种体验有点像写Rust代码时的“如果编译通过，基本就能运行”。

其次，它带来了无与伦比的开发体验和可维护性。当你定义一个返回SupportOutput模型的Agent时，在整个代码链路中，result.output的属性都是可被IDE自动补全和类型检查的。这对于构建复杂、多人协作的AI应用至关重要，大大降低了心智负担。

2.2 依赖注入：让Agent变得可测试、可组合

这是Pydantic AI让我拍案叫绝的设计。传统的Agent框架里，工具函数经常需要访问数据库连接、用户会话、配置信息等“上下文”。常见的做法是使用全局变量、或者通过复杂闭包来传递，导致代码耦合度高，难以单独测试。

Pydantic AI引入了明确的依赖注入（Dependency Injection）系统。你定义一个DepsType（比如一个dataclass或Pydantic模型），里面包含Agent运行所需的所有外部资源。然后，在工具函数或动态指令中，通过RunContext[DepsType]参数来安全地访问这些依赖。

这样做有几个巨大优势：

关注点分离：Agent的逻辑（LLM交互、工具调用）和外部资源管理（数据库、API客户端）被清晰地分开。
易于测试：你可以轻松地为测试创建一个DepsType的模拟（Mock）实例，注入假数据或模拟服务，从而对Agent的逻辑进行单元测试，而不需要启动整个数据库或调用真实的外部API。
可配置性：根据运行环境（开发、测试、生产），你可以注入不同的依赖实例，比如使用内存数据库还是云数据库。

2.3 一切皆可组合：Capability模式

很多框架提供了丰富的功能，但它们是“硬编码”进去的。Pydantic AI提出了“Capability”（能力）的概念。一个Capability是一个可复用的单元，它打包了一组相关的工具（Tools）、钩子（Hooks）、指令（Instructions）和模型设置。

例如，框架内置了WebSearch和Thinking这两个Capability。你只需要在创建Agent时，将capabilities=[Thinking(), WebSearch()]作为参数传入，你的Agent就立刻拥有了联网搜索和“链式思考”（CoT）的能力。这比手动去配置搜索API的密钥、设计提示词要优雅和高效得多。

更重要的是，这种设计是面向生态的。你可以构建自己的Capability（比如一个专门与内部CRM系统交互的能力包），或者安装社区发布的第三方Capability。这意味着功能的复用和共享变得极其简单，避免了重复造轮子。

3. 从零到一：构建你的第一个生产级Agent

理论说再多，不如动手写一行代码。我们用一个贴近实际的场景来演示：构建一个“智能天气旅行顾问”Agent。这个Agent能根据用户提供的城市和日期，查询天气，并结合一些简单的规则（比如下雨天建议带伞，高温天建议去室内场所），给出旅行建议。我们会逐步加入工具、依赖注入和结构化输出。

3.1 环境准备与安装

首先，确保你的Python版本在3.10以上。然后，安装pydantic-ai。我强烈建议使用虚拟环境。

# 创建并激活虚拟环境（以venv为例） python -m venv .venv source .venv/bin/activate # Linux/macOS # .venv\Scripts\activate # Windows # 安装pydantic-ai pip install pydantic-ai

注意：安装过程可能会同时安装pydantic和httpx等依赖。由于Pydantic AI深度集成Pydantic，请确保你安装的Pydantic版本是兼容的（通常最新版即可）。如果你项目中已有Pydantic，注意版本冲突。

接下来，你需要一个LLM的API密钥。Pydantic AI是模型无关的，这里我们以OpenAI为例。去OpenAI平台创建一个API Key。

# 将你的API Key设置为环境变量（更安全的方式） export OPENAI_API_KEY='sk-...' # Linux/macOS # set OPENAI_API_KEY=sk-... # Windows CMD # $env:OPENAI_API_KEY='sk-...' # Windows PowerShell

3.2 定义核心数据模型（Pydantic的用武之地）

在Pydantic AI里，先定义模型，再写逻辑，是一个好习惯。这能帮你理清思路。

from pydantic import BaseModel, Field from datetime import date from enum import Enum # 定义一个枚举，表示天气状况 class WeatherCondition(str, Enum): SUNNY = "sunny" CLOUDY = "cloudy" RAINY = "rainy" SNOWY = "snowy" STORMY = "stormy" # 定义工具函数的返回模型：天气查询结果 class WeatherData(BaseModel): city: str date: date condition: WeatherCondition temperature_high: float = Field(description="最高温度，摄氏度") temperature_low: float = Field(description="最低温度，摄氏度") humidity: int = Field(description="湿度百分比", ge=0, le=100) precipitation_prob: int = Field(description="降水概率百分比", ge=0, le=100) # 定义Agent的最终输出模型：旅行建议 class TravelAdvice(BaseModel): summary: str = Field(description="对天气的简要总结和总体建议") clothing: str = Field(description="着装建议") activity_suggestion: str = Field(description="活动建议") risk_level: int = Field(description="出行风险等级，1-10", ge=1, le=10) need_umbrella: bool = Field(description="是否需要带伞") need_sunscreen: bool = Field(description="是否需要涂防晒霜")

为什么要把WeatherCondition定义成Enum？因为枚举类型能为LLM提供清晰、有限的选项，减少它“胡编乱造”的可能性。Field描述符不仅用于文档，在有些模型（如OpenAI的JSON模式）下，这些描述会传递给LLM，帮助它更好地理解字段含义。ge和le是Pydantic的验证器，确保数据在合理范围内。

3.3 创建Agent并添加工具

现在，创建Agent并为其添加一个“查询天气”的工具。这里我们先模拟一个工具函数，实际应用中你会在这里调用真实的天气API（如OpenWeatherMap）。

from pydantic_ai import Agent, RunContext from typing import Any # 首先，定义Agent的依赖类型。这里我们先留空，用一个简单的字典，后续再扩展。 from typing import Dict DepsType = Dict[str, Any] # 临时使用，后续会替换为更具体的类型 # 创建Agent实例 weather_agent = Agent( model='openai:gpt-4o', # 使用OpenAI的gpt-4o模型 deps_type=DepsType, output_type=TravelAdvice, # 指定Agent输出必须符合TravelAdvice模型 instructions=( "你是一个专业的旅行天气顾问。根据用户提供的城市和日期，查询天气信息，并给出详细、贴心的旅行建议。" "建议需考虑温度、湿度、降水概率和天气状况。请确保建议具体、实用。" "风险等级根据极端天气（暴雨、暴雪、高温等）的可能性来评估。" ), ) # 注册一个工具：模拟查询天气 @weather_agent.tool async def get_weather(ctx: RunContext[DepsType], city: str, target_date: date) -> WeatherData: """ 根据城市和日期查询天气信息。 Args: city: 要查询的城市名称，例如“北京”、“New York”。 target_date: 要查询的日期。 Returns: 包含详细天气数据的WeatherData对象。 """ # 模拟数据 - 实际项目中这里应调用天气API # 注意：为了演示，我们根据城市名简单生成一些“假数据” print(f"[工具调用] 模拟查询 {city} 在 {target_date} 的天气") # 一个简单的模拟逻辑 import random condition = random.choice(list(WeatherCondition)) temp_high = round(random.uniform(15, 35), 1) temp_low = round(temp_high - random.uniform(5, 15), 1) humidity = random.randint(30, 90) precip_prob = 80 if condition in [WeatherCondition.RAINY, WeatherCondition.SNOWY, WeatherCondition.STORMY] else random.randint(0, 30) return WeatherData( city=city, date=target_date, condition=condition, temperature_high=temp_high, temperature_low=temp_low, humidity=humidity, precipitation_prob=precip_prob )

几点关键解释：

@agent.tool装饰器：这是注册工具的标准方式。被装饰的异步函数会自动暴露给LLM。
函数文档字符串（Docstring）：极其重要！LLM依靠函数的文档字符串来理解这个工具是做什么的、参数是什么意思。务必清晰、准确地描述。Pydantic AI会自动解析docstring，将其作为工具描述和参数描述发送给LLM。
参数类型：city: str和target_date: date。Pydantic AI会利用这些类型信息为LLM生成一个JSON Schema，指导LLM如何调用这个工具。date类型会被正确地处理。
返回值类型：-> WeatherData。这确保了工具返回的数据结构是明确的，也方便框架进行后续处理。

3.4 运行Agent并查看结果

现在，让我们运行这个初步的Agent。

import asyncio from datetime import date, timedelta async def main(): # 准备依赖，目前是空字典 deps = {} # 运行Agent tomorrow = date.today() + timedelta(days=1) result = await weather_agent.run( f"我明天想去上海，天气怎么样？有什么建议吗？今天是{tomorrow}。", deps=deps ) print("=== Agent运行结果 ===") print(f"最终输出: {result.output}") print(f"使用的消息数: {len(result.all_messages())}") print(f"工具调用历史:") for msg in result.all_messages(): if msg.kind == 'tool-call' or msg.kind == 'tool-result': print(f" - {msg.kind}: {msg.content}") # 运行异步主函数 if __name__ == "__main__": asyncio.run(main())

运行这段代码，你会看到类似以下的输出（由于天气数据是随机的，每次运行结果会不同）：

[工具调用] 模拟查询 上海 在 2024-05-27 的天气 === Agent运行结果 === 最终输出: summary='上海明天天气晴朗，最高气温28.5°C，最低气温18.3°C，湿度65%，降水概率10%。' clothing='建议穿着轻薄透气的衣物，如短袖T恤、衬衫搭配薄外套或长裤，早晚温差较大，注意增减衣物。' activity_suggestion='非常适合户外活动，如城市观光、公园散步或滨江骑行。' risk_level=2 need_umbrella=False need_sunscreen=True 使用的消息数: 5 工具调用历史: - tool-call: {'name': 'get_weather', 'arguments': {'city': '上海', 'target_date': '2024-05-27'}} - tool-result: {'city': '上海', 'date': '2024-05-27', 'condition': 'sunny', 'temperature_high': 28.5, 'temperature_low': 18.3, 'humidity': 65, 'precipitation_prob': 10}

发生了什么？

Agent收到了用户关于上海明天天气的询问。
LLM识别出需要调用get_weather工具来获取信息。
框架执行了工具函数，获得了结构化的WeatherData。
LLM根据工具返回的天气数据，结合你的指令（“给出详细、贴心的旅行建议”），生成了最终的回复。
最关键的一步：框架将LLM的回复，按照TravelAdvice模型的格式进行验证和解析。如果LLM返回的JSON不符合TravelAdvice的格式（比如缺少字段、类型不对），框架会自动要求LLM重试，直到得到一个有效的输出。这就是结构化输出的威力。

3.5 引入依赖注入：让Agent更“智能”

上面的Agent能工作，但工具里的模拟数据太假了。让我们引入真实的依赖：一个天气API客户端。同时，我们可能还想根据用户的历史偏好来微调建议。

from dataclasses import dataclass from pydantic_ai import Agent, RunContext import httpx from typing import Optional # 1. 定义更具体的依赖类型 @dataclass class TravelDependencies: weather_client: httpx.AsyncClient # 用于调用真实天气API的HTTP客户端 user_preference: Optional[dict] = None # 可选的用户偏好，例如“讨厌下雨”、“喜欢户外” # 2. 更新Agent，使用新的依赖类型 weather_agent_v2 = Agent( model='openai:gpt-4o', deps_type=TravelDependencies, # 关键变更：指定依赖类型 output_type=TravelAdvice, instructions=( "你是一个专业的旅行天气顾问。根据用户提供的城市和日期，查询天气信息，并给出详细、贴心的旅行建议。" "建议需考虑温度、湿度、降水概率和天气状况。请确保建议具体、实用。" "如果用户有已知偏好（如讨厌下雨），请在建议中予以考虑。" "风险等级根据极端天气（暴雨、暴雪、高温等）的可能性来评估。" ), ) # 3. 更新工具函数，使用注入的依赖 @weather_agent_v2.tool async def get_weather_real(ctx: RunContext[TravelDependencies], city: str, target_date: date) -> WeatherData: """ 根据城市和日期，调用真实天气API查询天气信息。 Args: city: 要查询的城市名称。 target_date: 要查询的日期。 Returns: 包含详细天气数据的WeatherData对象。 """ # 注意：这里使用了 ctx.deps.weather_client # 实际API调用代码，这里以OpenWeatherMap为例（需要注册获取API Key） api_key = "YOUR_OPENWEATHER_API_KEY" # 应从环境变量或配置中读取 # 构建请求URL (示例，请参考OpenWeatherMap API文档) # url = f"https://api.openweathermap.org/data/2.5/forecast?q={city}&appid={api_key}&units=metric" # 为了演示，我们仍然模拟，但展示了如何使用依赖 print(f"[工具调用] 使用注入的client查询 {city} 的天气。用户偏好: {ctx.deps.user_preference}") # 模拟API调用延迟 import asyncio await asyncio.sleep(0.5) # 更“真实”的模拟，可以考虑用户偏好来调整模拟结果（仅演示逻辑） condition = WeatherCondition.SUNNY if ctx.deps.user_preference and ctx.deps.user_preference.get('hates_rain'): # 如果用户讨厌下雨，我们“模拟”一个晴天结果（实际API不会这样） print(" -> 检测到用户讨厌下雨，模拟返回晴天数据。") import random return WeatherData( city=city, date=target_date, condition=condition, temperature_high=round(random.uniform(20, 30), 1), temperature_low=round(random.uniform(10, 20), 1), humidity=random.randint(40, 70), precipitation_prob=10 if condition == WeatherCondition.SUNNY else 60 ) # 4. 还可以添加动态指令，根据依赖生成不同的系统提示 @weather_agent_v2.instructions async def add_user_preference_context(ctx: RunContext[TravelDependencies]) -> str: """根据用户偏好动态添加指令。""" if not ctx.deps.user_preference: return "" prefs = [] if ctx.deps.user_preference.get('hates_rain'): prefs.append("该用户非常讨厌下雨天，请在建议中强烈避免推荐雨天户外活动，并强调防雨措施。") if ctx.deps.user_preference.get('likes_outdoor'): prefs.append("该用户喜爱户外活动，在天气允许的情况下优先推荐户外项目。") if prefs: return "用户已知偏好：\n" + "\n".join(prefs) return ""

现在，运行这个升级版的Agent：

async def main_v2(): # 创建真实的HTTP客户端（这里仍用于模拟） async with httpx.AsyncClient() as client: # 构造依赖，注入真实客户端和用户偏好 deps = TravelDependencies( weather_client=client, user_preference={'hates_rain': True, 'likes_outdoor': True} ) result = await weather_agent_v2.run( "我下周去杭州玩，天气如何？给点建议。", deps=deps ) print("=== 升级版Agent运行结果 ===") print(result.output.json(indent=2)) # 以格式化的JSON输出 asyncio.run(main_v2())

依赖注入带来的好处：

可测试性：在单元测试中，你可以创建一个TravelDependencies的模拟对象，其中weather_client是一个Mock对象，返回预设的天气数据，从而在不调用真实API的情况下测试Agent的逻辑。
灵活性：在生产环境和开发环境，你可以注入不同的httpx.AsyncClient实例（比如配置不同的超时、重试策略）。
上下文感知：Agent现在能根据user_preference动态调整其行为，这是通过@agent.instructions装饰器实现的动态指令。这使得Agent的提示词不再是静态的，而是可以根据运行时上下文进行定制。

4. 高级特性与生产实践

基础功能跑通后，我们来看看Pydantic AI那些能让你的应用真正达到“生产级”的特性。

4.1 可观测性（Observability）与Logfire集成

AI应用的黑盒特性是生产部署的主要挑战之一。你不知道Agent内部做了什么决策、调用了哪些工具、花了多少钱、哪里出了错。Pydantic AI原生集成了Pydantic Logfire，这是一个基于OpenTelemetry的可观测性平台。

# 首先，安装logfire # pip install logfire import logfire from pydantic_ai import Agent # 1. 配置Logfire（需要从Logfire网站获取token） logfire.configure(token='your-logfire-token') # 通常从环境变量读取 logfire.instrument_pydantic_ai() # 关键的一行：自动注入遥测 # 2. 像往常一样创建和使用Agent agent = Agent('openai:gpt-4o', instructions='...') # ... 添加工具等 # 3. 运行Agent。所有交互（LLM调用、工具调用、耗时、token消耗）都会被自动记录。 result = await agent.run('...')

完成这些后，打开Logfire控制台，你会看到一个清晰的追踪（Trace）视图。它展示了整个Agent运行的流水线：接收用户消息、LLM思考、工具调用（包括输入参数和返回值）、LLM再次思考、生成最终输出。每个步骤的耗时、Token使用量、成本（如果配置了）都一目了然。

这对于调试复杂的工作流、监控性能、进行成本分析以及评估（Evals）至关重要。如果某个工具调用频繁失败，或者某个提示词导致LLM陷入了长循环，你都能快速定位。

4.2 流式输出（Streaming）与实时体验

对于需要长时间运行或希望提供实时反馈的Agent，流式输出是必备功能。Pydantic AI的流式输出不仅支持文本流，还支持结构化数据的流式验证。

import asyncio from pydantic_ai import Agent from pydantic import BaseModel, Field class StreamedOutput(BaseModel): items: list[str] = Field(description="生成的条目列表") summary: str = Field(description="最终总结") agent = Agent('openai:gpt-4o', output_type=StreamedOutput) async def main_stream(): prompt = "列举5种适合在办公室养的盆栽植物，并简要总结它们的共同优点。" # 使用 run_stream 而不是 run async with agent.run_stream(prompt) as result_stream: async for chunk in result_stream: # chunk 可以是多种类型 if chunk.kind == 'output': # 结构化输出流：output模型的部分字段可能已经生成并验证 print(f"[结构化输出更新] {chunk.data}") elif chunk.kind == 'content': # 原始文本内容流（如果模型支持） print(f"[文本流] {chunk.text}", end='', flush=True) elif chunk.kind == 'tool-call': print(f"[工具调用] {chunk.name}") elif chunk.kind == 'tool-result': print(f"[工具结果] {chunk.result}") # 流结束后，可以获取完整的结果对象 final_result = result_stream.result() print(f"\n最终完整输出: {final_result.output}") asyncio.run(main_stream())

run_stream返回一个异步上下文管理器，它产生一个事件流。你可以在LLM生成内容的同时就收到部分输出（content），更重要的是，一旦LLM开始生成结构化的JSON并满足StreamedOutput模型的部分验证，output事件就会触发。这意味着你可以在最终结果完全生成之前，就开始处理已验证的部分数据，极大地提升了响应速度和应用体验。

4.3 持久化执行（Durable Execution）与可靠性

对于长时间运行、涉及多步工具调用或需要人工审批（Human-in-the-loop）的Agent，进程崩溃或网络抖动会导致整个工作流失败，需要重头再来。Pydantic AI的持久化执行功能就是为了解决这个问题。

其核心思想是将Agent的运行状态（对话历史、工具调用上下文、中间结果）自动保存到外部存储（如数据库、Redis）。当执行中断后，可以从断点恢复，而不是重新开始。

from pydantic_ai import Agent from pydantic_ai.durable_execution import DurableExecution, InMemoryDurableExecutionBackend # 1. 选择一个后端存储（这里用内存后端演示，生产环境需用数据库后端） backend = InMemoryDurableExecutionBackend() durable_exec = DurableExecution(backend=backend) # 2. 在创建Agent时启用持久化执行 agent = Agent( 'openai:gpt-4o', instructions='...', durable_execution=durable_exec, # 关键参数 ) # 3. 运行Agent时，提供一个唯一的run_id run_id = "my_unique_travel_plan_123" try: result = await agent.run("为我规划一个为期一周的日本关西旅行，详细到每天。", run_id=run_id) except SomeTransientError: # 模拟一个临时错误，如网络超时 print("执行中断...") # 一段时间后，可以从同一个run_id恢复 recovered_result = await agent.run("继续之前的规划。", run_id=run_id) # recovered_result 会从上次中断的地方继续，对话历史都在。

生产环境中，你会使用PostgresDurableExecutionBackend或RedisDurableExecutionBackend。这确保了即使你的应用服务器重启，那些运行到一半的、等待用户审批的Agent任务也不会丢失。

4.4 评估（Evals）与性能监控

如何知道你的Agent表现得好不好？光靠人工测试不行。Pydantic AI内置了评估（Evals）框架，允许你系统化地定义测试用例和评估标准。

from pydantic_ai import Agent from pydantic_ai.evals import Eval, EvalResult, run_evals from pydantic import BaseModel class QAPair(BaseModel): question: str expected_keywords: list[str] # 期望回答中包含的关键词 # 定义评估集 eval_dataset = [ QAPair(question="北京明天天气如何？", expected_keywords=["北京", "天气", "温度"]), QAPair(question="下雨天去上海要带什么？", expected_keywords=["伞", "雨具", "防水"]), # ... 更多测试用例 ] # 定义一个评估函数 def contains_keywords_eval(result, qa_pair: QAPair) -> EvalResult: """评估Agent的回答是否包含预期关键词。""" answer = result.output if hasattr(result, 'output') else result.data score = 0 found_keys = [] for keyword in qa_pair.expected_keywords: if keyword in answer: score += 1 found_keys.append(keyword) return EvalResult( score=score / len(qa_pair.expected_keywords), # 得分比例 details={'found_keywords': found_keys, 'question': qa_pair.question} ) # 创建要评估的Agent agent_to_eval = Agent('openai:gpt-4o', instructions='...') # 运行评估 eval_results = await run_evals( agent=agent_to_eval, dataset=eval_dataset, eval_fn=contains_keywords_eval, max_concurrency=2 # 并发运行评估 ) for res in eval_results: print(f"问题: {res.details['question']}") print(f" 得分: {res.score:.2%}, 找到的关键词: {res.details['found_keywords']}")

你可以将评估结果与Logfire集成，持续监控Agent在生产环境中的表现。如果某次代码更新导致评估分数下降，你能立即收到警报。

5. 避坑指南与最佳实践

在实际使用Pydantic AI构建了几个项目后，我总结了一些关键的经验和容易踩的坑。

5.1 工具设计：粒度、描述与错误处理

1. 工具粒度要适中：不要设计一个“巨无霸”工具，比如plan_entire_trip(city, dates, budget, interests...)。这会让LLM难以正确使用，且工具内部逻辑过于复杂。应该拆分成小工具：get_attractions(city),check_weather(city, date),book_hotel(name, dates)。每个工具职责单一。

2. 文档字符串是给LLM看的“API文档”：务必详细、准确。使用Args:和Returns:部分。LLM对参数描述非常敏感。好的描述能极大提升工具调用的准确率。

@agent.tool async def search_flights( ctx: RunContext[DepsType], origin: str = Field(description="出发城市机场代码，如'PEK'"), destination: str = Field(description="到达城市机场代码，如'SHA'"), date: date = Field(description="出发日期，格式YYYY-MM-DD"), max_price: Optional[float] = Field(None, description="最高可接受价格，人民币") ) -> List[FlightInfo]: """ 根据条件搜索航班信息。 Args: origin: 出发地机场IATA三字码。 destination: 目的地机场IATA三字码。 date: 出发日期。 max_price: 可选，价格过滤器。 Returns: 一个航班信息列表，按价格从低到高排序。如果没有符合的航班，返回空列表。 """ # ... 实现

3. 工具内部必须有坚实的错误处理：LLM调用工具时，参数可能不合法（比如不存在的城市代码）。工具函数内部应该进行验证，并抛出清晰的异常。Pydantic AI会捕获这些异常，将其信息反馈给LLM，让LLM有机会调整参数后重试。

@agent.tool async def get_city_code(ctx: RunContext[DepsType], city_name: str) -> str: """根据城市名获取机场IATA代码。""" code = await ctx.deps.db.get_city_code(city_name) if not code: # 抛出清晰的错误，LLM会看到这个信息 raise ValueError(f"未找到城市 '{city_name}' 对应的机场代码。请提供更准确的城市名或检查拼写。") return code

5.2 指令（Instructions）设计：静态与动态结合

静态指令（创建Agent时传入的instructions参数）用于定义Agent的固定角色和基础行为准则。

动态指令（通过@agent.instructions装饰器）用于注入运行时才能知道的信息，比如用户信息、会话上下文、实时数据。这是实现个性化Agent的关键。

@agent.instructions async def inject_conversation_history(ctx: RunContext[DepsType]) -> str: """将会话历史摘要注入指令。""" history = await ctx.deps.message_store.get_recent_messages(ctx.deps.user_id, limit=5) if not history: return "" summary = "\n".join([f"User: {m.user} | Assistant: {m.assistant[:50]}..." for m in history]) return f"最近的对话历史摘要：\n{summary}\n请参考以上历史进行回复，保持连贯性。"

避免指令冲突：静态指令和动态指令是合并的。确保它们不会互相矛盾。通常，静态指令定义“是什么”（角色），动态指令定义“当前上下文”（情境）。

5.3 模型选择与成本控制

Pydantic AI支持众多模型，但不同模型在工具调用、JSON模式遵循、成本方面差异巨大。

复杂工具调用/结构化输出：优先选择对工具调用和JSON模式支持最好的模型，如OpenAI的gpt-4o/gpt-4-turbo、Anthropic的claude-3-5-sonnet。虽然贵，但成功率高，节省调试时间。
简单任务/原型验证：可以使用更便宜或开源的模型，如gemini-flash、claude-haiku，或者通过Ollama运行的本地模型（如llama3.2）。但需要测试其对工具调用的支持度。
设置预算和超时：在生产中，务必在Agent或运行调用时设置max_messages（最大交互轮次）和timeout，防止LLM陷入死循环或产生过高费用。

from pydantic_ai.models.openai import OpenAIModel model = OpenAIModel( 'gpt-4o', max_retries=2, timeout=30.0, ) agent = Agent(model=model, ...) # 或者在运行时控制 result = await agent.run( prompt, deps=deps, max_messages=20, # 最多20轮消息交换（包括工具调用） )

5.4 调试与问题排查

当Agent行为不符合预期时，按以下步骤排查：

启用Logfire：这是第一选择。查看Trace，确认LLM收到了正确的消息、工具调用参数是否正确、工具返回了什么。
检查工具描述：LLM是否误解了工具功能？简化描述，确保每个参数的意义明确。
简化测试：用一个最简单的提示词和单一工具测试，排除复杂指令的干扰。
查看原始消息：result.all_messages()提供了完整的对话历史，包括所有LLM的原始请求和响应。这能帮你理解LLM的“思考过程”。
结构化输出失败：如果LLM始终无法生成符合output_type的JSON，尝试：
- 在指令中更明确地要求输出格式：“你必须返回一个JSON对象，包含summary和risk_level字段...”
- 使用更简单的输出模型，或者将某些字段设为Optional。
- 考虑使用支持JSON模式（JSON Mode）的模型，并在模型设置中启用它。

6. 总结：Pydantic AI的定位与未来

经过这一番深度探索，Pydantic AI给我的感觉非常明确：它不是一个试图囊括所有AI应用场景的“万能框架”，而是一个为Python开发者打造的、专注于构建可靠、可维护、类型安全的生产级Agent的工具箱。

它的优势在于其坚实的工程基础（Pydantic）、优秀的设计模式（依赖注入、Capability）、以及对生产环节的深度思考（可观测性、持久化执行、评估）。它可能不像一些框架那样有海量的预制“链”或“代理”，但它提供了更优雅、更可靠的方式来构建属于你自己的、独一无二的智能体。

如果你正在用Python构建需要与外部系统交互、有复杂逻辑、并且最终要部署上线的AI应用，Pydantic AI值得你投入时间学习。它带来的类型安全和开发体验提升，在项目复杂度增长时会变成巨大的生产力优势。从简单的天气顾问到复杂的多智能体协作系统，Pydantic AI提供了一套一致且强大的抽象，让你能更专注于业务逻辑，而不是框架的复杂性。

我个人在后续的项目中，会继续深入使用它的Graph功能来编排复杂工作流，并探索其与MCP（Model Context Protocol）的集成，为Agent连接更丰富的外部工具和数据源。这个框架的生态还在早期，但以其背后的团队和设计理念，我相信它会成为Python AI工程化领域的一个重要选择。

查看全文

http://www.jsqmd.com/news/702908/