使用LangChain自定义tools的三种方式

LLM对外部工具的使用现在可谓其必备技能之一。当我们使用LangChain/LangGraph等框架去编排定制一些或简单或复杂的LLM应用时，自定义LLM在处理用户提问时需要用到的工具是一个非常常见的步骤，熟练掌握自定义工具的过程是很有必要的。由此，本篇博客将尽可能详细地介绍如何基于LangChain相关python库中提供的模块组件进行工具的自定义，一起来学习吧！

本篇博客内容将参考 LangChain 关于创建自定义工具的官方文档，并结合案例代码运行效果展示和较为详细的文字解释（这部分大概率有 ChatGPT 参与帮忙），帮助大家快速了解自定义工具的构建方式。

工具的组成部分

• 名称 (name)（字符串）：工具的名称，必需提供且唯一

• 描述 (description)（字符串）：工具的功能描述。可选但推荐使用，因为LLM和agent会把该描述作为context，以实现对工具的准确选择。

• 参数模式 (args_schema)（Pydantic BaseModel）：可选但推荐使用，可以用于提供更多信息（例如提供少量示例）或验证传入参数是否符合预期。

• return_direct（布尔值）：仅对agents生效。当该参数设置为 True，在agent调用给定工具后，将直接把结果返回给用户。

工具的三种创建方式

1. 基于函数

@tool 装饰器

通过函数创建工具是一个最简单直接高效的创建方式，只需通过一个简单的 @tool 装饰器就可以完成，且足以满足大部分场景需求，但若需要对工具进行更多配置，如同时指定同步和异步实现，则可使用StructuredTool.from_function 类方法进行实现。

默认情况下，@tool 装饰器使用函数名作为工具名，但可以通过传递一个字符串作为第一个参数来覆盖。此外，@tool 装饰器会将函数的文档字符串用作工具的描述，因此文档字符串是必须要提供的。

代码示例（无传入参数）

比如我们希望编排一个非常简单的LLM应用，可以基于调用外部api获取到的黄金数据（黄金现货、黄金期货等品种的最新价、开盘价、最高价、最低价等价格信息）进行相关问题的回答，那我们就需要把调用该api获取返回信息的步骤封装到一个函数中作为工具。

1. 从 langchain_core.tools 导入 tool（pip install langchain-core）

from langchain_core.tools import tool
import requests

url = "https://api.jisuapi.com/gold/shgold"
params = {
"appkey": "你的appkey"
}

2. 用 @tool 装饰器 创建工具名为 get_current_gold_price 的函数（注意，函数的docstring，即工具的描述，是必须要提供的，否则会报错，因为装饰器中名为 parse_docstring 默认为 True，即@tool(parse_docstring=True)，想了解更多可查看 API 文档）

@tool
def get_current_gold_price():
    """
    Retrieves the current gold price from the specified API.

    This function sends a GET request to the gold price API and returns
    the result. If the request is successful, it returns the gold price data.
    Otherwise, it returns an error message with the status code.

    Returns:
    - dict: A dictionary containing the result or an error message.
    """
    response = requests.get(url, params=params)

    # 检查响应状态并输出结果
    if response.status_code == 200:
        data = response.json()
        result = data["result"]
    else:
        result = ["Failed to retrieve data, status code: {}".format(response.status_code)]

    return {"messages": result}

3. 打印出tool的名称、描述和参数

# 工具的名称
print(get_current_gold_price.name)
get_current_gold_price
# 工具的描述
print(get_current_gold_price.description)
Retrieves the current gold price from the specified API.

This function sends a GET request to the gold price API and returns
the result. If the request is successful, it returns the gold price data.
Otherwise, it returns an error message with the status code.

Returns:
- dict: A dictionary containing the result or an error message.
# 函数传入的参数信息，因为没有所以为空
print(get_current_gold_price.args)
{}

代码示例（有传入参数）

接下来我们看一个 官方文档 中提供的有传入参数，且传入参数需要符合一定规范的通过函数来自定义工具的例子。

1. 导入所需模块并创建继承自 BaseModel 的子类 CalculatorInput

这里让ChatGPT介绍一下从 pydantic 第三方python库中导出的 BaseModel 和 Field ：

pydantic.BaseModel 和 pydantic.Field：

• 这两者来自 pydantic 库，主要用于数据验证和结构化定义输入参数。通过使用 pydantic，你可以定义强类型的输入模型，并为每个字段添加描述、默认值、限制等。
• BaseModel：这是 pydantic 中的一个基础类，允许我们创建验证数据输入的类。
• Field：用于定义每个字段的属性和描述信息。在这段代码中，a 和 b 是两个整数字段，并分别有自己的描述信息。

from langchain_core.tools import tool
from pydantic import BaseModel, Field


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

2. 通过装饰器 @tool 将函数 multiply 声明为一个名为“multiplication-tool”的工具，并通过设置 args_schema 将 CalculatorInput 作为其输入参数的验证模式，即传入的两个参数必须为整型。

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

3. 打印出tool的名称、描述和参数

# 工具的名称
print(multiply.name)
multiplication-tool
# 工具的描述
print(multiply.description)
Multiply two numbers.
# 函数传入的参数信息
print(multiply.args)
{'a': {'description': 'first number', 'title': 'A', 'type': 'integer'}, 'b': {'description': 'second number', 'title': 'B', 'type': 'integer'}}
# 是否直接将结果返回，默认为False
print(multiply.return_direct)
False

StructuredTool.from_function 类方法

当然，除了使用 @tool 装饰器的方式去自定义工具，langchain还提供了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}))

以下为 ChatGPT-4o 对以上示例代码的详细解释：

这段代码展示了如何使用 StructuredTool.from_function 方法将一个同步函数和一个异步函数封装为可调用工具。让我们逐步解释这段代码：

代码解析

1. 导入模块:

   from langchain_core.tools import StructuredTool
   

   • 从 langchain_core.tools 模块中导入 StructuredTool 类。

2. 定义同步函数:

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

   • 定义了一个同步函数 multiply，它接收两个整数参数 a 和 b，返回它们的乘积。

3. 定义异步函数:

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

   • 定义了一个异步函数 amultiply，功能与 multiply 相同，但使用了 async 关键字使其可以在异步环境中运行。

4. 创建 StructuredTool 实例:

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

   • 使用 StructuredTool.from_function 创建一个工具实例 calculator，其中 func 参数指定同步函数 multiply，coroutine 参数指定异步函数 amultiply。这使得 calculator 可以在同步和异步环境中使用。

5. 调用同步工具:

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

   • 使用 calculator.invoke 方法调用同步函数 multiply，传入参数 a=2 和 b=3，输出结果为 6。

6. 调用异步工具:

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

   • 使用 await calculator.ainvoke 方法调用异步函数 amultiply，传入参数 a=2 和 b=5，输出结果为 10。因为是异步调用，使用了 await 关键字。

2. LangChain Runnables

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
# as_tool 的参数信息如下
{'answer_style': {'title': 'Answer Style', 'type': 'string'}}

代码解析

1. 导入模块:

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

   • 从 langchain_core.language_models 模块中导入 GenericFakeChatModel。
   • 从 langchain_core.output_parsers 模块中导入 StrOutputParser。
   • 从 langchain_core.prompts 模块中导入 ChatPromptTemplate。

2. 定义聊天提示模板:

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

   • 创建一个聊天提示模板 prompt，使用 from_messages 方法初始化。
   • 消息模板中包含一个占位符 {answer_style}，用于指定响应的风格。

3. 占位符语言模型:

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

   • 创建一个 GenericFakeChatModel 实例 llm，模拟一个简单的聊天模型。
   • 该模型被初始化为返回一个固定的消息 “hello matey”。

4. 创建输出解析器:

   chain = prompt | llm | StrOutputParser()
   

   • 使用管道符 | 将 prompt、llm 和 StrOutputParser() 连接成一个链 chain。
   • StrOutputParser() 用于将模型的输出解析为字符串格式。

5. 将链转化为工具:

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

   • 使用 chain.as_tool 方法将链封装为一个工具 as_tool。
   • 指定工具的名称为 “Style responder”，并提供了一个简单的描述 “Description of when to use tool.”。

该代码创建了一个简单的对话处理链，结合了提示模板、占位符语言模型和输出解析器；通过 as_tool 方法，将这个链封装成一个可以在更大应用中重用的工具。

如果要对创建的工具进行使用，代码如下：

# 使用 as_tool
input_params = {"answer_style": "gentle"}
# 调用工具，并传递输入参数
output = as_tool.invoke(input_params)
# 输出结果
print(output)

3. 通过继承 BaseTool 类

from typing import Optional, Type

from langchain_core.callbacks import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)
from langchain_core.tools import BaseTool
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: Type[BaseModel] = CalculatorInput
    return_direct: bool = True

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

    async def _arun(
        self,
        a: int,
        b: int,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> str:
        """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())

1. CalculatorInput 类

CalculatorInput 是一个继承自 pydantic.BaseModel 的类，用于定义计算器工具所需的输入参数。它包含两个整数字段：

• a: int：第一个数字，描述为“first number”。
• b: int：第二个数字，描述为“second number”。

Field 函数用于为每个字段提供额外的元数据，例如描述。

2. CustomCalculatorTool 类

CustomCalculatorTool 是一个自定义工具类，继承自 BaseTool。它定义了一个简单的计算器工具，能够同步和异步地运行乘法运算。

关键组件：

• name: str：工具的名称，这里为 "Calculator"。
• description: str：工具的描述，用于说明其用途，这里为 "useful for when you need to answer questions about math"。
• args_schema: Type[BaseModel]：指定工具所需参数的模式，这里为 CalculatorInput 类。
• return_direct: bool：指示工具在运行后是否直接返回结果，这里设置为 True。

关键方法：

• _run 方法：
  • 用于同步执行工具的主要功能。
  • 接受参数 a 和 b 以及可选的 run_manager。
  • 返回两个数字 a 和 b 的乘积。
• _arun 方法：
  • 用于异步执行工具的主要功能。
  • 接受与 _run 方法相同的参数。
  • 如果计算简单，则可以直接调用同步版本（_run 方法）。
  • 如果同步计算耗费资源，则可以删除整个 _arun 方法，此时 LangChain 将自动提供更好的实现，以便在不阻塞其他异步代码的情况下启动任务。

在通过继承 BaseTool 类来定义一个自定义工具 CustomCalculatorTool 类后 ，我们可以创建该类的对象（工具）并对之进行使用。

multiply = CustomCalculatorTool()
print(multiply.name)
Calculator
print(multiply.description)
useful for when you need to answer questions about math
print(multiply.args)
{'a': {'description': 'first number', 'title': 'A', 'type': 'integer'}, 'b': {'description': 'second number', 'title': 'B', 'type': 'integer'}}
print(multiply.return_direct)
True
print(multiply.invoke({"a": 2, "b": 3}))
6
print(await multiply.ainvoke({"a": 2, "b": 3}))
6