最近很多开发者朋友在尝试将大语言模型LLM集成到自己的应用中时都会遇到一个共同的“拦路虎”如何让模型输出的内容尤其是那些需要结构化、格式化的内容能够稳定、可靠地符合我们的要求比如让模型生成一个标准的 JSON 对象来返回天气数据或者输出一个格式工整的 Markdown 表格来总结会议纪要。你可能会发现直接让模型“生成一个 JSON”它有时会“放飞自我”——在 JSON 外面加上解释性文字或者漏掉引号甚至返回一段看似 JSON 但无法解析的文本。这种不稳定性让 LLM 在需要与下游系统如数据库、API无缝对接的生产环境中显得有些“不靠谱”。这背后的核心痛点就是LLM 输出的不可控性。它像一个才华横溢但有些随性的助手你需要花大量精力去“调教”和“后处理”它的输出才能让它融入严谨的工程流水线。今天要介绍的主角——Pydantic结合其强大的pydantic-ai库正是为了解决这个“最后一公里”的问题而生的。它不是一个新模型而是一套工程化框架其核心思想是用代码定义你期望的输出结构然后让 LLM 的生成过程被这个结构所约束和引导最终直接得到类型安全、格式正确的 Python 对象。简单来说它想让 LLM 的输出变得像调用一个普通函数一样可靠输入参数返回一个确定类型的对象。本文将深入探讨如何利用 Pydantic 来“管好” LLM 的输出让你告别繁琐的正则表达式匹配和字符串解析真正实现 AI 能力的即插即用。1. 这篇文章真正要解决的问题从“文本生成”到“函数调用”在传统开发中我们调用一个函数或 API返回值的数据类型和结构是预先定义好的。例如一个get_user_info(user_id: int) - User函数我们明确知道它会返回一个User对象里面有name、email等属性。但当我们将任务交给 LLM 时情况就变了。我们得到的是一段自由文本。为了从这段文本中提取结构化信息开发者通常需要精心设计提示词Prompt反复强调格式要求。在代码中编写复杂的后处理逻辑如正则表达式、字符串分割、JSON 解析并处理各种可能的异常格式。进行大量的测试和调试以覆盖模型可能产生的各种“创意”输出。这个过程不仅效率低下而且极其脆弱。提示词的微小改动或模型版本的更新都可能导致后处理逻辑失效。Pydantic pydantic-ai提供的解决方案是“结构化的生成”。它允许你用 Pydantic Model 定义输出像定义数据库表或 API 响应一样用 Python 类来定义你希望 LLM 生成的数据结构。将结构作为生成的一部分这个结构定义会被巧妙地融入到给 LLM 的提示词中引导模型在生成时就直接思考如何填充这个结构。直接得到类型化对象LLM 的原始输出会经过库的解析和验证直接转换为你定义的 Pydantic 模型实例。你可以立刻使用.操作符访问属性享受 IDE 的自动补全和静态类型检查。这本质上是在 LLM 的“自由创作”和程序的“严格接口”之间架起了一座坚固的桥梁。它解决的不是“生成什么内容”的问题而是“如何让生成的内容能被程序直接、可靠地使用”的问题。2. 基础概念与核心原理在深入代码之前我们先厘清几个关键概念理解pydantic-ai是如何工作的。2.1 Pydantic 是什么Pydantic 是一个 Python 库主要用于数据验证和设置管理。它利用 Python 的类型注解type hints来定义数据的形状Schema并自动验证传入的数据是否符合这个形状同时进行类型转换。一个简单的例子from pydantic import BaseModel class User(BaseModel): name: str age: int email: str # 有效数据 user1 User(nameAlice, age30, emailaliceexample.com) print(user1.name) # 输出: Alice # 无效数据会引发验证错误 try: user2 User(nameBob, agenot_a_number, emailbobexample.com) except Exception as e: print(e) # 输出验证错误信息Pydantic 确保了User对象的数据总是符合我们定义的规范。2.2pydantic-ai的核心思想pydantic-ai库将 Pydantic 的这种“定义-验证”能力逆向应用到了 LLM 的文本生成过程上。其核心流程可以概括为定义输出模型你创建一个 PydanticBaseModel描述你希望 LLM 生成的信息结构。创建智能体Agent你将这个输出模型“告诉”一个pydantic-ai的Agent并指定使用的 LLM如 OpenAI GPT-4, Anthropic Claude 等。运行并获取结构化结果你向 Agent 提问或下达指令。Agent 在内部会 a.构建增强提示将你的问题/指令与输出模型的结构描述结合生成一个更精确的提示词发送给 LLM。 b.解析与验证接收 LLM 的原始文本回复尝试将其解析并填充到预定义的输出模型中。 c.返回模型实例如果解析成功且通过 Pydantic 验证则直接返回该模型的一个实例。如果失败它可以进行重试或报错。2.3 与传统“函数调用Function Calling”的区别OpenAI 等厂商也提供了“函数调用”功能允许你描述函数让模型返回调用该函数所需的参数。pydantic-ai与它既有相似之处也有不同相似点两者都旨在让 LLM 输出结构化数据。不同点函数调用侧重于“让模型决定是否以及如何调用某个已知函数”。输出是函数名和参数核心是“动作”。pydantic-ai侧重于“让模型直接生成符合某个复杂结构的数据”。输出就是数据本身核心是“信息提取与结构化”。它更通用不限于函数参数可以描述任何复杂嵌套的对象。控制权pydantic-ai将结构定义完全放在开发者手中通过 Pydantic Model提供了更强的类型安全和代码集成度。3. 环境准备与前置条件开始实践前你需要准备好 Python 环境和一个可用的 LLM API。3.1 Python 环境建议使用 Python 3.8 及以上版本。使用虚拟环境是一个好习惯。# 创建并激活虚拟环境 (可选) python -m venv venv # Linux/macOS source venv/bin/activate # Windows venv\Scripts\activate3.2 安装依赖核心需要安装pydantic-ai。它将自动安装正确版本的pydantic。pip install pydantic-ai根据你计划使用的 LLM 提供商还需要安装对应的 SDK 并配置 API 密钥。本文以 OpenAI 为例pip install openai然后在环境变量中设置你的 OpenAI API Key# Linux/macOS export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here你也可以在代码中直接设置但出于安全考虑更推荐使用环境变量。4. 核心流程拆解第一个结构化输出让我们通过一个最简单的例子感受pydantic-ai的工作流程。4.1 第一步定义输出模型假设我们想让 LLM 从一个句子中提取人名和情绪。我们首先定义这个数据结构。from pydantic import BaseModel, Field from typing import Literal # 定义输出数据结构 class SentimentAnalysis(BaseModel): 分析句子中的情绪和提及的人物 person_name: str Field(description句子中提及的人物姓名) sentiment: Literal[POSITIVE, NEUTRAL, NEGATIVE] Field(description针对该人物的情绪倾向) confidence: float Field(description分析结果的置信度0到1之间, ge0, le1)BaseModel: 所有输出模型的基类。Field: 用于为字段提供更详细的描述和约束。这里的description非常重要它会帮助 LLM 理解每个字段的含义。ge和le是 Pydantic 的数值范围校验器。Literal: 表示该字段只能是列举值中的一个这为 LLM 提供了明确的选项。4.2 第二步创建并运行 Agent接下来我们创建一个 Agent让它使用这个模型来生成结果。from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel # 1. 选择模型。这里使用 OpenAI 的 gpt-4o-mini你也可以用 gpt-4-turbo 等。 model OpenAIModel(gpt-4o-mini) # 2. 创建 Agent并指定其输出模型为 SentimentAnalysis sentiment_agent Agent( modelmodel, result_typeSentimentAnalysis, # 关键绑定输出模型 ) # 3. 运行 Agent提出请求 async def main(): result await sentiment_agent.run( 从这句话中提取信息尽管项目延期了但张三仍然对团队的努力感到非常骄傲。 ) # result.data 就是 SentimentAnalysis 的一个实例 analysis: SentimentAnalysis result.data print(f人物: {analysis.person_name}) print(f情绪: {analysis.sentiment}) print(f置信度: {analysis.confidence:.2f}) # 你可以像使用普通对象一样访问其属性 if analysis.sentiment POSITIVE: print(检测到积极情绪) # 运行异步函数 import asyncio asyncio.run(main())关键点解析Agent: 核心执行器封装了与 LLM 的交互逻辑。result_type: 这是将 Agent 与 Pydantic 模型绑定的关键参数。它告诉 Agent“你每次运行的结果都应该符合这个模型”。agent.run(): 发送提示词并获取结果。返回的result对象包含原始响应、消耗的 Token 等信息而result.data就是我们需要的结构化对象。运行这段代码你可能会得到类似这样的输出人物: 张三 情绪: POSITIVE 置信度: 0.95 检测到积极情绪最重要的是analysis是一个SentimentAnalysis类型的对象analysis.person_name是字符串类型analysis.sentiment只能是POSITIVE,NEUTRAL,NEGATIVE之一。这一切都在代码层面得到了保证。5. 完整示例与代码实现构建一个天气查询助手让我们构建一个更实用的例子一个天气查询助手。用户输入一个城市名助手返回结构化的天气信息。5.1 定义复杂的输出模型天气信息通常包含多个数据点。我们设计一个嵌套的模型。from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime class Temperature(BaseModel): current: float Field(description当前温度单位摄氏度) feels_like: float Field(description体感温度单位摄氏度) min: Optional[float] Field(None, description今日最低温度) max: Optional[float] Field(None, description今日最高温度) class WeatherCondition(BaseModel): main: str Field(description主要天气状况如 Rain, Snow, Clouds, Clear) description: str Field(description详细的天气描述) class ForecastItem(BaseModel): time: datetime Field(description预报时间点) temp: float Field(description该时刻温度) condition: WeatherCondition Field(description天气状况) class WeatherReport(BaseModel): 针对某个城市的完整天气报告 city: str Field(description城市名称) country: str Field(description国家代码) timestamp: datetime Field(description数据更新时间) temperature: Temperature Field(description温度信息) humidity: int Field(description湿度百分比, ge0, le100) wind_speed: float Field(description风速米/秒, ge0) conditions: List[WeatherCondition] Field(description天气状况列表) forecast: Optional[List[ForecastItem]] Field(None, description未来几小时的预报)这个模型定义了嵌套关系WeatherReport包含Temperature和WeatherCondition列表还可以包含ForecastItem列表。Optional表示该字段可以为None。5.2 创建具有系统提示的 Agent我们可以给 Agent 一个系统角色让它更专注于特定任务。from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel model OpenAIModel(gpt-4o-mini) # 创建 Agent绑定复杂模型并设置系统提示 weather_agent Agent( modelmodel, result_typeWeatherReport, system_prompt( 你是一个专业的天气信息提取助手。 用户会给你一段包含某地天气信息的文本可能是从网页或对话中截取的。 你的任务是精确地从中提取信息并严格按照指定的 JSON 格式输出。 如果某些信息在文本中没有明确提及请将对应字段设为 null 或合理的默认值。 请确保数值类型正确时间格式化为 ISO 8601 字符串。 ), )5.3 运行并处理结果现在我们模拟一段包含天气信息的文本让 Agent 进行提取。import asyncio from pydantic_ai import RunContext async def get_weather_report(): # 模拟一段从网络爬取或用户提供的非结构化天气文本 unstructured_text 这里是北京中国的当前天气。 更新时间2023-10-27T14:30:0008:00。 现在气温 15°C体感温度 13°C。今天最高温18°C最低温10°C。 湿度是65%。风速每秒3.5米。 天气状况多云伴有轻度雾霾。 未来三小时预报 15:00: 16°C 多云。 16:00: 17°C 晴间多云。 17:00: 16°C 多云。 ctx RunContext(user_promptunstructured_text) result await weather_agent.run(ctx) report: WeatherReport result.data # 现在我们可以以编程方式轻松使用这些数据 print(f {report.city} ({report.country}) 天气报告 ) print(f更新时间: {report.timestamp}) print(f当前温度: {report.temperature.current}°C (体感 {report.temperature.feels_like}°C)) print(f温度范围: {report.temperature.min}°C ~ {report.temperature.max}°C) print(f湿度: {report.humidity}%) print(f风速: {report.wind_speed} m/s) print(天气状况:) for cond in report.conditions: print(f - {cond.main}: {cond.description}) if report.forecast: print(\n未来预报:) for fc in report.forecast: print(f {fc.time.strftime(%H:%M)}: {fc.temp}°C, {fc.condition.main}) # 数据可以轻松转换为字典或JSON用于API响应 # report_dict report.model_dump() # import json # report_json report.model_dump_json() asyncio.run(get_weather_report())运行这段代码pydantic-ai会驱动 LLM 从那段自由文本中精准地提取信息并填充到我们定义的WeatherReport模型中。最终result.data就是一个包含了所有层级数据的、类型正确的对象。6. 运行结果与效果验证执行上述get_weather_report函数预期的成功输出应该结构清晰、数据完整 北京 (中国) 天气报告 更新时间: 2023-10-27 14:30:0008:00 当前温度: 15.0°C (体感 13.0°C) 温度范围: 10.0°C ~ 18.0°C 湿度: 65% 风速: 3.5 m/s 天气状况: - Clouds: 多云 - Mist: 伴有轻度雾霾 未来预报: 15:00: 16.0°C, Clouds 16:00: 17.0°C, Clear 17:00: 16.0°C, Clouds如何验证成功程序无异常代码没有抛出pydantic.ValidationError或其他解析错误说明 LLM 的输出成功通过了模型验证。数据访问正常能够通过report.city、report.temperature.current等方式正常访问嵌套属性且类型正确如float、int。数据符合预期提取出的城市、温度、湿度等值与输入文本相符。如果运行失败第一步应该看哪里查看result对象或捕获异常。pydantic-ai在解析失败时会抛出异常。你可以检查API 密钥和网络是否配置正确是否有网络问题。模型能力过于复杂的结构或指令较弱的模型如gpt-3.5-turbo可能无法很好理解。尝试使用gpt-4或claude-3系列。提示词清晰度检查system_prompt和user_prompt是否清晰指明了任务和格式要求。字段的description是否足够明确。输出格式打开调试模式如设置Agent(..., debugTrue)查看实际发送给 LLM 的提示词和收到的原始响应看模型是否理解了结构化输出的要求。7. 高级用法与工程实践掌握了基础用法后我们来看一些提升可靠性和效率的高级技巧。7.1 依赖注入与动态上下文pydantic-ai支持依赖注入允许你在运行时为 Agent 提供额外的上下文信息这在处理需要实时数据的任务时非常有用。from pydantic_ai import Agent, RunContext, Depends from pydantic_ai.models.openai import OpenAIModel model OpenAIModel(gpt-4o-mini) # 定义一个依赖函数例如获取当前用户信息 def get_current_user(): # 这里可以从请求上下文、数据库或会话中获取 return {user_id: 123, role: premium_user} # 定义一个需要用户信息的输出模型 class PersonalizedResponse(BaseModel): greeting: str recommended_action: str # 创建 Agent并通过 deps 参数声明依赖 personal_agent Agent( modelmodel, result_typePersonalizedResponse, deps[get_current_user], # 声明依赖 system_prompt根据当前用户的身份生成个性化的问候和建议。 ) async def run_personalized(): # 运行时会自动调用 get_current_user() 并将结果注入上下文 result await personal_agent.run(我今天应该做什么) print(result.data.greeting) print(f建议{result.data.recommended_action}) # 输出可能为“尊敬的 premium_user 123您好” “建议您可以访问我们的专属高级功能区。”7.2 工具调用Tool Calling集成除了结构化输出pydantic-ai也支持让 LLM 决定调用你提供的工具函数并将工具执行结果纳入后续的思考。这实现了更复杂的多步推理和行动。from pydantic_ai import Agent, RunContext from pydantic_ai.models.openai import OpenAIModel model OpenAIModel(gpt-4o-mini) # 1. 定义工具函数。使用 tool 装饰器。 from pydantic_ai.tools import tool tool def get_stock_price(symbol: str) - float: 根据股票代码获取当前股价。 # 模拟一个数据库或 API 调用 mock_prices {AAPL: 175.25, GOOGL: 135.80, MSFT: 330.45} return mock_prices.get(symbol.upper(), 0.0) tool def calculate_investment_value(price: float, shares: int) - float: 计算投资总价值。 return price * shares # 2. 创建 Agent 并注册工具 investment_agent Agent( modelmodel, result_typestr, # 最终输出可以是简单文本 tools[get_stock_price, calculate_investment_value], # 注册工具 system_prompt你是一个投资助手可以查询股价并进行计算。请根据用户的问题决定是否需要调用工具。, ) async def ask_investment(): result await investment_agent.run( 如果我持有 10 股 AAPL 和 5 股 GOOGL我的投资组合总价值是多少 ) print(result.data) # Agent 会先调用 get_stock_price(AAPL) 和 get_stock_price(GOOGL) # 然后调用 calculate_investment_value最后总结输出。 # 输出可能为“您的投资组合总价值为 1752.5 (AAPL) 679.0 (GOOGL) 2431.5 美元。” asyncio.run(ask_investment())7.3 流式输出Streaming对于需要长时间生成或希望实时显示结果的应用可以使用流式输出。async def stream_structured_agent(): model OpenAIModel(gpt-4o-mini) stream_agent Agent(modelmodel, result_typeSentimentAnalysis) ctx RunContext(user_prompt这部电影的视觉效果令人惊叹但剧情拖沓。) async for chunk in stream_agent.run_stream(ctx): # chunk 可以是文本片段、工具调用请求或最终结果 if chunk.is_text: print(chunk.text, end, flushTrue) # 实时显示模型思考过程 elif chunk.is_result: final_result: SentimentAnalysis chunk.result.data print(f\n\n最终结构化结果: {final_result})8. 常见问题与排查思路在实际使用中你可能会遇到以下问题问题现象可能原因排查方式解决方案pydantic.ValidationError1. LLM 输出无法解析为模型。2. 字段类型不匹配如字符串传给了整型字段。3. 字段值为空但未标记Optional。1. 设置Agent(..., debugTrue)查看原始 LLM 输出。2. 检查模型字段的description是否清晰。3. 查看错误信息具体指出哪个字段验证失败。1. 简化输出模型或使用更强大的 LLM。2. 为可能为空的字段添加Optional或设置默认值Field(None)。3. 在system_prompt中更明确地强调输出格式。Agent 返回None或报错1. API 密钥错误或网络问题。2. 模型名称错误或不可用。3. 超出了速率限制或配额。1. 检查环境变量OPENAI_API_KEY。2. 尝试一个简单的纯文本请求测试连通性。3. 查看提供商控制台的用量和错误日志。1. 确认密钥有效且有余额。2. 使用正确的模型名称如gpt-4-turbo-preview。3. 添加重试逻辑或降低请求频率。输出结果不准确1. 提示词system_prompt和user_prompt不够清晰。2. 字段描述Field(description...)有歧义。3. 任务本身对当前模型太复杂。1. 在system_prompt中明确指令如“你必须输出一个有效的 JSON 对象”。2. 用更简单、无歧义的语言重写字段描述。3. 尝试将复杂任务拆解为多个简单 Agent 链式调用。1. 采用“角色-任务-格式”三段式系统提示。2. 提供少量示例Few-shot在提示词中。3. 升级到更强大的模型。工具调用不被触发1. 工具函数参数或返回值类型提示不明确。2. LLM 认为不需要调用工具。3. 工具描述docstring不够详细。1. 确保工具函数有完整的类型注解和清晰的文档字符串。2. 在user_prompt中明确要求使用工具或在system_prompt中强调。1. 完善工具函数的类型提示和文档。2. 使用tool装饰器的description参数提供更详细的工具描述。性能慢或 Token 消耗大1. 输出模型过于复杂导致提示词很长。2. 嵌套太深或列表字段可能产生很长输出。1. 使用Agent的result_type参数而不是在提示词中描述结构。2. 分析result.usage查看 Token 消耗分布。1. 优化模型设计只保留必要字段。2. 对于长列表考虑分页或让 LLM 只返回摘要。3. 使用gpt-4o-mini等性价比更高的模型。9. 最佳实践与工程建议要将pydantic-ai稳健地用于生产环境请遵循以下建议模型设计先行在编写 Agent 逻辑之前花时间精心设计你的 Pydantic 输出模型。清晰的字段名和详细的description是成功的一半。使用Optional和默认值来处理可能缺失的信息。强化系统提示system_prompt是引导 LLM 行为的关键。采用模板化提示词例如“你是一个 [角色]。你的任务是 [具体任务]。你必须将输出严格遵循以下 JSON 结构[简要说明结构]。不要添加任何额外的解释或注释。”实施重试与降级网络或 API 可能不稳定。为agent.run()添加重试机制如tenacity库。对于关键任务可以准备一个降级方案例如当结构化输出失败时回退到解析原始文本。设置超时与限制为异步调用设置合理的超时时间避免长时间阻塞。利用result.usage监控 Token 消耗对用户输入长度或模型输出长度进行限制以控制成本。进行充分的测试为你的 Agent 编写单元测试和集成测试。测试应包括正常用例验证典型输入能产生正确输出。边界用例测试缺失信息、极端值、模糊描述。错误恢复测试当 LLM 返回完全不相关内容时你的程序是否能优雅处理如捕获验证异常并返回友好错误。版本化与演进当你需要更改输出模型时如添加新字段要考虑向后兼容性。可以创建新的模型版本并通过 Agent 的配置或工厂模式来管理不同版本的模型避免直接破坏现有接口。安全与权限永远不要盲目信任 LLM 的输出即使它已被结构化。如果输出用于数据库查询、系统命令或金融交易必须进行额外的业务逻辑验证和权限检查。遵循最小权限原则。通过pydantic-ai我们将 LLM 从“黑盒文本生成器”变成了“可预测的数据生成服务”。它极大地减少了集成 AI 功能时的胶水代码和不确定性让开发者能够更专注于业务逻辑本身。下次当你需要从 LLM 获取一个干净的、程序可读的结果时不必再对着杂乱的文本发愁。定义一个 Pydantic 模型创建一个 Agent然后像调用本地函数一样获取类型安全的结果。这不仅是效率的提升更是工程思维在 AI 应用开发中的一次重要落地。
最近很多开发者朋友在尝试将大语言模型LLM集成到自己的应用中时都会遇到一个共同的“拦路虎”如何让模型输出的内容尤其是那些需要结构化、格式化的内容能够稳定、可靠地符合我们的要求比如让模型生成一个标准的 JSON 对象来返回天气数据或者输出一个格式工整的 Markdown 表格来总结会议纪要。你可能会发现直接让模型“生成一个 JSON”它有时会“放飞自我”——在 JSON 外面加上解释性文字或者漏掉引号甚至返回一段看似 JSON 但无法解析的文本。这种不稳定性让 LLM 在需要与下游系统如数据库、API无缝对接的生产环境中显得有些“不靠谱”。这背后的核心痛点就是LLM 输出的不可控性。它像一个才华横溢但有些随性的助手你需要花大量精力去“调教”和“后处理”它的输出才能让它融入严谨的工程流水线。今天要介绍的主角——Pydantic结合其强大的pydantic-ai库正是为了解决这个“最后一公里”的问题而生的。它不是一个新模型而是一套工程化框架其核心思想是用代码定义你期望的输出结构然后让 LLM 的生成过程被这个结构所约束和引导最终直接得到类型安全、格式正确的 Python 对象。简单来说它想让 LLM 的输出变得像调用一个普通函数一样可靠输入参数返回一个确定类型的对象。本文将深入探讨如何利用 Pydantic 来“管好” LLM 的输出让你告别繁琐的正则表达式匹配和字符串解析真正实现 AI 能力的即插即用。1. 这篇文章真正要解决的问题从“文本生成”到“函数调用”在传统开发中我们调用一个函数或 API返回值的数据类型和结构是预先定义好的。例如一个get_user_info(user_id: int) - User函数我们明确知道它会返回一个User对象里面有name、email等属性。但当我们将任务交给 LLM 时情况就变了。我们得到的是一段自由文本。为了从这段文本中提取结构化信息开发者通常需要精心设计提示词Prompt反复强调格式要求。在代码中编写复杂的后处理逻辑如正则表达式、字符串分割、JSON 解析并处理各种可能的异常格式。进行大量的测试和调试以覆盖模型可能产生的各种“创意”输出。这个过程不仅效率低下而且极其脆弱。提示词的微小改动或模型版本的更新都可能导致后处理逻辑失效。Pydantic pydantic-ai提供的解决方案是“结构化的生成”。它允许你用 Pydantic Model 定义输出像定义数据库表或 API 响应一样用 Python 类来定义你希望 LLM 生成的数据结构。将结构作为生成的一部分这个结构定义会被巧妙地融入到给 LLM 的提示词中引导模型在生成时就直接思考如何填充这个结构。直接得到类型化对象LLM 的原始输出会经过库的解析和验证直接转换为你定义的 Pydantic 模型实例。你可以立刻使用.操作符访问属性享受 IDE 的自动补全和静态类型检查。这本质上是在 LLM 的“自由创作”和程序的“严格接口”之间架起了一座坚固的桥梁。它解决的不是“生成什么内容”的问题而是“如何让生成的内容能被程序直接、可靠地使用”的问题。2. 基础概念与核心原理在深入代码之前我们先厘清几个关键概念理解pydantic-ai是如何工作的。2.1 Pydantic 是什么Pydantic 是一个 Python 库主要用于数据验证和设置管理。它利用 Python 的类型注解type hints来定义数据的形状Schema并自动验证传入的数据是否符合这个形状同时进行类型转换。一个简单的例子from pydantic import BaseModel class User(BaseModel): name: str age: int email: str # 有效数据 user1 User(nameAlice, age30, emailaliceexample.com) print(user1.name) # 输出: Alice # 无效数据会引发验证错误 try: user2 User(nameBob, agenot_a_number, emailbobexample.com) except Exception as e: print(e) # 输出验证错误信息Pydantic 确保了User对象的数据总是符合我们定义的规范。2.2pydantic-ai的核心思想pydantic-ai库将 Pydantic 的这种“定义-验证”能力逆向应用到了 LLM 的文本生成过程上。其核心流程可以概括为定义输出模型你创建一个 PydanticBaseModel描述你希望 LLM 生成的信息结构。创建智能体Agent你将这个输出模型“告诉”一个pydantic-ai的Agent并指定使用的 LLM如 OpenAI GPT-4, Anthropic Claude 等。运行并获取结构化结果你向 Agent 提问或下达指令。Agent 在内部会 a.构建增强提示将你的问题/指令与输出模型的结构描述结合生成一个更精确的提示词发送给 LLM。 b.解析与验证接收 LLM 的原始文本回复尝试将其解析并填充到预定义的输出模型中。 c.返回模型实例如果解析成功且通过 Pydantic 验证则直接返回该模型的一个实例。如果失败它可以进行重试或报错。2.3 与传统“函数调用Function Calling”的区别OpenAI 等厂商也提供了“函数调用”功能允许你描述函数让模型返回调用该函数所需的参数。pydantic-ai与它既有相似之处也有不同相似点两者都旨在让 LLM 输出结构化数据。不同点函数调用侧重于“让模型决定是否以及如何调用某个已知函数”。输出是函数名和参数核心是“动作”。pydantic-ai侧重于“让模型直接生成符合某个复杂结构的数据”。输出就是数据本身核心是“信息提取与结构化”。它更通用不限于函数参数可以描述任何复杂嵌套的对象。控制权pydantic-ai将结构定义完全放在开发者手中通过 Pydantic Model提供了更强的类型安全和代码集成度。3. 环境准备与前置条件开始实践前你需要准备好 Python 环境和一个可用的 LLM API。3.1 Python 环境建议使用 Python 3.8 及以上版本。使用虚拟环境是一个好习惯。# 创建并激活虚拟环境 (可选) python -m venv venv # Linux/macOS source venv/bin/activate # Windows venv\Scripts\activate3.2 安装依赖核心需要安装pydantic-ai。它将自动安装正确版本的pydantic。pip install pydantic-ai根据你计划使用的 LLM 提供商还需要安装对应的 SDK 并配置 API 密钥。本文以 OpenAI 为例pip install openai然后在环境变量中设置你的 OpenAI API Key# Linux/macOS export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here你也可以在代码中直接设置但出于安全考虑更推荐使用环境变量。4. 核心流程拆解第一个结构化输出让我们通过一个最简单的例子感受pydantic-ai的工作流程。4.1 第一步定义输出模型假设我们想让 LLM 从一个句子中提取人名和情绪。我们首先定义这个数据结构。from pydantic import BaseModel, Field from typing import Literal # 定义输出数据结构 class SentimentAnalysis(BaseModel): 分析句子中的情绪和提及的人物 person_name: str Field(description句子中提及的人物姓名) sentiment: Literal[POSITIVE, NEUTRAL, NEGATIVE] Field(description针对该人物的情绪倾向) confidence: float Field(description分析结果的置信度0到1之间, ge0, le1)BaseModel: 所有输出模型的基类。Field: 用于为字段提供更详细的描述和约束。这里的description非常重要它会帮助 LLM 理解每个字段的含义。ge和le是 Pydantic 的数值范围校验器。Literal: 表示该字段只能是列举值中的一个这为 LLM 提供了明确的选项。4.2 第二步创建并运行 Agent接下来我们创建一个 Agent让它使用这个模型来生成结果。from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel # 1. 选择模型。这里使用 OpenAI 的 gpt-4o-mini你也可以用 gpt-4-turbo 等。 model OpenAIModel(gpt-4o-mini) # 2. 创建 Agent并指定其输出模型为 SentimentAnalysis sentiment_agent Agent( modelmodel, result_typeSentimentAnalysis, # 关键绑定输出模型 ) # 3. 运行 Agent提出请求 async def main(): result await sentiment_agent.run( 从这句话中提取信息尽管项目延期了但张三仍然对团队的努力感到非常骄傲。 ) # result.data 就是 SentimentAnalysis 的一个实例 analysis: SentimentAnalysis result.data print(f人物: {analysis.person_name}) print(f情绪: {analysis.sentiment}) print(f置信度: {analysis.confidence:.2f}) # 你可以像使用普通对象一样访问其属性 if analysis.sentiment POSITIVE: print(检测到积极情绪) # 运行异步函数 import asyncio asyncio.run(main())关键点解析Agent: 核心执行器封装了与 LLM 的交互逻辑。result_type: 这是将 Agent 与 Pydantic 模型绑定的关键参数。它告诉 Agent“你每次运行的结果都应该符合这个模型”。agent.run(): 发送提示词并获取结果。返回的result对象包含原始响应、消耗的 Token 等信息而result.data就是我们需要的结构化对象。运行这段代码你可能会得到类似这样的输出人物: 张三 情绪: POSITIVE 置信度: 0.95 检测到积极情绪最重要的是analysis是一个SentimentAnalysis类型的对象analysis.person_name是字符串类型analysis.sentiment只能是POSITIVE,NEUTRAL,NEGATIVE之一。这一切都在代码层面得到了保证。5. 完整示例与代码实现构建一个天气查询助手让我们构建一个更实用的例子一个天气查询助手。用户输入一个城市名助手返回结构化的天气信息。5.1 定义复杂的输出模型天气信息通常包含多个数据点。我们设计一个嵌套的模型。from pydantic import BaseModel, Field from typing import List, Optional from datetime import datetime class Temperature(BaseModel): current: float Field(description当前温度单位摄氏度) feels_like: float Field(description体感温度单位摄氏度) min: Optional[float] Field(None, description今日最低温度) max: Optional[float] Field(None, description今日最高温度) class WeatherCondition(BaseModel): main: str Field(description主要天气状况如 Rain, Snow, Clouds, Clear) description: str Field(description详细的天气描述) class ForecastItem(BaseModel): time: datetime Field(description预报时间点) temp: float Field(description该时刻温度) condition: WeatherCondition Field(description天气状况) class WeatherReport(BaseModel): 针对某个城市的完整天气报告 city: str Field(description城市名称) country: str Field(description国家代码) timestamp: datetime Field(description数据更新时间) temperature: Temperature Field(description温度信息) humidity: int Field(description湿度百分比, ge0, le100) wind_speed: float Field(description风速米/秒, ge0) conditions: List[WeatherCondition] Field(description天气状况列表) forecast: Optional[List[ForecastItem]] Field(None, description未来几小时的预报)这个模型定义了嵌套关系WeatherReport包含Temperature和WeatherCondition列表还可以包含ForecastItem列表。Optional表示该字段可以为None。5.2 创建具有系统提示的 Agent我们可以给 Agent 一个系统角色让它更专注于特定任务。from pydantic_ai import Agent from pydantic_ai.models.openai import OpenAIModel model OpenAIModel(gpt-4o-mini) # 创建 Agent绑定复杂模型并设置系统提示 weather_agent Agent( modelmodel, result_typeWeatherReport, system_prompt( 你是一个专业的天气信息提取助手。 用户会给你一段包含某地天气信息的文本可能是从网页或对话中截取的。 你的任务是精确地从中提取信息并严格按照指定的 JSON 格式输出。 如果某些信息在文本中没有明确提及请将对应字段设为 null 或合理的默认值。 请确保数值类型正确时间格式化为 ISO 8601 字符串。 ), )5.3 运行并处理结果现在我们模拟一段包含天气信息的文本让 Agent 进行提取。import asyncio from pydantic_ai import RunContext async def get_weather_report(): # 模拟一段从网络爬取或用户提供的非结构化天气文本 unstructured_text 这里是北京中国的当前天气。 更新时间2023-10-27T14:30:0008:00。 现在气温 15°C体感温度 13°C。今天最高温18°C最低温10°C。 湿度是65%。风速每秒3.5米。 天气状况多云伴有轻度雾霾。 未来三小时预报 15:00: 16°C 多云。 16:00: 17°C 晴间多云。 17:00: 16°C 多云。 ctx RunContext(user_promptunstructured_text) result await weather_agent.run(ctx) report: WeatherReport result.data # 现在我们可以以编程方式轻松使用这些数据 print(f {report.city} ({report.country}) 天气报告 ) print(f更新时间: {report.timestamp}) print(f当前温度: {report.temperature.current}°C (体感 {report.temperature.feels_like}°C)) print(f温度范围: {report.temperature.min}°C ~ {report.temperature.max}°C) print(f湿度: {report.humidity}%) print(f风速: {report.wind_speed} m/s) print(天气状况:) for cond in report.conditions: print(f - {cond.main}: {cond.description}) if report.forecast: print(\n未来预报:) for fc in report.forecast: print(f {fc.time.strftime(%H:%M)}: {fc.temp}°C, {fc.condition.main}) # 数据可以轻松转换为字典或JSON用于API响应 # report_dict report.model_dump() # import json # report_json report.model_dump_json() asyncio.run(get_weather_report())运行这段代码pydantic-ai会驱动 LLM 从那段自由文本中精准地提取信息并填充到我们定义的WeatherReport模型中。最终result.data就是一个包含了所有层级数据的、类型正确的对象。6. 运行结果与效果验证执行上述get_weather_report函数预期的成功输出应该结构清晰、数据完整 北京 (中国) 天气报告 更新时间: 2023-10-27 14:30:0008:00 当前温度: 15.0°C (体感 13.0°C) 温度范围: 10.0°C ~ 18.0°C 湿度: 65% 风速: 3.5 m/s 天气状况: - Clouds: 多云 - Mist: 伴有轻度雾霾 未来预报: 15:00: 16.0°C, Clouds 16:00: 17.0°C, Clear 17:00: 16.0°C, Clouds如何验证成功程序无异常代码没有抛出pydantic.ValidationError或其他解析错误说明 LLM 的输出成功通过了模型验证。数据访问正常能够通过report.city、report.temperature.current等方式正常访问嵌套属性且类型正确如float、int。数据符合预期提取出的城市、温度、湿度等值与输入文本相符。如果运行失败第一步应该看哪里查看result对象或捕获异常。pydantic-ai在解析失败时会抛出异常。你可以检查API 密钥和网络是否配置正确是否有网络问题。模型能力过于复杂的结构或指令较弱的模型如gpt-3.5-turbo可能无法很好理解。尝试使用gpt-4或claude-3系列。提示词清晰度检查system_prompt和user_prompt是否清晰指明了任务和格式要求。字段的description是否足够明确。输出格式打开调试模式如设置Agent(..., debugTrue)查看实际发送给 LLM 的提示词和收到的原始响应看模型是否理解了结构化输出的要求。7. 高级用法与工程实践掌握了基础用法后我们来看一些提升可靠性和效率的高级技巧。7.1 依赖注入与动态上下文pydantic-ai支持依赖注入允许你在运行时为 Agent 提供额外的上下文信息这在处理需要实时数据的任务时非常有用。from pydantic_ai import Agent, RunContext, Depends from pydantic_ai.models.openai import OpenAIModel model OpenAIModel(gpt-4o-mini) # 定义一个依赖函数例如获取当前用户信息 def get_current_user(): # 这里可以从请求上下文、数据库或会话中获取 return {user_id: 123, role: premium_user} # 定义一个需要用户信息的输出模型 class PersonalizedResponse(BaseModel): greeting: str recommended_action: str # 创建 Agent并通过 deps 参数声明依赖 personal_agent Agent( modelmodel, result_typePersonalizedResponse, deps[get_current_user], # 声明依赖 system_prompt根据当前用户的身份生成个性化的问候和建议。 ) async def run_personalized(): # 运行时会自动调用 get_current_user() 并将结果注入上下文 result await personal_agent.run(我今天应该做什么) print(result.data.greeting) print(f建议{result.data.recommended_action}) # 输出可能为“尊敬的 premium_user 123您好” “建议您可以访问我们的专属高级功能区。”7.2 工具调用Tool Calling集成除了结构化输出pydantic-ai也支持让 LLM 决定调用你提供的工具函数并将工具执行结果纳入后续的思考。这实现了更复杂的多步推理和行动。from pydantic_ai import Agent, RunContext from pydantic_ai.models.openai import OpenAIModel model OpenAIModel(gpt-4o-mini) # 1. 定义工具函数。使用 tool 装饰器。 from pydantic_ai.tools import tool tool def get_stock_price(symbol: str) - float: 根据股票代码获取当前股价。 # 模拟一个数据库或 API 调用 mock_prices {AAPL: 175.25, GOOGL: 135.80, MSFT: 330.45} return mock_prices.get(symbol.upper(), 0.0) tool def calculate_investment_value(price: float, shares: int) - float: 计算投资总价值。 return price * shares # 2. 创建 Agent 并注册工具 investment_agent Agent( modelmodel, result_typestr, # 最终输出可以是简单文本 tools[get_stock_price, calculate_investment_value], # 注册工具 system_prompt你是一个投资助手可以查询股价并进行计算。请根据用户的问题决定是否需要调用工具。, ) async def ask_investment(): result await investment_agent.run( 如果我持有 10 股 AAPL 和 5 股 GOOGL我的投资组合总价值是多少 ) print(result.data) # Agent 会先调用 get_stock_price(AAPL) 和 get_stock_price(GOOGL) # 然后调用 calculate_investment_value最后总结输出。 # 输出可能为“您的投资组合总价值为 1752.5 (AAPL) 679.0 (GOOGL) 2431.5 美元。” asyncio.run(ask_investment())7.3 流式输出Streaming对于需要长时间生成或希望实时显示结果的应用可以使用流式输出。async def stream_structured_agent(): model OpenAIModel(gpt-4o-mini) stream_agent Agent(modelmodel, result_typeSentimentAnalysis) ctx RunContext(user_prompt这部电影的视觉效果令人惊叹但剧情拖沓。) async for chunk in stream_agent.run_stream(ctx): # chunk 可以是文本片段、工具调用请求或最终结果 if chunk.is_text: print(chunk.text, end, flushTrue) # 实时显示模型思考过程 elif chunk.is_result: final_result: SentimentAnalysis chunk.result.data print(f\n\n最终结构化结果: {final_result})8. 常见问题与排查思路在实际使用中你可能会遇到以下问题问题现象可能原因排查方式解决方案pydantic.ValidationError1. LLM 输出无法解析为模型。2. 字段类型不匹配如字符串传给了整型字段。3. 字段值为空但未标记Optional。1. 设置Agent(..., debugTrue)查看原始 LLM 输出。2. 检查模型字段的description是否清晰。3. 查看错误信息具体指出哪个字段验证失败。1. 简化输出模型或使用更强大的 LLM。2. 为可能为空的字段添加Optional或设置默认值Field(None)。3. 在system_prompt中更明确地强调输出格式。Agent 返回None或报错1. API 密钥错误或网络问题。2. 模型名称错误或不可用。3. 超出了速率限制或配额。1. 检查环境变量OPENAI_API_KEY。2. 尝试一个简单的纯文本请求测试连通性。3. 查看提供商控制台的用量和错误日志。1. 确认密钥有效且有余额。2. 使用正确的模型名称如gpt-4-turbo-preview。3. 添加重试逻辑或降低请求频率。输出结果不准确1. 提示词system_prompt和user_prompt不够清晰。2. 字段描述Field(description...)有歧义。3. 任务本身对当前模型太复杂。1. 在system_prompt中明确指令如“你必须输出一个有效的 JSON 对象”。2. 用更简单、无歧义的语言重写字段描述。3. 尝试将复杂任务拆解为多个简单 Agent 链式调用。1. 采用“角色-任务-格式”三段式系统提示。2. 提供少量示例Few-shot在提示词中。3. 升级到更强大的模型。工具调用不被触发1. 工具函数参数或返回值类型提示不明确。2. LLM 认为不需要调用工具。3. 工具描述docstring不够详细。1. 确保工具函数有完整的类型注解和清晰的文档字符串。2. 在user_prompt中明确要求使用工具或在system_prompt中强调。1. 完善工具函数的类型提示和文档。2. 使用tool装饰器的description参数提供更详细的工具描述。性能慢或 Token 消耗大1. 输出模型过于复杂导致提示词很长。2. 嵌套太深或列表字段可能产生很长输出。1. 使用Agent的result_type参数而不是在提示词中描述结构。2. 分析result.usage查看 Token 消耗分布。1. 优化模型设计只保留必要字段。2. 对于长列表考虑分页或让 LLM 只返回摘要。3. 使用gpt-4o-mini等性价比更高的模型。9. 最佳实践与工程建议要将pydantic-ai稳健地用于生产环境请遵循以下建议模型设计先行在编写 Agent 逻辑之前花时间精心设计你的 Pydantic 输出模型。清晰的字段名和详细的description是成功的一半。使用Optional和默认值来处理可能缺失的信息。强化系统提示system_prompt是引导 LLM 行为的关键。采用模板化提示词例如“你是一个 [角色]。你的任务是 [具体任务]。你必须将输出严格遵循以下 JSON 结构[简要说明结构]。不要添加任何额外的解释或注释。”实施重试与降级网络或 API 可能不稳定。为agent.run()添加重试机制如tenacity库。对于关键任务可以准备一个降级方案例如当结构化输出失败时回退到解析原始文本。设置超时与限制为异步调用设置合理的超时时间避免长时间阻塞。利用result.usage监控 Token 消耗对用户输入长度或模型输出长度进行限制以控制成本。进行充分的测试为你的 Agent 编写单元测试和集成测试。测试应包括正常用例验证典型输入能产生正确输出。边界用例测试缺失信息、极端值、模糊描述。错误恢复测试当 LLM 返回完全不相关内容时你的程序是否能优雅处理如捕获验证异常并返回友好错误。版本化与演进当你需要更改输出模型时如添加新字段要考虑向后兼容性。可以创建新的模型版本并通过 Agent 的配置或工厂模式来管理不同版本的模型避免直接破坏现有接口。安全与权限永远不要盲目信任 LLM 的输出即使它已被结构化。如果输出用于数据库查询、系统命令或金融交易必须进行额外的业务逻辑验证和权限检查。遵循最小权限原则。通过pydantic-ai我们将 LLM 从“黑盒文本生成器”变成了“可预测的数据生成服务”。它极大地减少了集成 AI 功能时的胶水代码和不确定性让开发者能够更专注于业务逻辑本身。下次当你需要从 LLM 获取一个干净的、程序可读的结果时不必再对着杂乱的文本发愁。定义一个 Pydantic 模型创建一个 Agent然后像调用本地函数一样获取类型安全的结果。这不仅是效率的提升更是工程思维在 AI 应用开发中的一次重要落地。
)
