Ai langchain

在Langchain中如何创建工具

How to create tools

Posted by Brian on Tuesday, July 29, 2025

概述

在上一篇文章中已经使用 @tool 装饰器来创建工具。这种方式足够大部分的应用场景。但是如果工具的需要执行很长时间或者想控制粒度在精细一些那这种方式显然就不太合适。在这一篇文章中我将与你一起深入了解工具如何在LLM下才能更好的工作。

当我们在构建一个工具时。你需要知道以下几个属性：

AttributeTypeDescription
namestr工具名称，必须保持惟一性
descriptionstr描述该工具的功能。由LLM或agent作为上下文件使用
args_schemapedantic.BaseModel可选但推荐使用，如果使用回调处理器则必须使用。它可用于提供更多信息（例如，少量示例）或对预期参数进行验证。
return_directboolean仅适用于 agent 当为 True 时，agent在调用给定工具后，将停止并直接将结果返回给用户。

了解了这些属性之后我将在下面的示例中给你演示他们在不同的定义中都是如何工作的。

在 Langchain 中我有有三种方法创建工具：

  • Function
  • Langchain Runnables
  • 通过继承 BaseTool。这是最高灵活度的方法。同时代码量也是最复杂最多的。

如果工具有很好的名字、描述和 JSON schemas 模型的表现会更好。

从函数创建工具

使用 @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)

结果如下：

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

可以看到 name description args 相关结果与我们希望的基本是一致的。请注意 @tool支持注解、嵌套和其它功能

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_max', 'type': 'object'}

使用 args_schema：

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

从结果中可以看到。我们第一个参数成功覆盖了使用函数名做工具名。args_schema与如预期那样子工作。

结构化工具

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

from langchain_core.tools import StructuredTool
import asyncio


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}))
# 执行异步方法
async def test():
    print(await calculator.ainvoke({"a": 2, "b": 5}))


asyncio.run(test())

6
10

请注意这里func是同步调用。而 coroutine 则是异步调用，也就是当使用 ainvoke 方法是将会使用 amultiply 方法，当然你还可以这样子配置：

from langchain_core.tools import StructuredTool

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'}}

这里我们使用了 name, description,args_schema,return_direct来配置该工具。

从Runnables创建工具

接受字符串或 dict 输入的 LangChain Runnables 可以使用 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
print(as_tool.args)

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
import asyncio

# 定义输入的类型
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())

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

print(multiply.invoke({"a": 2, "b": 3}))
# 运行异步
async def test():
    print(await multiply.ainvoke({"a": 2, "b": 5}))

asyncio.run(test())

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
10

创建异步工具

所有可运行对象都暴露了 invoke和ainvoke方法。关于异步你需要知道以下几点：

  • LangeChain 默认提供异步实现，假设你的工具耗时，它会将执行委托给另一个线程。
  • 如是你的使用场景是异步的，应该是实现异步工具而不是同步工具。
  • 如果需要异步与同步同时使用。使用 StructuredTool.from_function 或者继承 BaseTool
  • 如果同时实现了同步与异步，并且同步代码运行速度快。可以覆盖默认的 Langchain 异步实现，直接调用同步代码。
  • 不要使用 invoke 和 async 工具
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

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

如果工具定义为异步，而你使用 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.")

工具错误处理

如果在agent中使用工具。我们需要对工具的错误进行处理。以便 agent 能够从错误中恢复并继续执行。

工具的异常需要从工具内部抛出一个ToolException，并使用 handle_tool_error指定如何处理该异常。

handle_tool_error有如下几种方式：

  • True：错误将会以返回值的方式返回，同时不会中断程序。
  • False：中断程序。直接抛出堆椎信息
  • 指定字符串信息。只要工具返回了 ToolException ，返回值将永远是该指定的信息。同时不会中断程序。
  • 函数：如果是函数的方式，那么该函数将会收到一个 ToolException 类型的参数。你可以在该函数内决定如何处理你的错误。同时不会中断程序。

下面是几种 handle_tool_error 的示例：

设置为True时，不中断程序：

from langchain_core.tools import ToolException,StructuredTool

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}.")

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

err = get_weather_tool.invoke({"city": "foobar"})
print(err)
print("next execute")

# Output：
# Error: There is no city by the name of foobar.
# next execute

设置为False时，中断程序：

from langchain_core.tools import ToolException,StructuredTool

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}.")

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

err = get_weather_tool.invoke({"city": "foobar"})
print(err)
print("next execute")

Traceback (most recent call last):
  File "/xxx/langchain-learin/tools.py", line 345, in <module>
    err = get_weather_tool.invoke({"city": "foobar"})
  File "/xxx/langchain-learin/.venv/lib/python3.13/site-packages/langchain_core/tools/base.py", line 599, in invoke
    return self.run(tool_input, **kwargs)
           ~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^
  File "/xxx/langchain-learin/.venv/lib/python3.13/site-packages/langchain_core/tools/base.py", line 883, in run
    raise error_to_raise
  File "/xxx/langchain-learin/.venv/lib/python3.13/site-packages/langchain_core/tools/base.py", line 852, in run
    response = context.run(self._run, *tool_args, **tool_kwargs)
  File "/xxx/langchain-learin/.venv/lib/python3.13/site-packages/langchain_core/tools/structured.py", line 93, in _run
    return self.func(*args, **kwargs)
           ~~~~~~~~~^^^^^^^^^^^^^^^^^
  File "/xxx/langchain-learin/tools.py", line 338, in get_weather
    raise ToolException(f"Error: There is no city by the name of {city}.")
langchain_core.tools.base.ToolException: Error: There is no city by the name of foobar.

指定字符串信息:

from langchain_core.tools import ToolException,StructuredTool

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}.")

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

err = get_weather_tool.invoke({"city": "foobar"})
print(err)
print("next execute")

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

可以看到尽管我们在ToolException中设置了错误信息。但返回的错误信息始终是 There is no such city, but it's probably above 0K there!

使用函数处理错误：

from langchain_core.tools import ToolException,StructuredTool

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

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}.")

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

err = get_weather_tool.invoke({"city": "foobar"})
print(err)
print("next execute")

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

返回工具执行结果

有时，我们希望将工具执行产生的某些结果暴露给链或代理中的下游组件，但又不希望直接暴露给模型本身。例如，如果工具返回自定义对象（如文档），我们可能希望在不将原始输出直接传递给模型的情况下，向模型传递一些关于该输出的视图或元数据。同时，我们也可能希望能在其他地方访问这个完整的输出，例如在下游工具中。

工具和 ToolMessage 接口使得能够区分工具输出中供模型使用部分（这是 ToolMessage.content）和供模型外部使用部分（ToolMessage.artifact）。

如果我们希望我们的工具能够区分消息内容和其它工件，我们需要在定义工具时指定 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

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

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

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

如果我们使用 ToolCall（例如由调用工具模型生成的那些）来调用我们的工具，我们会得到一个包含工具生成的内容和工件的 ToolMessage：

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来实现。具体的请查看官方文档

总结

定义一个工具需要关注 name ,description,args_schema,return_direct四个属性。

创建工具有三种方式：

  • 使用 @tool从函数创建。使用简单
  • 从Runnables创建，最终使用 as_tool方法转换为工具。
  • 继承BaseTool创建，灵活度度，同时代码量还是最多的。

在使用工具时还需要注意处理工具错误的几种方式以及它们的区别。还需要注意异步工具与同步工具的它们的之前的区别。

    精选标签
    ai ci/cd fluentd gitlab-runner goframe kubernetes langchain laravel log pipeline pod skill tools