高级工具功能
本页涵盖了 Pydantic AI 中函数工具的高级功能。有关工具的基本用法，请参阅函数工具文档。
工具输出
工具可以返回任何 Pydantic 能序列化为 JSON 的内容，也可以是音频、视频、图像或文档内容，具体取决于模型支持的多模态输入类型。
from datetime import datetime
from pydantic import BaseModel
from pydantic_ai import Agent, DocumentUrl, ImageUrl
from pydantic_ai.models.openai import OpenAIResponsesModel
class User(BaseModel):
name: str
age: int
agent = Agent(model=OpenAIResponsesModel('gpt-4o'))
@agent.tool_plain
def get_current_time() -> datetime:
return datetime.now()
@agent.tool_plain
def get_user() -> User:
return User(name='John', age=30)
@agent.tool_plain
def get_company_logo() -> ImageUrl:
return ImageUrl(url='https://iili.io/3Hs4FMg.png')
@agent.tool_plain
def get_document() -> DocumentUrl:
return DocumentUrl(url='https://www.w3.org/WAI/ER/tests/xhtml/testfiles/resources/pdf/dummy.pdf')
result = agent.run_sync('What time is it?')
print(result.output)
#> The current time is 10:45 PM on April 17, 2025.
result = agent.run_sync('What is the user name?')
print(result.output)
#> The user's name is John.
result = agent.run_sync('What is the company name in the logo?')
print(result.output)
#> The company name in the logo is "Pydantic."
result = agent.run_sync('What is the main content of the document?')
print(result.output)
#> The document contains just the text "Dummy PDF file."
（这个例子是完整的，可以“按原样”运行）
一些模型（如 Gemini）原生支持半结构化的返回值，而另一些模型则期望文本（如 OpenAI），但它们在从数据中提取意义方面似乎同样出色。如果返回的是 Python 对象而模型期望字符串，该值将被序列化为 JSON。
高级工具返回
对于需要更精细地控制工具返回值和发送给模型的内容的场景，您可以使用 ToolReturn。这在您希望实现以下功能时特别有用：
- 为模型提供丰富的多模态内容（图像、文档等）作为上下文
- 将程序化的返回值与模型的上下文分离开
- 包含不应发送给大语言模型（LLM）的额外元数据
以下是一个计算机自动化工具的示例，该工具能捕获屏幕截图并提供视觉反馈：
import time
from pydantic_ai import Agent
from pydantic_ai.messages import ToolReturn, BinaryContent
agent = Agent('openai:gpt-4o')
@agent.tool_plain
def click_and_capture(x: int, y: int) -> ToolReturn:
"""Click at coordinates and show before/after screenshots."""
# Take screenshot before action
before_screenshot = capture_screen()
# Perform click operation
perform_click(x, y)
time.sleep(0.5) # Wait for UI to update
# Take screenshot after action
after_screenshot = capture_screen()
return ToolReturn(
return_value=f"Successfully clicked at ({x}, {y})",
content=[
f"Clicked at coordinates ({x}, {y}). Here's the comparison:",
"Before:",
BinaryContent(data=before_screenshot, media_type="image/png"),
"After:",
BinaryContent(data=after_screenshot, media_type="image/png"),
"Please analyze the changes and suggest next steps."
],
metadata={
"coordinates": {"x": x, "y": y},
"action_type": "click_and_capture",
"timestamp": time.time()
}
)
# The model receives the rich visual content for analysis
# while your application can access the structured return_value and metadata
result = agent.run_sync("Click on the submit button and tell me what happened")
print(result.output)
# The model can analyze the screenshots and provide detailed feedback
return_value：工具响应中使用的实际返回值。这部分内容会被序列化并作为工具结果发送回模型。content：一系列为模型提供额外上下文的内容（文本、图像、文档等）。这会作为一条独立的用户消息出现。metadata：可选的元数据，您的应用程序可以访问，但不会发送给大语言模型（LLM）。可用于日志记录、调试或其他额外处理。一些其他的 AI 框架称此功能为“artifacts”（工件）。
这种分离使您能够在为模型提供丰富上下文的同时，为您的应用程序逻辑保持干净、结构化的返回值。
自定义工具模式
如果您有一个函数缺乏适当的文档（例如，命名不佳、没有类型信息、文档字符串写得不好、使用了 *args 或 **kwargs 等），您仍然可以使用 Tool.from_schema 函数将其转换为一个能被智能体有效使用的工具。通过这种方式，您可以直接为该函数提供名称、描述、JSON 模式以及该函数是否接受 RunContext。
from pydantic_ai import Agent, Tool
from pydantic_ai.models.test import TestModel
def foobar(**kwargs) -> str:
return kwargs['a'] + kwargs['b']
tool = Tool.from_schema(
function=foobar,
name='sum',
description='Sum two numbers.',
json_schema={
'additionalProperties': False,
'properties': {
'a': {'description': 'the first number', 'type': 'integer'},
'b': {'description': 'the second number', 'type': 'integer'},
},
'required': ['a', 'b'],
'type': 'object',
},
takes_ctx=False,
)
test_model = TestModel()
agent = Agent(test_model, tools=[tool])
result = agent.run_sync('testing...')
print(result.output)
#> {"sum":0}
请注意，系统不会对工具参数进行验证，并且所有参数都将作为关键字参数传递。
动态工具
工具可以选择性地定义另一个函数：prepare。该函数在运行的每一步都会被调用，用于自定义传递给模型的工具定义，或者在该步骤中完全省略该工具。
prepare 方法可以通过 prepare 关键字参数注册到任何工具注册机制中：
@agent.tool装饰器@agent.tool_plain装饰器Tool数据类
prepare 方法的类型应为 ToolPrepareFunc，它是一个函数，接收 RunContext 和一个预先构建的 ToolDefinition，并应返回修改或未修改的 ToolDefinition、一个新的 ToolDefinition，或者返回 None 来表示该步骤不应注册此工具。
这是一个简单的 prepare 方法示例，仅当依赖项的值为 42 时才包含该工具。
与前面的示例一样，我们使用 TestModel 来演示此行为，而无需调用真实模型。
from pydantic_ai import Agent, RunContext, ToolDefinition
agent = Agent('test')
async def only_if_42(
ctx: RunContext[int], tool_def: ToolDefinition
) -> ToolDefinition | None:
if ctx.deps == 42:
return tool_def
@agent.tool(prepare=only_if_42)
def hitchhiker(ctx: RunContext[int], answer: str) -> str:
return f'{ctx.deps} {answer}'
result = agent.run_sync('testing...', deps=41)
print(result.output)
#> success (no tool calls)
result = agent.run_sync('testing...', deps=42)
print(result.output)
#> {"hitchhiker":"42 a"}
（这个例子是完整的，可以“按原样”运行）
这是一个更复杂的示例，我们根据 deps 的值来更改 name 参数的描述。
为了变化起见，我们使用 Tool 数据类来创建此工具。
from __future__ import annotations
from typing import Literal
from pydantic_ai import Agent, RunContext, Tool, ToolDefinition
from pydantic_ai.models.test import TestModel
def greet(name: str) -> str:
return f'hello {name}'
async def prepare_greet(
ctx: RunContext[Literal['human', 'machine']], tool_def: ToolDefinition
) -> ToolDefinition | None:
d = f'Name of the {ctx.deps} to greet.'
tool_def.parameters_json_schema['properties']['name']['description'] = d
return tool_def
greet_tool = Tool(greet, prepare=prepare_greet)
test_model = TestModel()
agent = Agent(test_model, tools=[greet_tool], deps_type=Literal['human', 'machine'])
result = agent.run_sync('testing...', deps='human')
print(result.output)
#> {"greet":"hello a"}
print(test_model.last_model_request_parameters.function_tools)
"""
[
ToolDefinition(
name='greet',
parameters_json_schema={
'additionalProperties': False,
'properties': {
'name': {'type': 'string', 'description': 'Name of the human to greet.'}
},
'required': ['name'],
'type': 'object',
},
)
]
"""
（这个例子是完整的，可以“按原样”运行）
智能体级别的动态工具
除了针对单个工具的 prepare 方法外，您还可以定义一个智能体级别的 prepare_tools 函数。此函数在运行的每一步都会被调用，允许您过滤或修改该步骤中智能体可用的所有工具定义列表。如果您想一次性启用或禁用多个工具，或者根据当前上下文应用全局逻辑，这将特别有用。
prepare_tools 函数的类型应为 ToolsPrepareFunc，它接收 RunContext 和一个 ToolDefinition 列表，并返回一个新的工具定义列表（或返回 None 以在该步骤禁用所有工具）。
要修改输出工具，您可以改为设置一个 prepare_output_tools 函数。
以下示例展示了如果模型是 OpenAI 模型，则将所有工具设置为严格模式：
from dataclasses import replace
from pydantic_ai import Agent, RunContext, ToolDefinition
from pydantic_ai.models.test import TestModel
async def turn_on_strict_if_openai(
ctx: RunContext[None], tool_defs: list[ToolDefinition]
) -> list[ToolDefinition] | None:
if ctx.model.system == 'openai':
return [replace(tool_def, strict=True) for tool_def in tool_defs]
return tool_defs
test_model = TestModel()
agent = Agent(test_model, prepare_tools=turn_on_strict_if_openai)
@agent.tool_plain
def echo(message: str) -> str:
return message
agent.run_sync('testing...')
assert test_model.last_model_request_parameters.function_tools[0].strict is None
# Set the system attribute of the test_model to 'openai'
test_model._system = 'openai'
agent.run_sync('testing with openai...')
assert test_model.last_model_request_parameters.function_tools[0].strict
（这个例子是完整的，可以“按原样”运行）
这是另一个示例，如果依赖项（ctx.deps）为 True，则根据名称有条件地过滤掉工具：
from pydantic_ai import Agent, RunContext, Tool, ToolDefinition
def launch_potato(target: str) -> str:
return f'Potato launched at {target}!'
async def filter_out_tools_by_name(
ctx: RunContext[bool], tool_defs: list[ToolDefinition]
) -> list[ToolDefinition] | None:
if ctx.deps:
return [tool_def for tool_def in tool_defs if tool_def.name != 'launch_potato']
return tool_defs
agent = Agent(
'test',
tools=[Tool(launch_potato)],
prepare_tools=filter_out_tools_by_name,
deps_type=bool,
)
result = agent.run_sync('testing...', deps=False)
print(result.output)
#> {"launch_potato":"Potato launched at a!"}
result = agent.run_sync('testing...', deps=True)
print(result.output)
#> success (no tool calls)
（这个例子是完整的，可以“按原样”运行）
您可以使用 prepare_tools 来：
- 根据当前模型、依赖项或其他上下文动态启用或禁用工具
- 全局修改工具定义（例如，将所有工具设置为严格模式、更改描述等）
如果同时使用了单个工具的 prepare 和智能体级别的 prepare_tools，会先对每个工具应用其 prepare 方法，然后再用生成的结果列表调用 prepare_tools。
工具执行与重试
当一个工具被执行时，其参数（由 LLM 提供）首先会使用 Pydantic 对照函数签名进行验证。如果验证失败（例如，由于类型不正确或缺少必需参数），会抛出 ValidationError，框架会自动生成一个包含验证详情的 RetryPromptPart。这个提示会发送回 LLM，告知其错误，并允许它修正参数后重试工具调用。
除了自动的验证错误外，工具自身的内部逻辑也可以通过抛出 ModelRetry 异常来明确请求重试。这对于参数在技术上有效，但在执行过程中出现问题（如瞬时网络错误，或工具判断初次尝试需要修改）的情况很有用。
from pydantic_ai import ModelRetry
def my_flaky_tool(query: str) -> str:
if query == 'bad':
# Tell the LLM the query was bad and it should try again
raise ModelRetry("The query 'bad' is not allowed. Please provide a different query.")
# ... process query ...
return 'Success!'
抛出 ModelRetry 同样会生成一个包含异常消息的 RetryPromptPart，并发送回 LLM 以指导其下一次尝试。ValidationError 和 ModelRetry 都会遵循在 Tool 或 Agent 上配置的 retries 设置。
并行工具调用与并发
当一个模型在一次响应中返回多个工具调用时，Pydantic AI 会使用 asyncio.create_task 来并发调度它们。
异步函数在事件循环上运行，而同步函数则会被移交到线程中。为获得最佳性能，务必使用异步函数，除非您正在进行阻塞式 I/O 操作（并且无法使用非阻塞库替代）或 CPU 密集型工作（如 numpy 或 scikit-learn 操作），这样可以避免不必要地将简单函数移交到线程中。
限制工具执行
您可以使用 UsageLimits(tool_calls_limit=...) 来限制单次运行中的工具执行次数。计数器仅在工具成功调用后才会增加。输出工具（用于结构化输出）不计入 tool_calls 指标。