跳转到内容

工具集

工具集(toolset)表示可以一次性注册给代理(agent)的一系列工具(tools)。它们可以被不同的代理重用,在运行时或测试期间进行替换,还可以通过组合来动态过滤可用的工具、修改工具定义或更改工具的执行行为。一个工具集可以包含本地定义的函数,也可以依赖外部服务来提供这些函数,或者实现自定义逻辑来列出可用工具并处理它们的调用。

工具集用于(除许多其他用途外)定义代理可用的 MCP 服务器。Pydantic AI 包含多种工具集,下文将进行描述,您也可以通过继承 AbstractToolset 类来定义一个自定义工具集

在代理运行期间可用的工具集可以通过以下四种不同方式指定:

  • 在代理构建时,通过 Agenttoolsets 关键字参数指定,该参数接受工具集实例以及动态生成工具集的函数,这些函数基于代理的运行上下文
  • 在代理运行时,通过 agent.run()agent.run_sync()agent.run_stream()agent.iter()toolsets 关键字参数指定。这些工具集将是注册在 Agent 上的工具集的补充
  • 动态地,通过 @agent.toolset 装饰器,它允许您基于代理的运行上下文来构建工具集
  • 作为上下文覆盖,通过 agent.override() 上下文管理器的 toolsets 关键字参数指定。在上下文管理器生命周期内,这些工具集将替换在代理构建时或运行时提供的工具集
toolsets.py
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai.toolsets import FunctionToolset


def agent_tool():
    return "I'm registered directly on the agent"


def extra_tool():
    return "I'm passed as an extra tool for a specific run"


def override_tool():
    return 'I override all other tools'


agent_toolset = FunctionToolset(tools=[agent_tool]) # (1)!
extra_toolset = FunctionToolset(tools=[extra_tool])
override_toolset = FunctionToolset(tools=[override_tool])

test_model = TestModel() # (2)!
agent = Agent(test_model, toolsets=[agent_toolset])

result = agent.run_sync('What tools are available?')
print([t.name for t in test_model.last_model_request_parameters.function_tools])
#> ['agent_tool']

result = agent.run_sync('What tools are available?', toolsets=[extra_toolset])
print([t.name for t in test_model.last_model_request_parameters.function_tools])
#> ['agent_tool', 'extra_tool']

with agent.override(toolsets=[override_toolset]):
    result = agent.run_sync('What tools are available?', toolsets=[extra_toolset]) # (3)!
    print([t.name for t in test_model.last_model_request_parameters.function_tools])
    #> ['override_tool']
  1. FunctionToolset 将在下一节详细解释。
  2. 我们在这里使用 TestModel,因为它能让我们很容易地看到每次运行中可用的工具。
  3. 这个 extra_toolset 将被忽略,因为我们在一个覆盖(override)上下文中。

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

函数工具集

顾名思义,FunctionToolset 将本地定义的函数作为工具提供。

函数可以通过以下三种方式添加为工具:

以任何这些方式注册的函数都可以定义一个初始的 ctx: RunContext 参数,以便接收代理的运行上下文add_function()add_tool() 方法也可以从一个工具函数内部使用,以便在一次运行期间动态注册新工具,供未来的运行步骤使用。

function_toolset.py
from datetime import datetime

from pydantic_ai import Agent, RunContext
from pydantic_ai.models.test import TestModel
from pydantic_ai.toolsets import FunctionToolset


def temperature_celsius(city: str) -> float:
    return 21.0


def temperature_fahrenheit(city: str) -> float:
    return 69.8


weather_toolset = FunctionToolset(tools=[temperature_celsius, temperature_fahrenheit])


@weather_toolset.tool
def conditions(ctx: RunContext, city: str) -> str:
    if ctx.run_step % 2 == 0:
        return "It's sunny"
    else:
        return "It's raining"


datetime_toolset = FunctionToolset()
datetime_toolset.add_function(lambda: datetime.now(), name='now')

test_model = TestModel()  # (1)!
agent = Agent(test_model)

result = agent.run_sync('What tools are available?', toolsets=[weather_toolset])
print([t.name for t in test_model.last_model_request_parameters.function_tools])
#> ['temperature_celsius', 'temperature_fahrenheit', 'conditions']

result = agent.run_sync('What tools are available?', toolsets=[datetime_toolset])
print([t.name for t in test_model.last_model_request_parameters.function_tools])
#> ['now']
  1. 我们在这里使用 TestModel,因为它能让我们很容易地看到每次运行中可用的工具。

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

工具集组合

工具集可以被组合,以动态过滤可用的工具、修改工具定义或更改工具的执行行为。多个工具集也可以合并成一个。

合并工具集

CombinedToolset 接收一个工具集列表,并让它们可以作为一个整体来使用。

combined_toolset.py
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai.toolsets import CombinedToolset

from function_toolset import datetime_toolset, weather_toolset

combined_toolset = CombinedToolset([weather_toolset, datetime_toolset])

test_model = TestModel() # (1)!
agent = Agent(test_model, toolsets=[combined_toolset])
result = agent.run_sync('What tools are available?')
print([t.name for t in test_model.last_model_request_parameters.function_tools])
#> ['temperature_celsius', 'temperature_fahrenheit', 'conditions', 'now']
  1. 我们在这里使用 TestModel,因为它能让我们很容易地看到每次运行中可用的工具。

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

过滤工具

FilteredToolset 包装一个工具集,并在每次运行步骤之前,根据一个用户定义的函数来过滤可用的工具。该函数接收代理的运行上下文和每个工具的 ToolDefinition,并返回一个布尔值,以指示给定的工具是否应该可用。

为了方便地链接不同的修改,您也可以在任何工具集上调用 filtered(),而不是直接构建一个 FilteredToolset

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

from combined_toolset import combined_toolset

filtered_toolset = combined_toolset.filtered(lambda ctx, tool_def: 'fahrenheit' not in tool_def.name)

test_model = TestModel() # (1)!
agent = Agent(test_model, toolsets=[filtered_toolset])
result = agent.run_sync('What tools are available?')
print([t.name for t in test_model.last_model_request_parameters.function_tools])
#> ['weather_temperature_celsius', 'weather_conditions', 'datetime_now']
  1. 我们在这里使用 TestModel,因为它能让我们很容易地看到每次运行中可用的工具。

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

为工具名称添加前缀

PrefixedToolset 包装一个工具集,并为每个工具名称添加一个前缀,以防止不同工具集之间的工具名称冲突。

为了方便地链接不同的修改,您也可以在任何工具集上调用 prefixed(),而不是直接构建一个 PrefixedToolset

combined_toolset.py
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai.toolsets import CombinedToolset

from function_toolset import datetime_toolset, weather_toolset

combined_toolset = CombinedToolset(
    [
        weather_toolset.prefixed('weather'),
        datetime_toolset.prefixed('datetime')
    ]
)

test_model = TestModel() # (1)!
agent = Agent(test_model, toolsets=[combined_toolset])
result = agent.run_sync('What tools are available?')
print([t.name for t in test_model.last_model_request_parameters.function_tools])
"""
[
    'weather_temperature_celsius',
    'weather_temperature_fahrenheit',
    'weather_conditions',
    'datetime_now',
]
"""
  1. 我们在这里使用 TestModel,因为它能让我们很容易地看到每次运行中可用的工具。

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

重命名工具

RenamedToolset 包装一个工具集,并允许您使用一个字典来重命名工具,该字典将新名称映射到原始名称。当一个工具集提供的名称有歧义或与其他工具集定义的工具冲突时,这个功能很有用,但为它们添加前缀会产生一个不必要地长或可能让模型混淆的名称。

为了方便地链接不同的修改,您也可以在任何工具集上调用 renamed(),而不是直接构建一个 RenamedToolset

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

from combined_toolset import combined_toolset

renamed_toolset = combined_toolset.renamed(
    {
        'current_time': 'datetime_now',
        'temperature_celsius': 'weather_temperature_celsius',
        'temperature_fahrenheit': 'weather_temperature_fahrenheit'
    }
)

test_model = TestModel() # (1)!
agent = Agent(test_model, toolsets=[renamed_toolset])
result = agent.run_sync('What tools are available?')
print([t.name for t in test_model.last_model_request_parameters.function_tools])
"""
['temperature_celsius', 'temperature_fahrenheit', 'weather_conditions', 'current_time']
"""
  1. 我们在这里使用 TestModel,因为它能让我们很容易地看到每次运行中可用的工具。

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

动态工具定义

PreparedToolset 允许您在代理运行的每个步骤之前,使用一个用户定义的函数来修改整个可用工具列表。该函数接收代理的运行上下文和一个 ToolDefinitions 列表,并返回一个修改后的 ToolDefinitions 列表。

这相当于 Agentprepare_tools 参数的工具集特定版本,该参数用于准备在代理上跨工具集注册的所有工具定义。

请注意,使用 PreparedToolset 无法添加或重命名工具。相反,您可以使用 FunctionToolset.add_function()RenamedToolset

为了方便地链接不同的修改,您也可以在任何工具集上调用 prepared(),而不是直接构建一个 PreparedToolset

prepared_toolset.py
from dataclasses import replace

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

from renamed_toolset import renamed_toolset

descriptions = {
    'temperature_celsius': 'Get the temperature in degrees Celsius',
    'temperature_fahrenheit': 'Get the temperature in degrees Fahrenheit',
    'weather_conditions': 'Get the current weather conditions',
    'current_time': 'Get the current time',
}

async def add_descriptions(ctx: RunContext, tool_defs: list[ToolDefinition]) -> list[ToolDefinition] | None:
    return [
        replace(tool_def, description=description)
        if (description := descriptions.get(tool_def.name, None))
        else tool_def
        for tool_def
        in tool_defs
    ]

prepared_toolset = renamed_toolset.prepared(add_descriptions)

test_model = TestModel() # (1)!
agent = Agent(test_model, toolsets=[prepared_toolset])
result = agent.run_sync('What tools are available?')
print(test_model.last_model_request_parameters.function_tools)
"""
[
    ToolDefinition(
        name='temperature_celsius',
        parameters_json_schema={
            'additionalProperties': False,
            'properties': {'city': {'type': 'string'}},
            'required': ['city'],
            'type': 'object',
        },
        description='Get the temperature in degrees Celsius',
    ),
    ToolDefinition(
        name='temperature_fahrenheit',
        parameters_json_schema={
            'additionalProperties': False,
            'properties': {'city': {'type': 'string'}},
            'required': ['city'],
            'type': 'object',
        },
        description='Get the temperature in degrees Fahrenheit',
    ),
    ToolDefinition(
        name='weather_conditions',
        parameters_json_schema={
            'additionalProperties': False,
            'properties': {'city': {'type': 'string'}},
            'required': ['city'],
            'type': 'object',
        },
        description='Get the current weather conditions',
    ),
    ToolDefinition(
        name='current_time',
        parameters_json_schema={
            'additionalProperties': False,
            'properties': {},
            'type': 'object',
        },
        description='Get the current time',
    ),
]
"""
  1. 我们在这里使用 TestModel,因为它能让我们很容易地看到每次运行中可用的工具。

要求工具批准

ApprovalRequiredToolset 包装一个工具集,并允许您根据一个用户定义的函数,动态地为给定的工具调用要求批准。该函数接收代理的运行上下文、工具的 ToolDefinition 以及已验证的工具调用参数。如果未提供函数,所有工具调用都将需要批准。

为了方便地链接不同的修改,您也可以在任何工具集上调用 approval_required(),而不是直接构建一个 ApprovalRequiredToolset

有关如何处理调用需要批准的工具的代理运行以及如何传入结果的更多信息,请参见人在环路中的工具批准文档。

approval_required_toolset.py
from pydantic_ai import Agent, DeferredToolRequests, DeferredToolResults
from pydantic_ai.models.test import TestModel

from prepared_toolset import prepared_toolset

approval_required_toolset = prepared_toolset.approval_required(lambda ctx, tool_def, tool_args: tool_def.name.startswith('temperature'))

test_model = TestModel(call_tools=['temperature_celsius', 'temperature_fahrenheit']) # (1)!
agent = Agent(
    test_model,
    toolsets=[approval_required_toolset],
    output_type=[str, DeferredToolRequests],
)
result = agent.run_sync('Call the temperature tools')
messages = result.all_messages()
print(result.output)
"""
DeferredToolRequests(
    calls=[],
    approvals=[
        ToolCallPart(
            tool_name='temperature_celsius',
            args={'city': 'a'},
            tool_call_id='pyd_ai_tool_call_id__temperature_celsius',
        ),
        ToolCallPart(
            tool_name='temperature_fahrenheit',
            args={'city': 'a'},
            tool_call_id='pyd_ai_tool_call_id__temperature_fahrenheit',
        ),
    ],
)
"""

result = agent.run_sync(
    message_history=messages,
    deferred_tool_results=DeferredToolResults(
        approvals={
            'pyd_ai_tool_call_id__temperature_celsius': True,
            'pyd_ai_tool_call_id__temperature_fahrenheit': False,
        }
    )
)
print(result.output)
#> {"temperature_celsius":21.0,"temperature_fahrenheit":"The tool call was denied."}
  1. 我们在这里使用 TestModel,因为它能让我们很容易地指定要调用哪些工具。

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

更改工具执行方式

WrapperToolset 包装另一个工具集,并将所有职责委托给它。

它默认是一个无操作(no-op)的类,但您可以通过子类化 WrapperToolset 并重写 call_tool() 方法来更改被包装工具集的工具执行行为。

logging_toolset.py
import asyncio

from typing_extensions import Any

from pydantic_ai import Agent, RunContext
from pydantic_ai.models.test import TestModel
from pydantic_ai.toolsets import ToolsetTool, WrapperToolset

from prepared_toolset import prepared_toolset

LOG = []

class LoggingToolset(WrapperToolset):
    async def call_tool(self, name: str, tool_args: dict[str, Any], ctx: RunContext, tool: ToolsetTool) -> Any:
        LOG.append(f'Calling tool {name!r} with args: {tool_args!r}')
        try:
            await asyncio.sleep(0.1 * len(LOG)) # (1)!

            result = await super().call_tool(name, tool_args, ctx, tool)
            LOG.append(f'Finished calling tool {name!r} with result: {result!r}')
        except Exception as e:
            LOG.append(f'Error calling tool {name!r}: {e}')
            raise e
        else:
            return result


logging_toolset = LoggingToolset(prepared_toolset)

agent = Agent(TestModel(), toolsets=[logging_toolset]) # (2)!
result = agent.run_sync('Call all the tools')
print(LOG)
"""
[
    "Calling tool 'temperature_celsius' with args: {'city': 'a'}",
    "Calling tool 'temperature_fahrenheit' with args: {'city': 'a'}",
    "Calling tool 'weather_conditions' with args: {'city': 'a'}",
    "Calling tool 'current_time' with args: {}",
    "Finished calling tool 'temperature_celsius' with result: 21.0",
    "Finished calling tool 'temperature_fahrenheit' with result: 69.8",
    'Finished calling tool \'weather_conditions\' with result: "It\'s raining"',
    "Finished calling tool 'current_time' with result: datetime.datetime(...)",
]
"""
  1. 所有文档示例都在 CI 中进行测试,并验证其输出,因此我们需要确保每当此代码运行时,LOG 的顺序始终相同。由于工具可能以任何顺序完成,我们根据工具调用的序号增加睡眠时间,以确保它们按照被调用的顺序完成(和记录日志)。
  2. 我们在这里使用 TestModel,因为它会自动调用每个工具。

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

外部工具集

如果您的代理需要能够调用由上游服务或前端提供和执行的外部工具,您可以从一个包含工具名称、参数 JSON 模式和描述的 ToolDefinition 列表来构建一个 ExternalToolset

当模型调用外部工具时,该调用被认为是“延迟的”,代理运行将以一个 DeferredToolRequests 输出对象结束,该对象有一个 calls 列表,其中包含 ToolCallParts,这些部分包含了工具名称、已验证的参数和唯一的工具调用 ID,这些信息预期会被传递给将产生结果的上游服务或前端。

当从上游服务或前端收到工具调用结果时,您可以构建一个 DeferredToolResults 对象,其 calls 字典将每个工具调用 ID 映射到一个任意值以返回给模型,这个值可以是一个 ToolReturn 对象,或者在工具调用失败且模型应重试的情况下,是一个 ModelRetry 异常。这个 DeferredToolResults 对象可以作为 deferred_tool_results 提供给代理的运行方法之一,同时提供原始运行的消息历史记录

请注意,您需要将 DeferredToolRequests 添加到 Agentagent.run()output_type 中,以便正确推断代理运行输出的可能类型。更多信息,请参见延迟工具文档。

为了演示,我们首先定义一个没有延迟工具的简单代理:

deferred_toolset_agent.py
from pydantic import BaseModel

from pydantic_ai import Agent
from pydantic_ai.toolsets.function import FunctionToolset

toolset = FunctionToolset()


@toolset.tool
def get_default_language():
    return 'en-US'


@toolset.tool
def get_user_name():
    return 'David'


class PersonalizedGreeting(BaseModel):
    greeting: str
    language_code: str


agent = Agent('openai:gpt-4o', toolsets=[toolset], output_type=PersonalizedGreeting)

result = agent.run_sync('Greet the user in a personalized way')
print(repr(result.output))
#> PersonalizedGreeting(greeting='Hello, David!', language_code='en-US')

接下来,我们定义一个函数,代表一个假设的“运行代理”API 端点,可以由前端调用,并接收要发送给模型的消息列表、前端工具定义列表以及可选的延迟工具结果。这就是 ExternalToolsetDeferredToolRequestsDeferredToolResults 发挥作用的地方:

deferred_toolset_api.py
from pydantic_ai import DeferredToolRequests, DeferredToolResults, ToolDefinition
from pydantic_ai.messages import ModelMessage
from pydantic_ai.toolsets import ExternalToolset

from deferred_toolset_agent import PersonalizedGreeting, agent


def run_agent(
    messages: list[ModelMessage] = [],
    frontend_tools: list[ToolDefinition] = {},
    deferred_tool_results: DeferredToolResults | None = None,
) -> tuple[PersonalizedGreeting | DeferredToolRequests, list[ModelMessage]]:
    deferred_toolset = ExternalToolset(frontend_tools)
    result = agent.run_sync(
        toolsets=[deferred_toolset], # (1)!
        output_type=[agent.output_type, DeferredToolRequests], # (2)!
        message_history=messages, # (3)!
        deferred_tool_results=deferred_tool_results,
    )
    return result.output, result.new_messages()
  1. 延迟工具文档中所述,这些 toolsets 是对提供给 Agent 构造函数的工具集的补充。
  2. 延迟工具文档中所述,这个 output_type 会覆盖提供给 Agent 构造函数的类型,所以我们必须确保不要丢失它。
  3. 我们没有包含 user_prompt 关键字参数,因为我们期望前端通过 messages 提供它。

现在,想象一下下面的代码是在前端实现的,而 run_agent 代表了对运行代理的后端的 API 调用。这就是我们实际执行延迟工具调用并使用新结果开始新一轮运行的地方:

deferred_tools.py
from pydantic_ai import (
    DeferredToolRequests,
    DeferredToolResults,
    ModelRetry,
    ToolDefinition,
)
from pydantic_ai.messages import ModelMessage, ModelRequest, UserPromptPart

from deferred_toolset_api import run_agent

frontend_tool_definitions = [
    ToolDefinition(
        name='get_preferred_language',
        parameters_json_schema={'type': 'object', 'properties': {'default_language': {'type': 'string'}}},
        description="Get the user's preferred language from their browser",
    )
]

def get_preferred_language(default_language: str) -> str:
    return 'es-MX' # (1)!

frontend_tool_functions = {'get_preferred_language': get_preferred_language}

messages: list[ModelMessage] = [
    ModelRequest(
        parts=[
            UserPromptPart(content='Greet the user in a personalized way')
        ]
    )
]

deferred_tool_results: DeferredToolResults | None = None

final_output = None
while True:
    output, new_messages = run_agent(messages, frontend_tool_definitions, deferred_tool_results)
    messages += new_messages

    if not isinstance(output, DeferredToolRequests):
        final_output = output
        break

    print(output.calls)
    """
    [
        ToolCallPart(
            tool_name='get_preferred_language',
            args={'default_language': 'en-US'},
            tool_call_id='pyd_ai_tool_call_id',
        )
    ]
    """
    deferred_tool_results = DeferredToolResults()
    for tool_call in output.calls:
        if function := frontend_tool_functions.get(tool_call.tool_name):
            result = function(**tool_call.args_as_dict())
        else:
            result = ModelRetry(f'Unknown tool {tool_call.tool_name!r}')
        deferred_tool_results.calls[tool_call.tool_call_id] = result

print(repr(final_output))
"""
PersonalizedGreeting(greeting='Hola, David! Espero que tengas un gran día!', language_code='es-MX')
"""
  1. 想象一下,这会返回前端的 navigator.language

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

动态构建工具集

工具集可以在每次代理运行或运行步骤之前动态构建,方法是使用一个接收代理运行上下文并返回一个工具集或 None 的函数。当一个工具集(如 MCP 服务器)依赖于特定于代理运行的信息(如其依赖项)时,这个功能很有用。

要注册一个动态工具集,您可以将一个接收 RunContext 的函数传递给 Agent 构造函数的 toolsets 参数,或者您可以用 @agent.toolset 装饰器来包装一个兼容的函数。

默认情况下,该函数将在每个代理运行步骤之前再次被调用。如果您使用装饰器,您可以选择性地提供一个 per_run_step=False 参数,以指示工具集在整个运行过程中只需要构建一次。

dynamic_toolset.py
from dataclasses import dataclass
from typing import Literal

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

from function_toolset import datetime_toolset, weather_toolset


@dataclass
class ToggleableDeps:
    active: Literal['weather', 'datetime']

    def toggle(self):
        if self.active == 'weather':
            self.active = 'datetime'
        else:
            self.active = 'weather'

test_model = TestModel()  # (1)!
agent = Agent(
    test_model,
    deps_type=ToggleableDeps  # (2)!
)

@agent.toolset
def toggleable_toolset(ctx: RunContext[ToggleableDeps]):
    if ctx.deps.active == 'weather':
        return weather_toolset
    else:
        return datetime_toolset

@agent.tool
def toggle(ctx: RunContext[ToggleableDeps]):
    ctx.deps.toggle()

deps = ToggleableDeps('weather')

result = agent.run_sync('Toggle the toolset', deps=deps)
print([t.name for t in test_model.last_model_request_parameters.function_tools])  # (3)!
#> ['toggle', 'now']

result = agent.run_sync('Toggle the toolset', deps=deps)
print([t.name for t in test_model.last_model_request_parameters.function_tools])
#> ['toggle', 'temperature_celsius', 'temperature_fahrenheit', 'conditions']
  1. 我们在这里使用 TestModel,因为它能让我们很容易地看到每次运行中可用的工具。
  2. 我们正在使用代理的依赖项,通过 RunContext 参数让 toggle 工具能够访问 active
  3. 这显示了在 toggle 工具执行之后可用的工具,因为“最后的模型请求”是将 toggle 工具的结果返回给模型的那一次。

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

构建自定义工具集

要定义一个完全自定义的工具集,拥有自己列出可用工具和处理工具调用的逻辑,您可以子类化 AbstractToolset 并实现 get_tools()call_tool() 方法。

如果您想在代理运行期间的工具列表和调用中重用网络连接或会话,您可以实现 __aenter__()__aexit__()

第三方工具集

MCP 服务器

有关如何将 MCP 服务器与 Pydantic AI 一起使用的信息,请参阅 MCP 客户端文档。

LangChain 工具

如果您想将 LangChain 的社区工具库中的工具或工具包与 Pydantic AI 一起使用,您可以使用 LangChainToolset,它接收一个 LangChain 工具列表。请注意,在这种情况下,Pydantic AI 不会验证参数——模型需要提供与 LangChain 工具指定的模式相匹配的参数,而 LangChain 工具则负责在参数无效时引发错误。

您需要安装 langchain-community 包以及相关工具所需的任何其他包。

from langchain_community.agent_toolkits import SlackToolkit

from pydantic_ai import Agent
from pydantic_ai.ext.langchain import LangChainToolset

toolkit = SlackToolkit()
toolset = LangChainToolset(toolkit.get_tools())

agent = Agent('openai:gpt-4o', toolsets=[toolset])
# ...

ACI.dev 工具

如果您想将 ACI.dev 工具库中的工具与 Pydantic AI 一起使用,您可以使用 ACIToolset 工具集,它接收一个 ACI 工具名称列表以及 linked_account_owner_id。请注意,在这种情况下,Pydantic AI 不会验证参数——模型需要提供与 ACI 工具指定的模式相匹配的参数,而 ACI 工具则负责在参数无效时引发错误。

您需要安装 aci-sdk 包,在 ACI_API_KEY 环境变量中设置您的 ACI API 密钥,并将您的 ACI “关联账户所有者 ID”(linked account owner ID)传递给函数。

import os

from pydantic_ai import Agent
from pydantic_ai.ext.aci import ACIToolset

toolset = ACIToolset(
    [
        'OPEN_WEATHER_MAP__CURRENT_WEATHER',
        'OPEN_WEATHER_MAP__FORECAST',
    ],
    linked_account_owner_id=os.getenv('LINKED_ACCOUNT_OWNER_ID'),
)

agent = Agent('openai:gpt-4o', toolsets=[toolset])