智能代理开发实战：使用Pydantic AI构建企业级AI应用

2026-04-03 09:27:31作者：蔡怀权

在当今快速发展的AI技术领域，企业级智能代理应用的开发面临着诸多挑战，如复杂的配置流程、繁琐的代码编写以及难以监控的运行状态。Pydantic AI作为一款强大的智能代理框架，为开发者提供了快速开发、低代码实现企业级应用的解决方案。本文将带你从零开始，通过实际案例掌握Pydantic AI的核心功能，避开常见陷阱，构建高效可靠的智能代理应用。

智能代理开发基础：核心概念与环境搭建

如何理解智能代理的核心组件

智能代理是Pydantic AI与大型语言模型交互的核心接口，它包含多个关键组件，这些组件协同工作，使AI应用能够理解指令、调用工具并生成结构化输出。

术语	通俗解释
Agent	智能代理的核心容器，整合了模型、工具和指令
Tool	代理可以调用的功能函数，扩展AI的能力范围
RunContext	运行上下文，保存代理执行过程中的状态和依赖
System Prompt	给AI的指令集，指导其行为和响应方式
Structured Output	AI必须返回的特定格式数据，通常通过Pydantic模型定义

实际应用场景：在客服系统中，Agent可以整合查询客户信息、处理订单、生成回复等工具，通过System Prompt定义客服话术风格，确保AI以一致的专业形象与用户交互。

从零开始搭建开发环境

Pydantic AI提供了灵活的安装方式，可根据项目需求选择不同的安装模式：

# 标准安装：包含所有核心功能和模型支持
pip install pydantic-ai

# 精简安装：仅包含基础功能，按需添加模型支持
pip install "pydantic-ai-slim[openai,anthropic]"

[!TIP] 对于生产环境，建议使用精简安装模式，仅包含项目所需的模型依赖，以减小部署体积并提高安全性。

环境验证：安装完成后，可以通过以下代码验证环境是否配置正确：

from pydantic_ai import Agent

# 创建一个简单的代理
agent = Agent('openai:gpt-3.5-turbo', instructions='你是一个帮助用户解答技术问题的助手。')

# 运行代理
async def main():
    result = await agent.run('什么是Pydantic AI？')
    print(result.output)

if __name__ == '__main__':
    import asyncio
    asyncio.run(main())

技术选型对比：如何选择适合的智能代理框架

主流智能代理框架特性对比

在选择智能代理框架时，需要考虑开发效率、性能、可扩展性等多个因素。以下是Pydantic AI与其他主流框架的对比：

特性	Pydantic AI	LangChain	AutoGPT
开发难度	低（Pydantic模型集成）	中（需学习特定API）	高（复杂配置）
性能	高（异步处理）	中（部分同步操作）	低（多轮思考）
工具集成	简单（装饰器模式）	中等（需定义工具类）	复杂（JSON配置）
结构化输出	原生支持（Pydantic模型）	需额外配置	有限支持
监控能力	内置Logfire集成	需第三方工具	基本日志

实际应用场景：对于需要快速开发且对结构化输出有严格要求的企业应用，Pydantic AI是理想选择；对于已有LangChain生态投资的团队，可以考虑两者结合使用。

Pydantic AI的独特优势

Pydantic AI相比其他框架的核心优势在于：

类型安全：基于Pydantic模型，提供严格的类型检查和数据验证
开发效率：简洁的API设计，减少样板代码
灵活性：支持多种模型和工具集成，易于扩展
可观测性：内置监控和日志功能，便于调试和优化

实战开发：构建企业级智能代理应用

实战一：构建产品信息查询代理

以下是一个产品信息查询代理的实现，它能够根据产品ID查询数据库并返回格式化信息：

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
from typing import Optional

# 定义产品信息数据模型
class ProductInfo(BaseModel):
    product_id: str
    name: str
    price: float
    stock: int
    description: Optional[str] = None

# 模拟数据库连接
class ProductDB:
    async def get_product(self, product_id: str) -> ProductInfo:
        # 实际应用中这里会连接真实数据库
        return ProductInfo(
            product_id=product_id,
            name="智能手表",
            price=1299.00,
            stock=42,
            description="多功能智能手表，支持心率监测和GPS定位"
        )

# 创建代理
product_agent = Agent(
    model='openai:gpt-4o',
    instructions='你是一个产品信息查询助手，使用提供的工具回答用户关于产品的问题。',
    deps_type=ProductDB,  # 依赖类型
)

# 定义工具函数
@product_agent.tool
async def query_product(ctx: RunContext[ProductDB], product_id: str) -> ProductInfo:
    """查询产品信息"""
    return await ctx.deps.get_product(product_id)

# 运行代理
async def main():
    db = ProductDB()
    result = await product_agent.run(
        "查询产品ID为P12345的信息，包括价格和库存",
        deps=db
    )
    print(f"产品信息: {result.output}")

if __name__ == '__main__':
    import asyncio
    asyncio.run(main())

这个代理展示了如何使用Pydantic模型定义结构化输出，以及如何集成外部依赖（如数据库连接）。完整案例：[examples/pydantic_ai_examples/product_agent.py]

实战二：构建多工具协作的客户支持代理

客户支持代理需要整合多种工具，如查询订单、处理退款和生成回复：

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
from typing import List, Optional

# 定义数据模型
class OrderStatus(BaseModel):
    order_id: str
    status: str  # 'pending', 'shipped', 'delivered', 'cancelled'
    estimated_delivery: Optional[str] = None

class RefundRequest(BaseModel):
    order_id: str
    reason: str
    amount: float

# 创建代理
support_agent = Agent(
    model='anthropic:claude-3-sonnet-20240229',
    instructions=(
        '作为客户支持助手，你可以查询订单状态和处理退款请求。'
        '保持回复友好且专业，提供清晰的下一步操作建议。'
    ),
)

# 定义工具
@support_agent.tool
async def check_order_status(ctx: RunContext, order_id: str) -> OrderStatus:
    """查询订单状态"""
    # 实际应用中这里会连接订单系统
    return OrderStatus(
        order_id=order_id,
        status="shipped",
        estimated_delivery="2023-12-15"
    )

@support_agent.tool
async def process_refund(ctx: RunContext, order_id: str, reason: str) -> RefundRequest:
    """处理退款请求"""
    # 实际应用中这里会调用退款系统
    return RefundRequest(
        order_id=order_id,
        reason=reason,
        amount=99.99
    )

# 运行代理
async def main():
    result = await support_agent.run("我想查询订单#A7B3921的状态，并申请退款，因为商品已损坏")
    print(result.output)

if __name__ == '__main__':
    import asyncio
    asyncio.run(main())

这个案例展示了如何在一个代理中集成多个工具，并让AI根据用户请求自动选择合适的工具。

代理工作原理流程图

上图展示了智能代理的典型工作流程，包括接收用户输入、工具调用决策、执行工具、处理结果和生成响应等步骤。通过Logfire监控面板，开发者可以清晰地看到代理的执行过程和每个步骤的耗时。

问题解决方案：企业应用中的常见挑战

如何处理代理的长时间运行任务

在企业应用中，智能代理经常需要处理长时间运行的任务，如数据分析或报告生成。以下是使用异步任务和进度跟踪的解决方案：

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
import asyncio
from typing import List

class AnalysisResult(BaseModel):
    task_id: str
    status: str
    progress: int  # 0-100
    result: Optional[str] = None

# 创建支持长时间任务的代理
analysis_agent = Agent(
    model='openai:gpt-4o',
    instructions='你是一个数据分析助手，能够处理长时间运行的分析任务并提供进度更新。',
)

@analysis_agent.tool
async def start_long_running_analysis(ctx: RunContext, task_id: str) -> AnalysisResult:
    """启动长时间运行的数据分析任务"""
    # 模拟长时间运行的任务
    for i in range(10):
        await asyncio.sleep(1)  # 模拟工作
        # 可以通过某种机制更新进度
        if i == 9:
            return AnalysisResult(
                task_id=task_id,
                status="completed",
                progress=100,
                result="分析完成：用户参与度提升了15.3%"
            )
    return AnalysisResult(
        task_id=task_id,
        status="in_progress",
        progress=50
    )

@analysis_agent.tool
async def check_task_status(ctx: RunContext, task_id: str) -> AnalysisResult:
    """检查长时间任务的状态"""
    # 实际应用中这里会查询任务状态
    return AnalysisResult(
        task_id=task_id,
        status="completed",
        progress=100,
        result="分析完成：用户参与度提升了15.3%"
    )

# 运行代理
async def main():
    result = await analysis_agent.run("开始分析销售数据，任务ID为SALES-2023-12")
    print(result.output)

if __name__ == '__main__':
    import asyncio
    asyncio.run(main())

如何实现多轮对话和上下文管理

对于需要记忆对话历史的应用，Pydantic AI提供了内置的对话历史管理：

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

async def main():
    agent = Agent('openai:gpt-4o', instructions='你是一个记住用户偏好的助手。')
    
    # 初始对话
    messages = [
        UserMessage(content='我喜欢科幻电影'),
        AssistantMessage(content='好的，我会记住你喜欢科幻电影。')
    ]
    
    # 继续对话，传递历史消息
    result = await agent.run(
        '有什么好电影推荐吗？',
        message_history=messages
    )
    
    print(result.output)  # 应该推荐科幻电影
    
    # 添加新消息到历史并继续对话
    messages.extend(result.new_messages)
    result = await agent.run(
        '还有其他类似的吗？',
        message_history=messages
    )
    
    print(result.output)

if __name__ == '__main__':
    import asyncio
    asyncio.run(main())

如何处理敏感数据和权限控制

在企业应用中，处理敏感数据时需要严格的权限控制：

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
from typing import Optional

class UserContext(BaseModel):
    user_id: str
    permissions: List[str]  # 如 ["view_orders", "process_refunds"]

class OrderDetails(BaseModel):
    order_id: str
    amount: float
    items: List[str]
    # 敏感字段
    customer_email: Optional[str] = None
    customer_phone: Optional[str] = None

# 创建带权限控制的代理
secure_agent = Agent(
    model='openai:gpt-4o',
    instructions='你是一个遵守数据访问权限的订单查询助手。',
    deps_type=UserContext,
)

@secure_agent.tool
async def get_order_details(ctx: RunContext[UserContext], order_id: str) -> OrderDetails:
    """获取订单详情"""
    # 模拟查询订单
    full_order = OrderDetails(
        order_id=order_id,
        amount=129.99,
        items=["商品A", "商品B"],
        customer_email="customer@example.com",
        customer_phone="123-456-7890"
    )
    
    # 根据权限过滤敏感信息
    if "view_personal_data" not in ctx.deps.permissions:
        full_order.customer_email = None
        full_order.customer_phone = None
        
    return full_order

# 运行代理
async def main():
    # 无敏感数据访问权限的用户
    limited_user = UserContext(user_id="user123", permissions=["view_orders"])
    result = await secure_agent.run(
        f"查询订单#O98765的详情",
        deps=limited_user
    )
    print("有限权限用户看到的订单:", result.output)
    
    # 有敏感数据访问权限的用户
    admin_user = UserContext(user_id="admin456", permissions=["view_orders", "view_personal_data"])
    result = await secure_agent.run(
        f"查询订单#O98765的详情",
        deps=admin_user
    )
    print("管理员看到的订单:", result.output)

if __name__ == '__main__':
    import asyncio
    asyncio.run(main())

部署与监控：确保企业级应用稳定运行

智能代理部署清单

部署企业级智能代理应用时，应遵循以下清单：

检查项	说明
环境配置	设置适当的环境变量，如API密钥、模型选择等
依赖管理	使用精简安装模式，仅包含必要依赖
日志配置	集成Logfire或其他日志系统
错误处理	实现全局异常捕获和优雅降级机制
资源限制	设置API调用速率限制和超时时间
安全措施	实施输入验证和敏感数据过滤
监控设置	配置关键指标监控和告警
文档生成	为API和工具函数生成文档
测试覆盖	编写单元测试和集成测试
部署流程	自动化部署脚本和版本控制

性能优化Checklist

为确保智能代理应用的高性能，可参考以下优化要点：

优化项	实施方法
连接池	复用HTTP客户端和数据库连接
缓存策略	缓存频繁使用的工具调用结果
异步处理	使用异步I/O提高并发能力
批处理	合并多个相似的工具调用
模型选择	根据任务复杂度选择合适的模型
输入优化	精简提示词，去除冗余信息
输出过滤	只返回必要的结果数据
资源监控	跟踪CPU、内存和网络使用情况
错误重试	实施智能重试策略，处理临时故障
负载均衡	在多实例部署中平衡请求负载

监控与调试工具

Pydantic AI集成了强大的监控工具，帮助开发者跟踪代理的运行状态和性能。

上图展示了Logfire监控面板，它提供了代理运行的详细视图，包括：

工具调用频率和耗时
模型响应时间
错误率和异常情况
资源使用情况

通过这些监控数据，开发者可以识别性能瓶颈，优化工具调用流程，并及时发现和解决问题。

常见误区解析：避坑指南

误区一：过度依赖大模型能力

许多开发者认为大型语言模型可以处理所有任务，而忽视了工具的重要性。实际上，将复杂计算和数据处理交给专用工具，只让LLM专注于决策和自然语言处理，能获得更好的性能和可靠性。

正确做法：

将数学计算、数据库查询等任务交给专用工具
LLM专注于理解用户意图、选择工具和生成自然语言响应
使用结构化输出确保工具调用的准确性

误区二：忽视错误处理和重试机制

在实际应用中，API调用失败、网络问题等情况时有发生。忽视错误处理会导致应用不稳定。

正确做法：

from pydantic_ai import Agent
from tenacity import retry, stop_after_attempt, wait_exponential

agent = Agent(
    model='openai:gpt-4o',
    retries=3,  # 全局重试次数
)

@agent.tool
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
async def unreliable_api_call():
    """可能会失败的API调用，需要重试机制"""
    # API调用代码
    pass

误区三：不恰当的工具设计

设计工具时过于复杂或不够通用，会降低代理的效率和灵活性。

正确做法：

工具应专注于单一功能
使用清晰的参数和返回类型
提供详细的工具描述，帮助LLM理解用途
工具之间保持低耦合

高级应用：构建实时聊天系统

实时聊天应用架构

实时聊天系统是Pydantic AI的一个重要应用场景，它需要处理流式响应和保持对话状态。以下是一个基于FastAPI的实时聊天应用实现：

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic_ai import Agent
from pydantic_ai.messages import Message, UserMessage
import asyncio
from datetime import datetime
import json
from typing import List

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

# 存储对话历史（实际应用中应使用数据库）
chat_history: dict[str, List[Message]] = {}

@app.post("/chat/{session_id}")
async def chat(session_id: str, request: Request) -> StreamingResponse:
    data = await request.json()
    user_message = data.get("message", "")
    
    # 获取或初始化对话历史
    history = chat_history.get(session_id, [])
    history.append(UserMessage(content=user_message))
    
    async def event_generator():
        # 流式传输AI响应
        async with chat_agent.run_stream(
            user_message,
            message_history=history[:-1]  # 排除当前消息
        ) as result:
            full_response = ""
            async for chunk in result.stream_output():
                full_response += chunk
                # 发送SSE事件
                yield f"data: {json.dumps({'chunk': chunk})}\n\n"
            
            # 更新对话历史
            history.append(result.assistant_message)
            chat_history[session_id] = history[-20:]  # 限制历史长度
    
    return StreamingResponse(event_generator(), media_type="text/event-stream")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)