探索pydantic-ai：构建智能代理的5个核心技术与实战应用

2026-04-12 09:09:46作者：盛欣凯Ernestine

在AI应用开发中，如何高效地将大型语言模型(LLM)与外部工具集成？如何确保AI代理的运行过程可监控、可调试？pydantic-ai作为一款专注于LLM交互的开发框架，为解决这些问题提供了优雅的解决方案。本文将深入剖析pydantic-ai的核心技术原理，通过实战案例展示其在智能代理开发中的独特优势，帮助技术探索者快速掌握这一强大工具。

环境准备：如何搭建高效的AI代理开发环境？

当我们开始构建AI代理应用时，首先面临的问题是如何配置一个既全面又灵活的开发环境。pydantic-ai提供了两种安装模式，满足不同场景的需求。

标准安装模式适合大多数开发者，它包含了所有核心依赖和模型支持，只需一条命令即可完成：

pip install pydantic-ai

这种安装方式确保你能直接使用所有内置模型和功能，无需额外配置。要求Python 3.10或更高版本，这是因为pydantic-ai充分利用了Python 3.10+的类型提示和异步特性。

轻量级配置：如何为特定场景优化依赖体积？

对于资源受限或需求明确的生产环境，轻量级配置成为必然选择。pydantic-ai的精简安装模式允许开发者仅安装所需组件，显著减少依赖体积。

例如，当我们只需使用OpenAI模型时，可以这样安装：

pip install "pydantic-ai-slim[openai]"

这种方式只安装核心框架和OpenAI相关依赖。如果需要同时支持多个模型或功能，可以组合安装不同的可选依赖组：

pip install "pydantic-ai-slim[openai,logfire]"

这种模块化设计让应用在保持功能完整的同时，实现了资源的最优利用。

核心价值：为什么选择pydantic-ai构建智能代理？

在众多AI代理开发框架中，pydantic-ai凭借其独特的设计理念脱颖而出。它将Pydantic的数据验证能力与LLM交互完美结合，创造出一种声明式的AI代理开发模式。

想象一下传统的AI代理开发流程：开发者需要手动处理API调用、响应解析、错误处理等繁琐任务。而pydantic-ai通过提供统一的Agent接口，将这些复杂性抽象化，让开发者可以专注于业务逻辑而非技术细节。

特别是在工具集成方面，pydantic-ai的装饰器模式使得工具函数的定义和注册变得异常简单，大大降低了LLM与外部系统交互的门槛。

技术解构：Agent组件如何协同工作？

当我们深入研究pydantic-ai的内部结构时，会发现Agent是整个框架的核心。Agent就像一位餐厅经理，协调着厨师(工具)、顾客(用户)和菜单(指令)之间的关系，确保整个服务流程顺畅高效。

Agent的核心组件包括：

Instructions：给LLM的系统提示，指导其行为方式
Function tool(s)：LLM可以调用的外部工具函数
Structured output type：定义LLM必须返回的数据结构
Dependency type：工具函数可能需要的依赖项类型
LLM model：关联的默认语言模型
Model Settings：用于微调模型请求的参数

这些组件协同工作，形成一个完整的智能代理系统。当用户输入查询时，Agent会根据系统提示决定是否调用工具，处理工具返回结果，并最终生成结构化输出。

场景实践：如何构建一个实用的天气查询代理？

需求分析

在日常生活中，我们经常需要查询不同地点的天气情况。一个智能的天气查询代理应该能够理解用户的自然语言请求，获取多个地点的天气信息，并以简洁的方式呈现结果。

方案设计

基于pydantic-ai，我们可以设计一个包含以下功能的天气代理：

解析用户查询中的地点信息
获取每个地点的经纬度坐标
根据经纬度查询天气数据
整合结果并以自然语言回答

编码实现

首先，我们需要定义代理的依赖项和工具函数：

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel
import asyncio
from httpx import AsyncClient
import logfire

# 配置日志监控
logfire.configure(send_to_logfire='if-token-present')
logfire.instrument_pydantic_ai()

# 定义依赖项类型，包含HTTP客户端
class Deps:
    client: AsyncClient

# 创建天气代理实例
# 这里我们指定了默认模型、指令和依赖类型
weather_agent = Agent(
    'openai:gpt-4.1-mini',
    instructions='Be concise, reply with one sentence.',
    deps_type=Deps,
    retries=2,  # 自动重试机制
)

# 定义经纬度数据模型
class LatLng(BaseModel):
    lat: float
    lng: float

# 定义获取经纬度的工具函数
@weather_agent.tool
async def get_lat_lng(ctx: RunContext[Deps], location_description: str) -> LatLng:
    """Get the latitude and longitude of a location."""
    # 使用依赖中的HTTP客户端发送请求
    r = await ctx.deps.client.get(
        'https://demo-endpoints.pydantic.workers.dev/latlng',
        params={'location': location_description},
    )
    r.raise_for_status()
    return LatLng.model_validate_json(r.content)

# 定义获取天气的工具函数
@weather_agent.tool
async def get_weather(ctx: RunContext[Deps], lat: float, lng: float) -> dict[str, str]:
    """Get the weather at a location."""
    # 并行获取温度和天气描述
    temp_response, descr_response = await asyncio.gather(
        ctx.deps.client.get('https://demo-endpoints.pydantic.workers.dev/number', params={'min': 10, 'max': 30}),
        ctx.deps.client.get('https://demo-endpoints.pydantic.workers.dev/weather', params={'lat': lat, 'lng': lng}),
    )
    return {
        'temperature': f'{temp_response.text} °C',
        'description': descr_response.text,
    }

# 运行代理的主函数
async def main():
    async with AsyncClient() as client:
        logfire.instrument_httpx(client, capture_all=True)
        deps = Deps(client=client)
        result = await weather_agent.run(
            'What is the weather like in London and in Wiltshire?', deps=deps
        )
        print('Response:', result.output)

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

效果验证

运行上述代码后，我们可以通过Logfire监控面板观察代理的执行过程：

从监控面板中，我们可以清晰地看到代理如何调用get_lat_lng工具获取经纬度，然后调用get_weather工具获取天气信息，最终生成用户所需的回答。这种可视化的执行流程大大简化了调试和优化过程。

进阶拓展：如何构建实时聊天系统？

当我们掌握了基本代理开发后，自然会思考如何构建更复杂的交互系统。实时聊天应用就是一个典型案例，它需要处理持续的对话流和状态管理。

基于pydantic-ai和FastAPI，我们可以构建一个功能完善的实时聊天系统：

from fastapi import FastAPI, Depends, Request
from fastapi.responses import StreamingResponse
from pydantic_ai import Agent
from datetime import datetime, timezone
import json
from typing import AsyncGenerator

app = FastAPI()
# 创建聊天代理，使用gpt-4o模型
agent = Agent('openai:gpt-4o')

# 数据库依赖
async def get_db():
    # 实际应用中这里会返回数据库连接
    pass

# 聊天消息模型
class ModelResponse:
    def __init__(self, parts, timestamp):
        self.parts = parts
        self.timestamp = timestamp

class TextPart:
    def __init__(self, text):
        self.text = text

# 聊天接口
@app.post('/chat/')
async def post_chat(prompt: str, database=Depends(get_db)) -> StreamingResponse:
    async def stream_messages() -> AsyncGenerator[bytes, None]:
        # 流式传输用户输入
        yield json.dumps({
            'role': 'user',
            'timestamp': datetime.now(tz=timezone.utc).isoformat(),
            'content': prompt,
        }).encode('utf-8') + b'\n'
        
        # 获取聊天历史
        messages = await database.get_messages()
        
        # 运行代理并流式传输响应
        async with agent.run_stream(prompt, message_history=messages) as result:
            # 实时流式输出AI响应
            async for text in result.stream_output(debounce_by=0.01):
                m = ModelResponse(parts=[TextPart(text)], timestamp=result.timestamp())
                yield json.dumps({
                    'role': 'assistant',
                    'timestamp': m.timestamp.isoformat(),
                    'content': text
                }).encode('utf-8') + b'\n'
        
        # 保存新消息到数据库
        await database.add_messages(result.new_messages_json())
    
    return StreamingResponse(stream_messages(), media_type='text/plain')