别再手动拼接Prompt了！Pydantic-AI让AI应用开发像写API一样优雅

为什么Pydantic-AI正在重新定义AI Agent开发范式

引言：AI开发的新范式

当你第一次尝试用Python构建AI应用时，是否经历过这样的痛苦？

Prompt字符串像意大利面一样缠绕，每次改需求都要小心翼翼地在某个角落里修改文字，LLM返回的数据格式完全不可控，一个不小心整个流程就崩溃了。更让人头疼的是，代码写完后自己都看不懂，更别说让同事维护了。

这些问题，Pydantic团队在构建了全球最流行的数据验证库Pydantic之后，给出了他们的答案：Pydantic-AI。

如果你还不知道Pydantic是什么，简而言之，它是一个Python库，通过Python类型注解来定义数据模型，自动进行数据验证和转换。在FastAPI、Django Ninja等现代Web框架中，你几乎都能看到Pydantic的身影。

而Pydantic-AI，则是将这种“类型驱动开发”的理念，带入了AI Agent开发领域。

为什么值得关注

解决的核心痛点

痛点一：Prompt管理的混乱

传统做法是把所有指令都塞进一个巨大的字符串：

# 传统做法 - 维护困难
prompt = f"""
你是一个客服助手。
用户说: {user_message}
请用专业、友好的语气回复。
如果用户抱怨，要道歉。
如果用户询问产品，提供详细信息。
...
"""

当项目变大时，这种做法会迅速失控。而Pydantic-AI允许你用结构化的方式定义Agent的行为：

from pydantic_ai import Agent

agent = Agent(
    'openai:gpt-4',
    system_prompt='你是一个专业的客服助手...',
    result_type=SupportResult  # 结果类型自动验证
)

痛点二：输出格式的不确定性

让LLM返回结构化数据一直是难题。JSON模式有时候不work，Parser经常返回格式错误的数据。Pydantic-AI通过Result Type机制，强制要求模型输出符合预期格式的数据，并且自动验证。

痛点三：工具调用的复杂性

要让AI调用外部工具，需要处理一堆JSON Schema、参数校验、结果解析的逻辑。Pydantic-AI简化了这个过程，直接用Python函数定义工具，自动处理所有底层细节。

与竞品的对比

特性	LangChain	LlamaIndex	Pydantic-AI
类型安全	弱	弱	强
学习曲线	陡峭	中等	平缓
数据验证	需额外集成	需额外集成	内置
代码可读性	中等	中等	高
维护成本	高	中	低

环境搭建

前置要求

在开始之前，请确保你的开发环境满足以下条件：

Python 3.10 或更高版本
已安装pip或poetry包管理器
拥有一个支持的LLM服务API密钥（OpenAI、Anthropic、Google等）

安装步骤

方式一：使用pip安装

pip install pydantic-ai

方式二：使用poetry安装

poetry add pydantic-ai

方式三：安装特定版本

pip install pydantic-ai==0.0.1b1

安装额外依赖

根据你使用的LLM提供商，可能需要安装额外的依赖：

# OpenAI支持
pip install 'pydantic-ai[openai]'

# Anthropic支持
pip install 'pydantic-ai[anthropic]'

# Google Gemini支持
pip install 'pydantic-ai[google-gemini]'

# 同时安装多个
pip install 'pydantic-ai[openai,anthropic]'

环境变量配置

创建一个.env文件来管理你的API密钥：

# .env文件内容
OPENAI_API_KEY=sk-your-openai-key-here
ANTHROPIC_API_KEY=sk-ant-your-anthropic-key-here
GOOGLE_API_KEY=your-google-api-key

在Python代码中加载环境变量：

from dotenv import load_dotenv

load_dotenv()  # 加载.env文件中的环境变量

验证安装

创建一个简单的测试脚本来验证安装是否成功：

import pydantic_ai

print(f"Pydantic-AI版本: {pydantic_ai.__version__}")

from pydantic_ai import Agent
print("Agent导入成功！")

核心概念详解

Agent：AI Agent的基本单元

在Pydantic-AI中，Agent是核心概念。你可以把它理解为一个“智能助手”，它具备以下特性：

有一个特定的角色和目标（通过system_prompt定义）
能够处理用户输入并返回结构化结果
可以调用外部工具来扩展能力
返回的结果会被自动验证

创建一个最基本的Agent：

from pydantic_ai import Agent

agent = Agent(
    model='openai:gpt-4',
    system_prompt='你是一个乐于助人的AI助手'
)

Model：支持多种LLM后端

Pydantic-AI支持多种LLM提供商，模型名称遵循统一的格式：provider:model-name

支持的模型示例：

# OpenAI系列
agent = Agent('openai:gpt-4')
agent = Agent('openai:gpt-4-turbo')
agent = Agent('openai:gpt-3.5-turbo')

# Anthropic系列
agent = Agent('anthropic:claude-3-5-sonnet-20241022')
agent = Agent('anthropic:claude-3-haiku-20240307')

# Google系列
agent = Agent('google-gemini:gemini-pro')
agent = Agent('google-gemini:gemini-1.5-flash')

# Ollama本地模型
agent = Agent('ollama:llama3')
agent = Agent('ollama:qwen2')

Result Type：结构化输出的保证

这是Pydantic-AI最强大的特性之一。你可以通过定义Pydantic模型来指定Agent的输出格式：

from pydantic import BaseModel
from pydantic_ai import Agent

class UserProfile(BaseModel):
    name: str
    age: int
    email: str

agent = Agent(
    'openai:gpt-4',
    result_type=UserProfile
)

当Agent执行完成后，返回的结果会自动转换为UserProfile类型。如果格式不匹配，会抛出明确的错误。

System Prompt：定义Agent角色

System Prompt定义了Agent的身份和行为规则。Pydantic-AI支持多种方式定义system prompt：

方式一：直接字符串

agent = Agent(
    'openai:gpt-4',
    system_prompt='你是一个专业的Python程序员，擅长编写高质量、易读的代码。'
)

方式二：多段落Prompt

system_prompt = """
你是一个数据分析助手。

你的职责：
1. 理解用户的数据分析需求
2. 提供清晰的数据解读
3. 生成可执行的建议

沟通风格：
- 使用简单易懂的语言
- 适当使用数据可视化
- 给出具体的行动建议
"""

agent = Agent('openai:gpt-4', system_prompt=system_prompt)

方式三：动态System Prompt

有时你需要在运行时动态调整system prompt：

from pydantic_ai import Agent

def get_system_prompt(context: dict) -> str:
    base = "你是一个数据分析助手。"
    if context.get('expert_mode'):
        base += " 以专业分析师的标准回答。"
    else:
        base += " 用通俗易懂的语言解释。"
    return base

agent = Agent(
    'openai:gpt-4',
    system_prompt=get_system_prompt({'expert_mode': True})
)

实战教程：从入门到精通

案例一：构建一个智能客服Agent

让我们从头开始，构建一个能够回答产品相关问题的智能客服。

第一步：定义数据结构

from pydantic import BaseModel, Field
from typing import Optional

class SupportResult(BaseModel):
    """客服回复的结果模型"""
    response: str = Field(description="回复用户的文本内容")
    category: str = Field(description="问题分类：product, shipping, refund, other")
    needs_escalation: bool = Field(description="是否需要人工介入")
    confidence: float = Field(description="回答的置信度，0-1之间", ge=0, le=1)

class Product(BaseModel):
    name: str
    price: float
    description: str
    in_stock: bool

第二步：创建Agent

from pydantic_ai import Agent

system_prompt = """
你是一个专业的产品客服助手。

你的职责：
1. 回答用户关于产品的问题
2. 提供准确的库存和价格信息
3. 判断是否需要人工客服介入

回答规则：
- 保持友好、专业的语气
- 如果信息不确定，请明确说明
- 当用户要求退款或投诉时，设置needs_escalation为true
"""

agent = Agent(
    'openai:gpt-4',
    result_type=SupportResult,
    system_prompt=system_prompt
)

第三步：执行查询

async def main():
    result = await agent.run("这件T恤有红色可选吗？多少钱？")

    print(f"回复: {result.data.response}")
    print(f"分类: {result.data.category}")
    print(f"需要人工: {result.data.needs_escalation}")
    print(f"置信度: {result.data.confidence}")

import asyncio
asyncio.run(main())

完整示例代码：

from pydantic import BaseModel, Field
from pydantic_ai import Agent
import asyncio

# ==================== 定义数据结构 ====================

class SupportResult(BaseModel):
    response: str = Field(description="回复用户的文本内容")
    category: str = Field(description="问题分类")
    needs_escalation: bool = Field(description="是否需要人工介入")
    confidence: float = Field(description="置信度", ge=0, le=1)

# ==================== 创建Agent ====================

agent = Agent(
    'openai:gpt-4',
    result_type=SupportResult,
    system_prompt=(
        "你是一个专业的产品客服助手。"
        "回答用户关于产品的问题，并判断是否需要人工介入。"
        "始终保持友好、专业的语气。"
    )
)

# ==================== 定义产品数据 ====================

products = {
    "T恤": {
        "price": 99.0,
        "colors": ["黑色", "白色", "蓝色", "红色"],
        "in_stock": True
    },
    "牛仔裤": {
        "price": 199.0,
        "sizes": ["S", "M", "L", "XL"],
        "in_stock": True
    }
}

# ==================== 执行查询 ====================

async def main():
    # 查询产品信息
    result = await agent.run("T恤有红色的吗？多少钱？")

    print("=" * 50)
    print(f"回复内容: {result.data.response}")
    print(f"问题分类: {result.data.category}")
    print(f"需要人工介入: {result.data.needs_escalation}")
    print(f"置信度: {result.data.confidence:.2f}")
    print("=" * 50)

    # 投诉场景
    complaint_result = await agent.run(
        "我上周买的商品质量很差，要求全额退款！"
    )

    print(f"投诉回复: {complaint_result.data.response}")
    print(f"需要人工: {complaint_result.data.needs_escalation}")

if __name__ == "__main__":
    asyncio.run(main())

案例二：构建带工具调用的Agent

让Agent能够调用外部函数是构建实用AI应用的关键。Pydantic-AI提供了简洁的方式来定义和使用工具。

第一步：定义工具函数

from pydantic_ai import Agent, Tool
from datetime import datetime

# ==================== 定义各种工具函数 ====================

def get_weather(city: str) -> str:
    """
    获取城市天气信息

    Args:
        city: 城市名称
    Returns:
        天气描述字符串
    """
    # 这里是模拟数据，实际应用中会调用天气API
    weather_data = {
        "北京": "多云，25°C，适宜出行",
        "上海": "晴天，28°C，注意防晒",
        "广州": "雷阵雨，26°C，记得带伞",
        "深圳": "阴天，27°C，空气湿润"
    }
    return weather_data.get(city, "抱歉，暂不支持该城市")

def get_current_time() -> str:
    """获取当前时间"""
    return datetime.now().strftime("%Y年%m月%d日 %H:%M:%S")

def calculate(expression: str) -> float:
    """
    执行数学计算

    Args:
        expression: 数学表达式，如 "2+3*4"
    Returns:
        计算结果
    """
    # 注意：实际应用中应该使用安全的计算器，避免eval
    allowed_chars = set("0123456789+-*/.() ")
    if all(c in allowed_chars for c in expression):
        return eval(expression)
    return "表达式包含非法字符"

第二步：创建带工具的Agent

# 创建Agent并注册工具
agent = Agent(
    'openai:gpt-4',
    tools=[
        Tool(get_weather, description="获取指定城市的天气信息"),
        Tool(get_current_time, description="获取当前日期和时间"),
        Tool(calculate, description="执行数学计算")
    ]
)

第三步：使用工具执行任务

async def tool_demo():
    # 示例1：查询天气
    result1 = await agent.run("北京今天天气怎么样？适合出门吗？")
    print("天气查询结果：")
    print(result1.data)
    print("-" * 40)

    # 示例2：获取时间并计算
    result2 = await agent.run("现在是几点？如果加上6小时是几点？")
    print("时间计算结果：")
    print(result2.data)
    print("-" * 40)

    # 示例3：纯计算
    result3 = await agent.run("帮我计算 (15 + 25) * 3 / 4")
    print("计算结果：")
    print(result3.data)

asyncio.run(tool_demo())

完整示例代码：

from pydantic_ai import Agent, Tool
from pydantic import BaseModel
from datetime import datetime
import asyncio

# ==================== 定义数据结构 ====================

class WeatherInfo(BaseModel):
    city: str
    weather: str
    temperature: str
    suggestion: str

class TimeInfo(BaseModel):
    current_time: str
    calculated_time: str
    calculation_expression: str
    result: str

# ==================== 定义工具函数 ====================

def get_weather(city: str) -> str:
    """获取城市天气信息"""
    weather_data = {
        "北京": "多云，25°C",
        "上海": "晴天，28°C",
        "广州": "雷阵雨，26°C",
        "深圳": "阴天，27°C"
    }
    return weather_data.get(city, f"暂不支持{city}的天气查询")

def get_current_time() -> str:
    """获取当前时间"""
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def add_hours(hours: int) -> str:
    """添加小时数到当前时间"""
    from datetime import timedelta
    new_time = datetime.now() + timedelta(hours=hours)
    return new_time.strftime("%Y-%m-%d %H:%M:%S")

def calculate(expression: str) -> str:
    """执行数学计算"""
    try:
        result = eval(expression)
        return str(result)
    except Exception as e:
        return f"计算错误: {str(e)}"

# ==================== 创建Agent ====================

agent = Agent(
    'openai:gpt-4',
    tools=[
        Tool(get_weather, description="获取城市天气，如'北京天气'"),
        Tool(get_current_time, description="获取当前时间"),
        Tool(add_hours, description="计算几小时后的时间"),
        Tool(calculate, description="计算数学表达式")
    ],
    system_prompt=(
        "你是一个智能助手，可以查询天气、时间和执行计算。"
        "根据用户需求选择合适的工具来回答问题。"
        "如果有多个工具相关，先调用一个再看结果。"
    )
)

# ==================== 执行演示 ====================

async def main():
    print("=" * 60)
    print("示例1：查询天气")
    print("=" * 60)
    result1 = await agent.run("帮我查一下上海的天气")
    print(f"AI回复: {result1.data}")

    print("\n" + "=" * 60)
    print("示例2：时间计算")
    print("=" * 60)
    result2 = await agent.run("现在几点了？如果过8小时是几点？")
    print(f"AI回复: {result2.data}")

    print("\n" + "=" * 60)
    print("示例3：数学计算")
    print("=" * 60)
    result3 = await agent.run("计算 125 * 8 + 300 等于多少？")
    print(f"AI回复: {result3.data}")

if __name__ == "__main__":
    asyncio.run(main())

案例三：构建多轮对话Agent

在实际应用中，AI通常需要记住之前的对话内容。Pydantic-AI提供了简洁的对话管理机制。

第一步：理解Message历史

from pydantic_ai import Agent
from pydantic_ai.messages import UserMessage, AssistantMessage

# 创建Agent
agent = Agent('openai:gpt-4')

async def conversation_demo():
    # 第一次对话
    result1 = await agent.run("我叫张三，是一名软件工程师")
    print(f"AI: {result1.data}")

    # 第二次对话（带历史）
    result2 = await agent.run(
        "根据我之前说的，我适合什么类型的工作？",
        message_history=agent.messages  # 传递历史消息
    )
    print(f"AI: {result2.data}")

第二步：更复杂的多轮对话

from pydantic_ai import Agent
from pydantic_ai.messages import UserMessage, AssistantMessage, SystemMessage
from dataclasses import dataclass

@dataclass
class ConversationContext:
    user_name: str = ""
    preferences: list = None

    def __post_init__(self):
        if self.preferences is None:
            self.preferences = []

# 创建带上下文的Agent
agent = Agent(
    'openai:gpt-4',
    system_prompt="""
    你是一个旅游规划助手。
    记住用户的基本信息和偏好，为他们推荐合适的旅游目的地。
    """
)

async def travel_planner():
    context = ConversationContext()

    # 第一轮：收集信息
    result1 = await agent.run(
        "我叫李明，我喜欢海边和美食，不喜欢太冷的地方"
    )
    print(f"助手: {result1.data}")

    # 更新上下文
    context.user_name = "李明"
    context.preferences = ["海边", "美食"]

    # 第二轮：基于偏好推荐
    result2 = await agent.run(
        f"你好{context.user_name}，根据我的偏好推荐一个目的地吧",
        message_history=agent.messages
    )
    print(f"助手: {result2.data}")

    # 第三轮：进一步规划
    result3 = await agent.run(
        "好，那就去那里吧。帮我规划3天的行程",
        message_history=agent.messages
    )
    print(f"助手: {result3.data}")

asyncio.run(travel_planner())

案例四：流式输出处理

对于需要实时显示AI回复的应用，流式输出是必须的。

from pydantic_ai import Agent
import asyncio

agent = Agent('openai:gpt-4')

async def streaming_demo():
    print("开始流式输出：\n")

    # 使用stream方法获取流式响应
    async with agent.run_stream("写一首关于春天的诗") as response:
        # 处理流式输出
        async for chunk in response.stream():
            print(chunk, end="", flush=True)

    print("\n\n流式输出完成！")

asyncio.run(streaming_demo())

带有进度指示的流式输出：

import asyncio
from pydantic_ai import Agent

agent = Agent('openai:gpt-4')

async def streaming_with_progress():
    print("正在生成代码，请稍候...\n")

    async with agent.run_stream(
        "写一个Python函数，实现快速排序算法"
    ) as response:
        full_response = ""
        async for chunk in response.stream():
            full_response += chunk
            # 可以在这里添加进度指示
            print(f"\r生成中... {len(full_response)} 字符", end="")

    print(f"\n\n完整代码：\n{full_response}")

asyncio.run(streaming_with_progress())

常见使用场景

场景一：数据提取与转换

从非结构化文本中提取结构化数据是AI的强项：

from pydantic import BaseModel, Field
from pydantic_ai import Agent
import asyncio

class InvoiceData(BaseModel):
    invoice_number: str = Field(description="发票号码")
    date: str = Field(description="开票日期")
    amount: float = Field(description="金额")
    items: list[str] = Field(description="购买物品列表")
    vendor: str = Field(description="供应商名称")

agent = Agent(
    'openai:gpt-4',
    result_type=InvoiceData,
    system_prompt=(
        "你是一个数据提取专家。"
        "从发票文本中提取结构化信息。"
        "所有金额必须是数字格式。"
    )
)

async def extract_invoice():
    invoice_text = """
    增值税专用发票
    发票号码：FP202401150001
    开票日期：2024年1月15日
    供应商：北京科技有限公司
    购买物品：
    1. 笔记本电脑 x1台 - 5999元
    2. 无线鼠标 x2个 - 198元
    3. 机械键盘 x1个 - 399元
    合计金额：6596元（含税）
    """

    result = await agent.run(invoice_text)

    print("提取的发票数据：")
    print(f"发票号: {result.data.invoice_number}")
    print(f"日期: {result.data.date}")
    print(f"供应商: {result.data.vendor}")
    print(f"总金额: ¥{result.data.amount}")
    print("购买物品:")
    for item in result.data.items:
        print(f"  - {item}")

asyncio.run(extract_invoice())

场景二：智能问答系统

构建一个基于知识库的问答系统：

from pydantic import BaseModel
from pydantic_ai import Agent
import asyncio

class QAResult(BaseModel):
    answer: str = Field(description="回答内容")
    source: str = Field(description="答案来源")
    confidence: float = Field(description="置信度", ge=0, le=1)

class QAKnowledge:
    """知识库模拟"""

    def __init__(self):
        self.knowledge = {
            "退款政策": "我们的退货政策是收到商品后30天内可申请退货，"
                      "商品需保持原包装完好。退款将在收到退货后3-5个工作日内处理。",
            "会员权益": "成为会员后可享受：1. 全场9折优惠 2. 积分翻倍 3. 专属客服服务 4. 优先发货",
            "配送时间": "标准配送3-5个工作日，加急配送次日达，部分地区支持当日达"
        }

    def search(self, query: str) -> str:
        """搜索相关知识"""
        for key, value in self.knowledge.items():
            if key in query:
                return f"[{key}]\n{value}"
        return "抱歉，我没有找到相关的答案。"

# 初始化知识库
knowledge_base = QAKnowledge()

agent = Agent(
    'openai:gpt-4',
    result_type=QAResult,
    system_prompt=f"""
    你是一个智能客服助手，使用以下知识库回答用户问题。

    知识库内容：
    {knowledge_base.search('退款')}
    {knowledge_base.search('会员')}
    {knowledge_base.search('配送')}

    回答规则：
    1. 优先使用知识库内容回答
    2. 如果知识库没有相关信息，说明无法回答
    3. 回答要清晰、有条理
    """
)

async def qa_demo():
    questions = [
        "退货要多久能拿到钱？",
        "成为会员有什么好处？",
        "一般几天能收到货？"
    ]

    for question in questions:
        result = await agent.run(question)
        print(f"问题: {question}")
        print(f"回答: {result.data.answer}")
        print(f"来源: {result.data.source}")
        print(f"置信度: {result.data.confidence}")
        print("-" * 50)

asyncio.run(qa_demo())

场景三：代码审查助手

构建一个自动化代码审查工具：

from pydantic import BaseModel, Field
from pydantic_ai import Agent
from typing import list
import asyncio

class CodeReviewResult(BaseModel):
    issues: list[dict] = Field(description="发现的问题列表")
    suggestions: list[str] = Field(description="改进建议")
    overall_rating: str = Field(description="整体评价：good/needs_work/poor")
    score: int = Field(description="评分1-10分", ge=1, le=10)

agent = Agent(
    'openai:gpt-4',
    result_type=CodeReviewResult,
    system_prompt="""
    你是一个资深的Python代码审查专家。
    审查以下代码，关注：
    1. 代码质量和可读性
    2. 潜在的bug和安全问题
    3. 性能优化建议
    4. Python最佳实践

    请给出具体的改进建议和代码示例。
    """
)

async def code_review_demo():
    code_to_review = '''
def process_user_data(data):
    # 处理用户数据
    result = eval(data)  # 危险！
    exec(result)
    return result
    '''

    result = await agent.run(f"请审查以下Python代码：\n{code_to_review}")

    print("=" * 60)
    print("代码审查报告")
    print("=" * 60)
    print(f"\n整体评分: {result.data.score}/10")
    print(f"评价: {result.data.overall_rating}")

    print("\n发现的问题：")
    for i, issue in enumerate(result.data.issues, 1):
        print(f"  {i}. [{issue.get('severity', 'N/A')}] {issue.get('message', '')}")
        if issue.get('line'):
            print(f"     行号: {issue['line']}")

    print("\n改进建议：")
    for i, suggestion in enumerate(result.data.suggestions, 1):
        print(f"  {i}. {suggestion}")

asyncio.run(code_review_demo())

进阶技巧与最佳实践

技巧一：自定义输出验证

有时候需要更复杂的输出验证逻辑：

from pydantic import BaseModel, field_validator, Field
from pydantic_ai import Agent
import asyncio

class ValidatedEmailResult(BaseModel):
    email: str = Field(description="邮箱地址")
    is_valid: bool = Field(description="是否有效")
    suggestion: str = Field(description="如果是无效邮箱，给出更正建议")

    @field_validator('email')
    @classmethod
    def validate_email_format(cls, v):
        """验证邮箱格式"""
        import re
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(pattern, v):
            raise ValueError(f"邮箱格式不正确: {v}")
        return v.lower()

agent = Agent(
    'openai:gpt-4',
    result_type=ValidatedEmailResult,
    system_prompt=(
        "你是一个邮箱验证助手。"
        "检查用户提供的邮箱地址是否有效。"
        "如果无效，给出合理的更正建议。"
    )
)

async def email_validation():
    test_emails = [
        "John@Example.com",
        "user+tag@gmail.com",
        "invalid-email",
        "test@company.co.uk"
    ]

    for email in test_emails:
        try:
            result = await agent.run(f"验证邮箱: {email}")
            print(f"✓ {email}")
            print(f"  有效: {result.data.is_valid}")
            if result.data.suggestion:
                print(f"  建议: {result.data.suggestion}")
        except Exception as e:
            print(f"✗ {email}: {e}")
        print()

asyncio.run(email_validation())

技巧二：并行工具调用

当有多个独立的工具调用时，可以让它们并行执行：

from pydantic_ai import Agent, Tool
import asyncio

def get_stock_price(symbol: str) -> float:
    """获取股票价格"""
    prices = {"AAPL": 175.50, "GOOGL": 140.25, "MSFT": 378.90}
    return prices.get(symbol, 0.0)

def get_company_info(symbol: str) -> str:
    """获取公司信息"""
    info = {
        "AAPL": "Apple Inc. - 全球领先的科技公司",
        "GOOGL": "Alphabet Inc. - 全球最大的搜索引擎公司",
        "MSFT": "Microsoft Corporation - 全球领先的软件公司"
    }
    return info.get(symbol, "未知公司")

agent = Agent(
    'openai:gpt-4',
    tools=[
        Tool(get_stock_price),
        Tool(get_company_info)
    ],
    system_prompt="你可以查询股票信息，请尽可能一次性获取所有需要的信息。"
)

async def parallel_demo():
    # 虽然这里表面上是顺序调用
    # 但Agent内部会根据需要决定是否并行
    result = await agent.run(
        "告诉我苹果(AAPL)、谷歌(GOOGL)和微软(MSFT)的股价和公司信息"
    )
    print(result.data)

asyncio.run(parallel_demo())

技巧三：错误处理与重试机制

from pydantic_ai import Agent
from pydantic_ai.exceptions import ModelRetry
import asyncio

agent = Agent(
    'openai:gpt-4',
    retries=3  # 设置重试次数
)

async def error_handling_demo():
    try:
        result = await agent.run("你的问题")
        print(result.data)
    except ModelRetry as e:
        print(f"模型需要重试: {e}")
    except Exception as e:
        print(f"发生错误: {e}")

技巧四：使用本地模型

对于有隐私要求或想要节省成本的场景，可以使用Ollama运行本地模型：

# 确保已安装并启动Ollama
# ollama serve
# ollama pull llama3

agent = Agent(
    'ollama:llama3',  # 使用本地模型
    system_prompt="你是一个友好的AI助手",
    # 可以指定Ollama服务的地址
    # infer_params={"base_url": "http://localhost:11434"}
)

async def local_model_demo():
    result = await agent.run("你好，请介绍一下你自己")
    print(result.data)

最佳实践总结

实践一：保持System Prompt简洁明确

# 不好的做法
agent = Agent(
    'openai:gpt-4',
    system_prompt="""
    你是一个助手。你要帮助用户。用户可能会问各种问题。
    你需要尽可能准确地回答问题。如果你不确定，就说不知道。
    不要编造信息。保持礼貌。回答要简洁但完整。
    ... 还有100行类似的废话 ...
    """
)

# 好的做法
agent = Agent(
    'openai:gpt-4',
    system_prompt="""
    你是一个数据分析助手。
    职责：分析数据，提供洞察，给出建议。
    规则：1. 只陈述有数据支撑的结论 2. 不确定时明确说明
    """
)

实践二：充分利用类型提示

from pydantic import BaseModel, Field
from typing import Optional, Literal

class SearchResult(BaseModel):
    title: str
    url: str
    snippet: str
    relevance_score: float = Field(ge=0, le=1, description="相关性评分")

class AdvancedSearchResult(SearchResult):
    published_date: Optional[str] = None
    author: Optional[str] = None

# 使用Literal限制可选值
class StatusResponse(BaseModel):
    status: Literal["success", "pending", "failed"]
    message: str

实践三：合理设置超时和重试

from pydantic_ai import Agent
import asyncio

agent = Agent(
    'openai:gpt-4',
    retries=2,
    timeout=30  # 30秒超时
)

async def with_timeout():
    try:
        result = await asyncio.wait_for(
            agent.run("复杂问题"),
            timeout=30
        )
        return result
    except asyncio.TimeoutError:
        print("请求超时，请重试")

与其他工具的集成

集成FastAPI

from fastapi import FastAPI
from pydantic_ai import Agent
from pydantic import BaseModel
import asyncio

app = FastAPI()
agent = Agent('openai:gpt-4')

class Question(BaseModel):
    text: str

class Answer(BaseModel):
    response: str

@app.post("/ask", response_model=Answer)
async def ask_question(question: Question):
    result = await agent.run(question.text)
    return Answer(response=result.data)

# 运行方式：uvicorn main:app --reload

集成Streamlit

import streamlit as st
from pydantic_ai import Agent
import asyncio

st.title("AI 助手")

agent = Agent('openai:gpt-4')

if "messages" not in st.session_state:
    st.session_state.messages = []

for message in st.session_state.messages:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

if prompt := st.chat_input("请输入您的问题"):
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    with st.chat_message("assistant"):
        result = asyncio.run(agent.run(prompt))
        st.markdown(result.data)

    st.session_state.messages.append(
        {"role": "assistant", "content": result.data}
    )

集成Django

# views.py
from django.http import JsonResponse
from pydantic_ai import Agent
from pydantic import BaseModel
import asyncio

agent = Agent('openai:gpt-4')

class ChatRequest(BaseModel):
    message: str

class ChatResponse(BaseModel):
    response: str

async def chat_view(request):
    data = ChatRequest.model_validate_json(request.body)
    result = await agent.run(data.message)
    return JsonResponse(ChatResponse(response=result.data).model_dump())

常见问题与解决方案

问题一：API密钥配置

问题描述： 代码运行时报错提示找不到API密钥

解决方案：

# 方式一：使用环境变量
import os
os.environ["OPENAI_API_KEY"] = "your-key-here"

# 方式二：使用python-dotenv
from dotenv import load_dotenv
load_dotenv()  # 加载.env文件

# 方式三：直接传入
agent = Agent('openai:gpt-4', api_key="your-key-here")

问题二：模型不支持工具调用

问题描述： 某些模型返回不支持工具的错误

解决方案： 确保使用支持工具调用的模型版本

# 确认模型支持工具调用
# OpenAI: gpt-4-turbo, gpt-4, gpt-3.5-turbo (需最新版本)
# Anthropic: claude-3系列
# Google: gemini-pro (需检查版本)

# 如果必须使用不支持的模型，可以禁用工具
agent = Agent(
    'openai:gpt-3.5-turbo',
    tools=[]  # 空列表，不使用工具
)

问题三：输出格式不匹配

问题描述： Agent返回的数据无法转换为指定的Result Type

解决方案：

from pydantic import BaseModel
from pydantic_ai import Agent

class FlexibleResult(BaseModel):
    text: str
    extra_data: dict = {}  # 接收额外未定义字段

agent = Agent(
    'openai:gpt-4',
    result_type=FlexibleResult,
    system_prompt="你的回复应该是一个包含text字段的JSON对象"
)

# 或者添加验证器处理特殊情况
class RobustResult(BaseModel):
    content: str

    @classmethod
    def validate_response(cls, data):
        # 自定义验证逻辑
        if isinstance(data, str):
            return cls(content=data)
        return cls(**data)

问题四：Token使用超限

问题描述： 对话历史太长导致Token超限

解决方案：

# 限制消息历史长度
async def limited_chat(user_input: str):
    MAX_MESSAGES = 10  # 保留最近10条消息

    # 截取最近的消息
    recent_messages = agent.messages[-MAX_MESSAGES:]

    result = await agent.run(
        user_input,
        message_history=recent_messages
    )
    return result

总结与展望

核心价值回顾

Pydantic-AI为Python开发者带来了几个关键价值：

类型安全：通过Python类型注解实现Prompt、数据验证和输出的全链路类型安全
简洁API：相比LangChain等框架，API更加直观易懂
结构化输出：内置的Result Type机制解决了LLM输出格式不确定的问题
工具集成：简洁的Tool装饰器让外部函数调用变得轻而易举

适用场景

构建需要返回结构化数据的AI应用
开发需要调用多个外部API的Agent
需要类型安全和代码自动补全的团队项目
从其他框架迁移希望降低维护成本的团队

不适用场景

极度简单的单次调用场景（直接调用API可能更简单）
需要复杂图结构的复杂工作流
对延迟极度敏感的生产环境（需要更多优化）

展望未来

Pydantic-AI作为一个新兴项目，正在快速发展。建议关注：

官方GitHub仓库的更新和版本发布
社区贡献的工具和插件
与更多LLM提供商的集成
性能优化和新特性

相关资源链接

官方资源

GitHub仓库：https://github.com/pydantic/pydantic-ai
官方文档：https://ai.pydantic.dev/
PyPI页面：https://pypi.org/project/pydantic-ai/
Pydantic官网：https://docs.pydantic.dev/

学习资源

Pydantic基础：如果你不熟悉Pydantic，建议先学习其基础用法
Python类型提示：熟练使用typing模块能让开发更顺畅

相关项目

Pydantic：数据验证库，Pydantic-AI的基础
FastAPI：现代Web框架，与Pydantic深度集成
Instructor：另一个用于结构化输出的库
DSPy：斯坦福的编程模型框架

写在最后

Pydantic-AI代表了AI应用开发的一种新思路：将我们在传统软件开发中积累的最佳实践（如类型安全、代码验证）引入到AI领域。这不仅仅是一个技术框架，更是一种开发范式的转变。

无论你是正在构建AI应用的开发者，还是希望了解AI工程化的学习者，Pydantic-AI都值得一试。它的设计理念简单而强大：让AI开发变得像写普通Python代码一样自然。

开始动手试试吧，也许你会发现，构建AI应用从未如此简单。

祝你开发愉快！

别再手动拼接Prompt了！Pydantic-AI让AI应用开发像写API一样优雅

☕ 如果内容对您有帮助，欢迎打赏

评论区

发表回复取消回复

☕ 如果内容对您有帮助，欢迎打赏

相关文章

别再盲目跟风了！这个开源项目重新定义人机交互的边界

**别再花大价钱买客服系统了！这款开源方案让中小企业也能拥有 Zendesk 级别的体验**

别再为库存管理头疼了！这个开源项目让电商库存控制变得如此简单

评论区

发表回复 取消回复

别再花大价钱买客服系统了！这款开源方案让中小企业也能拥有 Zendesk 级别的体验

发表回复取消回复