使用 PydanticAI 框架快速构建 Multi-Agent 系统 - AI Agent 协作触手可及

from pydantic_ai import Agent

agent = Agent(  
    'gemini-1.5-flash',
    system_prompt='Be concise, reply with one sentence.',  
)

result = agent.run_sync('Where does "hello world" come from?')  
print(result.data)
"""
The first known use of "hello, world" was in a 1974 textbook about the C programming language.
"""

Pydantic

from datetime import date
from pydantic import BaseModel
classUser(BaseModel):
    id: int
    name: str
    dob: date

user = User(id='123', name='Samuel Colvin', dob='1987-01-28')
#> User(id=123, name='Samuel Colvin', dob=date(1987, 1, 28))
user = User.model_validate_json('{"id: 123, "name": "Samuel Colvin", "dob": "1987-01-28"}')
#> User(id=123, name='Samuel Colvin', dob=date(1987, 1, 28))
print(User.model_json_schema())
s = {
    'properties': {
        'id': {'title': 'Id', 'type': 'integer'},
        'name': {'title': 'Name', 'type': 'string'},
        'dob': {'format': 'date', 'title': 'Dob', 'type': 'string'},
    },
    'required': ['id', 'name', 'dob'],
    'title': 'User',
    'type': 'object',
}

GenAI 中的 Pydantic

from datetime import date
from pydantic import BaseModel
from openai import OpenAI
classUser(BaseModel):
    """Definition of a user"""
    id: int
    name: str
    dob: date
response = OpenAI().chat.completions.create(
    model='gpt-4o',
    messages=[
        {'role': 'system', 'content': 'Extract information about the user'},
        {'role': 'user', 'content': 'The user with ID 123 is called Samuel, born on Jan 28th 87'}
    ],
    tools=[
        {
            'function': {
                'name': User.__name__,
                'description': User.__doc__,
                'parameters': User.model_json_schema(),
            },
            'type': 'function'
        }
    ]
)
user = User.model_validate_json(response.choices[0].message.tool_calls[0].function.arguments)
print(user)

PydanticAI

同样的例子还有 PydanticAI - 用于生产的Agent框架。

from datetime import date
from pydantic_ai import Agent
from pydantic import BaseModel
classUser(BaseModel):
    """Definition of a user"""
    id: int
    name: str
    dob: date
agent = Agent(
    'openai:gpt-4o',
    result_type=User,
    system_prompt='Extract information about the user',
)
result = agent.run_sync('The user with ID 123 is called Samuel, born on Jan 28th 87')
print(result.data)

假设你正在制作一个应用，用户可以提交姓名、年龄和电子邮件。你希望确保

• • 名称是一个字符串。
• • 年龄只是一个数字。
• • 电子邮件格式有效。

Pydantic 如何轻松做到这一点：

from pydantic import BaseModel, EmailStr
# Define the model
classUser(BaseModel):
    name: str
    age: int
    email: EmailStr
# Example input
user_data = {
    "name": "Alice",
    "age": 25,
    "email": ""
}
# Validate the input
user = User(**user_data)
print(user.name)  # Alice
print(user.age)   # 25
print(user.email) # 

如果用户提交的数据无效（如 "年龄"："25"），Pydantic 会自动出错：

user_data = {
    "name": "Alice",
    "age": "twenty-five",  # Invalid
    "email": ""
}
user = User(**user_data)
# Error: value is not a valid integer

Pydantic 在部署过程中起着至关重要的作用，因为它是必须遵守的，因为

由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 熟悉的控制流和Agent组成来构建人工智能驱动的项目，从而轻松应用标准 Python 最佳实践，您在任何其他项目（非人工智能项目）中都会用到这些实践。

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

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

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

其目的是解决 Agent 框架中的一个主要问题：缺乏专为生产用例设计的工具。

安装 PydanticAI

需要 Python 3.9 及以上版本

pip install pydantic-ai

这将安装 pydantic_ai 包、核心依赖项以及使用 PydanticAI 中包含的所有模型所需的库。如果您想使用某个特定模型，可以安装 "slim "版本的 PydanticAI。

PydanticAI 的 8 个重要组成部分

• • Agents
• • Models （模型）
• • Dependencies （依赖）
• • Function Tools  • 函数工具
• • Results  • 结果
• • 信息和聊天记录
• • 测试和评估
• • 调试和监控

Agents

Agent 是 PydanticAI 与 LLM 交互的主要媒介。

在某些使用案例中，单 Agent 可以控制整个应用程序或组件，但multi-agents也可以进行交互，以体现更复杂的工作流程。

Agent 类有完整的应用程序接口文档，但从概念上讲，你可以把 Agent 看作是一个容器

运行Agent有三种方法：

• • agent.run() - 返回包含已完成响应的 RunResult 的例程
• • loop.run_until_complete(self.run()))
• • agent.run_stream() - 返回 StreamedRunResult 的例行程序，其中包含将响应作为异步可迭代流的方法

下面是一个简单的例子，展示了这三点：

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o')

result_sync = agent.run_sync('What is the capital of Italy?')
print(result_sync.data)
#> Rome


asyncdefmain():
    result = await agent.run('What is the capital of France?')
    print(result.data)
    #> Paris

    asyncwith agent.run_stream('What is the capital of the UK?') as response:
        print(await response.get_data())
        #> London

模型

PydanticAI 与模型是解耦的，内置了对以下模型的支持：

• • OpenAI
• • Anthropic
• • Gemini 通过两个不同的应用程序接口：Generative Language API 和 VertexAI API
• • Ollama
• • Groq
• • Mistral

你还可以自定义对其他模型的支持。

PydanticAI 还提供用于测试和开发的 TestModel 和 FunctionModel。

要使用每个模型提供商，你需要配置本地环境，并确保安装了正确的包。需要要使用 GeminiModel 模型，只需安装 pydantic-ai 或 pydantic-ai-slim，无需额外的依赖项。

GeminiModel 可让您通过生成语言 API（generativelanguage.googleapis.com）使用谷歌的Gemini模型。

GeminiModelName 包含可通过此接口使用的可用Gemini模型列表。

要使用 GeminiModel，请访问 aistudio.google.com，然后按照提示找到生成 API 密钥的地方。

获得 API 密钥后，可以将其设置为环境变量：

from pydantic_ai import Agent
from pydantic_ai.models.gemini import GeminiModel
model = GeminiModel('gemini-1.5-flash', api_key=os.environ['GEMINI_API_KEY'])
agent = Agent(model)

依赖关系

PydanticAI 使用依赖注入系统为你的Agent的system prompt、工具和结果验证器提供数据和服务。

与 PydanticAI 的设计理念相匹配的是，依赖系统尝试使用 Python 开发中现有的最佳实践，而不是发明深奥晦涩的东西。

依赖关系可以是任何 python 类型。在简单的情况下，你可以传递单个对象作为依赖关系（如 HTTP 连接），但当你的依赖关系包括多个对象时，数据类通常是一个方便的容器。

下面是一个定义需要依赖关系的Agent例子。

from dataclasses import dataclass

import httpx

from pydantic_ai import Agent, RunContext


@dataclass
classMyDeps:
    api_key: str
    http_client: httpx.AsyncClient


agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
)


@agent.system_prompt  
asyncdefget_system_prompt(ctx: RunContext[MyDeps]) -> str:  
    response = await ctx.deps.http_client.get(  
        'https://example.com',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},  
    )
    response.raise_for_status()
    returnf'Prompt: {response.text}'


asyncdefmain():
    asyncwith httpx.AsyncClient() as client:
        deps = MyDeps('foobar', client)
        result = await agent.run('Tell me a joke.', deps=deps)
        print(result.data)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

MyDeps 是注入到 agent.run 方法中的依赖项

Function Tools  功能工具

Function Tools 为模型提供了一种检索额外信息的机制，以帮助它们生成响应。
当把Agent可能需要的所有上下文放入system prompt不切实际或不可能时，或者当你想通过把生成响应所需的一些逻辑延迟到另一个（不一定由AI驱动的）工具来使Agent的行为更确定或更可靠时，它们就很有用。
向Agent注册工具有多种方法：

• • 通过 @agent.tool 装饰器 - 用于需要访问Agent上下文的工具
• • 通过 @agent.tool_plain 装饰器 - 用于不需要访问Agent上下文的工具
• • 通过工具关键字参数，Agent 可以使用普通函数，也可以使用工具实例。
• • @agent.tool 被认为是默认装饰器，因为在大多数情况下，工具需要访问Agent上下文。

下面是一个同时使用这两种方法的例子：

import random

from pydantic_ai import Agent, RunContext

agent = Agent(
    'gemini-1.5-flash',  
    deps_type=str,  
    system_prompt=(
        "You're a dice game, you should roll the die and see if the number "
        "you get back matches the user's guess. If so, tell them they're a winner. "
        "Use the player's name in the response."
    ),
)


@agent.tool_plain  
defroll_die() -> str:
    """Roll a six-sided die and return the result."""
    returnstr(random.randint(1, 6))


@agent.tool  
defget_player_name(ctx: RunContext[str]) -> str:
    """Get the player's name."""
    return ctx.deps


dice_result = agent.run_sync('My guess is 4', deps='Anne')  
print(dice_result.data)
#> Congratulations Anne, you guessed correctly! You're a winner!

返回结果

Results是运行Agent时返回的最终值。结果值封装在 RunResult 和 StreamedRunResult 中，因此你可以访问其他数据，如运行的使用情况和消息历史记录。

RunResult 和 StreamedRunResult 所封装的数据都是通用的，因此会保留有关Agent返回的数据的键入信息

from pydantic import BaseModel

from pydantic_ai import Agent


classCityLocation(BaseModel):
    city: str
    country: str


agent = Agent('gemini-1.5-flash', result_type=CityLocation)
result = agent.run_sync('Where were the olympics held in 2012?')
print(result.data)
#> city='London' country='United Kingdom'
print(result.usage())
"""
Usage(requests=1, request_tokens=57, response_tokens=8, total_tokens=65, details=None)
"""

信息和聊天记录

PydanticAI 可访问Agent 运行期间交换的信息。这些信息既可用于继续连贯的对话，也可用于了解Agent的执行情况。

运行Agent后，可以通过结果对象访问运行期间交换的报文。

RunResult（由 Agent.run 和 Agent.run_sync 返回）和 StreamedRunResult（由 Agent.run_stream 返回）都有以下方法：

• • all_messages()：返回所有信息，包括之前运行的信息。还有一个返回 JSON 字节的变体，即 all_messages_json()。
• • new_messages()：只返回当前运行中的信息。还有一个返回 JSON 字节的变量，即 new_messages_json()。

将信息作为进一步运行Agent的输入

在 PydanticAI 中，消息历史记录的主要用途是在多个Agent运行时保持上下文。

要在运行中使用现有消息，请将它们传递给 Agent.run、Agent.run_sync 或 Agent.run_stream 的 message_history 参数。

如果 message_history 已设置且非空，则不会生成新的系统提示--假定现有的消息历史记录包含系统提示。

from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', system_prompt='Be a helpful assistant.')

result1 = agent.run_sync('Tell me a joke.')
print(result1.data)
#> Did you hear about the toothpaste scandal? They called it Colgate.

result2 = agent.run_sync('Explain?', message_history=result1.new_messages())
print(result2.data)
#> This is an excellent joke invent by Samuel Colvin, it needs no explanation.

print(result2.all_messages())
"""
[
    ModelRequest(
        parts=[
            SystemPromptPart(
                content='Be a helpful assistant.', part_kind='system-prompt'
            ),
            UserPromptPart(
                content='Tell me a joke.',
                timestamp=datetime.datetime(...),
                part_kind='user-prompt',
            ),
        ],
        kind='request',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='Did you hear about the toothpaste scandal? They called it Colgate.',
                part_kind='text',
            )
        ],
        timestamp=datetime.datetime(...),
        kind='response',
    ),
    ModelRequest(
        parts=[
            UserPromptPart(
                content='Explain?',
                timestamp=datetime.datetime(...),
                part_kind='user-prompt',
            )
        ],
        kind='request',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content='This is an excellent joke invent by Samuel Colvin, it needs no explanation.',
                part_kind='text',
            )
        ],
        timestamp=datetime.datetime(...),
        kind='response',
    ),
]
"""

测试和评估

PydanticAI 和 LLM 集成一般有两种不同的测试：

• • Unit tests--测试您的应用程序代码，以及其是否正确运行
• • Evals --对LLM的测试，以及其反应的好坏
  在大多数情况下，这两种测试的目标和考虑因素是截然不同的。

PydanticAI 代码的单元测试与其他 Python 代码的单元测试一样。

因为在大多数情况下，这些测试并不新鲜，有相当成熟的工具和模式来编写和运行这类测试。

除非你真的确信自己更了解情况，否则你可能会希望大致遵循这一策略：

• • 使用 pytest 作为测试工具
• • 如果你发现自己键入了很长的断言，请使用 inline-snapshot
• • 同样，dirty-equals 也可用于比较大型数据结构
• • 使用 TestModel 或 FunctionModel 代替实际模型，以避免实际 LLM 调用的使用量、延迟和可变性
• • 使用 Agent.override 替换应用程序逻辑中的模型
• • 全局设置 ALLOW_MODEL_REQUESTS=False，以阻止向非测试模型发出任何请求。

为下面的程序代码编写单元测试：

import asyncio
from datetime import date

from pydantic_ai import Agent, RunContext

from fake_database import DatabaseConn  
from weather_service import WeatherService  

weather_agent = Agent(
    'openai:gpt-4o',
    deps_type=WeatherService,
    system_prompt='Providing a weather forecast at the locations the user provides.',
)


@weather_agent.tool
defweather_forecast(
    ctx: RunContext[WeatherService], location: str, forecast_date: date
) -> str:
    if forecast_date < date.today():  
        return ctx.deps.get_historic_weather(location, forecast_date)
    else:
        return ctx.deps.get_forecast(location, forecast_date)


asyncdefrun_weather_forecast(  
    user_prompts: list[tuple[str, int]], conn: DatabaseConn
):
    """Run weather forecast for a list of user prompts and save."""
    asyncwith WeatherService() as weather_service:

        asyncdefrun_forecast(prompt: str, user_id: int):
            result = await weather_agent.run(prompt, deps=weather_service)
            await conn.store_forecast(user_id, result.data)

        # run all prompts in parallel
        await asyncio.gather(
            *(run_forecast(prompt, user_id) for (prompt, user_id) in user_prompts)
        )

这里有一个函数，它接收一个（user_prompt, user_id）元组列表，为每个提示获取天气预报，并将结果存储到数据库中。
我们想测试这段代码，而无需模拟某些对象或修改代码，这样我们就可以传递测试对象。

下面是使用 TestModel 编写测试的方法：

from datetime import timezone
import pytest

from dirty_equals import IsNow

from pydantic_ai import models, capture_run_messages
from pydantic_ai.models.test import TestModel
from pydantic_ai.messages import (
    ArgsDict,
    ModelResponse,
    SystemPromptPart,
    TextPart,
    ToolCallPart,
    ToolReturnPart,
    UserPromptPart,
    ModelRequest,
)

from fake_database import DatabaseConn
from weather_app import run_weather_forecast, weather_agent

pytestmark = pytest.mark.anyio  
models.ALLOW_MODEL_REQUESTS = False


asyncdeftest_forecast():
    conn = DatabaseConn()
    user_id = 1
    with capture_run_messages() as messages:
        with weather_agent.override(model=TestModel()):  
            prompt = 'What will the weather be like in London on 2024-11-28?'
            await run_weather_forecast([(prompt, user_id)], conn)  

    forecast = await conn.get_forecast(user_id)
    assert forecast == '{"weather_forecast":"Sunny with a chance of rain"}'

    assert messages == [  
        ModelRequest(
            parts=[
                SystemPromptPart(
                    content='Providing a weather forecast at the locations the user provides.',
                ),
                UserPromptPart(
                    content='What will the weather be like in London on 2024-11-28?',
                    timestamp=IsNow(tz=timezone.utc),  
                ),
            ]
        ),
        ModelResponse(
            parts=[
                ToolCallPart(
                    tool_name='weather_forecast',
                    args=ArgsDict(
                        args_dict={
                            'location': 'a',
                            'forecast_date': '2024-01-01',  
                        }
                    ),
                    tool_call_id=None,
                )
            ],
            timestamp=IsNow(tz=timezone.utc),
        ),
        ModelRequest(
            parts=[
                ToolReturnPart(
                    tool_name='weather_forecast',
                    content='Sunny with a chance of rain',
                    tool_call_id=None,
                    timestamp=IsNow(tz=timezone.utc),
                ),
            ],
        ),
        ModelResponse(
            parts=[
                TextPart(
                    content='{"weather_forecast":"Sunny with a chance of rain"}',
                )
            ],
            timestamp=IsNow(tz=timezone.utc),
        ),
    ]

上述测试是一个很好的开始，但细心的读者会发现，WeatherService.get_forecast 从未被调用过，因为 TestModel 调用 weather_forecast 时的日期是过去的。

为了充分发挥 weather_forecast 的作用，我们需要使用 FunctionModel 来定制工具的调用方式。
下面是使用 FunctionModel 测试带有自定义输入的天气预报工具的示例

import re

import pytest

from pydantic_ai import models
from pydantic_ai.messages import (
    ModelMessage,
    ModelResponse,
    ToolCallPart,
)
from pydantic_ai.models.function import AgentInfo, FunctionModel

from fake_database import DatabaseConn
from weather_app import run_weather_forecast, weather_agent

pytestmark = pytest.mark.anyio
models.ALLOW_MODEL_REQUESTS = False


defcall_weather_forecast(  
    messages: list[ModelMessage], info: AgentInfo
) -> ModelResponse:
    iflen(messages) == 1:
        # first call, call the weather forecast tool
        user_prompt = messages[0].parts[-1]
        m = re.search(r'\d{4}-\d{2}-\d{2}', user_prompt.content)
        assert m isnotNone
        args = {'location': 'London', 'forecast_date': m.group()}  
        return ModelResponse(
            parts=[ToolCallPart.from_raw_args('weather_forecast', args)]
        )
    else:
        # second call, return the forecast
        msg = messages[-1].parts[0]
        assert msg.part_kind == 'tool-return'
        return ModelResponse.from_text(f'The forecast is: {msg.content}')


asyncdeftest_forecast_future():
    conn = DatabaseConn()
    user_id = 1
    with weather_agent.override(model=FunctionModel(call_weather_forecast)):  
        prompt = 'What will the weather be like in London on 2032-01-01?'
        await run_weather_forecast([(prompt, user_id)], conn)

    forecast = await conn.get_forecast(user_id)
    assert forecast == 'The forecast is: Rainy with a chance of sun'

Evals

"Evals "指的是针对特定应用评估模型的性能。

Evals 通常更像基准而不是单元测试，它们虽然 "失败"，但永远不会 "通过"；你关心的主要是它们如何随时间而变化。

由于 evals 需要针对真实模型运行，运行速度较慢且成本较高，因此一般不希望每次提交都在 CI 中运行 evals。

评估中最难的部分是衡量模型的性能如何。

在某些情况下（如生成 SQL 的Agent），可以使用简单、易于运行的测试来衡量性能（如 SQL 是否有效？是否返回正确的结果？是否只返回正确的结果？）

在其他情况下（例如，提供戒烟建议的Agent），很难或不可能对绩效进行量化衡量--在吸烟的情况下，你确实需要进行为期数月的双盲试验，然后等待 40 年，观察健康结果，才能知道你的提示变化是否是一种改进。

可以使用几种不同的策略来衡量绩效：

• • 端到端、自成一体的测试--如 SQL 示例，我们可以近乎即时地测试Agent的最终结果
• • 合成独立测试--编写单元测试风格的检查程序，检查输出是否符合预期，检查响应是否像 "嚼口香糖 "一样，这些检查可能看起来很简单，但却很有帮助，一个很好的特点是，当这些检查失败时，很容易知道哪里出了问题。
• • LLM 评估 LLM--使用另一个模型，甚至是带有不同提示的同一个模型来评估Agent的表现（比如当全班同学因为老师宿醉而互相批改作业时），虽然这种方法的缺点和复杂性显而易见，但有些人认为在适当的情况下，它可能是一个有用的工具。
• • 在生产中进行评估--测量Agent在生产中的最终结果，然后创建性能的量化指标，这样您就可以在更改提示或所用模型时轻松测量随时间推移而发生的变化，logfire 在这种情况下非常有用，因为您可以编写自定义查询来测量Agent的性能。

System Prompt 自定义

System Prompt是开发人员控制Agent行为的主要工具，因此能够自定义系统提示符并查看性能如何变化往往非常有用。当系统提示包含一个示例列表，而您想了解更改该列表对模型性能的影响时，这一点尤为重要。

调试和监控

使用 LLM 的应用程序面临着一些众所周知的挑战：LLM 速度慢、不可靠、成本高。

这些应用还面临着一些大多数开发人员较少遇到的挑战：LLM 善变且不确定。提示符中的细微变化会完全改变模型的性能，而且无法运行 EXPLAIN 查询来了解原因。

要利用 LLM 构建成功的应用程序，我们需要新的工具来了解模型的性能以及依赖它们的应用程序的行为。

LLM 可观察性工具只能让你了解模型的性能，但毫无用处：对 LLM 进行 API 调用很容易，将其构建到应用程序中才是难事。

Pydantic Logfire

Pydantic Logfire 是一个可观察性平台，由创建和维护 Pydantic 和 PydanticAI 的团队开发。Logfire 旨在让你了解整个应用程序：Gen AI、经典预测性 AI、HTTP 流量、数据库查询以及现代应用所需的其他一切。

PydanticAI 通过 logfire-api no-op 软件包内置（但可选）支持 Logfire。

这意味着，如果安装并配置了日志火软件包，Agent运行的详细信息就会发送到Logfire。但如果没有安装Logfire 软件包，则几乎没有任何开销，也不会发送任何信息。

Example 1: A Simple AI Agent

下面介绍如何建立一个基本的AI Agent，使用 Pydantic 模型验证结构化数据。

上图是用户、Agent和模型的时序图

让我们先创建一些 python 代码，以便开始工作。

from pydantic_ai import Agent
from pydantic import BaseModel
# Define the structure of the response
classCityInfo(BaseModel):
 city: str
 country: str
# Create an agent
agent = Agent(
 model='openai:gpt-4o', # Specify your model
 result_type=CityInfo # Enforce the structure of the response
)
# Run the agent
if __name__ == '__main__':
 result = agent.run_sync("Tell me about Paris.")
print(result.data) # Outputs: {'city': 'Paris', 'country': 'France'}

简单吗？在这里，CityInfo 模型定义了我们期望从响应中获得的结构。然后对Agent进行配置，以便根据此结构验证 LLM 的输出，从而确保可预测的结果。简单快捷！

示例 2：为Agent添加工具

PydanticAI 中的工具是你的Agent在推理过程中可以调用的辅助函数。

上图是正在使用的用户、Agent和工具的序列图

让我们先创建一些 python 代码来展示如何使用工具。

from pydantic_ai import Agent, RunContext
import random
# Define the agent
agent = Agent('openai:gpt-4o')
# Add a tool to roll a die
@agent.tool
async def roll_die(ctx: RunContext, sides: int = 6) -> int:
    """Rolls a die with the specified number of sides."""
    return random.randint(1, sides)
# Run the agent
if __name__ == '__main__':
    result = agent.run_sync("Roll a 20-sided die.")
    print(result.data)  # Outputs a random number between 1 and 20

工具为Agent提供了工作资源，提高了工作效率。最后，它们通过扩展Agent的功能来提供帮助。它们可以与应用程序接口、数据库交互，甚至执行计算。

使用案例 ：研究员和作家

我们需要 Gemini API Key 和 Tavily API、

如果你想知道如何创建Gemini API 密钥，请访问 aistudio.google.com，然后按照说明操作，直到找到生成 API 密钥的地方。

如果您想创建一个 TAVILY_API_KEY，请从以下链接获取 https://app.tavily.com/

获得这两个密钥后，请在 .env 文件中进行设置，如下所示

GEMINI_API_KEY = "Your api key"
TVLY_API_KEY ="Your api key"

pip install 'pydantic-ai[examples]' \
pip install tavily-python 

import asyncio
from dataclasses import dataclass
import json
import os
from typing importList
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext 
from dotenv import load_dotenv
from pydantic_ai.models.gemini import GeminiModel
from httpx import AsyncClient
from tavily import TavilyClient
load_dotenv()

@dataclass
classDeps:
    content_strategist_agent:Agent[None,str]
    client: AsyncClient
    tvly_api_key: str | None
    content:str
    
@dataclass
classContent:
    points: str

classBlogPostBaseModel(BaseModel):
    content:str

model = GeminiModel('gemini-1.5-flash', api_key=os.environ['GEMINI_API_KEY'])

# Agent setup
search_agent = Agent(
    model= model ,
    result_type=Content,
    system_prompt=(
        """you are Senior Research Analyst and your work as a leading tech think tank.
  Your expertise lies in identifying emerging trends.
  You have a knack for dissecting complex data and presenting actionable insights.given topic pydantic AI.
  Full analysis report in bullet points"""
    ),
    retries=2
)

content_writer_agents = Agent(
    model= model ,
    deps_type=Content,
    result_type=BlogPostBaseModel,
    system_prompt=(
        """You are a renowned Content Strategist, known for your insightful and engaging articles.use search_web for getting the list of points
  You transform complex concepts into compelling narratives.Full blog post of at least 4 paragraphs include paragrahs,headings, bullet points include html tags, please remove '\\n\\n'}"""
    ),
    retries=2
)

# Web Search for your query
@search_agent.tool
asyncdefsearch_web(
    ctx: RunContext[Deps], web_query: str
) -> str:
    """Web Search for your query."""
    tavily_client = TavilyClient(api_key=ctx.deps.tvly_api_key)
    response =  tavily_client.search(web_query)
    return json.dumps(response)

@search_agent.tool
asyncdefcontent_writer_agent(
    ctx: RunContext[Deps], question: str
) -> str:
    """Use this tool to communicate with content strategist"""
    print(question)
    response = await ctx.deps.content_strategist_agent.run(user_prompt=question)
    ctx.deps.content = response.data
    print("contentstragist") 
    return response.data

asyncdefmain():
    asyncwith AsyncClient() as client:
        message_history =[]
        tvly_api_key = os.environ['TVLY_API_KEY']
        deps = Deps(client=client, tvly_api_key=tvly_api_key,content_strategist_agent=content_writer_agents,content="")
        result = await search_agent.run("blog article for Pydantic AI",message_history=message_history, deps=deps)
        message_history = result.all_messages()
        print("Blog:")
        print(deps.content)

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

输出是一个关于 Pydantic AI 的blog

我们设置了两个Agent search_agent 和 content_writer_agents , 两个工具 search_agent 和 content_writer_agent

搜索Agent：研究给定主题并使用搜索工具获取信息，然后将信息（上下文）传递给内容撰写Agent工具，因为我们使用的Agent属性的Agent将维护上下文，内容撰写Agent工具将运行内容撰写Agent，内容撰写Agent工具将设置内容撰写Agent生成的博客上下文。

内容撰写Agent：从搜索工具中提取的内容，可以很好地阐述一个blog。