Pydantic AI ：从安装到智能体开发（一）
1. 什么是 Pydantic AI？
Pydantic AI 是一个由 Pydantic 团队开发的 Python Agent 框架，旨在帮助开发者快速、自信且轻松地构建生产级的生成式 AI 应用程序和工作流。
1.1 核心概念解释
Agent（智能体）：在 AI 应用中，Agent 是一个能够理解用户请求、调用工具处理任务并返回结果的程序实体。可以把它想象成一个智能助手，能够执行复杂的多步骤任务。
Pydantic：一个强大的 Python 数据验证库，使用类型注解来验证数据。它被广泛应用于 OpenAI SDK、LangChain 等主流 AI 工具中。
框架优势：
- 模型无关：支持几乎所有主流模型提供商
- 完全类型安全：在编码时就能发现错误
- 无缝可观测性：内置监控和调试功能
- 生产级可靠性：支持持久化执行和错误恢复
2. 安装与环境配置
2.1 基础安装
# 安装 Pydantic AI pip install pydantic-ai # 或者使用 uv（更快的 Python 包管理器） uv add pydantic-ai 2.2 精简安装（按需选择）
# 如果你只需要特定模型，可以使用精简版本 pip install "pydantic-ai-slim[openai]" # 安装多个模型支持 pip install "pydantic-ai-slim[openai,anthropic,logfire]" 3. 基础使用：从 Hello World 开始
3.1 最简单的 Agent 示例
from pydantic_ai import Agent # 创建使用阿里云 Qwen3 模型的 Agent agent = Agent( 'qwen:qwen2.5-7b-instruct', # 使用阿里云 Qwen 模型 instructions='请用一句话简洁回复。', ) # 同步运行 result = agent.run_sync('"Hello World" 这个短语最早出现在哪里？') print(result.output) # 输出：第一个已知的 "Hello, World" 使用出现在 1974 年关于 C 编程语言的教科书中。 代码解释：
Agent(): 创建智能体实例，第一个参数指定使用的模型instructions: 给模型的系统指令，指导其行为方式run_sync(): 同步运行方法，适合简单的脚本使用result.output: 获取模型的文本输出
3.2 异步运行示例
import asyncio from pydantic_ai import Agent async def main(): # 创建使用 Qwen3 模型的 Agent agent = Agent('qwen:qwen2.5-7b-instruct') # 异步运行 result = await agent.run('Python 是什么？') print(result.output) # 运行异步函数 asyncio.run(main()) 4. 工具与依赖注入
4.1 银行客服 Agent 示例
from dataclasses import dataclass from pydantic import BaseModel, Field from pydantic_ai import Agent, RunContext # 模拟数据库连接 class DatabaseConn: async def customer_name(self, id: int) -> str: # 模拟数据库查询 names = {123: "张三", 456: "李四"} return names.get(id, "未知用户") async def customer_balance(self, id: int, include_pending: bool) -> float: # 模拟余额查询 return 1234.56 # 依赖项定义 @dataclass class SupportDependencies: customer_id: int db: DatabaseConn # 输出模型定义 class SupportOutput(BaseModel): support_advice: str = Field(description='给客户的建议') block_card: bool = Field(description="是否冻结银行卡") risk: int = Field(description='风险等级', ge=0, le=10) # 创建客服 Agent support_agent = Agent( 'qwen:qwen2.5-7b-instruct', deps_type=SupportDependencies, output_type=SupportOutput, instructions='你是一个银行客服助手，为客户提供支持并评估查询的风险等级。' ) # 动态指令：添加客户姓名 @support_agent.instructions async def add_customer_name(ctx: RunContext[SupportDependencies]) -> str: customer_name = await ctx.deps.db.customer_name(id=ctx.deps.customer_id) return f"客户姓名是 {customer_name}" # 工具函数：查询余额 @support_agent.tool async def customer_balance( ctx: RunContext[SupportDependencies], include_pending: bool ) -> float: """返回客户的当前账户余额""" return await ctx.deps.db.customer_balance( id=ctx.deps.customer_id, include_pending=include_pending, ) # 运行 Agent async def main(): deps = SupportDependencies(customer_id=123, db=DatabaseConn()) # 查询余额 result = await support_agent.run('我的余额是多少？', deps=deps) print(f"建议: {result.output.support_advice}") print(f"冻结卡片: {result.output.block_card}") print(f"风险等级: {result.output.risk}") asyncio.run(main()) 代码解释：
deps_type: 定义依赖项类型，可以在工具和指令中访问output_type: 定义结构化输出，确保返回数据格式一致@agent.tool: 将函数注册为工具，Agent 可以调用RunContext: 提供运行时的上下文信息
5. 流式输出与实时监控
5.1 基础流式输出
import asyncio from pydantic_ai import Agent async def stream_example(): agent = Agent('qwen:qwen2.5-7b-instruct') # 使用流式输出 async with agent.run_stream('介绍一下人工智能的发展历史') as response: async for text in response.stream_text(): print(text,, flush=True) print() # 换行 asyncio.run(stream_example()) 5.2 高级事件流处理
import asyncio from collections.abc import AsyncIterable from datetime import date from pydantic_ai import Agent, RunContext from pydantic_ai.messages import ( AgentStreamEvent, FunctionToolCallEvent, FunctionToolResultEvent, PartDeltaEvent, TextPartDelta, ) # 创建天气查询 Agent weather_agent = Agent( 'qwen:qwen2.5-7b-instruct', system_prompt='根据用户提供的位置提供天气预报。' ) # 天气查询工具 @weather_agent.tool async def weather_forecast( ctx: RunContext, location: str, forecast_date: date, ) -> str: return f'{location} 在 {forecast_date} 的天气预报是 24°C，晴天。' # 事件流处理器 async def event_stream_handler( ctx: RunContext, event_stream: AsyncIterable[AgentStreamEvent], ): async for event in event_stream: if isinstance(event, FunctionToolCallEvent): print(f"[工具调用] 调用 {event.part.tool_name}，参数: {event.part.args}") elif isinstance(event, FunctionToolResultEvent): print(f"[工具结果] 返回: {event.result.content}") elif isinstance(event, PartDeltaEvent): if isinstance(event.delta, TextPartDelta): print(f"[文本生成] {event.delta.content_delta}",, flush=True) async def main(): user_prompt = '巴黎周二天气怎么样？' async with weather_agent.run_stream( user_prompt, event_stream_handler=event_stream_handler ) as run: async for output in run.stream_text(): pass # 我们已经通过事件处理器输出了 asyncio.run(main()) 6. 多模态处理 with Qwen3-VL
6.1 图像理解示例
import asyncio from pydantic_ai import Agent from pydantic_ai.messages import ImagePart async def multimodal_example(): # 使用 Qwen3-VL 多模态模型 vl_agent = Agent('qwen:qwen3-vl-7b-instruct') # 构建包含图像的消息 result = await vl_agent.run([ "描述这张图片的内容", ImagePart.from_url("https://example.com/sample-image.jpg") # 替换为实际图片URL ]) print("图像描述:", result.output) asyncio.run(multimodal_example()) 6.2 本地图像处理
import asyncio from pydantic_ai import Agent from pydantic_ai.messages import ImagePart async def local_image_example(): vl_agent = Agent('qwen:qwen3-vl-7b-instruct') # 从本地文件加载图像 try: image_part = ImagePart.from_file("path/to/your/image.jpg") result = await vl_agent.run([ "分析这张图片，描述其中的主要物体和场景", image_part ]) print("分析结果:", result.output) except FileNotFoundError: print("请提供有效的图像文件路径") asyncio.run(local_image_example()) 7. 配置与优化
7.1 模型设置与限制
from pydantic_ai import Agent, UsageLimits, ModelSettings # 配置模型参数 agent = Agent( 'qwen:qwen2.5-7b-instruct', model_settings=ModelSettings( temperature=0.3, # 控制随机性 (0-1) max_tokens=500, # 最大输出长度 timeout=30, # 超时时间 ) ) # 使用用量限制 result = agent.run_sync( '写一个关于 Python 的简短介绍', usage_limits=UsageLimits( response_tokens_limit=100, # 限制输出token数 request_limit=3, # 限制请求次数 tool_calls_limit=5 # 限制工具调用次数 ) ) 7.2 错误处理与重试
from pydantic_ai import Agent, ModelRetry, UnexpectedModelBehavior agent = Agent('qwen:qwen2.5-7b-instruct', retries=2) @agent.tool(retries=3) def risky_operation(data: str) -> str: """一个可能失败的操作""" if len(data) < 5: raise ModelRetry('输入数据太短，请提供更详细的信息') return f"处理结果: {data}" try: result = agent.run_sync('处理这个数据: "hi"') except UnexpectedModelBehavior as e: print(f"操作失败: {e}") print(f"失败原因: {e.__cause__}") 8. 实际应用案例
8.1 智能数据分析助手
import pandas as pd from pydantic import BaseModel from pydantic_ai import Agent, RunContext class AnalysisResult(BaseModel): summary: str insights: list[str] recommendations: list[str] class DataAnalyzer: def __init__(self): self.df = None def load_data(self, data_dict: dict): self.df = pd.DataFrame(data_dict) return f"数据加载成功，共 {len(self.df)} 行 {len(self.df.columns)} 列" def basic_stats(self) -> str: if self.df is None: return "请先加载数据" return self.df.describe().to_string() analyzer_agent = Agent( 'qwen:qwen2.5-7b-instruct', deps_type=DataAnalyzer, output_type=AnalysisResult, instructions='你是一个数据分析助手，帮助用户分析数据并提供见解' ) @analyzer_agent.tool def load_dataset(ctx: RunContext[DataAnalyzer], data: dict) -> str: """加载数据集""" return ctx.deps.load_data(data) @analyzer_agent.tool def get_statistics(ctx: RunContext[DataAnalyzer]) -> str: """获取数据的基本统计信息""" return ctx.deps.basic_stats() # 使用示例 async def analyze_data(): analyzer = DataAnalyzer() result = await analyzer_agent.run( "请分析这个销售数据并给出总结", deps=analyzer ) print("分析总结:", result.output.summary) print("关键洞察:", result.output.insights) print("建议:", result.output.recommendations) 9. 总结
Pydantic AI 提供了一个强大而灵活的框架来构建生产级的 AI 应用。通过本文的介绍，您应该已经掌握了：
- 基础概念：理解 Agent、工具、依赖注入等核心概念
- 模型集成：如何使用阿里云 Qwen 系列模型
- 多模态处理：利用 Qwen3-VL 处理图像和文本
- 流式输出：实现实时响应和监控
- 错误处理：构建健壮的 AI 应用
- 实际应用：将框架应用于真实场景
Pydantic AI 的设计理念强调类型安全、生产就绪和开发者体验，使得构建复杂的 AI 工作流变得更加简单和可靠。无论是简单的聊天机器人还是复杂的企业级应用，Pydantic AI 都能提供强大的支持。
