字数 1431，阅读大约需 8 分钟
Python：学习Pydantic AI 的多智能体模式
超级好用的AI Agent开发框架：PydanticAI。在Python构建人工智能驱动的应用程序的时候，往往会出现非结构化输出、类型不匹配和生产可靠性问题。PydanticAI将帮助我们解决：将大型语言模型集成到Python应用程序缺乏生产系统所需的结构和验证的问题。本系列文章持续分享pydantic ai的用法和最新更新内容！ #PydanticAI
✨ 刚刷到的朋友注意啦！点击【关注】，从此升职加薪不迷路
🌟 若觉得内容有用，长按点赞！你的每次互动，都是我深夜码字的星光
✅ 点击「关注」→ 持续收获成长能量
✅ 点亮「点赞」→ 为干货内容打call
✅ 设为「星标」⭐️→ 微信公众号算法优先推送，更新不错过
多智能体agent
Pydantic AI 构建应用程序时，大致有四个复杂性级别
1. 单智能体工作流 — pydantic_ai 文档的大部分内容涵盖了这一点
2. 智能体委托 — 智能体通过工具使用另一个智能体
3. 程序化智能体交接 — 一个智能体运行后，由应用程序代码调用另一个智能体
4. 基于图的控制流 — 对于最复杂的情况，可以使用基于图的状态机来控制多个智能体的执行
智能体工作流
agent.run() 就是典型的工作流，以下三种是之前文章没有提及的智能体交互方式。
智能委托
“智能体委托”指的是一个智能体将工作委托给另一个智能体，然后在受委托的智能体（从工具内部调用的智能体）完成后收回控制权的场景。
说明要求
1. ctx.usage 传递给受委托智能体运行的 usage 关键字参数，这样该运行中的用量就会计入父智能体运行的总用量
2. 智能体委托不要求每个智能体使用相同的模型
3. 智能体是无状态的并且设计为全局的，因此不能在智能体依赖里包含智能体
委托示例
from pydantic_ai import Agent, RunContext, UsageLimits
joke_selection_agent = Agent(
'openai:gpt-4o',
system_prompt=(
'Use the `joke_factory` to generate some jokes, then choose the best. '
'You must return just a single joke.'
),
)
joke_generation_agent = Agent(
'google-gla:gemini-1.5-flash', output_type=list[str]
)
@joke_selection_agent.tool
async def joke_factory(ctx: RunContext[None], count: int) -> list[str]:
r = await joke_generation_agent.run(
f'Please generate {count} jokes.',
usage=ctx.usage,
)
return r.output
result = joke_selection_agent.run_sync(
'Tell me a joke.',
usage_limits=UsageLimits(request_limit=5, total_tokens_limit=500),
)
print(result.output)
#> Did you hear about the toothpaste scandal? They called it Colgate.
print(result.usage())
#> RunUsage(input_tokens=204, output_tokens=24, requests=3, tool_calls=1)智能体委托和依赖
通常，受委托的智能体需要与调用它的智能体具有相同的依赖，或者其依赖是调用智能体依赖的子集。
from dataclasses import dataclass
import httpx
from pydantic_ai import Agent, RunContext
@dataclass
class ClientAndKey:
http_client: httpx.AsyncClient
api_key: str
joke_selection_agent = Agent(
'openai:gpt-4o',
deps_type=ClientAndKey,
system_prompt=(
'Use the `joke_factory` tool to generate some jokes on the given subject, '
'then choose the best. You must return just a single joke.'
),
)
joke_generation_agent = Agent(
'google-gla:gemini-1.5-flash',
deps_type=ClientAndKey,
output_type=list[str],
system_prompt=(
'Use the "get_jokes" tool to get some jokes on the given subject, '
'then extract each joke into a list.'
),
)
@joke_selection_agent.tool
async def joke_factory(ctx: RunContext[ClientAndKey], count: int) -> list[str]:
r = await joke_generation_agent.run(
f'Please generate {count} jokes.',
deps=ctx.deps,
usage=ctx.usage,
)
return r.output
@joke_generation_agent.tool
async def get_jokes(ctx: RunContext[ClientAndKey], count: int) -> str:
response = await ctx.deps.http_client.get(
'https://example.com',
params={'count': count},
headers={'Authorization': f'Bearer {ctx.deps.api_key}'},
)
response.raise_for_status()
return response.text
async def main():
async with httpx.AsyncClient() as client:
deps = ClientAndKey(client, 'foobar')
result = await joke_selection_agent.run('Tell me a joke.', deps=deps)
print(result.output)
#> Did you hear about the toothpaste scandal? They called it Colgate.
print(result.usage())
#> RunUsage(input_tokens=309, output_tokens=32, requests=4, tool_calls=2)智能体交接
“程序化智能体交接”指的是多个智能体被相继调用的场景，其中应用程序代码和/或人工介入负责决定接下来调用哪个智能体。
from typing import Literal
from pydantic import BaseModel, Field
from rich.prompt import Prompt
from pydantic_ai import Agent, RunContext, RunUsage, UsageLimits
from pydantic_ai.messages import ModelMessage
class FlightDetails(BaseModel):
flight_number: str
class Failed(BaseModel):
"""Unable to find a satisfactory choice."""
flight_search_agent = Agent[None, FlightDetails | Failed](
'openai:gpt-4o',
output_type=FlightDetails | Failed, # type: ignore
system_prompt=(
'Use the "flight_search" tool to find a flight '
'from the given origin to the given destination.'
),
)
@flight_search_agent.tool
async def flight_search(
ctx: RunContext[None], origin: str, destination: str
) -> FlightDetails | None:
# in reality, this would call a flight search API or
# use a browser to scrape a flight search website
return FlightDetails(flight_number='AK456')
usage_limits = UsageLimits(request_limit=15)
async def find_flight(usage: RunUsage) -> FlightDetails | None:
message_history: list[ModelMessage] | None = None
for _ in range(3):
prompt = Prompt.ask(
'Where would you like to fly from and to?',
)
result = await flight_search_agent.run(
prompt,
message_history=message_history,
usage=usage,
usage_limits=usage_limits,
)
if isinstance(result.output, FlightDetails):
return result.output
else:
message_history = result.all_messages(
output_tool_return_content='Please try again.'
)
class SeatPreference(BaseModel):
row: int = Field(ge=1, le=30)
seat: Literal['A', 'B', 'C', 'D', 'E', 'F']
# This agent is responsible for extracting the user's seat selection
seat_preference_agent = Agent[None, SeatPreference | Failed](
'openai:gpt-4o',
output_type=SeatPreference | Failed, # type: ignore
system_prompt=(
"Extract the user's seat preference. "
'Seats A and F are window seats. '
'Row 1 is the front row and has extra leg room. '
'Rows 14, and 20 also have extra leg room. '
),
)
async def find_seat(usage: RunUsage) -> SeatPreference:
message_history: list[ModelMessage] | None = None
while True:
answer = Prompt.ask('What seat would you like?')
result = await seat_preference_agent.run(
answer,
message_history=message_history,
usage=usage,
usage_limits=usage_limits,
)
if isinstance(result.output, SeatPreference):
return result.output
else:
print('Could not understand seat preference. Please try again.')
message_history = result.all_messages()
async def main():
usage: RunUsage = RunUsage()
opt_flight_details = await find_flight(usage)
if opt_flight_details is not None:
print(f'Flight found: {opt_flight_details.flight_number}')
#> Flight found: AK456
seat_preference = await find_seat(usage)
print(f'Seat preference: {seat_preference}')
#> Seat preference: row=1 seat='A'智能体基于图的控制流
后面的文章单独讲：概述 - Pydantic AI 框架[1]
彩蛋结尾
嘿，别滑了！手指停一停，听我说句悄悄话👇
🌟 关注我：下次更新，系统会自动弹窗提醒你；
📌 收藏本文：点个收藏，让它成为你的知识库，随时挖宝；
❤️ 点赞在看：你的每个赞都是我熬夜写文的“鸡血”；
更多内容
访问我的专属博客：https://www.funkygod.vip/
引用链接
[1] 概述 - Pydantic AI 框架: https://ai.pydantic.org.cn/graph/
评论区