如何创建工具

在构建代理时，您需要为其提供一个可供其使用的工具列表。除了实际调用的函数外，工具还包含几个组件：

属性	类型	描述
name	str	在提供给 LLM 或代理的工具集中必须是唯一的。
description	str	描述工具的功能。作为 LLM 或代理的上下文使用。
args_schema	pydantic.BaseModel	可选但推荐，如果使用回调处理程序则必需。可用于提供更多信息（例如，少量示例）或对预期参数进行验证。
return_direct	boolean	仅与代理相关。如果设置为 True，则在调用给定工具后，代理将停止并将结果直接返回给用户。

LangChain 支持从以下来源创建工具：

函数；
LangChain 可运行对象；
通过继承 BaseTool -- 这是最灵活的方法，它提供了最大的控制度，但需要更多的精力和代码。

对于大多数用例，从函数创建工具可能就足够了，可以通过简单的 `tool` 装饰器来实现。如果需要更多配置，例如指定同步和异步实现，也可以使用 StructuredTool.from_function 类方法。

在本指南中，我们将概述这些方法。

tip

如果工具具有精心选择的名称、描述和 JSON 模式，模型的性能会更好。

从函数创建工具

`@tool` 装饰器

@tool 装饰器是定义自定义工具最简单的方式。该装饰器默认使用函数名作为工具名，但也可以通过传递一个字符串作为第一个参数来覆盖默认设置。此外，该装饰器将使用函数的文档字符串作为工具的描述——因此必须提供文档字符串。

from langchain_core.tools import tool


@tool
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


# Let's inspect some of the attributes associated with the tool.
print(multiply.name)
print(multiply.description)
print(multiply.args)

API Reference:tool

multiply
Multiply two numbers.
{'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}

或者创建一个异步实现，如下所示：

from langchain_core.tools import tool


@tool
async def amultiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

API Reference:tool

请注意，@tool 支持注解的解析、嵌套 schemas 以及其他功能：

from typing import Annotated, List


@tool
def multiply_by_max(
    a: Annotated[int, "scale factor"],
    b: Annotated[List[int], "list of ints over which to take maximum"],
) -> int:
    """Multiply a by the maximum of b."""
    return a * max(b)


print(multiply_by_max.args_schema.model_json_schema())

{'description': 'Multiply a by the maximum of b.',
 'properties': {'a': {'description': 'scale factor',
   'title': 'A',
   'type': 'integer'},
  'b': {'description': 'list of ints over which to take maximum',
   'items': {'type': 'integer'},
   'title': 'B',
   'type': 'array'}},
 'required': ['a', 'b'],
 'title': 'multiply_by_maxSchema',
 'type': 'object'}

您还可以通过将它们传递到工具装饰器中来定制工具名称和 JSON 参数。

from pydantic import BaseModel, Field


class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")


@tool("multiplication-tool", args_schema=CalculatorInput, return_direct=True)
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


# Let's inspect some of the attributes associated with the tool.
print(multiply.name)
print(multiply.description)
print(multiply.args)
print(multiply.return_direct)

multiplication-tool
Multiply two numbers.
{'a': {'description': 'first number', 'title': 'A', 'type': 'integer'}, 'b': {'description': 'second number', 'title': 'B', 'type': 'integer'}}
True

Docstring 解析

@tool 可选择性地解析 Google 风格文档字符串，并将文档字符串组件（如参数描述）与工具模式的相关部分关联起来。要启用此行为，请指定 parse_docstring：

@tool(parse_docstring=True)
def foo(bar: str, baz: int) -> str:
    """The foo.

    Args:
        bar: The bar.
        baz: The baz.
    """
    return bar


print(foo.args_schema.model_json_schema())

{'description': 'The foo.',
 'properties': {'bar': {'description': 'The bar.',
   'title': 'Bar',
   'type': 'string'},
  'baz': {'description': 'The baz.', 'title': 'Baz', 'type': 'integer'}},
 'required': ['bar', 'baz'],
 'title': 'fooSchema',
 'type': 'object'}

caution

默认情况下，@tool(parse_docstring=True) 如果 docstring 未能正确解析，将引发 ValueError。有关详细信息和示例，请参阅 API 参考。

StructuredTool

StructuredTool.from_function 类方法比 @tool 装饰器提供了更多的可配置性，而无需编写大量额外的代码。

from langchain_core.tools import StructuredTool

def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

async def amultiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b

calculator = StructuredTool.from_function(func=multiply, coroutine=amultiply)

print(calculator.invoke({"a": 2, "b": 3}))
print(await calculator.ainvoke({"a": 2, "b": 5}))

API Reference:StructuredTool

6
10

要配置它：

class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")


def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


calculator = StructuredTool.from_function(
    func=multiply,
    name="Calculator",
    description="multiply numbers",
    args_schema=CalculatorInput,
    return_direct=True,
    # coroutine= ... <- you can specify an async method if desired as well
)

print(calculator.invoke({"a": 2, "b": 3}))
print(calculator.name)
print(calculator.description)
print(calculator.args)

6
Calculator
multiply numbers
{'a': {'description': 'first number', 'title': 'A', 'type': 'integer'}, 'b': {'description': 'second number', 'title': 'B', 'type': 'integer'}}

使用 Runnable 创建工具

LangChain 的 Runnables（可接受字符串或 dict 作为输入的）可以通过 as_tool 方法转换为工具，该方法允许指定名称、描述和附加的参数模式信息。

用法示例：

from langchain_core.language_models import GenericFakeChatModel
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages(
    [("human", "Hello. Please respond in the style of {answer_style}.")]
)

# Placeholder LLM
llm = GenericFakeChatModel(messages=iter(["hello matey"]))

chain = prompt | llm | StrOutputParser()

as_tool = chain.as_tool(
    name="Style responder", description="Description of when to use tool."
)
as_tool.args

API Reference:GenericFakeChatModel | StrOutputParser | ChatPromptTemplate

/var/folders/4j/2rz3865x6qg07tx43146py8h0000gn/T/ipykernel_95770/2548361071.py:14: LangChainBetaWarning: This API is in beta and may change in the future.
  as_tool = chain.as_tool(

{'answer_style': {'title': 'Answer Style', 'type': 'string'}}

请参阅本指南了解更多详情。

继承 `BaseTool`

你可以通过继承 BaseTool 来定义自定义工具。这提供了对工具定义的最高控制，但需要编写更多代码。

from typing import Optional

from langchain_core.callbacks import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)
from langchain_core.tools import BaseTool
from langchain_core.tools.base import ArgsSchema
from pydantic import BaseModel, Field


class CalculatorInput(BaseModel):
    a: int = Field(description="first number")
    b: int = Field(description="second number")


# Note: It's important that every field has type hints. BaseTool is a
# Pydantic class and not having type hints can lead to unexpected behavior.
class CustomCalculatorTool(BaseTool):
    name: str = "Calculator"
    description: str = "useful for when you need to answer questions about math"
    args_schema: Optional[ArgsSchema] = CalculatorInput
    return_direct: bool = True

    def _run(
        self, a: int, b: int, run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> int:
        """Use the tool."""
        return a * b

    async def _arun(
        self,
        a: int,
        b: int,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> int:
        """Use the tool asynchronously."""
        # If the calculation is cheap, you can just delegate to the sync implementation
        # as shown below.
        # If the sync calculation is expensive, you should delete the entire _arun method.
        # LangChain will automatically provide a better implementation that will
        # kick off the task in a thread to make sure it doesn't block other async code.
        return self._run(a, b, run_manager=run_manager.get_sync())

API Reference:AsyncCallbackManagerForToolRun | CallbackManagerForToolRun | BaseTool

multiply = CustomCalculatorTool()
print(multiply.name)
print(multiply.description)
print(multiply.args)
print(multiply.return_direct)

print(multiply.invoke({"a": 2, "b": 3}))
print(await multiply.ainvoke({"a": 2, "b": 3}))

Calculator
useful for when you need to answer questions about math
{'a': {'description': 'first number', 'title': 'A', 'type': 'integer'}, 'b': {'description': 'second number', 'title': 'B', 'type': 'integer'}}
True
6
6

如何创建异步工具

LangChain 工具实现了 Runnable 接口 🏃。

所有 Runnable 都公开 invoke 和 ainvoke 方法（以及 batch、abatch、astream 等其他方法）。

因此，即使您只提供工具的同步实现，您仍然可以使用 ainvoke 接口，但有几点需要注意：

LangChain 默认提供了一个异步实现，它假设函数计算成本很高，因此会将执行委托给另一个线程。
如果您正在使用异步代码库，您应该创建异步工具而不是同步工具，以避免由于那个线程而产生少量开销。
如果您需要同步和异步实现，请使用 StructuredTool.from_function 或继承自 BaseTool。
如果同时实现同步和异步，并且同步代码运行速度很快，请覆盖默认的 LangChain 异步实现并简单调用同步代码。
您不能也不应该对异步工具使用同步的 invoke。

from langchain_core.tools import StructuredTool


def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


calculator = StructuredTool.from_function(func=multiply)

print(calculator.invoke({"a": 2, "b": 3}))
print(
    await calculator.ainvoke({"a": 2, "b": 5})
)  # Uses default LangChain async implementation incurs small overhead

API Reference:StructuredTool

6
10

from langchain_core.tools import StructuredTool


def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


async def amultiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


calculator = StructuredTool.from_function(func=multiply, coroutine=amultiply)

print(calculator.invoke({"a": 2, "b": 3}))
print(
    await calculator.ainvoke({"a": 2, "b": 5})
)  # Uses use provided amultiply without additional overhead

API Reference:StructuredTool

6
10

当仅提供异步定义时，你不应该也不能使用 .invoke。

@tool
async def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


try:
    multiply.invoke({"a": 2, "b": 3})
except NotImplementedError:
    print("Raised not implemented error. You should not be doing this.")

Raised not implemented error. You should not be doing this.

处理工具错误

如果您正在使用带有 Agent 的工具，您将需要一个错误处理策略，以便 Agent 可以从错误中恢复并继续执行。

一个简单的策略是从工具内部抛出 ToolException，并使用 handle_tool_error 指定错误处理器。

当指定了错误处理器时，异常将被捕获，并且错误处理器将决定从工具返回哪个输出。

您可以将 handle_tool_error 设置为 True、一个字符串值或一个函数。如果它是一个函数，该函数应以 ToolException 作为参数并返回一个值。

请注意，仅抛出 ToolException 是无效的。您需要先设置工具的 handle_tool_error，因为它的默认值是 False。

from langchain_core.tools import ToolException


def get_weather(city: str) -> int:
    """Get weather for the given city."""
    raise ToolException(f"Error: There is no city by the name of {city}.")

API Reference:ToolException

以下是具有默认 handle_tool_error=True 行为的示例。

get_weather_tool = StructuredTool.from_function(
    func=get_weather,
    handle_tool_error=True,
)

get_weather_tool.invoke({"city": "foobar"})

'Error: There is no city by the name of foobar.'

我们可以将 handle_tool_error 设置为一个始终会被返回的字符串。

get_weather_tool = StructuredTool.from_function(
    func=get_weather,
    handle_tool_error="There is no such city, but it's probably above 0K there!",
)

get_weather_tool.invoke({"city": "foobar"})

"There is no such city, but it's probably above 0K there!"

使用函数处理错误：

def _handle_error(error: ToolException) -> str:
    return f"The following errors occurred during tool execution: `{error.args[0]}`"


get_weather_tool = StructuredTool.from_function(
    func=get_weather,
    handle_tool_error=_handle_error,
)

get_weather_tool.invoke({"city": "foobar"})

'The following errors occurred during tool execution: `Error: There is no city by the name of foobar.`'

返回工具执行的构件

有时，我们希望工具执行的某些构件（artifacts）能够被链或代理中的下游组件访问，同时又不暴露给模型本身。例如，如果一个工具返回自定义对象（如 Documents），我们可能希望将一些关于此输出的视图或元数据传递给模型，而不将原始输出传递给模型。同时，我们可能希望能够在其他地方访问此完整输出，例如在下游工具中。

Tool 和 [ToolMessage](https://python.langchain.com/api_reference/core/messages/langchain_core.messages.tool.ToolMessage.html) 接口使得区分旨在提供给模型使用的工具输出部分（ToolMessage.content）和旨在模型外部使用的部分（ToolMessage.artifact）成为可能。

要求 langchain-core >= 0.2.19

此功能已在 langchain-core == 0.2.19 中添加。请确保您的包已更新到最新版本。

如果希望我们的工具区分消息内容和其他构件，我们需要在定义工具时指定 response_format="content_and_artifact"，并确保返回一个元组 (content, artifact)：

import random
from typing import List, Tuple

from langchain_core.tools import tool


@tool(response_format="content_and_artifact")
def generate_random_ints(min: int, max: int, size: int) -> Tuple[str, List[int]]:
    """Generate size random ints in the range [min, max]."""
    array = [random.randint(min, max) for _ in range(size)]
    content = f"Successfully generated array of {size} random ints in [{min}, {max}]."
    return content, array

API Reference:tool

如果我们直接使用工具参数来调用我们的工具，我们只会得到输出的内容部分：

generate_random_ints.invoke({"min": 0, "max": 9, "size": 10})

'Successfully generated array of 10 random ints in [0, 9].'

如果我们调用我们的工具，并传入一个 ToolCall（类似由支持工具调用的模型生成的那些），我们会收到一个 ToolMessage，其中包含工具生成的 content 和 artifact：

generate_random_ints.invoke(
    {
        "name": "generate_random_ints",
        "args": {"min": 0, "max": 9, "size": 10},
        "id": "123",  # required
        "type": "tool_call",  # required
    }
)

ToolMessage(content='Successfully generated array of 10 random ints in [0, 9].', name='generate_random_ints', tool_call_id='123', artifact=[4, 8, 2, 4, 1, 0, 9, 5, 8, 1])

我们可以通过继承 BaseTool 来实现相同的效果：

from langchain_core.tools import BaseTool


class GenerateRandomFloats(BaseTool):
    name: str = "generate_random_floats"
    description: str = "Generate size random floats in the range [min, max]."
    response_format: str = "content_and_artifact"

    ndigits: int = 2

    def _run(self, min: float, max: float, size: int) -> Tuple[str, List[float]]:
        range_ = max - min
        array = [
            round(min + (range_ * random.random()), ndigits=self.ndigits)
            for _ in range(size)
        ]
        content = f"Generated {size} floats in [{min}, {max}], rounded to {self.ndigits} decimals."
        return content, array

    # Optionally define an equivalent async method

    # async def _arun(self, min: float, max: float, size: int) -> Tuple[str, List[float]]:
    #     ...

API Reference:BaseTool

rand_gen = GenerateRandomFloats(ndigits=4)

rand_gen.invoke(
    {
        "name": "generate_random_floats",
        "args": {"min": 0.1, "max": 3.3333, "size": 3},
        "id": "123",
        "type": "tool_call",
    }
)

ToolMessage(content='Generated 3 floats in [0.1, 3.3333], rounded to 4 decimals.', name='generate_random_floats', tool_call_id='123', artifact=[1.5566, 0.5134, 2.7914])

从函数创建工具​

@tool 装饰器​

Docstring 解析​

StructuredTool​

使用 Runnable 创建工具​

继承 BaseTool​

如何创建异步工具​

处理工具错误​

返回工具执行的构件​

从函数创建工具

`@tool` 装饰器

Docstring 解析

StructuredTool

使用 Runnable 创建工具

继承 `BaseTool`

如何创建异步工具

处理工具错误

返回工具执行的构件