最近,很多开发者朋友在尝试将大语言模型(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(name="Alice", age=30, email="alice@example.com") print(user1.name) # 输出: Alice # 无效数据会引发验证错误 try: user2 = User(name="Bob", age="not_a_number", email="bob@example.com") except Exception as e: print(e) # 输出验证错误信息Pydantic 确保了User对象的数据总是符合我们定义的规范。
2.2pydantic-ai的核心思想
pydantic-ai库将 Pydantic 的这种“定义-验证”能力,逆向应用到了 LLM 的文本生成过程上。其核心流程可以概括为:
- 定义输出模型:你创建一个 Pydantic
BaseModel,描述你希望 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_KEY='your-api-key-here' # Windows (PowerShell) $env:OPENAI_API_KEY='your-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之间", ge=0, le=1)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( model=model, result_type=SentimentAnalysis, # 关键:绑定输出模型 ) # 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="湿度百分比", ge=0, le=100) wind_speed: float = Field(description="风速,米/秒", ge=0) 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( model=model, result_type=WeatherReport, 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:00+08: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_prompt=unstructured_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:00+08: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(..., debug=True))查看实际发送给 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( model=model, result_type=PersonalizedResponse, 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( model=model, result_type=str, # 最终输出可以是简单文本 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(model=model, result_type=SentimentAnalysis) ctx = RunContext(user_prompt="这部电影的视觉效果令人惊叹,但剧情拖沓。") async for chunk in stream_agent.run_stream(ctx): # chunk 可以是文本片段、工具调用请求或最终结果 if chunk.is_text: print(chunk.text, end='', flush=True) # 实时显示模型思考过程 elif chunk.is_result: final_result: SentimentAnalysis = chunk.result.data print(f"\n\n最终结构化结果: {final_result}")8. 常见问题与排查思路
在实际使用中,你可能会遇到以下问题:
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
pydantic.ValidationError | 1. LLM 输出无法解析为模型。 2. 字段类型不匹配(如字符串传给了整型字段)。 3. 字段值为空但未标记 Optional。 | 1. 设置Agent(..., debug=True)查看原始 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 应用开发中的一次重要落地。