Pydantic 是 Python 生态系统中的强大平台，每月下载量超过 2.85 亿次。现在，Pydantic的创始人也正在通过 Pydantic AI 涉足 AI 的前沿领域，Pydantic AI 是一个专为构建由生成式 AI 提供支持的生产级应用程序的框架。在本文中，我们将深入探讨 Pydantic AI 的独特之处、它的主要功能以及它与其他代理框架的比较。
一、Pydantic、GenAI 中的 Pydantic、PydanticAI对比
1.1 Pydantic
from datetime import date
from pydantic import BaseModel
class User(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',
}
1.2 Pydantic in GenAI
from datetime import date
from pydantic import BaseModel
from openai import OpenAI
class User(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)
1.3 PydanticAI
from datetime import date
from pydantic_ai import Agent
from pydantic import BaseModel
class User(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)
二、为什么是PydanticAI****
假设您正在制作一个应用程序，用户可在其中提交他们的姓名、年龄和电子邮件。您希望确保：
名称是一个字符串；
年龄是一个数字；
电子邮件的格式有效；
从以下示例可以看出 Pydantic 是如何简化此操作：
from pydantic import BaseModel, EmailStr
# Define the model
class User(BaseModel):
name: str
age: int
email: EmailStr
# Example input
user_data = {
"name": "Alice",
"age": 25,
"email": "alice@example.com"
}
# Validate the input
user = User(**user_data)
print(user.name) # Alice
print(user.age) # 25
print(user.email) # alice@example.com
如果用户提交了无效数据（例如，“age”：“twenty-five”），Pydantic 将自动抛出错误：
user_data = {
"name": "Alice",
"age": "twenty-five", # Invalid
"email": "alice@example.com"
}
user = User(**user_data)
# Error: value is not a valid integer
三、PydanticAI特点
Pydantic 在部署中起着关键作用，因为大多数情况下必须遵循 Pydantic：
-
开发团队 ：由Pydantic背后的团队构建（比如OpenAI SDK、Anthropic SDK、LangChain、LlamaIndex、AutoGPT、Transformers、CrewAI、Instructor 等）；
-
与模型无关 ：支持 OpenAI、Anthropic、Gemini、Ollama、Groq 和 Mistral，并且有一个简单的界面来实现对其他模型的支持；
-
与Pydantic Logfire无缝集成 ：用于实时调试、性能监控和行为跟踪 LLM 驱动的应用程序；
-
以python为中心设计 ：利用 Python 熟悉的控制流和代理组合来构建 AI 驱动的项目，从而轻松应用在任何其他（非 AI）项目中使用的标准 Python 最佳实践；
-
结构化输出 ：利用 Pydantic 的强大功能来验证和构建模型输出，确保响应在运行之间保持一致；
-
流式输出 ：能够连续流式传输 LLM 输出，并立即进行验证，确保快速准确的结果。
四、如何使用PydanticAI****
4.1 安装PydanticAI（需要python3.9+）
pip install pydantic-ai
这将安装使用 PydanticAI 中包含的所有模型所需的 pydantic\_ai 包、核心依赖项和库。如果你想使用特定的模型，你可以安装 PydanticAI 的 “slim” 版本。
4.2 PydanticAI 的 8 个重要组成部分：
-
Agents（代理）
-
Models（模型）
-
Dependencies（依赖）
-
Function Tools（函数工具）
-
Results（结果）
-
Messages and chat history（ 消息和聊天记录）
-
Testing and Evals（测试和评估）
-
Debugging and Monitoring（调试和监控）
4.2.1 Agents（代理）
有三种方式运行Agent：
- agent.run() ：是协程，返回一个包含已完成响应的 RunResult；
- agent.run_sync() ：一个普通的同步函数，它返回一个包含已完成响应的 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
async def main():
result = await agent.run('What is the capital of France?')
print(result.data)
#> Paris
async with agent.run_stream('What is the capital of the UK?') as response:
print(await response.get_data())
#> London
4.2.2 Models（模型）
PydanticAI 与模型无关，并内置了对以下模型提供程序的支持：
- OpenAI
- Anthropic
- Gemini 通过两个不同的 API：生成语言 API 和 VertexAI API
- Ollama
- Groq
- Mistral
用户还可以添加对其他模型的支持。
PydanticAI 还附带了用于测试和开发的 TestModel 和 FunctionModel。要使用每个模型提供程序，需要配置本地环境并确保安装了正确的软件包。比如要使用 GeminiModel 模型，只需要安装 pydantic-ai 或 pydantic-ai-slim，不需要额外的依赖项。
a）配置
GeminiModel 允许通过其生成语言 API 使用 Google 的 Gemini 模型。generativelanguage.googleapis.com
GeminiModelName 包含可通过此接口使用的可用 Gemini 模型列表。
获得 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)
4.2.3 Dependencies（依赖）
PydanticAI 使用依赖注入系统为代理的系统提示、工具和结果验证器提供数据和服务。
依赖项可以是任何 python 类型。在简单情况下，可能将单个对象作为依赖项传递（例如 HTTP 连接），但当依赖项包含多个对象时，数据类通常是一个方便的容器。
下面是定义需要依赖项的代理的示例：
from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext
@dataclass
class MyDeps:
api_key: str
http_client: httpx.AsyncClient
agent = Agent(
'openai:gpt-4o',
deps_type=MyDeps,
)
@agent.system_prompt
async def get_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()
return f'Prompt: {response.text}'
async def main():
async with 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 方法中的依赖项
4.2.4 Function Tools（函数工具）
函数工具为模型提供了一种用于检索额外信息以生成响应的机制。
当将代理可能需要的所有上下文放入系统提示符中不切实际或不可能时，或者当您希望通过将生成响应所需的一些逻辑延迟到另一个（不一定是 AI 驱动的）工具来使代理的行为更具确定性或可靠性时，它们非常有用。
有多种方法可以向Agent注册工具：
- @agent.tool：用于需要访问代理上下文的工具；
- @agent.tool_plain：用于不需要访问代理上下文的工具；
- Tools 关键字参数传递给 Agent，它可以采用普通函数或 Tool 的实例
@agent.tool 被视为默认装饰器，因为在大多数情况下，工具需要访问代理上下文。
下面是一个同时使用两者的示例：
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
def roll_die() -> str:
"""Roll a six-sided die and return the result."""
return str(random.randint(1, 6))
@agent.tool
def get_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!
4.2.5 Results（结果）
Result是运行Agent的最终返回值。Result值包装在RunResult和 StreamedRunResult中，这样可以像使用run和message histroy一样访问其他数据。
RunResult 和 StreamedRunResult 在它们包装的数据中都是通用的，因此会保留有关Agent返回的数据的输入信息。
from pydantic import BaseModel
from pydantic_ai import Agent
class CityLocation(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)
"""
4.2.6 Messages and chat history（ 消息和聊天记录）
PydanticAI 可以在Agent运行期间交换的消息，这些消息既可用于继续连贯的对话，也可用于掌握Agent的性能。
a）从Results访问消息 ：
运行代理后，可以从 result 对象访问该运行期间交换的消息。
RunResult(由Agent.run 返回，Agent.run_sync）和 StreamedRunResult（由 Agent.run_stream 返回）都有以下方法：
-
all_messages（）：返回所有消息，包括之前运行的消息。还有一个返回 JSON 字节的变体 all_messages_json（）
-
new_messages（）：仅返回当前运行的消息。还有一个返回 JSON 字节的变体 new_messages_json（）
b）使用Messages作为输入给Agent
PydanticAI 中消息历史记录的主要用途是在多个代理运行中维护上下文。
要在运行中使用历史消息，请将它们传递给 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',
),
]
"""
4.2.7 Testing and Evals（测试和评估）
使用 PydanticAI 和 LLM 集成，通常有两种不同的测试：
- 单元测试 — 测试应用程序代码是否可以正确运行；
- Evals — 测试LLM的整体性能；
在大多数情况下，这两种类型的测试具有相当不同的目标和注意事项。
a）单元测试 ：
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
def weather_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)
async def run_weather_forecast(
user_prompts: list[tuple[str, int]], conn: DatabaseConn
):
"""Run weather forecast for a list of user prompts and save."""
async with WeatherService() as weather_service:
async def run_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
async def test_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),
),
]
使用 FunctionModel 进行单元测试
上述测试是一个很好的开始，但细心的读者会注意到，由于 TestModel 调用的日期是过去的日期，因此从未调用 weather\_forecast WeatherService.get\_forecast。
要充分实践weather\_forecast，我们需要使用 FunctionModel 来自定义工具的调用方式。
下面是使用 FunctionModel 测试具有自定义输入的 weather\_forecast 工具的示例
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
def call_weather_forecast(
messages: list[ModelMessage], info: AgentInfo
) -> ModelResponse:
if len(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 is not None
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}')
async def test_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'
4.2.8 Debugging and Monitoring（调试和监控）
使用 LLMs存在一些众所周知的挑战： **LLMs 速度慢** 、 **不可靠** 且 **成本高昂** 。
这些应用程序还存在大多数开发人员很少遇到的一些挑战：LLMs 是变化无常且不确定的。提示中的细微变化可能会完全改变模型的性能，并且无法解释原因。
要使用 LLMs，我们需要新的工具来了解模型性能以及依赖它们的应用程序的行为。
a）Pydantic Logfire
Pydantic Logfire 是由创建和维护 Pydantic 和 PydanticAI 的团队开发的可观测性平台。Logfire 旨在让您了解整个应用程序：Gen AI、经典预测 AI、HTTP 流量、数据库查询以及现代应用程序所需的一切。
PydanticAI 通过 logfire-api no-op 包内置了对 Logfire 的（但可选）支持。
这意味着，如果安装并配置了 logfire 软件包，则会将有关代理运行的详细信息发送到 Logfire。但是，如果未安装 logfire 包，则几乎没有开销，也不会发送任何内容。
以下示例显示了在 Logfire 中运行 Weather Agent 的详细信息：
五、示例：
示例1：简单的AI Agent
下面介绍如何设置一个基本的 AI 代理，使用 Pydantic 模型验证结构化数据。
图 1：显示 User、Agent 和 Model 的序列图
下面是具体的代码实现：
from pydantic_ai import Agent
from pydantic import BaseModel
# Define the structure of the response
class CityInfo(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 中的工具是您的代理在其推理过程中可以调用的辅助函数。
图 2：显示正在使用的 User、Agent 和 Tool 的序列图
下面是具体的实现代码：
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
Tools扩展了Agent的功能并提高工作效率，比如可以与 API、数据库交互，甚至执行计算。
示例3：研究人员和作家
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 import List
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
class Deps:
content_strategist_agent:Agent[None,str]
client: AsyncClient
tvly_api_key: str | None
content:str
@dataclass
class Content:
points: str
class BlogPostBaseModel(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
async def search_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
async def content_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
async def main():
async with 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 的好博客......
Pydantic AI: A Robust Foundation for Your AI Applications
In the dynamic world of Artificial Intelligence (AI), where data integrity and model reliability are paramount, choosing the right tools is crucial. Pydantic, a Python library for data parsing and validation, emerges as a powerful ally in building robust and efficient AI systems. Its ability to define data models and enforce data validation rules significantly improves the development process and reduces the risk of errors caused by inconsistent or malformed data.
Data Validation: The Cornerstone of Reliable AI
Data validation is a critical step in any AI project. Inaccurate or incomplete data can lead to flawed models, erroneous predictions, and ultimately, unreliable AI systems. Pydantic excels in this area by allowing you to define strict data schemas. This ensures that all data processed by your AI application conforms to predefined rules. No more unexpected type errors or missing values crashing your model – Pydantic acts as a gatekeeper, ensuring only valid data proceeds.
Type checking: Pydantic automatically validates data types, ensuring that integers remain integers, strings stay strings, and so on.
Value constraints: Define constraints such as minimum/maximum values, allowed strings, regular expressions, etc., to enforce precise data integrity.
Custom validation: Implement custom validation logic to handle complex scenarios and business rules specific to your AI application.
Data Modeling: Structure and Organization for Your AI Projects
AI projects often involve numerous data structures and configurations. Pydantic provides a powerful and elegant way to model your data. By creating data models, you not only improve the organization of your code but also enable early detection of inconsistencies. This structured approach simplifies data processing and manipulation, making your AI application cleaner and easier to maintain.
Structured Data: Easily define classes representing the structure of your input and output data.
Complex Data Types: Handle nested structures, lists, and dictionaries seamlessly, enforcing validation at every level.
Improved Readability: Clearly defined data models make your code easier to understand and maintain for both you and your collaborators.
Configuration Management: Centralized and Validated Settings
Managing configurations, especially in complex AI systems, can quickly become challenging. Pydantic offers a solution for efficient configuration management. You can create Pydantic models for your application's settings, ensuring that all configurations conform to predefined types and constraints. This centralized approach simplifies the management and modification of settings, thereby improving the overall workflow of your AI project.
Environment Variables: Easily load configuration settings from environment variables with validation.
Configuration Files: Use configuration files (YAML, JSON, etc.) to store and load settings while enforcing type and constraint validation.
Version Control Integration: Easily track configuration changes in your version control system, ensuring reproducibility and traceability.
Conclusion
Incorporating Pydantic into your AI development pipeline offers numerous advantages, from robust data validation to streamlined configuration management. By leveraging Pydantic's data modeling capabilities and validation features, you can create more reliable, maintainable, and ultimately, successful AI applications. It’s a simple yet powerful addition that significantly enhances the overall robustness and efficiency of your AI development efforts.
我们设置了两个代理search\_agent content\_writer\_agents两个工具 search\_agent 和 content\_writer\_agent
search_agent ： 研究给定主题并使用search_web工具获取信息并将信息（上下文）传递给content_writer_agent工具，因为我们使用了工具属性代理，因此content_writer_agent工具将运行content_writer_agents代理，content_writer_agent工具将设置由代理生成的博客上下文content_writer_agents content_writer_agents：它拥有的内容是从 search_web 工具中输出的，它将很好地表达一个博客。
至此，应该掌握如何使用Pydantic AI 框架了，那么，接下来，就可以使用 Pydantic AI 在生产环境中创建AI应用程序了。