如何在10分钟内向 PydanticAI 代理添加工具

发布: (2026年3月16日 GMT+8 05:01)
4 分钟阅读
原文: Dev.to

Source: Dev.to

代码

import httpx
from dataclasses import dataclass
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

@dataclass
class Deps:
    http_client: httpx.Client

class CityWeather(BaseModel):
    city: str
    temperature_f: float
    summary: str
    recommendation: str

agent = Agent(
    "openai:gpt-4o-mini",
    deps_type=Deps,
    output_type=CityWeather,
    instructions="You help users check weather conditions. Use the tools provided to fetch real data before answering.",
)

@agent.tool
def get_coordinates(ctx: RunContext[Deps], city: str) -> str:
    """Get latitude and longitude for a city."""
    response = ctx.deps.http_client.get(
        "https://geocoding-api.open-meteo.com/v1/search",
        params={"name": city, "count": 1},
    )
    data = response.json()
    if not data.get("results"):
        return f"City '{city}' not found."
    result = data["results"][0]
    return f"{result['name']}: lat={result['latitude']}, lon={result['longitude']}"

@agent.tool
def get_weather(ctx: RunContext[Deps], latitude: float, longitude: float) -> str:
    """Get current weather for coordinates."""
    response = ctx.deps.http_client.get(
        "https://api.open-meteo.com/v1/forecast",
        params={
            "latitude": latitude,
            "longitude": longitude,
            "current": "temperature_2m,wind_speed_10m",
            "temperature_unit": "fahrenheit",
        },
    )
    data = response.json()["current"]
    return f"Temperature: {data['temperature_2m']}F, Wind: {data['wind_speed_10m']} km/h"

result = agent.run_sync(
    "What's the weather like in Tokyo?",
    deps=Deps(http_client=httpx.Client()),
)

print(result.output.city)
print(f"{result.output.temperature_f}F")
print(result.output.summary)
print(result.output.recommendation)

安装并运行

pip install pydantic-ai httpx
export OPENAI_API_KEY="sk-..."
python weather_agent.py

How It Works

Define dependencies with a dataclass.
Deps 类保存你的 HTTP 客户端。PydanticAI 通过 RunContext 将其注入到每一次工具调用中——没有全局变量,也没有单例。可以在测试时换成 mock 客户端,工具的行为保持一致。

Set the output type with a Pydantic model.
CityWeather 定义了代理响应的精确结构。PydanticAI 强制 LLM 返回符合该 schema 的数据,直接得到一个已类型化的 Python 对象——而不是需要自行解析的字符串。

Register tools with @agent.tool.
每个被装饰的函数都会成为 LLM 可以调用的工具。函数签名会转化为工具的参数 schema,函数的 docstring 会成为工具的描述——请写清楚,以便模型知道 何时 调用它。

The agent chains tools automatically.
询问 “What’s the weather in Tokyo?” 时,代理会先调用 get_coordinates 获取纬度/经度,然后再使用这些坐标调用 get_weather。你不需要编写这些编排逻辑——LLM 会根据工具描述自行推断调用顺序。

您将看到

Tokyo
58.4F
Mild and calm conditions with light winds.
Light jacket weather -- comfortable for walking around the city.

响应是一个已验证的 CityWeather 对象。可直接访问 .city.temperature_f.summary.recommendation——无需 JSON 解析,也无需字符串提取。

关键细节

  • Docstrings 很重要。 LLM 使用你的工具的 docstring 来决定何时调用它。像 “获取城市的纬度和经度” 这样的清晰描述可以正确引导模型。
  • 类型提示驱动 schema。 city: strlatitude: float 会成为工具的 JSON schema。PydanticAI 在函数运行前验证 LLM 的参数——无需额外的类型检查代码。
  • 依赖使工具可测试。 与其在 get_weather 中创建 HTTP 客户端,不如通过 ctx.deps 传入。在测试时,传入返回固定响应的 mock 客户端。这种模式对 FastAPI 开发者来说很熟悉。

进一步

  • Agent() 构造函数中添加 retries=3,以在验证失败时自动重试。
  • 使用 async 工具函数与 httpx.AsyncClient 进行并发 API 调用。
  • agent.run_stream() 结合,实现实时流式响应。

查看 AI Agent Quick Tips 系列的其他文章:

需要进行工具编排、调度和记忆而不想处理基础设施工作吗?Nebula 为您处理,让您专注于工具。

0 浏览
Back to Blog

相关文章

阅读更多 »