Pydantic 是 Python 生态系统中的强大工具，月下载量超过 2.85 亿次。它一直是 Python 项目中，强大的、用于数据验证的基石。现在，它的创造者正通过 Pydantic AI 向人工智能的前沿领域进军。Pydantic AI 是一个框架，专为构建生产级应用而设计，旨在驱动生成式人工智能的落地。在本文中，我们将深入探讨 Pydantic AI 的独特之处、主要功能以及与其他Agent框架的比较。
本文较长，建议PC阅读，如果可能，打开IDE一起操作，会更有收获。
frompydantic_aiimportAgent
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.
"""fromdatetimeimportdate
frompydanticimportBaseModel
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',
}
fromdatetimeimportdate
frompydanticimportBaseModel
fromopenaiimportOpenAI
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)
ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;padding-left: 8px;color: rgb(63, 63, 63);">PydanticAIingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-size: 14px;letter-spacing: 0.1em;color: rgb(63, 63, 63);">同样的例子还有 PydanticAI - 用于生产的Agent框架。ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;font-feature-settings: normal;font-variation-settings: normal;font-size: 14px;margin: 10px 8px;background: rgb(46, 52, 64);color: rgb(216, 222, 233);text-align: left;line-height: 1.5;overflow-x: auto;border-radius: 8px;padding: 0px !important;">fromdatetimeimportdate
frompydantic_aiimportAgent
frompydanticimportBaseModel
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)
ingFang SC", "Hiragino Sans GB", "Microsoft YaHei UI", "Microsoft YaHei", Arial, sans-serif;letter-spacing: normal;border-width: 0px 0px 2px;border-style: solid;border-bottom-color: rgb(250, 81, 81);text-align: center;line-height: 1.75;display: table;">为什么选择 PydanticAI？假设你正在制作一个应用，用户可以提交姓名、年龄和电子邮件。你希望确保
Pydantic 如何轻松做到这一点：
frompydanticimportBaseModel, EmailStr
# Define the model
classUser(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
如果用户提交的数据无效（如 "年龄"："25"），Pydantic 会自动出错：
user_data = {
"name":"Alice",
"age":"twenty-five", # Invalid
"email":"alice@example.com"
}
user = User(**user_data)
# Error: value is not a valid integerPydantic 在部署过程中起着至关重要的作用，因为它是必须遵守的，因为
由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 框架中的一个主要问题：缺乏专为生产用例设计的工具。
需要 Python 3.9 及以上版本
pipinstallpydantic-ai
这将安装 pydantic_ai 包、核心依赖项以及使用 PydanticAI 中包含的所有模型所需的库。如果您想使用某个特定模型，可以安装 "slim "版本的 PydanticAI。
Agent 是 PydanticAI 与 LLM 交互的主要媒介。
在某些使用案例中，单 Agent 可以控制整个应用程序或组件，但multi-agents也可以进行交互，以体现更复杂的工作流程。
Agent 类有完整的应用程序接口文档，但从概念上讲，你可以把 Agent 看作是一个容器
运行Agent有三种方法：
下面是一个简单的例子，展示了这三点：
frompydantic_aiimportAgent
agent = Agent('openai:gpt-4o')
result_sync = agent.run_sync('What is the capital of Italy?')
print(result_sync.data)
#> Rome
asyncdefmain():
result =awaitagent.run('What is the capital of France?')
print(result.data)
#> Paris
asyncwithagent.run_stream('What is the capital of the UK?')asresponse:
print(awaitresponse.get_data())
#> LondonPydanticAI 与模型是解耦的，内置了对以下模型的支持：
你还可以自定义对其他模型的支持。
PydanticAI 还提供用于测试和开发的 TestModel 和 FunctionModel。
要使用每个模型提供商，你需要配置本地环境，并确保安装了正确的包。需要要使用GeminiModel模型，只需安装pydantic-ai或pydantic-ai-slim，无需额外的依赖项。
GeminiModel 可让您通过生成语言 API（generativelanguage.googleapis.com）使用谷歌的Gemini模型。
GeminiModelName 包含可通过此接口使用的可用Gemini模型列表。
要使用 GeminiModel，请访问 aistudio.google.com，然后按照提示找到生成 API 密钥的地方。
获得 API 密钥后，可以将其设置为环境变量：
frompydantic_aiimportAgent
frompydantic_ai.models.geminiimportGeminiModel
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例子。
fromdataclassesimportdataclass
importhttpx
frompydantic_aiimportAgent, 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 =awaitctx.deps.http_client.get(
'https://example.com',
headers={'Authorization':f'Bearer{ctx.deps.api_key}'},
)
response.raise_for_status()
returnf'Prompt:{response.text}'
asyncdefmain():
asyncwithhttpx.AsyncClient()asclient:
deps = MyDeps('foobar', client)
result =awaitagent.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 为模型提供了一种检索额外信息的机制，以帮助它们生成响应。
当把Agent可能需要的所有上下文放入system prompt不切实际或不可能时，或者当你想通过把生成响应所需的一些逻辑延迟到另一个（不一定由AI驱动的）工具来使Agent的行为更确定或更可靠时，它们就很有用。
向Agent注册工具有多种方法：
下面是一个同时使用这两种方法的例子：
importrandom
frompydantic_aiimportAgent, 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."""
returnctx.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返回的数据的键入信息
frompydanticimportBaseModel
frompydantic_aiimportAgent
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()。在 PydanticAI 中，消息历史记录的主要用途是在多个Agent运行时保持上下文。
要在运行中使用现有消息，请将它们传递给Agent.run、Agent.run_sync或Agent.run_stream的message_history参数。
如果message_history已设置且非空，则不会生成新的系统提示--假定现有的消息历史记录包含系统提示。
frompydantic_aiimportAgent
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 集成一般有两种不同的测试：
PydanticAI 代码的单元测试与其他 Python 代码的单元测试一样。
因为在大多数情况下，这些测试并不新鲜，有相当成熟的工具和模式来编写和运行这类测试。
除非你真的确信自己更了解情况，否则你可能会希望大致遵循这一策略：
inline-snapshotTestModel或FunctionModel代替实际模型，以避免实际 LLM 调用的使用量、延迟和可变性Agent.override替换应用程序逻辑中的模型ALLOW_MODEL_REQUESTS=False，以阻止向非测试模型发出任何请求。为下面的程序代码编写单元测试：
importasyncio
fromdatetimeimportdate
frompydantic_aiimportAgent, RunContext
fromfake_databaseimportDatabaseConn
fromweather_serviceimportWeatherService
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:
ifforecast_date < date.today():
returnctx.deps.get_historic_weather(location, forecast_date)
else:
returnctx.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."""
asyncwithWeatherService()asweather_service:
asyncdefrun_forecast(prompt:str, user_id:int):
result =awaitweather_agent.run(prompt, deps=weather_service)
awaitconn.store_forecast(user_id, result.data)
# run all prompts in parallel
awaitasyncio.gather(
*(run_forecast(prompt, user_id)for(prompt, user_id)inuser_prompts)
)这里有一个函数，它接收一个（user_prompt, user_id）元组列表，为每个提示获取天气预报，并将结果存储到数据库中。
我们想测试这段代码，而无需模拟某些对象或修改代码，这样我们就可以传递测试对象。
下面是使用TestModel编写测试的方法：
fromdatetimeimporttimezone
importpytest
fromdirty_equalsimportIsNow
frompydantic_aiimportmodels, capture_run_messages
frompydantic_ai.models.testimportTestModel
frompydantic_ai.messagesimport(
ArgsDict,
ModelResponse,
SystemPromptPart,
TextPart,
ToolCallPart,
ToolReturnPart,
UserPromptPart,
ModelRequest,
)
fromfake_databaseimportDatabaseConn
fromweather_appimportrun_weather_forecast, weather_agent
pytestmark = pytest.mark.anyio
models.ALLOW_MODEL_REQUESTS =False
asyncdeftest_forecast():
conn = DatabaseConn()
user_id =1
withcapture_run_messages()asmessages:
withweather_agent.override(model=TestModel()):
prompt ='What will the weather be like in London on 2024-11-28?'
awaitrun_weather_forecast([(prompt, user_id)], conn)
forecast =awaitconn.get_forecast(user_id)
assertforecast =='{"weather_forecast":"Sunny with a chance of rain"}'
assertmessages == [
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测试带有自定义输入的天气预报工具的示例
importre
importpytest
frompydantic_aiimportmodels
frompydantic_ai.messagesimport(
ModelMessage,
ModelResponse,
ToolCallPart,
)
frompydantic_ai.models.functionimportAgentInfo, FunctionModel
fromfake_databaseimportDatabaseConn
fromweather_appimportrun_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)
assertmisnotNone
args = {'location':'London','forecast_date': m.group()}
returnModelResponse(
parts=[ToolCallPart.from_raw_args('weather_forecast', args)]
)
else:
# second call, return the forecast
msg = messages[-1].parts[0]
assertmsg.part_kind =='tool-return'
returnModelResponse.from_text(f'The forecast is:{msg.content}')
asyncdeftest_forecast_future():
conn = DatabaseConn()
user_id =1
withweather_agent.override(model=FunctionModel(call_weather_forecast)):
prompt ='What will the weather be like in London on 2032-01-01?'
awaitrun_weather_forecast([(prompt, user_id)], conn)
forecast =awaitconn.get_forecast(user_id)
assertforecast =='The forecast is: Rainy with a chance of sun'"Evals "指的是针对特定应用评估模型的性能。
Evals 通常更像基准而不是单元测试，它们虽然 "失败"，但永远不会 "通过"；你关心的主要是它们如何随时间而变化。
由于 evals 需要针对真实模型运行，运行速度较慢且成本较高，因此一般不希望每次提交都在 CI 中运行 evals。
评估中最难的部分是衡量模型的性能如何。
在某些情况下（如生成 SQL 的Agent），可以使用简单、易于运行的测试来衡量性能（如 SQL 是否有效？是否返回正确的结果？是否只返回正确的结果？）
在其他情况下（例如，提供戒烟建议的Agent），很难或不可能对绩效进行量化衡量--在吸烟的情况下，你确实需要进行为期数月的双盲试验，然后等待 40 年，观察健康结果，才能知道你的提示变化是否是一种改进。
可以使用几种不同的策略来衡量绩效：
System Prompt是开发人员控制Agent行为的主要工具，因此能够自定义系统提示符并查看性能如何变化往往非常有用。当系统提示包含一个示例列表，而您想了解更改该列表对模型性能的影响时，这一点尤为重要。
使用 LLM 的应用程序面临着一些众所周知的挑战：LLM 速度慢、不可靠、成本高。
这些应用还面临着一些大多数开发人员较少遇到的挑战：LLM 善变且不确定。提示符中的细微变化会完全改变模型的性能，而且无法运行 EXPLAIN 查询来了解原因。
要利用 LLM 构建成功的应用程序，我们需要新的工具来了解模型的性能以及依赖它们的应用程序的行为。
LLM 可观察性工具只能让你了解模型的性能，但毫无用处：对 LLM 进行 API 调用很容易，将其构建到应用程序中才是难事。
Pydantic Logfire 是一个可观察性平台，由创建和维护 Pydantic 和 PydanticAI 的团队开发。Logfire 旨在让你了解整个应用程序：Gen AI、经典预测性 AI、HTTP 流量、数据库查询以及现代应用所需的其他一切。
PydanticAI 通过 logfire-api no-op 软件包内置（但可选）支持 Logfire。
这意味着，如果安装并配置了日志火软件包，Agent运行的详细信息就会发送到Logfire。但如果没有安装Logfire 软件包，则几乎没有任何开销，也不会发送任何信息。
下面的示例显示了在 Logfire 中运行气象Agent的详细信息：下面介绍如何建立一个基本的AI Agent，使用 Pydantic 模型验证结构化数据。
让我们先创建一些 python 代码，以便开始工作。
frompydantic_aiimportAgent
frompydanticimportBaseModel
# 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 的输出，从而确保可预测的结果。简单快捷！
PydanticAI 中的工具是你的Agent在推理过程中可以调用的辅助函数。
让我们先创建一些 python 代码来展示如何使用工具。
frompydantic_aiimportAgent, RunContext
importrandom
# Define the agent
agent = Agent('openai:gpt-4o')
# Add a tool to roll a die
@agent.tool
asyncdefroll_die(ctx: RunContext, sides:int=6) ->int:
"""Rolls a die with the specified number of sides."""
returnrandom.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-pythonimportasyncio
fromdataclassesimportdataclass
importjson
importos
fromtypingimportList
frompydanticimportBaseModel
frompydantic_aiimportAgent, RunContext
fromdotenvimportload_dotenv
frompydantic_ai.models.geminiimportGeminiModel
fromhttpximportAsyncClient
fromtavilyimportTavilyClient
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)
returnjson.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 =awaitctx.deps.content_strategist_agent.run(user_prompt=question)
ctx.deps.content = response.data
print("contentstragist")
returnresponse.data
asyncdefmain():
asyncwithAsyncClient()asclient:
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 =awaitsearch_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
我们设置了两个Agentsearch_agent和content_writer_agents, 两个工具search_agent和content_writer_agent
搜索Agent：研究给定主题并使用搜索工具获取信息，然后将信息（上下文）传递给内容撰写Agent工具，因为我们使用的Agent属性的Agent将维护上下文，内容撰写Agent工具将运行内容撰写Agent，内容撰写Agent工具将设置内容撰写Agent生成的博客上下文。
内容撰写Agent：从搜索工具中提取的内容，可以很好地阐述一个blog。
通过这些内容，你将了解 Pydantic AI 如何使用 Pydantic AI 框架创建生产就绪的 AI 应用程序。
| 欢迎光临 链载Ai (https://www.lianzai.com/) | Powered by Discuz! X3.5 |