探索pydantic Ai：提升python数据验证与ai集成的创新框架

Rifx.Online
Generative AI , Large Language Models , AI Applications
23 Feb, 2025

如果我问你 Pydantic 是什么，你可能已经知道它是 Python 的领先数据验证库。如果你不知道这一点或者之前没有听说过 Pydantic，我将在本文末尾留下一个链接，提供关于其功能和用例的良好概述。

你可能会问，为什么一个数据验证库背后的组织会在其最新项目 Pydantic AI 中涉足 AI 领域，这是一个代理 AI 框架。

这是个好问题，但你可能没有意识到，原始的 Pydantic 库已经深深嵌入到大多数领先的 LLM 提供商的当前 APIs 和框架中，包括 OpenAI、Ollama、Anthropic、CrewAI 等。

许多 LLM APIs 中最近的一个改进——称为 结构化输出——如果没有 Pydantic，LLM 提供商将更难以实现。

所以，我认为我问题的答案是，Pydantic 看到市场中存在一个他们相信自己独特能够填补的空白。

那么，为什么使用 Pydantic AI？

首先，Pydantic AI 背后的团队有着良好的背景。他们的 Pydantic 数据验证库的稳健性和实用性众所周知，它是许多现代 Python 框架的重要组成部分，包括备受欢迎的 FastAPI 网络前端库。

Pydantic AI 也是模型无关的。它开箱即用地支持 OpenAI、Gemini 和 Groq，并且即将支持 Anthropic。此外，它提供了一个简单的接口，可以与其他模型兼容。

该框架强调类型安全，并使用标准 Python 实现控制流和代理组合。这意味着您可以将传统软件开发中的熟悉 Python 最佳实践应用到您的 AI 项目中。

该系统包括由 Pydantic 提供支持的结构化响应验证，支持验证流式响应。它还引入了一种新颖的、类型安全的依赖注入机制，增强了基于测试和评估的迭代工作流程。

最后，Logfire 集成提供了强大的工具，用于调试、监控和优化基于 LLM 的应用程序的性能。

在本文的其余部分，我将通过代码查看使用 Pydantic AI 的具体示例。这样，您可以比较它如何实现代理行为与其他如 Anthropic、OpenAI 等的不同。

设置开发环境

注意：在撰写时，Pydantic AI 仍处于测试阶段。如果您在阅读本文时仍然如此，其 API 可能会发生变化。

好的，在我们开始编码之前，让我们设置我们的开发环境。最佳实践是创建一个单独的 Python 环境，在这个环境中您可以安装所需的任何软件并进行编码实验，知道您在此环境中所做的任何操作都不会影响系统的其他部分。

我使用 conda 来实现这一点，但您可以使用任何您认为最适合您的方法。请注意，我使用的是 Linux。

创建我们的测试环境

(base) $ conda create -n pydantic_ai_test python=3.12 -y

现在激活它

(base) $ conda activate pydantic_ai_test

现在我们的环境已经设置好，我们可以安装所需的库和软件。

(pydantic_ai_test) $ pip install pydantic-ai jupyter nest-asyncio pydantic httpx

现在在命令提示符中输入 jupyter notebook。你应该会看到一个 Jupyter Notebook 在你的浏览器中打开。如果没有自动打开，你可能会在执行 jupyter notebook 命令后看到一屏的信息，在底部会有一个 URL，你应该复制并粘贴到浏览器中以启动 Jupyter Notebook。

你的 URL 会与你的不同，但应该类似于以下内容：

http://127.0.0.1:8888/tree?token=3b9f7bd07b6966b41b68e2350721b2d0b6f388d248cc69da

在我所有的示例中，我将使用 GROQ 提供的 LLM 推理与 Pydantic AI 进行交互。

代码示例 1 — Pydantic AI 的 hello_world 等效

import nest_asyncio
nest_asyncio.apply()

from pydantic_ai import Agent
from pydantic_ai.models.groq import GroqModel

model = GroqModel('llama-3.3-70b-versatile')
agent = Agent(model)

result = agent.run_sync('When was the first computer bug found?')  
print(result.data)

这是我收到的输出。

第一个计算机错误是在 1947 年 9 月 9 日被发现的。它是一个真正的昆虫，卡在了哈佛 Mark II 计算机中。操作员在调试系统时发现了一只蛾子卡在一个继电器中。他们把这只蛾子贴在计算机日志上，并在旁边写下了“第一次实际发现错误案例”。

代码示例 2 — 获取航班信息

在这个更复杂的示例中，我们将使用 Pydantic AI 来让 LLM 模型为用户提供关于航班的一些结构化输出。要使用此代码，您需要从 https://aviationstack.com/ 获取一个 (免费) API 密钥。

所以，您现在可以去获取您的 API 密钥。我在写这篇文章时恰好知道航班 BA1455 正在从爱丁堡前往伦敦，因此这是我将要使用的示例航班号。

from pydantic_ai import Agent, RunContext
from httpx import AsyncClient
from dataclasses import dataclass
import os
import json

@dataclass
class FlightDeps:
    client: AsyncClient
    flight_api_key: str | None

定义模型和航班代理

flight_agent = Agent(
    model=GroqModel('llama-3.3-70b-versatile'),
    system_prompt="""
        You are a helpful flight assistant. If flight information is 
        available, provide it in the following form:

Flight IATA: {flight.iata}
        Flight Date: {flight_date}
        Departure Airport: {departure.airport} ({departure.iata})
        Departure Time: {departure.scheduled_time}
        Arrival Airport: {arrival.airport} ({arrival.iata})
        Arrival Time: {arrival.scheduled_time}

If no information is available, respond with: "No information found 
        for this flight."
    """,
    deps_type=FlightDeps,
)

@flight_agent.tool
async def get_flight_info(ctx: RunContext[FlightDeps], flight_iata: str) -> dict:
    """
    使用 AviationStack API 检索航班信息。
    """
    if ctx.deps.flight_api_key is None:
        return {'error': '未提供 API 密钥'}

    url = "https://api.aviationstack.com/v1/flights"
    params = {
        "access_key": ctx.deps.flight_api_key,
        "flight_iata": flight_iata,
    }

    try:
        response = await ctx.deps.client.get(url, params=params)
        response.raise_for_status()

        if not response.text.strip():  # 检查空响应
            return {"error": "API 返回了空响应"}

        data = response.json()  # 解析 JSON
        if data.get("data"):
            flight_data = data["data"][0]  # 使用第一个航班记录
            return {
                "flight_date": flight_data.get("flight_date"),
                "flight_status": flight_data.get("flight_status"),
                "departure": {
                    "airport": flight_data["departure"].get("airport"),
                    "iata": flight_data["departure"].get("iata"),
                    "scheduled_time": flight_data["departure"].get("scheduled"),
                },
                "arrival": {
                    "airport": flight_data["arrival"].get("airport"),
                    "iata": flight_data["arrival"].get("iata"),
                    "scheduled_time": flight_data["arrival"].get("scheduled"),
                },
                "airline": {
                    "name": flight_data["airline"].get("name"),
                    "iata": flight_data["airline"].get("iata"),
                },
                "flight": {
                    "number": flight_data["flight"].get("number"),
                    "iata": flight_data["flight"].get("iata"),
                },
            }
        else:
            return {"error": "未找到给定航班号的数据"}
    except Exception as e:
        return {"error": f"API 错误: {str(e)}"}

执行航班代理的主函数

async def main():
    """
    主函数，用于运行航班代理。
    """
    async with AsyncClient() as client:
        # 将'your_api_key_here'替换为您的AviationStack API 密钥
        deps = FlightDeps(client, "YOUR_AVIATION_STACK_API_KEY")
        flight_iata = "BA1455"  # 示例航班号
        result = await flight_agent.run(f"检查航班状态：{flight_iata}。", deps=deps)

        # 调试原始结果数据
        print(result.data)

处理和显示航班信息

try:
    flight_info = result.data  # Treat result.data as plain text

    # Check if an error occurred
    if "No information found" in flight_info or "error" in flight_info:
        print("Error:", flight_info)

except Exception as e:
    print("An unexpected error occurred:", e)

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

输出为：

Flight IATA: BA1455
Flight Date: 2024-12-09
Departure Airport: Edinburgh (EDI)
Departure Time: 2024-12-09T16:10:00+00:00
Arrival Airport: Heathrow (LHR)
Arrival Time: 2024-12-09T17:40:00+00:00

为了证明模型没有产生幻觉，我尝试输入一个虚假的航班号，例如 BA98765，并收到了以下输出。

No information found for this flight.
Error: No information found for this flight.

代码示例 3 — 网络搜索

在这个例子中，我将使用 Bing 搜索 API 在网上搜索指定主题。然后，将使用 LLM 来结构化 Bing 返回的输出。要像这样使用 Bing，您需要一个 API 密钥。

您可以通过 Microsoft Azure（Bing Search V7）进行设置。

如果您选择最低可用的 “F” 级别，可能会安排为无成本选项。然而，这限制了每秒搜索次数为 3，并限制了每月的总搜索调用次数。

设置这个有点复杂，但基本上，您需要按照以下步骤进行：

如果您还没有 Microsoft Azure 账号，请注册一个免费的 Microsoft Azure 账号。
在 Azure 门户中创建一个 Bing 搜索资源；确保选择最低的 F 级别，这个级别是免费的，但如上所述有点受限。
从资源概览中获取您的 API 密钥。

在代码中，我要求 Bing 搜索有关 OpenAI 最近发布的最新 SORA 文本到视频模型的信息。

from pydantic_ai import Agent, RunContext
from httpx import AsyncClient
from dataclasses import dataclass
import asyncio
from datetime import datetime
from pydantic_ai.models.groq import GroqModel
from pydantic import BaseModel, ValidationError
from typing import List

@dataclass
class Deps:
    client: AsyncClient
    bing_api_key: str | None

class SearchResult(BaseModel):
    website_url: str
    content_summary: str

class SearchResults(BaseModel):
    results: List[SearchResult]

web_search_agent = Agent(
    model=GroqModel('llama-3.3-70b-versatile'),
    system_prompt=f"""
        You are an expert at summarizing search results. 
        Always provide the response as JSON in the following format:

{{
            "results": [
                {{
                    "website_url": "string",
                    "content_summary": "string"
                }},
                ...
            ]
        }}
        Summarize the content from the search results provided, ensuring full sentences and cohesive summaries.
        If no valid results are provided, return:
        {{
            "results": []
        }}
        The current date is: {datetime.now().strftime("%Y-%m-%d")}
    """,
    deps_type=Deps,
    retries=2
)

@web_search_agent.tool
async def search_web(ctx: RunContext[Deps], web_query: str) -> str:
    """
    Perform a web search and return the raw results as a formatted string for LLM processing.
    """

    if ctx.deps.bing_api_key is None:
        return "No API key provided. Cannot perform search."

    headers = {
        "Ocp-Apim-Subscription-Key": ctx.deps.bing_api_key,
        'Accept': 'application/json',
    }

    r = await ctx.deps.client.get(
        'https://api.bing.microsoft.com/v7.0/search',
        params={
            'q': web_query,
            'count': 5,
            'text_decorations': True,
            'text_format': 'Raw',
            'search_lang': 'en'
        },
        headers=headers
    )
    r.raise_for_status()
    data = r.json()

    web_results = data.get('webPages', {}).get('value', [])
    formatted_results = []
    for item in web_results[:5]:  
        title = item.get('name', 'No title provided')
        description = item.get('snippet', 'No description available')
        url = item.get('url', 'No URL provided')
        formatted_results.append(f"Title: {title}\\nDescription: {description}\\nURL: {url}\\n")

    return "\\n".join(formatted_results)

async def main():
    async with AsyncClient() as client:

        deps = Deps(client=client, bing_api_key='YOUR_BING_API_KEY')

        search_query = """Search the web for articles talking about the new release of SORA by OpenAI."""
        search_results = await search_web(RunContext(deps=deps, retry=None), search_query)

        summarization_query = f"""
            Based on the following search results, summarize the key points about SORA by OpenAI in a cohesive manner:

{search_results}
        """

        result = await web_search_agent.run(summarization_query, deps=deps)

        try:
            structured_response = SearchResults.model_validate_json(result.data)
            print("结构化响应:")
            print(structured_response.model_dump_json(indent=2))
        except ValidationError as e:
            print("来自 LLM 的无效结构化响应:")
            print(e)
            print("原始响应:", result.data)

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

我收到的输出如下。

结构化响应:

{
  "results": [
    {
      "website_url": "https://openai.com/index/sora-is-here/",
      "content_summary": "OpenAI's Sora is a video generator that provides transparency through C2PA metadata and visible watermarks by default, and has an internal search tool to help verify the origin of generated content."
    },
    {
      "website_url": "https://techcrunch.com/2024/12/09/openais-sora-is-launching-today-heres-highlights-from-the-first-review/",
      "content_summary": "Sora, OpenAI's video generator, is launching and will be available to some users, with YouTuber Marques Brownlee being one of the first to review it and share his experience with the tool."
    },
    {
      "website_url": "https://www.theverge.com/2024/12/9/24317090/openai-sora-text-to-video-ai-generator-confirmed-release",
      "content_summary": "OpenAI's Sora, a highly-anticipated AI text-to-video generator, is being released to the public, with Marques Brownlee confirming its launch and sharing details about the tool in a video."
    },
    {
      "website_url": "https://www.theatlantic.com/technology/archive/2024/12/openai-sora-release/680944/",
      "content_summary": "Sora, OpenAI's video generator, is generating significant hype and is expected to have a major impact, with early-access users already sharing their generated videos on social media and a large number of users expected to follow suit."
    },
    {
      "website_url": "https://www.latimes.com/entertainment-arts/business/story/2024-12-09/openais-controversial-text-to-video-tool-sora-is-widely-released",
      "content_summary": "OpenAI's Sora, a controversial text-to-video tool, is being released to the broader public, with consumers in the US able to use it with a ChatGPT Plus subscription, and the tool facing its next big test as it becomes widely available."
    }
  ]
}