pydantic_ai.tools

AgentDepsT module-attribute

AgentDepsT = TypeVar(
    "AgentDepsT", default=None, contravariant=True
)

代理依赖项的类型变量。

RunContext dataclass

Bases: Generic[AgentDepsT]

关于当前调用的信息。

Source code in pydantic_ai_slim/pydantic_ai/tools.py

@dataclasses.dataclass
class RunContext(Generic[AgentDepsT]):
    """Information about the current call."""

    deps: AgentDepsT
    """Dependencies for the agent."""
    model: models.Model
    """The model used in this run."""
    usage: Usage
    """LLM usage associated with the run."""
    prompt: str | Sequence[_messages.UserContent]
    """The original user prompt passed to the run."""
    messages: list[_messages.ModelMessage] = field(default_factory=list)
    """Messages exchanged in the conversation so far."""
    tool_call_id: str | None = None
    """The ID of the tool call."""
    tool_name: str | None = None
    """Name of the tool being called."""
    retry: int = 0
    """Number of retries so far."""
    run_step: int = 0
    """The current step in the run."""

    def replace_with(
        self, retry: int | None = None, tool_name: str | None | _utils.Unset = _utils.UNSET
    ) -> RunContext[AgentDepsT]:
        # Create a new `RunContext` a new `retry` value and `tool_name`.
        kwargs = {}
        if retry is not None:
            kwargs['retry'] = retry
        if tool_name is not _utils.UNSET:
            kwargs['tool_name'] = tool_name
        return dataclasses.replace(self, **kwargs)

deps instance-attribute

deps: AgentDepsT

代理的依赖项。

model instance-attribute

model: Model

在此运行中使用的模型。

usage instance-attribute

usage: Usage

与运行相关的 LLM 使用情况。

prompt instance-attribute

prompt: str | Sequence[UserContent]

传递给运行的原始用户提示。

messages class-attribute instance-attribute

messages: list[ModelMessage] = field(default_factory=list)

到目前为止在对话中交换的消息。

tool_call_id class-attribute instance-attribute

tool_call_id: str | None = None

工具调用的 ID。

tool_name class-attribute instance-attribute

tool_name: str | None = None

正在调用的工具的名称。

retry class-attribute instance-attribute

retry: int = 0

到目前为止的重试次数。

run_step class-attribute instance-attribute

run_step: int = 0

运行中的当前步骤。

ToolParams module-attribute

ToolParams = ParamSpec('ToolParams', default=...)

检索函数参数规范。

SystemPromptFunc module-attribute

SystemPromptFunc = Union[
    Callable[[RunContext[AgentDepsT]], str],
    Callable[[RunContext[AgentDepsT]], Awaitable[str]],
    Callable[[], str],
    Callable[[], Awaitable[str]],
]

一个可能接受或不接受 RunContext 作为参数，并且可能是异步或非异步的函数。

用法 SystemPromptFunc[AgentDepsT]。

ToolFuncContext module-attribute

ToolFuncContext = Callable[
    Concatenate[RunContext[AgentDepsT], ToolParams], Any
]

一个将 RunContext 作为第一个参数的工具函数。

用法 ToolContextFunc[AgentDepsT, ToolParams]。

ToolFuncPlain module-attribute

ToolFuncPlain = Callable[ToolParams, Any]

一个不将 RunContext 作为第一个参数的工具函数。

用法 ToolPlainFunc[ToolParams]。

ToolFuncEither module-attribute

ToolFuncEither = Union[
    ToolFuncContext[AgentDepsT, ToolParams],
    ToolFuncPlain[ToolParams],
]

任何一种工具函数。

这只是 ToolFuncContext 和 ToolFuncPlain 的联合。

用法 ToolFuncEither[AgentDepsT, ToolParams]。

ToolPrepareFunc module-attribute

ToolPrepareFunc: TypeAlias = (
    "Callable[[RunContext[AgentDepsT], ToolDefinition], Awaitable[ToolDefinition | None]]"
)

可以在调用时准备工具定义的函数的定义。

有关更多信息，请参阅 工具文档。

示例 — 此处 only_if_42 作为 ToolPrepareFunc 有效

from typing import Union

from pydantic_ai import RunContext, Tool
from pydantic_ai.tools import ToolDefinition

async def only_if_42(
    ctx: RunContext[int], tool_def: ToolDefinition
) -> Union[ToolDefinition, None]:
    if ctx.deps == 42:
        return tool_def

def hitchhiker(ctx: RunContext[int], answer: str) -> str:
    return f'{ctx.deps} {answer}'

hitchhiker = Tool(hitchhiker, prepare=only_if_42)

用法 ToolPrepareFunc[AgentDepsT]。

DocstringFormat module-attribute

DocstringFormat = Literal[
    "google", "numpy", "sphinx", "auto"
]

支持的文档字符串格式。

• 'google' — Google 风格 文档字符串。
• 'numpy' — Numpy 风格 文档字符串。
• 'sphinx' — Sphinx 风格 文档字符串。
• 'auto' — 根据文档字符串的结构自动推断格式。

Tool dataclass

Bases: Generic[AgentDepsT]

代理的工具函数。

Source code in pydantic_ai_slim/pydantic_ai/tools.py

@dataclass(init=False)
class Tool(Generic[AgentDepsT]):
    """A tool function for an agent."""

    function: ToolFuncEither[AgentDepsT]
    takes_ctx: bool
    max_retries: int | None
    name: str
    description: str
    prepare: ToolPrepareFunc[AgentDepsT] | None
    docstring_format: DocstringFormat
    require_parameter_descriptions: bool
    _is_async: bool = field(init=False)
    _single_arg_name: str | None = field(init=False)
    _positional_fields: list[str] = field(init=False)
    _var_positional_field: str | None = field(init=False)
    _validator: SchemaValidator = field(init=False, repr=False)
    _parameters_json_schema: ObjectJsonSchema = field(init=False)

    # TODO: Move this state off the Tool class, which is otherwise stateless.
    #   This should be tracked inside a specific agent run, not the tool.
    current_retry: int = field(default=0, init=False)

    def __init__(
        self,
        function: ToolFuncEither[AgentDepsT],
        *,
        takes_ctx: bool | None = None,
        max_retries: int | None = None,
        name: str | None = None,
        description: str | None = None,
        prepare: ToolPrepareFunc[AgentDepsT] | None = None,
        docstring_format: DocstringFormat = 'auto',
        require_parameter_descriptions: bool = False,
        schema_generator: type[GenerateJsonSchema] = GenerateToolJsonSchema,
    ):
        """Create a new tool instance.

        Example usage:

        ```python {noqa="I001"}
        from pydantic_ai import Agent, RunContext, Tool

        async def my_tool(ctx: RunContext[int], x: int, y: int) -> str:
            return f'{ctx.deps} {x} {y}'

        agent = Agent('test', tools=[Tool(my_tool)])
        ```

        or with a custom prepare method:

        ```python {noqa="I001"}
        from typing import Union

        from pydantic_ai import Agent, RunContext, Tool
        from pydantic_ai.tools import ToolDefinition

        async def my_tool(ctx: RunContext[int], x: int, y: int) -> str:
            return f'{ctx.deps} {x} {y}'

        async def prep_my_tool(
            ctx: RunContext[int], tool_def: ToolDefinition
        ) -> Union[ToolDefinition, None]:
            # only register the tool if `deps == 42`
            if ctx.deps == 42:
                return tool_def

        agent = Agent('test', tools=[Tool(my_tool, prepare=prep_my_tool)])
        ```


        Args:
            function: The Python function to call as the tool.
            takes_ctx: Whether the function takes a [`RunContext`][pydantic_ai.tools.RunContext] first argument,
                this is inferred if unset.
            max_retries: Maximum number of retries allowed for this tool, set to the agent default if `None`.
            name: Name of the tool, inferred from the function if `None`.
            description: Description of the tool, inferred from the function if `None`.
            prepare: custom method to prepare the tool definition for each step, return `None` to omit this
                tool from a given step. This is useful if you want to customise a tool at call time,
                or omit it completely from a step. See [`ToolPrepareFunc`][pydantic_ai.tools.ToolPrepareFunc].
            docstring_format: The format of the docstring, see [`DocstringFormat`][pydantic_ai.tools.DocstringFormat].
                Defaults to `'auto'`, such that the format is inferred from the structure of the docstring.
            require_parameter_descriptions: If True, raise an error if a parameter description is missing. Defaults to False.
            schema_generator: The JSON schema generator class to use. Defaults to `GenerateToolJsonSchema`.
        """
        if takes_ctx is None:
            takes_ctx = _pydantic.takes_ctx(function)

        f = _pydantic.function_schema(
            function, takes_ctx, docstring_format, require_parameter_descriptions, schema_generator
        )
        self.function = function
        self.takes_ctx = takes_ctx
        self.max_retries = max_retries
        self.name = name or function.__name__
        self.description = description or f['description']
        self.prepare = prepare
        self.docstring_format = docstring_format
        self.require_parameter_descriptions = require_parameter_descriptions
        self._is_async = inspect.iscoroutinefunction(self.function)
        self._single_arg_name = f['single_arg_name']
        self._positional_fields = f['positional_fields']
        self._var_positional_field = f['var_positional_field']
        self._validator = f['validator']
        self._parameters_json_schema = f['json_schema']

    async def prepare_tool_def(self, ctx: RunContext[AgentDepsT]) -> ToolDefinition | None:
        """Get the tool definition.

        By default, this method creates a tool definition, then either returns it, or calls `self.prepare`
        if it's set.

        Returns:
            return a `ToolDefinition` or `None` if the tools should not be registered for this run.
        """
        tool_def = ToolDefinition(
            name=self.name,
            description=self.description,
            parameters_json_schema=self._parameters_json_schema,
        )
        if self.prepare is not None:
            return await self.prepare(ctx, tool_def)
        else:
            return tool_def

    async def run(
        self, message: _messages.ToolCallPart, run_context: RunContext[AgentDepsT]
    ) -> _messages.ToolReturnPart | _messages.RetryPromptPart:
        """Run the tool function asynchronously."""
        try:
            if isinstance(message.args, str):
                args_dict = self._validator.validate_json(message.args)
            else:
                args_dict = self._validator.validate_python(message.args)
        except ValidationError as e:
            return self._on_error(e, message)

        args, kwargs = self._call_args(args_dict, message, run_context)
        try:
            if self._is_async:
                function = cast(Callable[[Any], Awaitable[str]], self.function)
                response_content = await function(*args, **kwargs)
            else:
                function = cast(Callable[[Any], str], self.function)
                response_content = await _utils.run_in_executor(function, *args, **kwargs)
        except ModelRetry as e:
            return self._on_error(e, message)

        self.current_retry = 0
        return _messages.ToolReturnPart(
            tool_name=message.tool_name,
            content=response_content,
            tool_call_id=message.tool_call_id,
        )

    def _call_args(
        self,
        args_dict: dict[str, Any],
        message: _messages.ToolCallPart,
        run_context: RunContext[AgentDepsT],
    ) -> tuple[list[Any], dict[str, Any]]:
        if self._single_arg_name:
            args_dict = {self._single_arg_name: args_dict}

        ctx = dataclasses.replace(
            run_context,
            retry=self.current_retry,
            tool_name=message.tool_name,
            tool_call_id=message.tool_call_id,
        )
        args = [ctx] if self.takes_ctx else []
        for positional_field in self._positional_fields:
            args.append(args_dict.pop(positional_field))
        if self._var_positional_field:
            args.extend(args_dict.pop(self._var_positional_field))

        return args, args_dict

    def _on_error(
        self, exc: ValidationError | ModelRetry, call_message: _messages.ToolCallPart
    ) -> _messages.RetryPromptPart:
        self.current_retry += 1
        if self.max_retries is None or self.current_retry > self.max_retries:
            raise UnexpectedModelBehavior(f'Tool exceeded max retries count of {self.max_retries}') from exc
        else:
            if isinstance(exc, ValidationError):
                content = exc.errors(include_url=False)
            else:
                content = exc.message
            return _messages.RetryPromptPart(
                tool_name=call_message.tool_name,
                content=content,
                tool_call_id=call_message.tool_call_id,
            )

__init__

__init__(
    function: ToolFuncEither[AgentDepsT],
    *,
    takes_ctx: bool | None = None,
    max_retries: int | None = None,
    name: str | None = None,
    description: str | None = None,
    prepare: ToolPrepareFunc[AgentDepsT] | None = None,
    docstring_format: DocstringFormat = "auto",
    require_parameter_descriptions: bool = False,
    schema_generator: type[
        GenerateJsonSchema
    ] = GenerateToolJsonSchema
)

创建一个新的工具实例。

用法示例

from pydantic_ai import Agent, RunContext, Tool

async def my_tool(ctx: RunContext[int], x: int, y: int) -> str:
    return f'{ctx.deps} {x} {y}'

agent = Agent('test', tools=[Tool(my_tool)])

或使用自定义准备方法

from typing import Union

from pydantic_ai import Agent, RunContext, Tool
from pydantic_ai.tools import ToolDefinition

async def my_tool(ctx: RunContext[int], x: int, y: int) -> str:
    return f'{ctx.deps} {x} {y}'

async def prep_my_tool(
    ctx: RunContext[int], tool_def: ToolDefinition
) -> Union[ToolDefinition, None]:
    # only register the tool if `deps == 42`
    if ctx.deps == 42:
        return tool_def

agent = Agent('test', tools=[Tool(my_tool, prepare=prep_my_tool)])

参数

名称	类型	描述	默认值
函数	ToolFuncEither[AgentDepsT]	要作为工具调用的 Python 函数。	必需
takes_ctx	bool | None	该函数是否将 RunContext 作为第一个参数，如果未设置，则会推断出来。	None
max_retries	int | None	此工具允许的最大重试次数，如果为 None，则设置为代理默认值。	None
名称	str | None	工具的名称，如果为 None，则从函数推断。	None
描述	str | None	工具的描述，如果为 None，则从函数推断。	None
prepare	ToolPrepareFunc[AgentDepsT] | None	用于为每个步骤准备工具定义的自定义方法，返回 None 以从给定步骤中省略此工具。 如果你想在调用时自定义工具，或者从步骤中完全省略它，这将非常有用。 请参阅 ToolPrepareFunc。	None
docstring_format	DocstringFormat	文档字符串的格式，请参阅 DocstringFormat。 默认为 'auto'，这样格式会从文档字符串的结构中推断出来。	'auto'
require_parameter_descriptions	bool	如果为 True，则在缺少参数描述时引发错误。 默认为 False。	False
schema_generator	type[GenerateJsonSchema]	要使用的 JSON schema 生成器类。 默认为 GenerateToolJsonSchema。	GenerateToolJsonSchema

Source code in pydantic_ai_slim/pydantic_ai/tools.py

def __init__(
    self,
    function: ToolFuncEither[AgentDepsT],
    *,
    takes_ctx: bool | None = None,
    max_retries: int | None = None,
    name: str | None = None,
    description: str | None = None,
    prepare: ToolPrepareFunc[AgentDepsT] | None = None,
    docstring_format: DocstringFormat = 'auto',
    require_parameter_descriptions: bool = False,
    schema_generator: type[GenerateJsonSchema] = GenerateToolJsonSchema,
):
    """Create a new tool instance.

    Example usage:

    ```python {noqa="I001"}
    from pydantic_ai import Agent, RunContext, Tool

    async def my_tool(ctx: RunContext[int], x: int, y: int) -> str:
        return f'{ctx.deps} {x} {y}'

    agent = Agent('test', tools=[Tool(my_tool)])
    ```

    or with a custom prepare method:

    ```python {noqa="I001"}
    from typing import Union

    from pydantic_ai import Agent, RunContext, Tool
    from pydantic_ai.tools import ToolDefinition

    async def my_tool(ctx: RunContext[int], x: int, y: int) -> str:
        return f'{ctx.deps} {x} {y}'

    async def prep_my_tool(
        ctx: RunContext[int], tool_def: ToolDefinition
    ) -> Union[ToolDefinition, None]:
        # only register the tool if `deps == 42`
        if ctx.deps == 42:
            return tool_def

    agent = Agent('test', tools=[Tool(my_tool, prepare=prep_my_tool)])
    ```


    Args:
        function: The Python function to call as the tool.
        takes_ctx: Whether the function takes a [`RunContext`][pydantic_ai.tools.RunContext] first argument,
            this is inferred if unset.
        max_retries: Maximum number of retries allowed for this tool, set to the agent default if `None`.
        name: Name of the tool, inferred from the function if `None`.
        description: Description of the tool, inferred from the function if `None`.
        prepare: custom method to prepare the tool definition for each step, return `None` to omit this
            tool from a given step. This is useful if you want to customise a tool at call time,
            or omit it completely from a step. See [`ToolPrepareFunc`][pydantic_ai.tools.ToolPrepareFunc].
        docstring_format: The format of the docstring, see [`DocstringFormat`][pydantic_ai.tools.DocstringFormat].
            Defaults to `'auto'`, such that the format is inferred from the structure of the docstring.
        require_parameter_descriptions: If True, raise an error if a parameter description is missing. Defaults to False.
        schema_generator: The JSON schema generator class to use. Defaults to `GenerateToolJsonSchema`.
    """
    if takes_ctx is None:
        takes_ctx = _pydantic.takes_ctx(function)

    f = _pydantic.function_schema(
        function, takes_ctx, docstring_format, require_parameter_descriptions, schema_generator
    )
    self.function = function
    self.takes_ctx = takes_ctx
    self.max_retries = max_retries
    self.name = name or function.__name__
    self.description = description or f['description']
    self.prepare = prepare
    self.docstring_format = docstring_format
    self.require_parameter_descriptions = require_parameter_descriptions
    self._is_async = inspect.iscoroutinefunction(self.function)
    self._single_arg_name = f['single_arg_name']
    self._positional_fields = f['positional_fields']
    self._var_positional_field = f['var_positional_field']
    self._validator = f['validator']
    self._parameters_json_schema = f['json_schema']

prepare_tool_def async

prepare_tool_def(
    ctx: RunContext[AgentDepsT],
) -> ToolDefinition | None

获取工具定义。

默认情况下，此方法创建一个工具定义，然后返回它，或者在设置了 self.prepare 时调用它。

返回

类型	描述
ToolDefinition | None	如果不应为此运行注册工具，则返回 ToolDefinition 或 None。

Source code in pydantic_ai_slim/pydantic_ai/tools.py

async def prepare_tool_def(self, ctx: RunContext[AgentDepsT]) -> ToolDefinition | None:
    """Get the tool definition.

    By default, this method creates a tool definition, then either returns it, or calls `self.prepare`
    if it's set.

    Returns:
        return a `ToolDefinition` or `None` if the tools should not be registered for this run.
    """
    tool_def = ToolDefinition(
        name=self.name,
        description=self.description,
        parameters_json_schema=self._parameters_json_schema,
    )
    if self.prepare is not None:
        return await self.prepare(ctx, tool_def)
    else:
        return tool_def

run async

run(
    message: ToolCallPart,
    run_context: RunContext[AgentDepsT],
) -> ToolReturnPart | RetryPromptPart

异步运行工具函数。

Source code in pydantic_ai_slim/pydantic_ai/tools.py

async def run(
    self, message: _messages.ToolCallPart, run_context: RunContext[AgentDepsT]
) -> _messages.ToolReturnPart | _messages.RetryPromptPart:
    """Run the tool function asynchronously."""
    try:
        if isinstance(message.args, str):
            args_dict = self._validator.validate_json(message.args)
        else:
            args_dict = self._validator.validate_python(message.args)
    except ValidationError as e:
        return self._on_error(e, message)

    args, kwargs = self._call_args(args_dict, message, run_context)
    try:
        if self._is_async:
            function = cast(Callable[[Any], Awaitable[str]], self.function)
            response_content = await function(*args, **kwargs)
        else:
            function = cast(Callable[[Any], str], self.function)
            response_content = await _utils.run_in_executor(function, *args, **kwargs)
    except ModelRetry as e:
        return self._on_error(e, message)

    self.current_retry = 0
    return _messages.ToolReturnPart(
        tool_name=message.tool_name,
        content=response_content,
        tool_call_id=message.tool_call_id,
    )

ObjectJsonSchema module-attribute

ObjectJsonSchema: TypeAlias = dict[str, Any]

表示对象 JSON schema 的类型，例如 "type": "object" 的情况。

此类型用于在 ToolDefinition 中定义工具参数（又名参数）。

使用 PEP-728，这应该是一个带有 type: Literal['object'] 和 extra_parts=Any 的 TypedDict。

ToolDefinition dataclass

传递给模型的工具的定义。

这用于函数工具和结果工具。

Source code in pydantic_ai_slim/pydantic_ai/tools.py

@dataclass
class ToolDefinition:
    """Definition of a tool passed to a model.

    This is used for both function tools result tools.
    """

    name: str
    """The name of the tool."""

    description: str
    """The description of the tool."""

    parameters_json_schema: ObjectJsonSchema
    """The JSON schema for the tool's parameters."""

    outer_typed_dict_key: str | None = None
    """The key in the outer [TypedDict] that wraps a result tool.

    This will only be set for result tools which don't have an `object` JSON schema.
    """

name instance-attribute

name: str

工具的名称。

description instance-attribute

description: str

工具的描述。

parameters_json_schema instance-attribute

parameters_json_schema: ObjectJsonSchema

工具参数的 JSON schema。

outer_typed_dict_key class-attribute instance-attribute

outer_typed_dict_key: str | None = None

包装结果工具的外部 [TypedDict] 中的键。

这仅会为没有 object JSON schema 的结果工具设置。