Type something to search...
to navigateto selectESC to close
构建高效ai应用的利器：探索pydanticai框架和代理构建工具

构建高效ai应用的利器：探索pydanticai框架和代理构建工具

PydanticAI

PydanticAI 是一个 Python Agent 框架，旨在减少使用生成式 AI 构建生产级应用程序时的痛苦。

FastAPI 通过提供创新和符合人体工程学的设计，彻底改变了 web 开发，建立在 Pydantic 的基础上。

同样，几乎每个 Python 中的 agent 框架和 LLM 库都使用 Pydantic，但当我们开始在 Pydantic Logfire 中使用 LLM 时，我们找不到任何能给我们带来相同感觉的东西。

PydanticAI 的构建目标很简单：将这种 FastAPI 的感觉带入 GenAI 应用程序开发中。

为什么使用 PydanticAI

由 Pydantic 团队构建
由 Pydantic 背后的团队构建（OpenAI SDK、Anthropic SDK、LangChain、LlamaIndex、AutoGPT、Transformers、CrewAI、Instructor 及更多的验证层）。

模型无关
支持 OpenAI、Anthropic、Gemini、Ollama、Groq 和 Mistral，并且有一个简单的接口来实现对 其他模型 的支持。

Pydantic Logfire 集成
与 Pydantic Logfire 无缝 集成，实现对您的 LLM 驱动应用程序的实时调试、性能监控和行为跟踪。

类型安全
旨在使类型检查对您尽可能有用，因此它与静态类型检查器（如 mypy 和 pyright）良好 集成。

以 Python 为中心的设计
利用 Python 熟悉的控制流和代理组合来构建您的 AI 驱动项目，使您能够轻松应用在任何其他（非 AI）项目中使用的标准 Python 最佳实践。

结构化响应
利用 Pydantic 的强大功能来 验证和结构化 模型输出，确保响应在多次运行中保持一致。

依赖注入系统
提供一个可选的 依赖注入 系统，为您的代理的 系统提示、工具 和 结果验证器 提供数据和服务。这对于测试和评估驱动的迭代开发非常有用。

流式响应
提供连续 流式 LLM 输出的能力，进行即时验证，确保快速和准确的结果。

模型

PydanticAI 是 Model-agnostic，内置支持以下模型提供者：

您还可以 添加对其他模型的支持。

PydanticAI 还配备了 TestModel 和 FunctionModel 用于测试和开发。

要使用每个模型提供者，您需要配置本地环境，并确保安装了正确的包。

一些代码示例：

让我解释这段代码，它演示了 PydanticAI 的基本用法，这是一个构建 LLM 驱动应用程序的框架：

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = None

item = Item(name="Sample Item", price=19.99, is_offer=True)
print(item)

解释

  1. BaseModel: 这是 Pydantic 提供的用于创建数据模型的基类。
  2. Item 类: 一个表示项目的模型，具有三个属性：name、price 和 is_offer。
  3. 创建实例: 使用示例数据创建 Item 类的一个实例。
  4. 打印项目: 创建的项目实例被打印到控制台。

示例 — 1:

  1. 导入和设置:
from pydantic import BaseModel, Field
from pydantic_ai import Agent, ModelRetry, RunContext, Tool
from pydantic_ai.models.openai import OpenAIModel
  • 从 Pydantic 和 PydanticAI 导入必要的组件
  • nest_asyncio.apply() 在 Jupyter 笔记本中启用异步操作
  • load_dotenv() 加载环境变量（可能包含 API 密钥）
  1. 模型初始化:
model = OpenAIModel("gpt-4o-mini")

创建 OpenAI 模型接口的实例

  1. 基本代理创建:
agent = Agent(  
    model,
    system_prompt='Be concise, reply with one sentence.',  
)

创建一个简单的代理，具有：

  • 指定的 OpenAI 模型
  • 指示代理简洁回复的系统提示
  1. 代理执行:
result = agent.run_sync('Where does "hello world" come from?')
print(result.data)
  • 使用问题同步运行代理
  • 打印响应

输出:

Image 3

关键特性:

  1. 类型安全: 使用 Pydantic 进行类型验证和安全性
  2. 结构化响应: 通过 Pydantic 模型启用结构化输出
  3. 生产就绪: 设计用于生产使用，具有适当的错误处理
  4. 异步支持: 支持同步和异步操作

这是一个最小示例，演示了 PydanticAI 的基本设置和用法，展示了如何创建一个可以处理查询并返回结构化响应的简单代理。该框架特别适用于构建具有类型安全和结构化数据处理的强大 AI 应用程序。

示例 — 2:

""" 本示例演示了 PydanticAI 代理的基本用法。
关键概念：

  • 创建一个带有系统提示的基本代理
  • 运行同步查询
  • 访问响应数据、消息历史和成本 """
agent1 = Agent(
    model=model,
    system_prompt="You are a helpful customer support agent. Be concise and friendly.",
)

response = agent1.run_sync("How can I track my order #12345?")
print(response.data)

输出:

示例 — 3：

此示例展示了如何从代理获取结构化的、类型安全的响应。
关键概念：

  • 使用 Pydantic 模型定义响应结构
  • 类型验证和安全性
  • 字段描述以便更好地理解模型
class ResponseModel(BaseModel):
    """Structured response with metadata."""

    response: str
    needs_escalation: bool
    follow_up_required: bool
    sentiment: str = Field(description="Customer sentiment analysis")

agent2 = Agent(
    model=model,
    result_type=ResponseModel,
    system_prompt=(
        "You are an intelligent customer support agent. "
        "Analyze queries carefully and provide structured responses."
    ),
)

response = agent2.run_sync("How can I track my order #12345?")
print(response.data.model_dump_json(indent=2))

  1. Pydantic 模型定义：

    使用 Pydantic 的 BaseModel 创建一个结构化响应模型

    定义四个具有特定类型的字段：

    • response: 包含答案的字符串
    • needs_escalation: 用于升级的布尔标志
    • follow_up_required: 用于跟进的布尔标志
    • sentiment: 用于情感分析的字符串

  2. 高级代理创建：

    这里我们创建了 ‘agent2’

  3. 响应处理：

    这里我们创建了 ‘response’。

    • 执行查询并将响应格式化为 JSON
    • 输出将结构化为：
    {
  "response": "You can track your order #12345 by...",
  "needs_escalation": false,
  "follow_up_required": true,
  "sentiment": "neutral"
}

4. 输出：

示例 — 4:

""" 本示例演示了如何在代理中使用依赖关系和上下文。
关键概念：

  • 使用 Pydantic 定义复杂数据模型
  • 注入运行时依赖关系
  • 使用动态系统提示
    """
class Order(BaseModel):
    """Structure for order details."""

    order_id: str
    status: str
    items: List[str]

class CustomerDetails(BaseModel):
    """Structure for incoming customer queries."""

    customer_id: str
    name: str
    email: str
    orders: Optional[List[Order]] = None

class ResponseModel(BaseModel):
    """Structured response with metadata."""

    response: str
    needs_escalation: bool
    follow_up_required: bool
    sentiment: str = Field(description="Customer sentiment analysis")

agent5 = Agent(
    model=model,
    result_type=ResponseModel,
    deps_type=CustomerDetails,
    retries=3,
    system_prompt=(
        "You are an intelligent customer support agent. "
        "Analyze queries carefully and provide structured responses. "
        "Always greet the customer and provide a helpful response."
    ),  
)

@agent5.system_prompt
async def add_customer_name(ctx: RunContext[CustomerDetails]) -> str:
    return f"Customer details: {to_markdown(ctx.deps)}"

customer = CustomerDetails(
    customer_id="1",
    name="John Doe",
    email="[email protected]",
    orders=[
        Order(order_id="12345", status="shipped", items=["Blue Jeans", "T-Shirt"]),
    ],
)

response = agent5.run_sync(user_prompt="What did I order?", deps=customer)

response.all_messages()
print(response.data.model_dump_json(indent=2))

print(
    "Customer Details:\n"
    f"Name: {customer.name}\n"
    f"Email: {customer.email}\n\n"
    "Response Details:\n"
    f"{response.data.response}\n\n"
    "Status:\n"
    f"Follow-up Required: {response.data.follow_up_required}\n"
    f"Needs Escalation: {response.data.needs_escalation}"
)
  1. 数据模型定义：
  • 定义三个 Pydantic 模型以处理结构化数据
  • 创建模型层次结构（CustomerDetails 包含 Orders） — 类对象（Order、CustomerDetails、ResponseModel）

2. 高级代理配置：

  • 指定响应类型和依赖类型
  • 包含重试逻辑（3 次尝试）
  • 设置基本系统提示

3. 动态系统提示：

  • 为系统提示添加运行时上下文
  • 使用装饰器模式进行动态提示
  • 将客户详情转换为 markdown 格式

4. 响应处理：

  • 打印格式化的 JSON 响应
  • 显示客户详情和响应状态

5. 输出：

  1. 依赖注入：通过 deps 提供运行时上下文
  2. 复杂数据建模：具有关系的嵌套模型
  3. 动态提示：上下文感知的系统提示
  4. 类型安全：全面的类型检查
  5. 错误处理：用于可靠性的重试机制

示例 — 5:

此示例展示了如何通过自定义工具增强代理。
关键概念：

  • 创建和注册工具
  • 在工具中访问上下文
shipping_info_db: Dict[str, str] = {
    "12345": "Shipped on 2024-12-01",
    "67890": "Out for delivery",
}

class Order(BaseModel):
    """订单详情结构。"""

    order_id: str
    status: str
    items: List[str]

class CustomerDetails(BaseModel):
    """传入客户查询的结构。"""

    customer_id: str
    name: str
    email: str
    orders: Optional[List[Order]] = None

class ResponseModel(BaseModel):
    """带有元数据的结构化响应。"""

    response: str
    needs_escalation: bool
    follow_up_required: bool
    sentiment: str = Field(description="客户情感分析")

customer = CustomerDetails(
    customer_id="1",
    name="John Doe",
    email="[email protected]",
    orders=[
        Order(order_id="12345", status="shipped", items=["Blue Jeans", "T-Shirt"]),
    ],
)

def get_shipping_info(ctx: RunContext[CustomerDetails]) -> str:
    """获取客户的运输信息。"""
    return shipping_info_db[ctx.deps.orders[0].order_id]

agent5 = Agent(
    model=model,
    result_type=ResponseModel,
    deps_type=CustomerDetails,
    retries=3,
    system_prompt=(
        "您是一个智能客户支持代理。"
        "仔细分析查询并提供结构化响应。"
        "使用工具查找相关信息。"
        "始终问候客户并提供有帮助的响应。"
    ),  
    tools=[Tool(get_shipping_info, takes_ctx=True)],  
)

@agent5.system_prompt
async def add_customer_name(ctx: RunContext[CustomerDetails]) -> str:
    return f"客户详情: {to_markdown(ctx.deps)}"

response = agent5.run_sync(
    user_prompt="我最后一个订单的状态是什么？", deps=customer
)

response.all_messages()
print(response.data.model_dump_json(indent=2))

print(
    "客户详情:\n"
    f"姓名: {customer.name}\n"
    f"邮箱: {customer.email}\n\n"
    "响应详情:\n"
    f"{response.data.response}\n\n"
    "状态:\n"
    f"需要跟进: {response.data.follow_up_required}\n"
    f"需要升级: {response.data.needs_escalation}"
)

  1. 数据库模拟:

    • 模拟一个简单的运输信息数据库

  2. 数据模型（与上一个示例相同）:

  3. 自定义工具定义：（这里 — get_shipping_info）

    • 创建一个查找运输信息的工具
    • 接受上下文参数以访问客户详情
    • 返回运输状态字符串

  4. 增强的代理配置:

    • 通过 tools 参数添加工具
    • 指定工具需要上下文（takes_ctx=True）

  5. 动态系统提示:

    • add_customer_name 方法负责接受动态系统提示。

  6. 输出:

示例 — 6：

此示例展示了具有自我纠正能力的高级代理功能。

关键概念：

  • 实现自我反思
  • 优雅地处理错误并进行重试
  • 使用 ModelRetry 进行自动重试
  • 基于装饰器的工具注册
shipping_info_db: Dict[str, str] = {
    "#12345": "Shipped on 2024-12-01",
    "#67890": "Out for delivery",
}

class Order(BaseModel):
    """Structure for order details."""
    order_id: str
    status: str
    items: List[str]

class ResponseModel(BaseModel):
    """Structured response with metadata."""
    response: str
    needs_escalation: bool
    follow_up_required: bool
    sentiment: str = Field(description="Customer sentiment analysis")

class CustomerDetails(BaseModel):
    """Structure for incoming customer queries."""
    customer_id: str
    name: str
    email: str
    orders: Optional[List[Order]] = None

customer = CustomerDetails(
    customer_id="1",
    name="John Doe",
    email="[email protected]",
    orders=[
        Order(order_id="12345", status="shipped", items=["Blue Jeans", "T-Shirt"]),
    ],
)

agent5 = Agent(
    model=model,
    result_type=ResponseModel,
    deps_type=CustomerDetails,
    retries=3,
    system_prompt=(
        "You are an intelligent customer support agent. "
        "Analyze queries carefully and provide structured responses. "
        "Use tools to look up relevant information. "
        "Always greet the customer and provide a helpful response."
    ),
)

@agent5.tool_plain()  
def get_shipping_status(order_id: str) -> str:
    """Get the shipping status for a given order ID."""
    shipping_status = shipping_info_db.get(order_id)
    if shipping_status is None:
        raise ModelRetry(
            f"No shipping information found for order ID {order_id}. "
            "Make sure the order ID starts with a #: e.g, #624743 "
            "Self-correct this if needed and try again."
        )
    return shipping_info_db[order_id]

response = agent5.run_sync(
    user_prompt="What's the status of my last order 12345?", deps=customer
)

response.all_messages()
print(response.data.model_dump_json(indent=2))

关键特性：

  1. 自我纠正机制：

    • 使用 ModelRetry 进行自动重试
    • 提供有助于自我纠正的错误信息
    • 验证订单 ID 格式

  2. 基于装饰器的工具：

    • 使用 @agent5.tool_plain() 装饰器
    • 简化工具注册
    • 提供清晰的工具定义语法

  3. 错误处理：

    • 优雅地处理缺失数据
    • 自动重试机制
    • 提供信息丰富的错误信息

  4. 输入验证：

    • 检查正确的订单 ID 格式
    • 提供纠正指导
    • 维护数据完整性

  5. 输出：

Image 8

在这里，我们探索了使用 PydanticAI 框架的不同类型的代理。完全归功于 (daveebbelaar)。

参考文献:

  1. https://ai.pydantic.dev/
  2. https://github.com/daveebbelaar/pydantic-ai-tutorial/blob/main/src/introduction.py
Tags :
Share :

Related Posts

结合chatgpt-o3-mini与perplexity Deep Research的3步提示：提升论文写作质量的终极指南

结合chatgpt-o3-mini与perplexity Deep Research的3步提示：提升论文写作质量的终极指南

AI 研究报告和论文写作 合并两个系统指令以获得两个模型的最佳效果 Perplexity AI 的 Deep Research 工具提供专家级的研究报告，而 OpenAI 的 ChatGPT-o3-mini-high 擅长推理。我发现你可以将它们结合起来生成令人难以置信的论文，这些论文比任何一个模型单独撰写的都要好。你只需要将这个一次性提示复制到 **

阅读更多
让 Excel 过时的 10 种 Ai 工具：实现数据分析自动化，节省手工作业时间

让 Excel 过时的 10 种 Ai 工具：实现数据分析自动化，节省手工作业时间

Non members click here作为一名软件开发人员，多年来的一个发现总是让我感到惊讶，那就是人们还在 Excel

阅读更多
使用 ChatGPT 搜索网络功能的 10 种创意方法

使用 ChatGPT 搜索网络功能的 10 种创意方法

例如，提示和输出 你知道可以使用 ChatGPT 的“搜索网络”功能来完成许多任务，而不仅仅是基本的网络搜索吗？ 对于那些不知道的人，ChatGPT 新的“搜索网络”功能提供实时信息。 截至撰写此帖时，该功能仅对使用 ChatGPT 4o 和 4o-mini 的付费会员开放。 ![](https://images.weserv.nl/?url=https://cdn-im

阅读更多
掌握Ai代理：解密Google革命性白皮书的10个关键问题解答

掌握Ai代理：解密Google革命性白皮书的10个关键问题解答

10 个常见问题解答 本文是我推出的一个名为“10 个常见问题解答”的新系列的一部分。在本系列中，我旨在通过回答关于该主题的十个最常见问题来分解复杂的概念。我的目标是使用简单的语言和相关的类比，使这些想法易于理解。 图片来自 [Solen Feyissa](https://unsplash.com/@solenfeyissa?utm_source=medium&utm_medi

阅读更多
在人工智能和技术领域保持领先地位的 10 项必学技能 📚

在人工智能和技术领域保持领先地位的 10 项必学技能 📚

在人工智能和科技这样一个动态的行业中，保持领先意味着不断提升你的技能。无论你是希望深入了解人工智能模型性能、掌握数据分析，还是希望通过人工智能转变传统领域如法律，这些课程都是你成功的捷径。以下是一个精心策划的高价值课程列表，可以助力你的职业发展，并让你始终处于创新的前沿。 1. 生成性人工智能简介课程: [生成性人工智能简介](https://genai.works

阅读更多
揭开真相！深度探悉DeepSeek AI的十大误区，您被误导了吗？

揭开真相！深度探悉DeepSeek AI的十大误区，您被误导了吗？

在AI军备竞赛中分辨事实与虚构 DeepSeek AI真的是它所宣传的游戏规则改变者，还是仅仅聪明的营销和战略炒作？👀 虽然一些人将其视为AI效率的革命性飞跃，但另一些人则认为它的成功建立在借用（甚至窃取的）创新和可疑的做法之上。传言称，DeepSeek的首席执行官在疫情期间像囤积卫生纸一样囤积Nvidia芯片——这只是冰山一角。 从其声称的550万美元培训预算到使用Open

阅读更多
Type something to search...
to navigateto selectESC to close