拓冰网站定制 
如果你正在寻找一个能真正落地、能快速构建生产级 AI 应用的开源框架而不是停留在玩具 Demo 或复杂的学术概念上那么 Pydantic AI 绝对值得你花时间深入了解。过去一年我们见证了 LangChain、LlamaIndex 等框架的崛起它们极大地降低了 AI 应用开发的门槛。然而随着项目复杂度提升开发者们开始面临新的痛点如何优雅地管理复杂的提示词如何确保 LLM 输出的结构化数据能被程序可靠地使用如何构建一个状态清晰、易于调试的智能体Agent这些问题正是 Pydantic AI 试图从根源上解决的。Pydantic AI 的核心思想非常直接将 Python 的类型系统Type Hints和强大的数据验证库 Pydantic 深度集成到 AI 应用开发流程中。它不是一个试图包罗万象的“全家桶”而是一个专注于“结构化输出”和“智能体状态管理”的利器。这意味着你可以用编写普通 Python 类和方法的方式来定义 AI 的行为、输入和输出并获得编译时级别的类型安全和运行时数据验证。本文将带你从零开始通过一个完整的实战项目深入理解 Pydantic AI 的核心概念、工作流和最佳实践。你将学会如何构建一个具备记忆、工具调用和结构化输出能力的智能体并理解它为何能在生产环境中提供更高的可靠性和开发效率。1. Pydantic AI 解决了什么核心问题在深入代码之前我们必须先理解 Pydantic AI 诞生的背景和它要解决的核心矛盾。1.1 传统 AI 应用开发的痛点假设你要开发一个“智能客服助手”它需要根据用户描述的问题自动分类技术问题、账单问题、一般咨询并提取关键实体如订单号、错误代码。使用传统方式你可能会这样写提示词prompt f 请分析以下用户问题 {user_query} 请按以下格式回复 问题分类[技术问题/账单问题/一般咨询] 关键实体[提取出的订单号、错误代码等若无则写“无”] response llm.invoke(prompt) # 然后你需要手动解析 response 的文本用正则表达式或字符串分割来提取“问题分类”和“关键实体”。这种方式存在几个明显问题输出不稳定LLM 可能不严格按照你指定的格式回复导致解析失败。解析代码脆弱你的解析逻辑正则、字符串处理与提示词强耦合提示词稍作修改解析代码就可能崩溃。缺乏类型安全问题分类和关键实体在代码中是字符串编译器无法帮你检查它们的取值是否合法也无法提供自动补全。状态管理混乱如果这是一个多轮对话的智能体你需要手动维护对话历史、用户信息等状态代码会变得难以维护。1.2 Pydantic AI 的解决方案类型即契约Pydantic AI 将上述过程彻底转变。你首先定义数据模型继承自pydantic.BaseModel这个模型清晰地描述了 AI 应该输出什么。from pydantic import BaseModel, Field from typing import List, Optional class ProblemAnalysis(BaseModel): 对用户问题的分析结果 category: str Field(description问题分类, choices[技术问题, 账单问题, 一般咨询]) key_entities: List[str] Field(description提取出的关键实体如订单号、错误码) requires_human_intervention: bool Field(description是否需要人工介入)然后你创建一个Agent并告诉它当调用你的analyze_problem方法时请使用指定的模型如 GPT-4和提示词模板并且必须返回一个ProblemAnalysis类型的实例。from pydantic_ai import Agent agent Agent( modelopenai:gpt-4-turbo, system_prompt你是一个专业的客服问题分析助手。, ) agent.tool async def analyze_problem(query: str) - ProblemAnalysis: 分析用户问题并返回结构化结果。 return await agent.run(query) # 注意这里只是示意实际调用方式略有不同当agent.run()执行时Pydantic AI 会在后台将你的提示词和用户输入组合。调用 LLM。要求 LLM 必须返回一个符合ProblemAnalysis模型的 JSON 对象。自动用 Pydantic 验证这个 JSON 对象如果不符合定义例如category的值不在choices中会抛出清晰的验证错误。最终返回给你一个ProblemAnalysis的实例对象你可以像使用普通 Python 对象一样访问其属性result.category,result.key_entities。这种“定义输出模型框架负责验证”的模式将不可靠的自然语言输出转化为了可靠的、类型化的程序对象。这是 Pydantic AI 最核心的价值。2. 核心概念快速理解在动手之前我们先厘清 Pydantic AI 的几个核心抽象它们构成了整个框架的骨架。概念类比在 Pydantic AI 中的角色关键点Agent智能体/机器人核心执行单元。它封装了 LLM 模型、系统提示词、工具集以及运行时的状态管理。一个应用可以由多个 Agent 协作完成。不是单次对话而是一个有状态的执行上下文。Model大脑/引擎负责实际调用 LLM如 OpenAI GPT, Anthropic Claude, 本地 Ollama 模型。Agent 在创建时需要指定一个 Model。支持多种后端通过类似openai:gpt-4o的字符串指定。Tool技能/函数Agent 可以调用的外部函数。用于获取实时信息如天气、股票、执行操作如发邮件、查数据库或进行计算。用agent.tool装饰器定义。工具的参数和返回值也推荐用 Pydantic Model 定义以实现结构化调用。Result任务结果agent.run()的返回对象。它包含了本次运行的最终输出数据data、使用的消息历史messages、消耗的 Token 数usage等元信息。通过result.data获取你定义的 Pydantic Model 实例。State记忆/上下文Agent 在多次run调用之间保持的数据。例如你可以让 State 记录对话历史、用户 ID、会话步骤等。State 也是一个 Pydantic Model。状态管理是构建复杂、多轮交互智能体的关键。Dependency依赖项注入到 Agent 工具或运行上下文中的对象例如数据库连接池、HTTP 客户端、配置对象等。用于实现依赖注入使代码更易测试和维护。类似于 FastAPI 的Depends机制。理解这些概念后你会发现 Pydantic AI 的编程模型非常直观定义数据模型描述输入输出和状态 - 创建 Agent配置大脑和基础设定 - 定义工具赋予其能力 - 运行并获取结构化结果。3. 环境准备与安装我们将在 Python 3.10 的环境中进行实战。确保你的环境已就绪。3.1 创建虚拟环境与安装核心库强烈建议使用虚拟环境来管理依赖。# 1. 创建并进入项目目录 mkdir pydantic-ai-tutorial cd pydantic-ai-tutorial # 2. 创建虚拟环境以 venv 为例 python -m venv .venv # 3. 激活虚拟环境 # Linux/macOS source .venv/bin/activate # Windows .venv\Scripts\activate # 4. 安装 Pydantic AI 和 OpenAI 库我们将使用 OpenAI 作为示例模型后端 pip install pydantic-ai openai # 5. 安装可选的开发依赖如用于结构化输出的 pydantic-settings pip install pydantic-settings3.2 配置 API 密钥Pydantic AI 本身不提供 LLM你需要一个模型提供商。这里以 OpenAI 为例。获取 OpenAI API Key访问 OpenAI Platform 创建密钥。设置环境变量推荐方式# Linux/macOS export OPENAI_API_KEY你的-api-key # Windows (PowerShell) $env:OPENAI_API_KEY你的-api-key或者在代码中直接设置不推荐用于生产环境import os os.environ[OPENAI_API_KEY] 你的-api-key3.3 验证安装创建一个简单的测试文件test_install.py# test_install.py import asyncio from pydantic_ai import Agent from pydantic import BaseModel class Joke(BaseModel): setup: str punchline: str async def main(): # 创建一个简单的 Agent agent Agent( modelopenai:gpt-3.5-turbo, system_prompt你是一个讲笑话的助手。, ) # 运行并指定输出类型为 Joke result await agent.run(讲一个关于编程的笑话。, response_typeJoke) joke: Joke result.data print(f问题{joke.setup}) print(f笑点{joke.punchline}) if __name__ __main__: asyncio.run(main())运行它python test_install.py如果看到类似以下的输出说明环境配置成功问题为什么程序员分不清万圣节和圣诞节 笑点因为 Oct 31 Dec 25。4. 实战项目构建智能旅行规划助手我们将通过一个完整的项目来学习 Pydantic AI。这个助手能根据用户偏好预算、兴趣、时间推荐目的地规划大致行程并能调用“工具”查询实时天气。4.1 第一步定义数据模型Models这是 Pydantic AI 开发的起点。我们先在models.py中定义所有需要用到的数据结构。# models.py from pydantic import BaseModel, Field, validator from typing import List, Optional from datetime import date from enum import Enum # 1. 用户输入偏好 class UserPreference(BaseModel): budget_level: str Field(description预算等级, choices[经济, 舒适, 豪华]) interests: List[str] Field(description兴趣列表如[自然风光, 历史古迹, 美食, 购物]) travel_days: int Field(ge1, le30, description旅行天数) start_date: Optional[date] Field(defaultNone, description出发日期) validator(interests) def validate_interests(cls, v): allowed [自然风光, 历史古迹, 美食, 购物, 冒险运动, 艺术文化, 休闲度假] for interest in v: if interest not in allowed: raise ValueError(f兴趣{interest}不在允许的列表中: {allowed}) return v # 2. 推荐的目的地 class Destination(BaseModel): name: str Field(description目的地名称) country: str Field(description所在国家) reason: str Field(description推荐理由需关联用户兴趣) estimated_cost_per_day: float Field(ge0, description日均预估花费当地货币) # 3. 每日行程规划 class DailyPlan(BaseModel): day: int Field(ge1, description第几天) morning: str Field(description上午活动) afternoon: str Field(description下午活动) evening: str Field(description晚上活动) meal_suggestion: str Field(description餐饮建议) # 4. 完整的旅行计划 class TravelPlan(BaseModel): destination: Destination daily_plans: List[DailyPlan] Field(description每日行程安排) total_estimated_cost: float Field(description总预估花费) packing_tips: List[str] Field(description行李准备建议) # 5. 用于工具调用的天气查询结果 class WeatherInfo(BaseModel): city: str date: date condition: str # e.g., Sunny, Rainy high_temp: float # 最高温 low_temp: float # 最低温 humidity: int # 湿度百分比关键点我们使用了Field来添加描述和约束如choices,ge,le这些描述会帮助 LLM 更好地理解字段含义。validator装饰器用于自定义验证逻辑确保输入数据的有效性。模型之间可以嵌套使用如TravelPlan包含Destination和DailyPlan。4.2 第二步创建 Agent 并定义工具接下来在agent_builder.py中创建我们的核心智能体。# agent_builder.py import asyncio from typing import List from pydantic_ai import Agent, RunContext from pydantic_ai.models.openai import OpenAIModel from models import UserPreference, Destination, TravelPlan, WeatherInfo # 初始化 Agent使用 GPT-4 模型以获得更好的推理和结构化输出能力 # 注意实际使用时请根据你的需求选择合适的模型 travel_agent Agent( modelOpenAIModel(gpt-4-turbo), # 等价于 modelopenai:gpt-4-turbo system_prompt 你是一个专业的旅行规划专家。你的任务是根据用户的偏好为他们推荐合适的目的地并制定详细的旅行计划。 你必须严格遵守输出格式确保所有推荐都基于用户的预算和兴趣。 在规划行程时请考虑交通、景点的合理时间安排。 , depsNone, # 暂时没有外部依赖 ) # 定义第一个工具获取目的地推荐模拟 travel_agent.tool async def recommend_destinations( ctx: RunContext[None], # RunContext 提供了运行上下文目前状态为 None preference: UserPreference ) - List[Destination]: 根据用户偏好推荐最多3个潜在旅行目的地。 这是一个模拟工具实际项目中应接入目的地数据库或第三方API。 # 模拟一些逻辑和 API 调用 print(f[工具调用] recommend_destinations: 正在为偏好 {preference.interests} 的用户寻找目的地...) await asyncio.sleep(0.5) # 模拟网络延迟 # 这里应该是真实的业务逻辑例如查询数据库。 # 为了演示我们让 LLM 根据偏好来“生成”推荐。 # 注意在实际工具中我们通常从固定数据源获取但这里我们巧妙地将任务交还给 LLM 来演示。 # 更常见的做法是工具从数据库查询返回固定列表。这里为了展示LLM的灵活性我们让它动态生成。 prompt_for_llm f 用户偏好预算{preference.budget_level}兴趣{preference.interests}旅行{preference.travel_days}天。 请生成3个符合上述条件的真实世界旅行目的地。 每个目的地需要包含名称、国家、推荐理由关联兴趣、日均预估花费请符合其预算等级。 请以 JSON 列表格式回复严格匹配 Destination 模型。 # 我们让同一个 Agent 来生成这个列表但指定输出类型为 List[Destination] result await ctx.agent.run(prompt_for_llm, response_typeList[Destination]) return result.data # 定义第二个工具查询天气预报模拟 travel_agent.tool async def get_weather_forecast( ctx: RunContext[None], city: str, for_date: date ) - WeatherInfo: 查询指定城市在指定日期的天气预报。 这是一个模拟工具返回模拟数据。 print(f[工具调用] get_weather_forecast: 查询{city}在{for_date}的天气...) await asyncio.sleep(0.3) # 模拟数据 - 实际应调用如 OpenWeatherMap 的 API # 这里简单根据城市名生成一个假天气 import random conditions [晴朗, 多云, 小雨, 大雨, 阴天] return WeatherInfo( citycity, datefor_date, conditionrandom.choice(conditions), high_tempround(random.uniform(20, 35), 1), low_tempround(random.uniform(10, 25), 1), humidityrandom.randint(40, 90) )关键点agent.tool装饰器将一个异步函数注册为 Agent 的工具。工具函数的第一个参数通常是RunContext它提供了访问当前 Agent、状态和依赖的入口。工具的参数和返回值都使用了我们之前定义的 Pydantic Model这保证了工具间数据传递的结构化和安全。在recommend_destinations工具中我们演示了一种模式工具内部再次调用ctx.agent.run并指定response_type让 LLM 来帮助完成复杂的逻辑。这在需要 LLM 进行创造性思考的场景下非常有用。4.3 第三步实现核心规划逻辑现在我们创建一个函数它利用上面定义的 Agent 和工具完成从用户输入到完整旅行规划的流程。在planner.py中# planner.py import asyncio from datetime import date, timedelta from agent_builder import travel_agent from models import UserPreference, TravelPlan async def create_travel_plan(user_input: str) - TravelPlan: 主函数接收用户自然语言描述生成完整的旅行计划。 # 第一步让 Agent 从用户描述中提取结构化偏好 print(步骤1: 解析用户偏好...) pref_result await travel_agent.run( user_input, response_typeUserPreference, system_prompt_extra请从用户的描述中提取预算、兴趣、天数和日期信息。如果日期未指定可以忽略。 ) user_pref: UserPreference pref_result.data print(f解析出的偏好{user_pref.dict()}) # 第二步调用工具获取目的地推荐 print(\n步骤2: 获取目的地推荐...) destinations await travel_agent.recommend_destinations(user_pref) # 直接调用工具方法 print(f推荐了 {len(destinations)} 个目的地) for d in destinations: print(f - {d.name} ({d.country}), 理由{d.reason}) # 第三步让用户或模拟选择一个目的地。这里我们简单选第一个。 selected_dest destinations[0] print(f\n步骤3: 选择目的地 {selected_dest.name} 进行详细规划。) # 第四步生成详细的旅行计划包含每日行程 print(\n步骤4: 生成详细旅行计划...) plan_prompt f 基于以下信息为用户制定一份详细的{user_pref.travel_days}天旅行计划 目的地{selected_dest.name}, {selected_dest.country} 用户兴趣{user_pref.interests} 预算等级{user_pref.budget_level} 日均花费参考{selected_dest.estimated_cost_per_day} (当地货币) 请规划每一天的上午、下午、晚上活动和餐饮建议。行程要合理符合兴趣。 最后计算总花费日均花费*天数并给出行李准备建议。 plan_result await travel_agent.run(plan_prompt, response_typeTravelPlan) travel_plan: TravelPlan plan_result.data # 第五步可选如果用户提供了出发日期为行程中的每一天查询天气 if user_pref.start_date: print(\n步骤5: 查询行程天气...) for i, day_plan in enumerate(travel_plan.daily_plans): target_date user_pref.start_date timedelta(daysi) weather await travel_agent.get_weather_forecast(selected_dest.name, target_date) # 我们可以把天气信息附加到每日计划中这里简单打印 print(f 第{day_plan.day}天 ({target_date}): {weather.condition}, {weather.low_temp}-{weather.high_temp}°C) return travel_plan if __name__ __main__: # 模拟用户输入 sample_input 我想下个月15号左右出发玩5天。预算比较宽松可以舒适一点。 我喜欢自然风光和美食想去个有山有水还能吃好的地方。 print(用户请求:, sample_input) print(*50) final_plan asyncio.run(create_travel_plan(sample_input)) print(\n *50) print(✅ 旅行计划生成完毕) print(f目的地{final_plan.destination.name}) print(f总预估花费{final_plan.total_estimated_cost:.2f} (当地货币)) print(行程概览) for day in final_plan.daily_plans: print(f 第{day.day}天) print(f 上午{day.morning}) print(f 下午{day.afternoon}) print(f 晚上{day.evening}) print(f 餐饮{day.meal_suggestion}) print(行李建议, , .join(final_plan.packing_tips))4.4 第四步运行与验证运行我们的主程序python planner.py你应该能看到类似以下的输出清晰地展示了智能体工作的每一步用户请求: 我想下个月15号左右出发玩5天。预算比较宽松可以舒适一点。我喜欢自然风光和美食想去个有山有水还能吃好的地方。 步骤1: 解析用户偏好... 解析出的偏好{budget_level: 舒适, interests: [自然风光, 美食], travel_days: 5, start_date: 2024-06-15} 步骤2: 获取目的地推荐... [工具调用] recommend_destinations: 正在为偏好 [自然风光, 美食] 的用户寻找目的地... 推荐了 3 个目的地 - 杭州 (中国), 理由拥有西湖等自然风光同时以杭帮菜闻名符合美食兴趣。 - 瑞士因特拉肯 (瑞士), 理由坐落于少女峰山脚下自然风光绝美同时可品尝瑞士奶酪火锅等美食。 - 日本京都 (日本), 理由周边有岚山等自然景观同时是日本传统怀石料理的发源地之一。 步骤3: 选择目的地 杭州 进行详细规划。 步骤4: 生成详细旅行计划... [工具调用] get_weather_forecast: 查询杭州在2024-06-15的天气... ... ✅ 旅行计划生成完毕 目的地杭州 总预估花费XXXX.XX (当地货币) 行程概览 第1天 上午抵达杭州入住酒店漫步西湖苏堤。 下午游览雷峰塔了解白蛇传传说。 晚上在楼外楼品尝西湖醋鱼、龙井虾仁。 ...5. 进阶状态管理State与多轮对话上面的例子是单次任务。真正的智能体往往需要记住对话历史。Pydantic AI 通过State概念来优雅地管理状态。让我们升级旅行助手使其能记住用户姓名并在多轮对话中调整计划。5.1 定义对话状态模型在models.py中添加# models.py (追加) class ConversationState(BaseModel): 记录与用户对话的状态 user_name: Optional[str] None discussed_destinations: List[Destination] [] # 讨论过的目的地 current_plan: Optional[TravelPlan] None # 当前生成的计划 conversation_history: List[str] [] # 简化的历史记录5.2 创建有状态的 Agent修改agent_builder.py创建一个新的、有状态的 Agent# agent_builder.py (追加) from models import ConversationState # 创建一个新的、有状态的 Agent。注意 state_type 参数。 stateful_agent Agent( modelopenai:gpt-4-turbo, system_prompt 你是旅行助手小游。请友好地与用户交流记住他们的名字和之前的对话。 你的目标是逐步了解用户需求并最终生成他们满意的旅行计划。 , state_typeConversationState, # 关键指定状态模型 ) stateful_agent.tool async def update_user_name(ctx: RunContext[ConversationState], name: str): 更新用户姓名到状态中。 ctx.state.user_name name return f好的{name}我已经记住你了。 stateful_agent.tool async def save_current_plan(ctx: RunContext[ConversationState], plan: TravelPlan): 将当前计划保存到状态中。 ctx.state.current_plan plan return 旅行计划已保存。你可以随时让我基于这个计划进行调整。 # 这个工具的上下文 RunContext 是 ConversationState stateful_agent.tool async def recommend_with_memory( ctx: RunContext[ConversationState], preference: UserPreference ) - List[Destination]: 推荐目的地并考虑之前讨论过的避免重复。 # 模拟获取推荐列表 result await ctx.agent.run( f根据偏好{preference}推荐目的地。, response_typeList[Destination] ) new_destinations result.data # 过滤掉已经讨论过的 discussed_names {d.name for d in ctx.state.discussed_destinations} filtered_destinations [d for d in new_destinations if d.name not in discussed_names] # 更新状态 ctx.state.discussed_destinations.extend(filtered_destinations) return filtered_destinations[:3] # 返回最多3个5.3 进行多轮对话创建一个新的对话脚本conversation.py# conversation.py import asyncio from agent_builder import stateful_agent from models import UserPreference, TravelPlan async def multi_turn_conversation(): # 初始化一个状态对象 initial_state ConversationState() # 第一轮用户打招呼并告知姓名 print(用户: 你好我是小明。) result1 await stateful_agent.run( 你好我是小明。, stateinitial_state ) print(f助手: {result1.data}\n) # 注意这里 result1.data 是 LLM 的自然语言回复。 # 我们需要在工具中显式调用 update_user_name或者让 LLM 在回复中触发工具。 # 为了简化我们假设 LLM 的回复触发了 update_user_name 工具实际需通过 prompt 引导。 # 更自动化的方式需要使用 Agent 的 tool_result_chain 或更精细的提示工程此处为演示概念。 # 手动调用工具来更新状态模拟工具被触发 from pydantic_ai import RunContext # 创建一个新的运行上下文使用上一轮的结果状态 new_state result1.new_state # 在实际中工具调用应由 LLM 决定。这里我们手动模拟。 # 我们直接修改状态来模拟效果。 new_state.user_name 小明 new_state.conversation_history.append(用户告知姓名小明) # 第二轮用户表达需求 print(用户: 我想找个暖和的地方度个短假3天左右预算经济些。) result2 await stateful_agent.run( 我想找个暖和的地方度个短假3天左右预算经济些。, statenew_state ) print(f助手: {result2.data}\n) # 解析偏好并调用推荐工具同样实际中应由LLM决定 # ... 这里省略复杂的工具调用派发逻辑直接使用之前的功能 # 重点是展示状态 (new_state) 如何在多次 run 之间传递和更新。 # 假设经过几轮交互状态中已经保存了一个计划 final_state result2.new_state if final_state.current_plan: print( 最终确认的计划 ) plan: TravelPlan final_state.current_plan print(f目的地: {plan.destination.name}) print(f总花费: {plan.total_estimated_cost}) print(f\n对话结束时状态: 用户名{final_state.user_name}, 讨论过{len(final_state.discussed_destinations)}个目的地) if __name__ __main__: asyncio.run(multi_turn_conversation())状态管理的核心agent.run()接受一个state参数并返回一个包含new_state的Result对象。每次运行都可能修改状态你需要将更新后的状态传递给下一次运行从而实现有记忆的对话。6. 常见问题与排查思路在开发和使用 Pydantic AI 过程中你可能会遇到以下典型问题。问题现象可能原因排查方式解决方案运行时报错pydantic_core.ValidationErrorLLM 的输出无法通过 Pydantic 模型验证。1. 查看错误详情找到是哪个字段验证失败。2. 检查模型的Field定义类型、约束、choices。3. 打印result.messages查看 LLM 的实际回复。1. 优化提示词更明确地指导 LLM 输出格式。2. 放宽模型约束如将str改为Optional[str]。3. 使用response_format强制 LLM 输出 JSON部分模型支持。Agent 不调用我定义的 Tool1. 提示词未引导 Agent 使用工具。2. 工具描述不够清晰。3. LLM 认为不需要调用工具。1. 在system_prompt中明确说明可用工具及其用途。2. 使用agent.tool装饰器的description参数提供清晰描述。3. 检查result.messages看 LLM 的思考过程。1. 在system_prompt中加入“你可以使用以下工具...”。2. 使用Agent的tool_result_chain或tool_choice参数进行更精细控制。3. 考虑使用OpenAIToolModel等对工具调用支持更好的模型。异步函数报错RuntimeWarning: coroutine was never awaited在同步代码中直接调用了异步函数 (async def)。检查是否在非async函数内直接调用了agent.run()或工具函数。确保在异步上下文async def函数内使用await或者使用asyncio.run()包装主逻辑。输出结果不稳定时好时坏1. LLM 温度 (temperature) 参数过高。2. 提示词不够精确。3. 模型能力不足。1. 检查创建Agent或Model时是否设置了temperature。2. 对比多次运行的输入和输出。1. 降低temperature如设为 0以获得更确定性的输出。2. 细化system_prompt和工具描述。3. 升级到更强大的模型如从gpt-3.5-turbo到gpt-4-turbo。处理速度慢1. 网络延迟。2. 模型本身响应慢。3. 工具中有同步阻塞操作。1. 使用异步 HTTP 客户端如httpx。2. 检查工具函数是否使用了time.sleep()等同步阻塞调用。1. 确保所有工具和外部调用都是异步的 (async/await)。2. 考虑使用流式响应 (streamTrue) 改善用户体验。3. 对耗时操作进行缓存。7. 最佳实践与工程建议将 Pydantic AI 用于生产项目时遵循以下建议可以大幅提升代码质量和可维护性。7.1 模型设计单一职责每个 Pydantic Model 应只代表一个明确的业务概念如User,Order,AnalysisResult。充分使用 Field利用Field(description...)为每个字段添加清晰的描述这本身就是给 LLM 的绝佳文档。定义验证器对于复杂业务规则使用validator确保数据在进入流程前就是正确的。使用嵌套和复用通过嵌套模型来构建复杂的数据结构避免庞大的扁平化模型。7.2 Agent 与工具组织功能拆分不要创建一个“上帝 Agent”。根据领域如TravelPlannerAgent,WeatherQueryAgent,CustomerServiceAgent创建多个 Agent让它们各司其职。工具粒度工具应该做一件事并做好。一个工具“查询天气”另一个工具“预订酒店”而不是一个“处理旅行相关的一切”。依赖注入通过deps参数向 Agent 注入数据库连接、配置、HTTP 会话等依赖而不是在工具内部全局获取。这使得测试更容易。错误处理在工具内部进行细致的错误处理如网络超时、API 限流并返回结构化的错误信息给 Agent而不是抛出未处理的异常。7.3 提示词工程系统提示词是宪法system_prompt定义了 Agent 的角色、行为边界和能力范围。写得越清晰Agent 行为越可控。在提示词中引用工具明确列出可用的工具及其用途。例如“你可以使用get_weather工具查询实时天气。”提供示例对于复杂的结构化输出在system_prompt或user_prompt中提供一两个输出示例Few-Shot Learning能显著提升输出质量。7.4 测试与监控单元测试工具函数像测试普通函数一样测试你的工具确保其逻辑正确。集成测试 Agent 运行模拟 LLM 的响应测试从输入到结构化输出的完整流程。Pydantic AI 支持使用MockModel进行测试。记录与监控记录每次agent.run()的输入、输出、Token 使用量和耗时。这对于调试、成本控制和性能优化至关重要。7.5 生产环境部署配置管理将模型类型、API Key、温度等参数放在环境变量或配置文件中不要硬编码。超时与重试为 LLM 调用和工具调用设置合理的超时和重试策略。速率限制遵守所用模型 API 的速率限制在客户端实现限流。结构化日志使用 JSON 格式等结构化日志便于后续聚合和分析。Pydantic AI 通过将 Python 强大的类型系统与 LLM 相结合为构建可靠、可维护的 AI 应用提供了一条优雅的路径。它可能不是功能最繁多的框架但在“确保 AI 输出能被程序可靠消费”这一核心需求上它做得非常出色。从简单的数据提取到复杂的有状态智能体这套范式都能提供坚实的保障。开始你的项目时建议从一个明确的小功能点切入定义好输入输出模型然后逐步添加工具和状态管理。你会发现这种“类型优先”的开发方式能让你的 AI 应用代码像传统软件一样清晰、健壮。 Pydantic AI实战：基于类型系统的AI应用开发框架解析
如果你正在寻找一个能真正落地、能快速构建生产级 AI 应用的开源框架而不是停留在玩具 Demo 或复杂的学术概念上那么 Pydantic AI 绝对值得你花时间深入了解。过去一年我们见证了 LangChain、LlamaIndex 等框架的崛起它们极大地降低了 AI 应用开发的门槛。然而随着项目复杂度提升开发者们开始面临新的痛点如何优雅地管理复杂的提示词如何确保 LLM 输出的结构化数据能被程序可靠地使用如何构建一个状态清晰、易于调试的智能体Agent这些问题正是 Pydantic AI 试图从根源上解决的。Pydantic AI 的核心思想非常直接将 Python 的类型系统Type Hints和强大的数据验证库 Pydantic 深度集成到 AI 应用开发流程中。它不是一个试图包罗万象的“全家桶”而是一个专注于“结构化输出”和“智能体状态管理”的利器。这意味着你可以用编写普通 Python 类和方法的方式来定义 AI 的行为、输入和输出并获得编译时级别的类型安全和运行时数据验证。本文将带你从零开始通过一个完整的实战项目深入理解 Pydantic AI 的核心概念、工作流和最佳实践。你将学会如何构建一个具备记忆、工具调用和结构化输出能力的智能体并理解它为何能在生产环境中提供更高的可靠性和开发效率。1. Pydantic AI 解决了什么核心问题在深入代码之前我们必须先理解 Pydantic AI 诞生的背景和它要解决的核心矛盾。1.1 传统 AI 应用开发的痛点假设你要开发一个“智能客服助手”它需要根据用户描述的问题自动分类技术问题、账单问题、一般咨询并提取关键实体如订单号、错误代码。使用传统方式你可能会这样写提示词prompt f 请分析以下用户问题 {user_query} 请按以下格式回复 问题分类[技术问题/账单问题/一般咨询] 关键实体[提取出的订单号、错误代码等若无则写“无”] response llm.invoke(prompt) # 然后你需要手动解析 response 的文本用正则表达式或字符串分割来提取“问题分类”和“关键实体”。这种方式存在几个明显问题输出不稳定LLM 可能不严格按照你指定的格式回复导致解析失败。解析代码脆弱你的解析逻辑正则、字符串处理与提示词强耦合提示词稍作修改解析代码就可能崩溃。缺乏类型安全问题分类和关键实体在代码中是字符串编译器无法帮你检查它们的取值是否合法也无法提供自动补全。状态管理混乱如果这是一个多轮对话的智能体你需要手动维护对话历史、用户信息等状态代码会变得难以维护。1.2 Pydantic AI 的解决方案类型即契约Pydantic AI 将上述过程彻底转变。你首先定义数据模型继承自pydantic.BaseModel这个模型清晰地描述了 AI 应该输出什么。from pydantic import BaseModel, Field from typing import List, Optional class ProblemAnalysis(BaseModel): 对用户问题的分析结果 category: str Field(description问题分类, choices[技术问题, 账单问题, 一般咨询]) key_entities: List[str] Field(description提取出的关键实体如订单号、错误码) requires_human_intervention: bool Field(description是否需要人工介入)然后你创建一个Agent并告诉它当调用你的analyze_problem方法时请使用指定的模型如 GPT-4和提示词模板并且必须返回一个ProblemAnalysis类型的实例。from pydantic_ai import Agent agent Agent( modelopenai:gpt-4-turbo, system_prompt你是一个专业的客服问题分析助手。, ) agent.tool async def analyze_problem(query: str) - ProblemAnalysis: 分析用户问题并返回结构化结果。 return await agent.run(query) # 注意这里只是示意实际调用方式略有不同当agent.run()执行时Pydantic AI 会在后台将你的提示词和用户输入组合。调用 LLM。要求 LLM 必须返回一个符合ProblemAnalysis模型的 JSON 对象。自动用 Pydantic 验证这个 JSON 对象如果不符合定义例如category的值不在choices中会抛出清晰的验证错误。最终返回给你一个ProblemAnalysis的实例对象你可以像使用普通 Python 对象一样访问其属性result.category,result.key_entities。这种“定义输出模型框架负责验证”的模式将不可靠的自然语言输出转化为了可靠的、类型化的程序对象。这是 Pydantic AI 最核心的价值。2. 核心概念快速理解在动手之前我们先厘清 Pydantic AI 的几个核心抽象它们构成了整个框架的骨架。概念类比在 Pydantic AI 中的角色关键点Agent智能体/机器人核心执行单元。它封装了 LLM 模型、系统提示词、工具集以及运行时的状态管理。一个应用可以由多个 Agent 协作完成。不是单次对话而是一个有状态的执行上下文。Model大脑/引擎负责实际调用 LLM如 OpenAI GPT, Anthropic Claude, 本地 Ollama 模型。Agent 在创建时需要指定一个 Model。支持多种后端通过类似openai:gpt-4o的字符串指定。Tool技能/函数Agent 可以调用的外部函数。用于获取实时信息如天气、股票、执行操作如发邮件、查数据库或进行计算。用agent.tool装饰器定义。工具的参数和返回值也推荐用 Pydantic Model 定义以实现结构化调用。Result任务结果agent.run()的返回对象。它包含了本次运行的最终输出数据data、使用的消息历史messages、消耗的 Token 数usage等元信息。通过result.data获取你定义的 Pydantic Model 实例。State记忆/上下文Agent 在多次run调用之间保持的数据。例如你可以让 State 记录对话历史、用户 ID、会话步骤等。State 也是一个 Pydantic Model。状态管理是构建复杂、多轮交互智能体的关键。Dependency依赖项注入到 Agent 工具或运行上下文中的对象例如数据库连接池、HTTP 客户端、配置对象等。用于实现依赖注入使代码更易测试和维护。类似于 FastAPI 的Depends机制。理解这些概念后你会发现 Pydantic AI 的编程模型非常直观定义数据模型描述输入输出和状态 - 创建 Agent配置大脑和基础设定 - 定义工具赋予其能力 - 运行并获取结构化结果。3. 环境准备与安装我们将在 Python 3.10 的环境中进行实战。确保你的环境已就绪。3.1 创建虚拟环境与安装核心库强烈建议使用虚拟环境来管理依赖。# 1. 创建并进入项目目录 mkdir pydantic-ai-tutorial cd pydantic-ai-tutorial # 2. 创建虚拟环境以 venv 为例 python -m venv .venv # 3. 激活虚拟环境 # Linux/macOS source .venv/bin/activate # Windows .venv\Scripts\activate # 4. 安装 Pydantic AI 和 OpenAI 库我们将使用 OpenAI 作为示例模型后端 pip install pydantic-ai openai # 5. 安装可选的开发依赖如用于结构化输出的 pydantic-settings pip install pydantic-settings3.2 配置 API 密钥Pydantic AI 本身不提供 LLM你需要一个模型提供商。这里以 OpenAI 为例。获取 OpenAI API Key访问 OpenAI Platform 创建密钥。设置环境变量推荐方式# Linux/macOS export OPENAI_API_KEY你的-api-key # Windows (PowerShell) $env:OPENAI_API_KEY你的-api-key或者在代码中直接设置不推荐用于生产环境import os os.environ[OPENAI_API_KEY] 你的-api-key3.3 验证安装创建一个简单的测试文件test_install.py# test_install.py import asyncio from pydantic_ai import Agent from pydantic import BaseModel class Joke(BaseModel): setup: str punchline: str async def main(): # 创建一个简单的 Agent agent Agent( modelopenai:gpt-3.5-turbo, system_prompt你是一个讲笑话的助手。, ) # 运行并指定输出类型为 Joke result await agent.run(讲一个关于编程的笑话。, response_typeJoke) joke: Joke result.data print(f问题{joke.setup}) print(f笑点{joke.punchline}) if __name__ __main__: asyncio.run(main())运行它python test_install.py如果看到类似以下的输出说明环境配置成功问题为什么程序员分不清万圣节和圣诞节 笑点因为 Oct 31 Dec 25。4. 实战项目构建智能旅行规划助手我们将通过一个完整的项目来学习 Pydantic AI。这个助手能根据用户偏好预算、兴趣、时间推荐目的地规划大致行程并能调用“工具”查询实时天气。4.1 第一步定义数据模型Models这是 Pydantic AI 开发的起点。我们先在models.py中定义所有需要用到的数据结构。# models.py from pydantic import BaseModel, Field, validator from typing import List, Optional from datetime import date from enum import Enum # 1. 用户输入偏好 class UserPreference(BaseModel): budget_level: str Field(description预算等级, choices[经济, 舒适, 豪华]) interests: List[str] Field(description兴趣列表如[自然风光, 历史古迹, 美食, 购物]) travel_days: int Field(ge1, le30, description旅行天数) start_date: Optional[date] Field(defaultNone, description出发日期) validator(interests) def validate_interests(cls, v): allowed [自然风光, 历史古迹, 美食, 购物, 冒险运动, 艺术文化, 休闲度假] for interest in v: if interest not in allowed: raise ValueError(f兴趣{interest}不在允许的列表中: {allowed}) return v # 2. 推荐的目的地 class Destination(BaseModel): name: str Field(description目的地名称) country: str Field(description所在国家) reason: str Field(description推荐理由需关联用户兴趣) estimated_cost_per_day: float Field(ge0, description日均预估花费当地货币) # 3. 每日行程规划 class DailyPlan(BaseModel): day: int Field(ge1, description第几天) morning: str Field(description上午活动) afternoon: str Field(description下午活动) evening: str Field(description晚上活动) meal_suggestion: str Field(description餐饮建议) # 4. 完整的旅行计划 class TravelPlan(BaseModel): destination: Destination daily_plans: List[DailyPlan] Field(description每日行程安排) total_estimated_cost: float Field(description总预估花费) packing_tips: List[str] Field(description行李准备建议) # 5. 用于工具调用的天气查询结果 class WeatherInfo(BaseModel): city: str date: date condition: str # e.g., Sunny, Rainy high_temp: float # 最高温 low_temp: float # 最低温 humidity: int # 湿度百分比关键点我们使用了Field来添加描述和约束如choices,ge,le这些描述会帮助 LLM 更好地理解字段含义。validator装饰器用于自定义验证逻辑确保输入数据的有效性。模型之间可以嵌套使用如TravelPlan包含Destination和DailyPlan。4.2 第二步创建 Agent 并定义工具接下来在agent_builder.py中创建我们的核心智能体。# agent_builder.py import asyncio from typing import List from pydantic_ai import Agent, RunContext from pydantic_ai.models.openai import OpenAIModel from models import UserPreference, Destination, TravelPlan, WeatherInfo # 初始化 Agent使用 GPT-4 模型以获得更好的推理和结构化输出能力 # 注意实际使用时请根据你的需求选择合适的模型 travel_agent Agent( modelOpenAIModel(gpt-4-turbo), # 等价于 modelopenai:gpt-4-turbo system_prompt 你是一个专业的旅行规划专家。你的任务是根据用户的偏好为他们推荐合适的目的地并制定详细的旅行计划。 你必须严格遵守输出格式确保所有推荐都基于用户的预算和兴趣。 在规划行程时请考虑交通、景点的合理时间安排。 , depsNone, # 暂时没有外部依赖 ) # 定义第一个工具获取目的地推荐模拟 travel_agent.tool async def recommend_destinations( ctx: RunContext[None], # RunContext 提供了运行上下文目前状态为 None preference: UserPreference ) - List[Destination]: 根据用户偏好推荐最多3个潜在旅行目的地。 这是一个模拟工具实际项目中应接入目的地数据库或第三方API。 # 模拟一些逻辑和 API 调用 print(f[工具调用] recommend_destinations: 正在为偏好 {preference.interests} 的用户寻找目的地...) await asyncio.sleep(0.5) # 模拟网络延迟 # 这里应该是真实的业务逻辑例如查询数据库。 # 为了演示我们让 LLM 根据偏好来“生成”推荐。 # 注意在实际工具中我们通常从固定数据源获取但这里我们巧妙地将任务交还给 LLM 来演示。 # 更常见的做法是工具从数据库查询返回固定列表。这里为了展示LLM的灵活性我们让它动态生成。 prompt_for_llm f 用户偏好预算{preference.budget_level}兴趣{preference.interests}旅行{preference.travel_days}天。 请生成3个符合上述条件的真实世界旅行目的地。 每个目的地需要包含名称、国家、推荐理由关联兴趣、日均预估花费请符合其预算等级。 请以 JSON 列表格式回复严格匹配 Destination 模型。 # 我们让同一个 Agent 来生成这个列表但指定输出类型为 List[Destination] result await ctx.agent.run(prompt_for_llm, response_typeList[Destination]) return result.data # 定义第二个工具查询天气预报模拟 travel_agent.tool async def get_weather_forecast( ctx: RunContext[None], city: str, for_date: date ) - WeatherInfo: 查询指定城市在指定日期的天气预报。 这是一个模拟工具返回模拟数据。 print(f[工具调用] get_weather_forecast: 查询{city}在{for_date}的天气...) await asyncio.sleep(0.3) # 模拟数据 - 实际应调用如 OpenWeatherMap 的 API # 这里简单根据城市名生成一个假天气 import random conditions [晴朗, 多云, 小雨, 大雨, 阴天] return WeatherInfo( citycity, datefor_date, conditionrandom.choice(conditions), high_tempround(random.uniform(20, 35), 1), low_tempround(random.uniform(10, 25), 1), humidityrandom.randint(40, 90) )关键点agent.tool装饰器将一个异步函数注册为 Agent 的工具。工具函数的第一个参数通常是RunContext它提供了访问当前 Agent、状态和依赖的入口。工具的参数和返回值都使用了我们之前定义的 Pydantic Model这保证了工具间数据传递的结构化和安全。在recommend_destinations工具中我们演示了一种模式工具内部再次调用ctx.agent.run并指定response_type让 LLM 来帮助完成复杂的逻辑。这在需要 LLM 进行创造性思考的场景下非常有用。4.3 第三步实现核心规划逻辑现在我们创建一个函数它利用上面定义的 Agent 和工具完成从用户输入到完整旅行规划的流程。在planner.py中# planner.py import asyncio from datetime import date, timedelta from agent_builder import travel_agent from models import UserPreference, TravelPlan async def create_travel_plan(user_input: str) - TravelPlan: 主函数接收用户自然语言描述生成完整的旅行计划。 # 第一步让 Agent 从用户描述中提取结构化偏好 print(步骤1: 解析用户偏好...) pref_result await travel_agent.run( user_input, response_typeUserPreference, system_prompt_extra请从用户的描述中提取预算、兴趣、天数和日期信息。如果日期未指定可以忽略。 ) user_pref: UserPreference pref_result.data print(f解析出的偏好{user_pref.dict()}) # 第二步调用工具获取目的地推荐 print(\n步骤2: 获取目的地推荐...) destinations await travel_agent.recommend_destinations(user_pref) # 直接调用工具方法 print(f推荐了 {len(destinations)} 个目的地) for d in destinations: print(f - {d.name} ({d.country}), 理由{d.reason}) # 第三步让用户或模拟选择一个目的地。这里我们简单选第一个。 selected_dest destinations[0] print(f\n步骤3: 选择目的地 {selected_dest.name} 进行详细规划。) # 第四步生成详细的旅行计划包含每日行程 print(\n步骤4: 生成详细旅行计划...) plan_prompt f 基于以下信息为用户制定一份详细的{user_pref.travel_days}天旅行计划 目的地{selected_dest.name}, {selected_dest.country} 用户兴趣{user_pref.interests} 预算等级{user_pref.budget_level} 日均花费参考{selected_dest.estimated_cost_per_day} (当地货币) 请规划每一天的上午、下午、晚上活动和餐饮建议。行程要合理符合兴趣。 最后计算总花费日均花费*天数并给出行李准备建议。 plan_result await travel_agent.run(plan_prompt, response_typeTravelPlan) travel_plan: TravelPlan plan_result.data # 第五步可选如果用户提供了出发日期为行程中的每一天查询天气 if user_pref.start_date: print(\n步骤5: 查询行程天气...) for i, day_plan in enumerate(travel_plan.daily_plans): target_date user_pref.start_date timedelta(daysi) weather await travel_agent.get_weather_forecast(selected_dest.name, target_date) # 我们可以把天气信息附加到每日计划中这里简单打印 print(f 第{day_plan.day}天 ({target_date}): {weather.condition}, {weather.low_temp}-{weather.high_temp}°C) return travel_plan if __name__ __main__: # 模拟用户输入 sample_input 我想下个月15号左右出发玩5天。预算比较宽松可以舒适一点。 我喜欢自然风光和美食想去个有山有水还能吃好的地方。 print(用户请求:, sample_input) print(*50) final_plan asyncio.run(create_travel_plan(sample_input)) print(\n *50) print(✅ 旅行计划生成完毕) print(f目的地{final_plan.destination.name}) print(f总预估花费{final_plan.total_estimated_cost:.2f} (当地货币)) print(行程概览) for day in final_plan.daily_plans: print(f 第{day.day}天) print(f 上午{day.morning}) print(f 下午{day.afternoon}) print(f 晚上{day.evening}) print(f 餐饮{day.meal_suggestion}) print(行李建议, , .join(final_plan.packing_tips))4.4 第四步运行与验证运行我们的主程序python planner.py你应该能看到类似以下的输出清晰地展示了智能体工作的每一步用户请求: 我想下个月15号左右出发玩5天。预算比较宽松可以舒适一点。我喜欢自然风光和美食想去个有山有水还能吃好的地方。 步骤1: 解析用户偏好... 解析出的偏好{budget_level: 舒适, interests: [自然风光, 美食], travel_days: 5, start_date: 2024-06-15} 步骤2: 获取目的地推荐... [工具调用] recommend_destinations: 正在为偏好 [自然风光, 美食] 的用户寻找目的地... 推荐了 3 个目的地 - 杭州 (中国), 理由拥有西湖等自然风光同时以杭帮菜闻名符合美食兴趣。 - 瑞士因特拉肯 (瑞士), 理由坐落于少女峰山脚下自然风光绝美同时可品尝瑞士奶酪火锅等美食。 - 日本京都 (日本), 理由周边有岚山等自然景观同时是日本传统怀石料理的发源地之一。 步骤3: 选择目的地 杭州 进行详细规划。 步骤4: 生成详细旅行计划... [工具调用] get_weather_forecast: 查询杭州在2024-06-15的天气... ... ✅ 旅行计划生成完毕 目的地杭州 总预估花费XXXX.XX (当地货币) 行程概览 第1天 上午抵达杭州入住酒店漫步西湖苏堤。 下午游览雷峰塔了解白蛇传传说。 晚上在楼外楼品尝西湖醋鱼、龙井虾仁。 ...5. 进阶状态管理State与多轮对话上面的例子是单次任务。真正的智能体往往需要记住对话历史。Pydantic AI 通过State概念来优雅地管理状态。让我们升级旅行助手使其能记住用户姓名并在多轮对话中调整计划。5.1 定义对话状态模型在models.py中添加# models.py (追加) class ConversationState(BaseModel): 记录与用户对话的状态 user_name: Optional[str] None discussed_destinations: List[Destination] [] # 讨论过的目的地 current_plan: Optional[TravelPlan] None # 当前生成的计划 conversation_history: List[str] [] # 简化的历史记录5.2 创建有状态的 Agent修改agent_builder.py创建一个新的、有状态的 Agent# agent_builder.py (追加) from models import ConversationState # 创建一个新的、有状态的 Agent。注意 state_type 参数。 stateful_agent Agent( modelopenai:gpt-4-turbo, system_prompt 你是旅行助手小游。请友好地与用户交流记住他们的名字和之前的对话。 你的目标是逐步了解用户需求并最终生成他们满意的旅行计划。 , state_typeConversationState, # 关键指定状态模型 ) stateful_agent.tool async def update_user_name(ctx: RunContext[ConversationState], name: str): 更新用户姓名到状态中。 ctx.state.user_name name return f好的{name}我已经记住你了。 stateful_agent.tool async def save_current_plan(ctx: RunContext[ConversationState], plan: TravelPlan): 将当前计划保存到状态中。 ctx.state.current_plan plan return 旅行计划已保存。你可以随时让我基于这个计划进行调整。 # 这个工具的上下文 RunContext 是 ConversationState stateful_agent.tool async def recommend_with_memory( ctx: RunContext[ConversationState], preference: UserPreference ) - List[Destination]: 推荐目的地并考虑之前讨论过的避免重复。 # 模拟获取推荐列表 result await ctx.agent.run( f根据偏好{preference}推荐目的地。, response_typeList[Destination] ) new_destinations result.data # 过滤掉已经讨论过的 discussed_names {d.name for d in ctx.state.discussed_destinations} filtered_destinations [d for d in new_destinations if d.name not in discussed_names] # 更新状态 ctx.state.discussed_destinations.extend(filtered_destinations) return filtered_destinations[:3] # 返回最多3个5.3 进行多轮对话创建一个新的对话脚本conversation.py# conversation.py import asyncio from agent_builder import stateful_agent from models import UserPreference, TravelPlan async def multi_turn_conversation(): # 初始化一个状态对象 initial_state ConversationState() # 第一轮用户打招呼并告知姓名 print(用户: 你好我是小明。) result1 await stateful_agent.run( 你好我是小明。, stateinitial_state ) print(f助手: {result1.data}\n) # 注意这里 result1.data 是 LLM 的自然语言回复。 # 我们需要在工具中显式调用 update_user_name或者让 LLM 在回复中触发工具。 # 为了简化我们假设 LLM 的回复触发了 update_user_name 工具实际需通过 prompt 引导。 # 更自动化的方式需要使用 Agent 的 tool_result_chain 或更精细的提示工程此处为演示概念。 # 手动调用工具来更新状态模拟工具被触发 from pydantic_ai import RunContext # 创建一个新的运行上下文使用上一轮的结果状态 new_state result1.new_state # 在实际中工具调用应由 LLM 决定。这里我们手动模拟。 # 我们直接修改状态来模拟效果。 new_state.user_name 小明 new_state.conversation_history.append(用户告知姓名小明) # 第二轮用户表达需求 print(用户: 我想找个暖和的地方度个短假3天左右预算经济些。) result2 await stateful_agent.run( 我想找个暖和的地方度个短假3天左右预算经济些。, statenew_state ) print(f助手: {result2.data}\n) # 解析偏好并调用推荐工具同样实际中应由LLM决定 # ... 这里省略复杂的工具调用派发逻辑直接使用之前的功能 # 重点是展示状态 (new_state) 如何在多次 run 之间传递和更新。 # 假设经过几轮交互状态中已经保存了一个计划 final_state result2.new_state if final_state.current_plan: print( 最终确认的计划 ) plan: TravelPlan final_state.current_plan print(f目的地: {plan.destination.name}) print(f总花费: {plan.total_estimated_cost}) print(f\n对话结束时状态: 用户名{final_state.user_name}, 讨论过{len(final_state.discussed_destinations)}个目的地) if __name__ __main__: asyncio.run(multi_turn_conversation())状态管理的核心agent.run()接受一个state参数并返回一个包含new_state的Result对象。每次运行都可能修改状态你需要将更新后的状态传递给下一次运行从而实现有记忆的对话。6. 常见问题与排查思路在开发和使用 Pydantic AI 过程中你可能会遇到以下典型问题。问题现象可能原因排查方式解决方案运行时报错pydantic_core.ValidationErrorLLM 的输出无法通过 Pydantic 模型验证。1. 查看错误详情找到是哪个字段验证失败。2. 检查模型的Field定义类型、约束、choices。3. 打印result.messages查看 LLM 的实际回复。1. 优化提示词更明确地指导 LLM 输出格式。2. 放宽模型约束如将str改为Optional[str]。3. 使用response_format强制 LLM 输出 JSON部分模型支持。Agent 不调用我定义的 Tool1. 提示词未引导 Agent 使用工具。2. 工具描述不够清晰。3. LLM 认为不需要调用工具。1. 在system_prompt中明确说明可用工具及其用途。2. 使用agent.tool装饰器的description参数提供清晰描述。3. 检查result.messages看 LLM 的思考过程。1. 在system_prompt中加入“你可以使用以下工具...”。2. 使用Agent的tool_result_chain或tool_choice参数进行更精细控制。3. 考虑使用OpenAIToolModel等对工具调用支持更好的模型。异步函数报错RuntimeWarning: coroutine was never awaited在同步代码中直接调用了异步函数 (async def)。检查是否在非async函数内直接调用了agent.run()或工具函数。确保在异步上下文async def函数内使用await或者使用asyncio.run()包装主逻辑。输出结果不稳定时好时坏1. LLM 温度 (temperature) 参数过高。2. 提示词不够精确。3. 模型能力不足。1. 检查创建Agent或Model时是否设置了temperature。2. 对比多次运行的输入和输出。1. 降低temperature如设为 0以获得更确定性的输出。2. 细化system_prompt和工具描述。3. 升级到更强大的模型如从gpt-3.5-turbo到gpt-4-turbo。处理速度慢1. 网络延迟。2. 模型本身响应慢。3. 工具中有同步阻塞操作。1. 使用异步 HTTP 客户端如httpx。2. 检查工具函数是否使用了time.sleep()等同步阻塞调用。1. 确保所有工具和外部调用都是异步的 (async/await)。2. 考虑使用流式响应 (streamTrue) 改善用户体验。3. 对耗时操作进行缓存。7. 最佳实践与工程建议将 Pydantic AI 用于生产项目时遵循以下建议可以大幅提升代码质量和可维护性。7.1 模型设计单一职责每个 Pydantic Model 应只代表一个明确的业务概念如User,Order,AnalysisResult。充分使用 Field利用Field(description...)为每个字段添加清晰的描述这本身就是给 LLM 的绝佳文档。定义验证器对于复杂业务规则使用validator确保数据在进入流程前就是正确的。使用嵌套和复用通过嵌套模型来构建复杂的数据结构避免庞大的扁平化模型。7.2 Agent 与工具组织功能拆分不要创建一个“上帝 Agent”。根据领域如TravelPlannerAgent,WeatherQueryAgent,CustomerServiceAgent创建多个 Agent让它们各司其职。工具粒度工具应该做一件事并做好。一个工具“查询天气”另一个工具“预订酒店”而不是一个“处理旅行相关的一切”。依赖注入通过deps参数向 Agent 注入数据库连接、配置、HTTP 会话等依赖而不是在工具内部全局获取。这使得测试更容易。错误处理在工具内部进行细致的错误处理如网络超时、API 限流并返回结构化的错误信息给 Agent而不是抛出未处理的异常。7.3 提示词工程系统提示词是宪法system_prompt定义了 Agent 的角色、行为边界和能力范围。写得越清晰Agent 行为越可控。在提示词中引用工具明确列出可用的工具及其用途。例如“你可以使用get_weather工具查询实时天气。”提供示例对于复杂的结构化输出在system_prompt或user_prompt中提供一两个输出示例Few-Shot Learning能显著提升输出质量。7.4 测试与监控单元测试工具函数像测试普通函数一样测试你的工具确保其逻辑正确。集成测试 Agent 运行模拟 LLM 的响应测试从输入到结构化输出的完整流程。Pydantic AI 支持使用MockModel进行测试。记录与监控记录每次agent.run()的输入、输出、Token 使用量和耗时。这对于调试、成本控制和性能优化至关重要。7.5 生产环境部署配置管理将模型类型、API Key、温度等参数放在环境变量或配置文件中不要硬编码。超时与重试为 LLM 调用和工具调用设置合理的超时和重试策略。速率限制遵守所用模型 API 的速率限制在客户端实现限流。结构化日志使用 JSON 格式等结构化日志便于后续聚合和分析。Pydantic AI 通过将 Python 强大的类型系统与 LLM 相结合为构建可靠、可维护的 AI 应用提供了一条优雅的路径。它可能不是功能最繁多的框架但在“确保 AI 输出能被程序可靠消费”这一核心需求上它做得非常出色。从简单的数据提取到复杂的有状态智能体这套范式都能提供坚实的保障。开始你的项目时建议从一个明确的小功能点切入定义好输入输出模型然后逐步添加工具和状态管理。你会发现这种“类型优先”的开发方式能让你的 AI 应用代码像传统软件一样清晰、健壮。