跳过内容

功能工具

函数工具为模型提供了一种机制,以便获取额外的信息来帮助它们生成响应。

当将代理可能需要的所有上下文放入系统提示变得不切实际或不可行时,它们是有用的,或者当您希望通过将生成响应所需的一些逻辑推迟到另一个(不一定是AI驱动的)工具来使代理的行为更加确定性或可靠时。

功能工具与RAG

函数工具基本上是RAG(检索增强生成)的“R”——它们通过允许模型请求额外的信息来增强模型的能力。

PydanticAI工具和RAG之间的主要语义差异在于RAG与向量搜索同义,而PydanticAI工具则更具通用性。(注意:我们可能会在未来添加对向量搜索功能的支持,特别是生成嵌入的API。请参见#58

有多种方法可以将工具注册到代理中:

@agent.tool 被认为是默认的装饰器,因为在大多数情况下,工具需要访问代理上下文。

这是一个同时使用两者的例子:

dice_game.py
import random

from pydantic_ai import Agent, RunContext

agent = Agent(
    'google-gla:gemini-1.5-flash',  # (1)!
    deps_type=str,  # (2)!
    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  # (3)!
def roll_die() -> str:
    """Roll a six-sided die and return the result."""
    return str(random.randint(1, 6))


@agent.tool  # (4)!
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')  # (5)!
print(dice_result.data)
#> Congratulations Anne, you guessed correctly! You're a winner!
  1. 这是一个相当简单的任务,因此我们可以使用快速且便宜的Gemini闪存模型。
  2. 我们将用户的名字作为依赖,为了简单起见,我们仅使用名字作为字符串作为依赖。
  3. 该工具不需要任何上下文,它只是返回一个随机数字。在这种情况下,您可能可以使用一个动态系统提示。
  4. 该工具需要玩家的名字,因此它使用 RunContext 来访问依赖项,这里依赖项只是玩家的名字。
  5. 运行代理,将玩家的名称作为依赖。

(这个例子是完整的,可以“原样”运行)

让我们打印出那个游戏的信息,看看发生了什么:

dice_game_messages.py
from dice_game import dice_result

print(dice_result.all_messages())
"""
[
    ModelRequest(
        parts=[
            SystemPromptPart(
                content="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.",
                dynamic_ref=None,
                part_kind='system-prompt',
            ),
            UserPromptPart(
                content='My guess is 4',
                timestamp=datetime.datetime(...),
                part_kind='user-prompt',
            ),
        ],
        kind='request',
    ),
    ModelResponse(
        parts=[
            ToolCallPart(
                tool_name='roll_die', args={}, tool_call_id=None, part_kind='tool-call'
            )
        ],
        model_name='function:model_logic',
        timestamp=datetime.datetime(...),
        kind='response',
    ),
    ModelRequest(
        parts=[
            ToolReturnPart(
                tool_name='roll_die',
                content='4',
                tool_call_id=None,
                timestamp=datetime.datetime(...),
                part_kind='tool-return',
            )
        ],
        kind='request',
    ),
    ModelResponse(
        parts=[
            ToolCallPart(
                tool_name='get_player_name',
                args={},
                tool_call_id=None,
                part_kind='tool-call',
            )
        ],
        model_name='function:model_logic',
        timestamp=datetime.datetime(...),
        kind='response',
    ),
    ModelRequest(
        parts=[
            ToolReturnPart(
                tool_name='get_player_name',
                content='Anne',
                tool_call_id=None,
                timestamp=datetime.datetime(...),
                part_kind='tool-return',
            )
        ],
        kind='request',
    ),
    ModelResponse(
        parts=[
            TextPart(
                content="Congratulations Anne, you guessed correctly! You're a winner!",
                part_kind='text',
            )
        ],
        model_name='function:model_logic',
        timestamp=datetime.datetime(...),
        kind='response',
    ),
]
"""

我们可以用一个图示来表示这一点:

sequenceDiagram
    participant Agent
    participant LLM

    Note over Agent: Send prompts
    Agent ->> LLM: System: "You're a dice game..."<br>User: "My guess is 4"
    activate LLM
    Note over LLM: LLM decides to use<br>a tool

    LLM ->> Agent: Call tool<br>roll_die()
    deactivate LLM
    activate Agent
    Note over Agent: Rolls a six-sided die

    Agent -->> LLM: ToolReturn<br>"4"
    deactivate Agent
    activate LLM
    Note over LLM: LLM decides to use<br>another tool

    LLM ->> Agent: Call tool<br>get_player_name()
    deactivate LLM
    activate Agent
    Note over Agent: Retrieves player name
    Agent -->> LLM: ToolReturn<br>"Anne"
    deactivate Agent
    activate LLM
    Note over LLM: LLM constructs final response

    LLM ->> Agent: ModelResponse<br>"Congratulations Anne, ..."
    deactivate LLM
    Note over Agent: Game session complete

通过 kwarg 注册函数工具

除了使用装饰器,我们还可以通过Agent构造函数中的tools参数注册工具。这在您想要重用工具时非常有用,并且也可以提供对工具的更细粒度控制。

dice_game_tool_kwarg.py
import random

from pydantic_ai import Agent, RunContext, Tool


def roll_die() -> str:
    """Roll a six-sided die and return the result."""
    return str(random.randint(1, 6))


def get_player_name(ctx: RunContext[str]) -> str:
    """Get the player's name."""
    return ctx.deps


agent_a = Agent(
    'google-gla:gemini-1.5-flash',
    deps_type=str,
    tools=[roll_die, get_player_name],  # (1)!
)
agent_b = Agent(
    'google-gla:gemini-1.5-flash',
    deps_type=str,
    tools=[  # (2)!
        Tool(roll_die, takes_ctx=False),
        Tool(get_player_name, takes_ctx=True),
    ],
)
dice_result = agent_b.run_sync('My guess is 4', deps='Anne')
print(dice_result.data)
#> Congratulations Anne, you guessed correctly! You're a winner!
  1. 通过 Agent 构造函数注册工具的最简单方法是传递一个函数列表,函数签名会被检查以确定该工具是否接受 RunContext
  2. agent_aagent_b 是相同的 — 但我们可以使用 Tool 来重用工具定义,并对工具的定义进行更细粒度的控制,例如设置它们的名称或描述,或使用自定义的 prepare 方法。

(这个例子是完整的,可以“原样”运行)

函数工具与结构化结果

顾名思义,函数工具使用模型的“工具”或“函数”API,让模型知道可以调用哪些内容。工具或函数还用于定义结构化响应的模式,因此模型可能会访问许多工具,其中一些调用函数工具,而其他工具则结束运行并返回结果。

函数工具和架构

函数参数是从函数签名中提取的,除了 RunContext 以外的所有参数用于构建该工具调用的模式。

更好的是,PydanticAI 从函数中提取文档字符串,并且(多亏了 griffe)从文档字符串中提取参数描述并将其添加到模式中。

Griffe 支持googlenumpysphinx 风格的文档字符串中提取参数描述。PydanticAI 将根据文档字符串推断要使用的格式,但您可以通过使用 docstring_format 显式设置它。您还可以通过设置 require_parameter_descriptions=True 来强制要求参数描述。如果缺少参数描述,这将引发一个 UserError

为了演示一个工具的架构,这里我们使用 FunctionModel 来打印一个模型将接收到的架构:

tool_schema.py
from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart
from pydantic_ai.models.function import AgentInfo, FunctionModel

agent = Agent()


@agent.tool_plain(docstring_format='google', require_parameter_descriptions=True)
def foobar(a: int, b: str, c: dict[str, list[float]]) -> str:
    """Get me foobar.

    Args:
        a: apple pie
        b: banana cake
        c: carrot smoothie
    """
    return f'{a} {b} {c}'


def print_schema(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
    tool = info.function_tools[0]
    print(tool.description)
    #> Get me foobar.
    print(tool.parameters_json_schema)
    """
    {
        'properties': {
            'a': {'description': 'apple pie', 'title': 'A', 'type': 'integer'},
            'b': {'description': 'banana cake', 'title': 'B', 'type': 'string'},
            'c': {
                'additionalProperties': {'items': {'type': 'number'}, 'type': 'array'},
                'description': 'carrot smoothie',
                'title': 'C',
                'type': 'object',
            },
        },
        'required': ['a', 'b', 'c'],
        'type': 'object',
        'additionalProperties': False,
    }
    """
    return ModelResponse(parts=[TextPart('foobar')])


agent.run_sync('hello', model=FunctionModel(print_schema))

(这个例子是完整的,可以“原样”运行)

工具的返回类型可以是任何Pydantic可以序列化为JSON的东西,因为某些模型(例如:Gemini)支持半结构化的返回值,某些则期望文本(OpenAI),但似乎同样擅长从数据中提取意义。如果返回一个Python对象而模型期望一个字符串,则该值将序列化为JSON。

如果一个工具有一个可以在JSON模式中表示为对象的单一参数(例如dataclass、TypedDict、pydantic模型),则该工具的模式简化为仅该对象。

这是一个示例,我们使用 TestModel.last_model_request_parameters 来检查将传递给模型的工具架构。

single_parameter_tool.py
from pydantic import BaseModel

from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel

agent = Agent()


class Foobar(BaseModel):
    """This is a Foobar"""

    x: int
    y: str
    z: float = 3.14


@agent.tool_plain
def foobar(f: Foobar) -> str:
    return str(f)


test_model = TestModel()
result = agent.run_sync('hello', model=test_model)
print(result.data)
#> {"foobar":"x=0 y='a' z=3.14"}
print(test_model.last_model_request_parameters.function_tools)
"""
[
    ToolDefinition(
        name='foobar',
        description='This is a Foobar',
        parameters_json_schema={
            'properties': {
                'x': {'title': 'X', 'type': 'integer'},
                'y': {'title': 'Y', 'type': 'string'},
                'z': {'default': 3.14, 'title': 'Z', 'type': 'number'},
            },
            'required': ['x', 'y'],
            'title': 'Foobar',
            'type': 'object',
        },
        outer_typed_dict_key=None,
    )
]
"""

(这个例子是完整的,可以“原样”运行)

动态函数工具

工具可以选择性地通过另一个函数定义:prepare,该函数在运行的每一步被调用,以自定义传递给模型的工具定义,或完全省略该步骤中的工具。

可以通过 prepare 关键字参数将 prepare 方法注册到任何工具注册机制:

prepare 方法应为ToolPrepareFunc类型,该函数接受RunContext和预构建的ToolDefinition,并应返回该ToolDefinition(可以修改也可以不修改),返回一个新的ToolDefinition,或者返回None以指示此工具不应在该步骤中注册。

这是一个简单的 prepare 方法,它仅在依赖项的值为 42 时才包含该工具。

与前面的示例一样,我们使用 TestModel 来演示行为,而不调用真实模型。

tool_only_if_42.py
from typing import Union

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

agent = Agent('test')


async def only_if_42(
    ctx: RunContext[int], tool_def: ToolDefinition
) -> Union[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.data)
#> success (no tool calls)
result = agent.run_sync('testing...', deps=42)
print(result.data)
#> {"hitchhiker":"42 a"}

(这个例子是完整的,可以“原样”运行)

这是一个更复杂的例子,我们根据deps的值来更改name参数的描述

为了变化,我们使用Tool数据类创建了这个工具。

customize_name.py
from __future__ import annotations

from typing import Literal

from pydantic_ai import Agent, RunContext
from pydantic_ai.models.test import TestModel
from pydantic_ai.tools import Tool, ToolDefinition


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.data)
#> {"greet":"hello a"}
print(test_model.last_model_request_parameters.function_tools)
"""
[
    ToolDefinition(
        name='greet',
        description='',
        parameters_json_schema={
            'properties': {
                'name': {
                    'title': 'Name',
                    'type': 'string',
                    'description': 'Name of the human to greet.',
                }
            },
            'required': ['name'],
            'type': 'object',
            'additionalProperties': False,
        },
        outer_typed_dict_key=None,
    )
]
"""

(这个例子是完整的,可以“原样”运行)