Pydantic-AI:用结构化数据模型驱动AI应用开发
1. 项目概述:当Pydantic遇上AI,数据验证的范式革命
如果你在过去几年里写过Python,尤其是涉及Web API、数据管道或者配置管理的代码,那么“Pydantic”这个名字对你来说一定不陌生。它凭借基于Python类型注解的、运行时强制的数据验证和序列化能力,几乎重塑了我们对Python数据处理的认识。从FastAPI的爆火到各类配置管理库的底层依赖,Pydantic已经成为了现代Python生态中不可或缺的一块基石。它的核心哲学是“用类型定义数据,让数据自己说话”,将开发者从繁琐的手动校验和错误处理中解放出来。
然而,当我们把目光投向另一个同样炙手可热的领域——人工智能(AI)应用开发时,情况变得复杂起来。AI模型,特别是大语言模型(LLM),其输入和输出本质上是非结构化的文本流。我们向模型发送一个提示词(Prompt),模型返回一段文本。如何确保我们发送的提示词包含了所有必要信息且格式正确?如何可靠地、结构化地解析模型那充满不确定性的回复,从中提取出我们想要的、可供程序后续使用的数据?传统的字符串拼接、正则表达式匹配和手写解析逻辑,不仅容易出错,而且难以维护和扩展。
正是在这样的背景下,pydantic-ai应运而生。它不是要取代Pydantic,而是Pydantic思想在AI应用领域的深度延伸和范式创新。简单来说,pydantic-ai是一个旨在让开发者能够像使用Pydantic定义数据模型一样,去定义和控制与AI模型的交互过程的库。它将一次AI调用抽象为一个“智能体”(Agent),而这个智能体的输入、输出、中间步骤乃至工具调用,都可以用Pydantic模型来严格定义和约束。这意味着一场从“字符串魔术”到“结构化工程”的转变。
对于谁来说,pydantic-ai价值最大?我认为是以下几类开发者:
- 正在构建生产级AI应用的工程师:你需要可靠、可测试、可监控的AI集成,而不是脆弱的脚本。
- 希望将AI能力嵌入现有系统的开发者:你的系统有清晰的数据边界,AI的输入输出必须符合这些边界。
- 厌倦了手动解析LLM输出的所有人:你受够了写一堆
if “关键词” in response的胶水代码,渴望一种声明式的、类型安全的方式来处理AI响应。
接下来,我将深入拆解pydantic-ai的核心设计、实操要点,并分享如何用它来构建一个从简单到复杂的AI智能体。
2. 核心设计哲学:将非结构化交互转化为结构化流程
pydantic-ai的设计并非凭空而来,它深刻回应了当前AI应用开发中的几个核心痛点,并将其解决思路建筑在Pydantic这一已被验证的坚实基础上。
2.1 问题根源:AI交互的“不确定性沼泽”
在与LLM交互的传统模式中,我们通常面临一个“不确定性沼泽”:
- 输入侧:我们构造提示词(Prompt),可能混合了系统指令、用户消息、历史对话和上下文信息。这个过程往往是字符串模板的拼接,一旦结构复杂,极易出错,且难以复用和测试。
- 输出侧:LLM返回的是纯文本。要从中提取结构化信息(如日期、人名、JSON对象),我们需要编写解析器。但LLM的输出格式不稳定,可能多一个换行,少一个逗号,或者用自然语言描述而不是直接给出数据,导致解析器异常脆弱。
- 流程侧:复杂的AI任务往往需要多轮对话、条件分支或调用外部工具(如查询数据库、执行计算)。管理这些状态和流程通常需要大量的胶水代码,逻辑分散,可维护性差。
2.2pydantic-ai的解决方案:智能体(Agent)与结构化消息
pydantic-ai引入了“智能体”(Agent)作为核心抽象。一个智能体封装了一次完整的、目标驱动的AI交互过程。其核心思想是:用Pydantic模型来定义智能体交互中的所有结构化部分。
结构化输入(Agent State):智能体的输入不再仅仅是字符串。你可以定义一个Pydantic模型作为智能体的“状态”(State)。这个状态模型可以包含任务目标、用户信息、上下文数据、配置参数等任何你需要传递给AI的信息。智能体在运行时,会自动将这个状态模型序列化并融入提示词中。
结构化输出(Result Model):你可以为智能体指定一个Pydantic模型作为期望的输出类型。智能体不仅会要求LLM生成符合该模型的数据(通常是JSON),还会在返回结果后,自动用这个模型去验证和解析LLM的响应。如果解析失败,你可以配置重试或降级策略。从此,你的代码拿到的是一个类型明确的Pydantic对象,而不是一串需要小心处理的文本。
结构化工具(Tools):智能体可以调用外部函数(工具)。
pydantic-ai要求这些工具函数有明确的输入和输出类型注解(最好是Pydantic模型)。当智能体决定调用一个工具时,它会自动从对话上下文中提取参数,并验证其是否符合工具函数的输入模型。工具执行后的结果,也会被结构化地返回给智能体。这确保了工具调用的类型安全和可靠性。结构化消息流:智能体与LLM的对话被建模为一系列结构化的“消息”(Message),如
UserMessage、SystemMessage、AIMessage、ToolMessage等。这比原始的字符串列表更易于操作、过滤和持久化。
通过这套设计,pydantic-ai将一次充满不确定性的AI调用,转变为一个输入明确、输出可靠、过程可控的“函数调用”。开发者从处理文本的泥潭中跳脱出来,回到了熟悉的定义数据类型、编写业务逻辑的舒适区。
注意:
pydantic-ai并不绑定于某个特定的LLM提供商。它通过“运行器”(Runner)抽象来支持不同的后端,如OpenAI、Anthropic、Google Gemini等,甚至本地模型。这使得你的智能体逻辑与底层模型实现解耦。
3. 从零到一:构建你的第一个Pydantic-AI智能体
理论说得再多,不如动手一试。让我们从一个最简单的例子开始,感受pydantic-ai如何改变我们的编码方式。假设我们要构建一个“天气查询助手”智能体,用户告诉它城市名,它返回一个结构化的天气简报。
3.1 环境准备与安装
首先,确保你有一个Python环境(建议3.8以上),然后安装pydantic-ai及其可选依赖。由于我们需要调用真实的LLM,这里以OpenAI为例。
# 安装 pydantic-ai 核心库 pip install pydantic-ai # 安装OpenAI运行器依赖 pip install 'pydantic-ai[openai]' # 当然,Pydantic本身也是必须的,通常pydantic-ai会依赖它 # pip install pydantic接下来,你需要设置你的OpenAI API密钥。通常可以通过环境变量设置:
export OPENAI_API_KEY='你的-api-key'或者在代码中直接设置(不推荐用于生产环境):
import os os.environ['OPENAI_API_KEY'] = '你的-api-key'3.2 定义数据模型:输入与输出的契约
这是pydantic-ai最核心的一步。我们需要定义智能体输出什么。对于天气简报,我们可能关心温度、天气状况和提示。
from pydantic import BaseModel, Field from typing import Literal # 定义智能体的输出模型 class WeatherReport(BaseModel): """天气简报数据模型""" city: str = Field(description="查询的城市名称") temperature_celsius: float = Field(description="摄氏温度") condition: Literal['sunny', 'cloudy', 'rainy', 'snowy', 'windy'] = Field(description="天气状况") advisory: str = Field(description="给用户的出行建议,例如'建议带伞'") # 可以添加自定义验证或计算方法 @property def temperature_fahrenheit(self) -> float: """返回华氏温度(计算属性)""" return self.temperature_celsius * 9/5 + 32这个WeatherReport类就是一个标准的Pydantic模型。Field(description=...)的描述字段非常重要,pydantic-ai会利用这些描述来指导LLM生成符合字段含义的数据。Literal类型则限定了condition字段只能取那几个枚举值,进一步约束了输出。
3.3 创建智能体与运行
现在,我们创建智能体,并将输出模型与之绑定。
from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel # 1. 选择模型。这里使用OpenAI的gpt-4o-mini,性价比高。 model = OpenAIModel('gpt-4o-mini') # 2. 创建智能体,并指定其输出模型为 WeatherReport # `result_type`参数是关键,它告诉智能体我们期望什么。 weather_agent = Agent( model=model, result_type=WeatherReport, system_prompt="你是一个专业的天气助手。根据用户提供的城市名,生成一份结构化的天气简报。如果无法确定城市,请合理推测或告知不确定性。", ) # 3. 运行智能体 async def main(): # 使用 .run() 方法执行,传入用户消息 result = await weather_agent.run("今天巴黎的天气怎么样?") # result.data 就是解析后的 WeatherReport 对象! report: WeatherReport = result.data print(f"城市: {report.city}") print(f"温度: {report.temperature_celsius}°C ({report.temperature_fahrenheit:.1f}°F)") print(f"状况: {report.condition}") print(f"建议: {report.advisory}") # 你也可以访问原始的AI消息和消耗情况 print(f"\n本次消耗token数: {result.usage.total_tokens}") print(f"AI回复的原始消息: {result.all_messages()}") # 如果是脚本运行,需要异步执行 import asyncio asyncio.run(main())执行这段代码,你会得到类似这样的输出:
城市: 巴黎 温度: 18.5°C (65.3°F) 状况: cloudy 建议: 天气多云,气温舒适,适合外出散步,但建议带一件薄外套。发生了什么?
- 智能体将系统提示、用户消息(“今天巴黎的天气怎么样?”)以及隐含的指令(“请以JSON格式输出,符合
WeatherReport模型”)组合成最终的提示词发送给LLM。 - LLM(如GPT-4)理解任务后,会尝试生成一个符合
WeatherReport模式的JSON对象。 pydantic-ai接收到LLM的回复后,自动尝试解析其中的JSON,并用WeatherReport模型进行验证和实例化。- 如果解析和验证成功,
result.data就是一个强类型的WeatherReport对象。如果失败(例如LLM返回了非JSON文本或字段类型错误),默认会抛出异常。
实操心得一:描述(description)是你的朋友。在定义Pydantic模型的字段时,务必提供清晰、准确的
description。这是指导LLM生成正确数据的最重要线索。好的描述应该像给一个实习生写工作说明一样明确。
3.4 处理不确定性:降级与重试策略
LLM的输出并不总是可靠的。pydantic-ai提供了优雅的机制来处理解析失败。
from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel from pydantic import ValidationError model = OpenAIModel('gpt-4o-mini') # 创建智能体时,可以配置重试和降级逻辑 weather_agent_robust = Agent( model=model, result_type=WeatherReport, system_prompt="...", # 最多重试2次 retries=2, # 设置一个降级结果,当所有重试都失败后返回 result_default=WeatherReport( city="未知", temperature_celsius=0.0, condition='cloudy', advisory='无法获取天气信息。' ) ) async def robust_query(city_query: str): try: result = await weather_agent_robust.run(city_query) return result.data except Exception as e: # 即使配置了result_default,某些错误(如网络问题)仍可能抛出异常 print(f"查询失败: {e}") return None通过retries和result_default,我们为智能体增加了韧性。在实际生产中,你还可以结合更复杂的逻辑,例如在重试时调整提示词,或者根据异常类型选择不同的降级策略。
4. 进阶实战:构建具备工具调用能力的多功能智能体
简单的问答只是开始。真正的威力在于让智能体能够调用外部工具,从而突破LLM的知识和时效限制,执行具体操作。让我们构建一个更复杂的“旅行规划助手”,它能查询天气(模拟)、计算汇率(模拟)并生成总结。
4.1 定义工具(Tools)
首先,我们定义两个模拟的工具函数。注意,工具函数的参数和返回值最好都用Pydantic模型来定义,以实现最佳的类型集成。
from pydantic import BaseModel from pydantic_ai import Tool # 1. 查询天气工具 class WeatherQuery(BaseModel): city: str date: str # 格式 YYYY-MM-DD class WeatherResult(BaseModel): city: str date: str high_temp_c: float low_temp_c: float condition: str @Tool async def get_weather(query: WeatherQuery) -> WeatherResult: """ 根据城市和日期查询天气预报。 这是一个模拟工具,实际应用中应接入真实天气API。 """ # 模拟API调用延迟 import asyncio await asyncio.sleep(0.5) # 模拟返回数据 return WeatherResult( city=query.city, date=query.date, high_temp_c=22.5, low_temp_c=15.0, condition='partly cloudy' ) # 2. 查询汇率工具 class CurrencyQuery(BaseModel): from_currency: str # 如 'USD' to_currency: str # 如 'EUR' amount: float = 1.0 class CurrencyResult(BaseModel): from_currency: str to_currency: str amount: float converted_amount: float rate: float @Tool async def get_exchange_rate(query: CurrencyQuery) -> CurrencyResult: """ 查询货币汇率。 这是一个模拟工具。 """ await asyncio.sleep(0.3) # 模拟一个固定汇率 rate = 0.85 if query.from_currency == 'USD' and query.to_currency == 'EUR' else 1.0 return CurrencyResult( from_currency=query.from_currency, to_currency=query.to_currency, amount=query.amount, converted_amount=query.amount * rate, rate=rate )@Tool装饰器是关键,它告诉pydantic-ai这个函数是一个可供智能体调用的工具。工具函数的文档字符串(""" ... """)同样会被用于指导LLM何时以及如何使用该工具。
4.2 定义智能体状态与结果模型
对于旅行规划,我们的智能体需要知道用户的基本信息(如出发地、预算货币)作为上下文。这可以通过“智能体状态”(Agent State)来实现。同时,规划的结果也是一个复杂的结构。
from pydantic import BaseModel, Field from datetime import date from typing import List, Optional # 智能体状态:包含会话的上下文信息 class TravelPlannerState(BaseModel): user_name: Optional[str] = None home_currency: str = 'USD' # 用户的本地货币,默认为美元 # 状态可以在对话中被更新 # 旅行规划的输出模型 class DayPlan(BaseModel): date: str morning: str = Field(description="上午的活动安排") afternoon: str = Field(description="下午的活动安排") evening: str = Field(description="晚上的活动安排") weather_forecast: Optional[str] = Field(None, description="当天的天气情况简报") class TravelItinerary(BaseModel): destination: str = Field(description="旅行目的地") travel_dates: str = Field(description="出行日期范围,例如'2024-07-01 至 2024-07-05'") total_budget_local: float = Field(description="总预算(当地货币)") total_budget_home: Optional[float] = Field(None, description="总预算(换算为用户本国货币后)") daily_plans: List[DayPlan] = Field(description="每日详细计划") general_advice: str = Field(description="整体旅行建议,如交通、文化提示等")4.3 组装并运行多功能智能体
现在,我们将工具、状态和输出模型组合起来,创建一个功能强大的智能体。
from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel # 创建模型 model = OpenAIModel('gpt-4o', model_settings={'temperature': 0.2}) # 创建智能体 travel_agent = Agent( model=model, # 指定状态模型 state_type=TravelPlannerState, # 指定输出模型 result_type=TravelItinerary, # 注册可用的工具 tools=[get_weather, get_exchange_rate], system_prompt=""" 你是一个专业的旅行规划助手。请根据用户的需求,为他们制定详细的旅行 itinerary。 你可以调用工具来获取目的地的天气预报,以及进行货币汇率换算,以便为用户提供更准确的信息。 在生成最终规划时,请确保包含天气信息和预算换算(如果用户提供了预算)。 请以友好、详尽的方式组织你的回复,但最终输出必须严格符合 TravelItinerary 数据模型。 """, ) async def plan_trip(): # 初始化智能体状态 initial_state = TravelPlannerState(user_name="小明", home_currency='CNY') # 运行智能体,传入用户请求和初始状态 result = await travel_agent.run( "我想在七月的第一周去巴黎旅行5天,总预算大概是8000人民币。请帮我做个规划,考虑一下天气。", state=initial_state ) # 获取结构化结果 itinerary: TravelItinerary = result.data print(f"目的地: {itinerary.destination}") print(f"日期: {itinerary.travel_dates}") print(f"总预算(当地货币EUR): {itinerary.total_budget_local}") print(f"总预算(您的货币CNY): {itinerary.total_budget_home}") print(f"\n每日计划:") for i, day in enumerate(itinerary.daily_plans, 1): print(f" 第{i}天 ({day.date}):") print(f" 上午: {day.morning}") print(f" 下午: {day.afternoon}") print(f" 晚上: {day.evening}") if day.weather_forecast: print(f" 天气: {day.weather_forecast}") print(f"\n整体建议: {itinerary.general_advice}") # 查看工具调用历史(调试非常有用) print(f"\n=== 工具调用记录 ===") for msg in result.all_messages(): if hasattr(msg, 'tool_calls') and msg.tool_calls: for tc in msg.tool_calls: print(f" 调用了工具: {tc.name}") if hasattr(msg, 'content') and isinstance(msg.content, str) and 'tool_result' in msg.content.lower(): # 简化打印工具结果 print(f" 工具返回了结果") asyncio.run(plan_trip())在这个例子中,智能体会根据用户请求,自动判断是否需要调用get_weather来查询巴黎七月初的天气,以及调用get_exchange_rate将8000人民币换算成欧元(假设当地货币),并将这些信息整合到最终的旅行规划中。整个过程完全由pydantic-ai驱动,开发者无需编写任何工具调用调度或结果解析的胶水代码。
实操心得二:善用状态(State)管理上下文。对于多轮对话或需要记住用户信息的场景,
state_type非常有用。你可以在每次run()之后,从result.new_state()获取更新后的状态,并在下一次调用时传入,从而实现有状态的会话。这比手动维护一个对话历史列表要清晰和类型安全得多。
5. 生产级考量:性能、监控与最佳实践
将pydantic-ai用于实际项目时,除了核心功能,还需要关注一些工程化方面的细节。
5.1 性能优化与成本控制
LLM API调用通常是应用中最耗时的部分,也是成本的主要来源。
设置超时与重试:网络和API服务可能不稳定。为智能体的
run操作配置合理的超时和重试策略至关重要。pydantic-ai的Agent可以接受一个run_settings参数,或者你可以使用像tenacity这样的库在更外层实现重试逻辑。from pydantic_ai import Agent, RunSettings import httpx agent = Agent(...) # 通过RunSettings配置 settings = RunSettings( timeout=httpx.Timeout(30.0), # 30秒超时 # 其他运行设置... ) result = await agent.run("查询...", run_settings=settings)利用流式响应(Streaming):对于生成内容较长的场景,使用流式响应可以提升用户体验。
pydantic-ai支持流式输出,你可以通过agent.run_stream()来获取一个异步生成器,实时处理返回的token。async for chunk in agent.run_stream("请写一篇长文..."): if chunk.is_content: print(chunk.content, end='', flush=True) # 实时打印内容 # 还可以处理delta状态、工具调用等缓存(Caching):对于内容相对固定或可重复的查询(例如,基于模板生成的提示词),可以考虑对LLM的响应进行缓存,以节省成本和延迟。
pydantic-ai可以与langchain.cache或自定义缓存逻辑集成。
5.2 可观测性与调试
当智能体行为不符合预期时,强大的调试工具是救命稻草。
记录完整的消息流:
result.all_messages()返回了本次交互中的所有消息,包括系统提示、用户消息、AI回复、工具调用请求和工具调用结果。这是分析问题最直接的资料。在生产环境中,应该将此日志持久化。result = await agent.run(...) conversation_log = result.all_messages() # 可以将conversation_log转换为字典或JSON保存到数据库或日志系统 for msg in conversation_log: print(f"{msg.__class__.__name__}: {msg.content}")使用“裸”模式进行调试:有时你需要查看发送给LLM的原始提示词。可以临时使用一个不绑定结果模型的智能体,或者直接检查消息列表。
debug_agent = Agent(model=model, system_prompt="...") # 不指定result_type result = await debug_agent.run("你的问题") print(result.data) # 这里data是Message列表验证与测试:由于智能体的核心是Pydantic模型,你可以像测试普通函数一样为它编写单元测试。模拟工具调用,并断言输出模型符合预期。
import pytest from unittest.mock import AsyncMock @pytest.mark.asyncio async def test_weather_agent(): # 模拟工具返回 mock_weather_tool = AsyncMock(return_value=WeatherResult(...)) agent = Agent(..., tools=[mock_weather_tool]) result = await agent.run("...") assert isinstance(result.data, WeatherReport) assert result.data.city == "Paris" # 验证工具被以正确的参数调用 mock_weather_tool.assert_called_once_with(...)
5.3 架构模式与最佳实践
智能体组合:复杂的应用可以由多个单一职责的智能体组合而成。例如,一个“路由智能体”先分析用户意图,然后调用专门的“查询智能体”、“规划智能体”或“总结智能体”。
pydantic-ai的智能体本身可以作为“工具”被另一个智能体调用,或者通过编排逻辑(如工作流引擎)来组合。提示词工程:
pydantic-ai不限制你的提示词设计。系统提示词(system_prompt)是控制智能体行为的关键。好的提示词应清晰定义角色、任务边界、输出格式要求以及可用的工具。将提示词模板化、外部化(如存储在配置文件或数据库中)有利于维护和A/B测试。错误处理与降级:如前所述,充分利用
retries和result_default。对于关键任务,考虑实现更复杂的降级链路,例如,当主要模型(如GPT-4)失败或超时时,自动切换到更便宜、更快的模型(如GPT-3.5-Turbo)或规则引擎。版本化管理数据模型:你的
result_type和state_typePydantic模型是API契约。对它们的修改(如增加字段、改变字段类型)可能会影响已有智能体的行为。考虑使用版本化的模型或向后兼容的变更策略。
pydantic-ai代表了一种构建AI应用的更稳健、更可维护的范式。它将Pydantic的“数据验证优先”理念带入AI领域,迫使开发者从一开始就思考数据的输入输出结构,从而大幅减少了后期集成和调试的麻烦。虽然它增加了一些前期定义模型的开销,但换来的却是整个应用生命周期内清晰的接口、自动化的验证和显著提升的代码可靠性。对于任何计划将AI能力深度集成到产品中的团队来说,这都是一项值得投入的基础设施投资。
