LangChain大模型应用开发：自定义工具调用

介绍

大家好，博主又来给大家分享知识了。今天要给大家分享的是使用LangChain进行大模型应用开发中的自定义工具调用。

在大模型应用开发的领域里，LangChain凭借其强大的功能和灵活性，成为了开发者们的得力助手。自定义工具调用则是LangChain众多亮点特性之一，它赋予了开发者根据特定需求打造专属工具的能力。

好了，我们直接进入正题。

自定义工具

在LangChain的应用开发中，自定义工具是实现灵活且强大功能的关键要素之一。尤其是在构建智能体(Agent)时，工具的合理配置和使用至关重要。智能体就像是大模型的 “助手”，能够代表大模型执行各种特定任务。而自定义工具则为这个 “助手” 提供了多样化的 “工作器具”，让其能够更好地应对不同场景下的需求。

接下来我介绍在构建智能体时自定义工具的相关内容。在构建智能体时，我们需要为其提供一个Tool列表，以便智能体可以使用这些工具。除了实际调用的函数之外，Tool由几个组件组成：

属性	类型	描述
name	str	在提供给LLM或智能体工具中必须是唯一的。
description	str	描述工具的功能。LLM或智能体将使用此描述作为上下文。
args_schema	Pydantic BaseModel	可选但建议，可用于提供更多信息(例如，few-shot示例)验证预期参数。
return_direct	boolean	仅对智能体相关。当为True时，在调用给定工具后，智能体将停止并将结果直接返回给用户。

LangChain提供了三种创建工具的方式：

1. 使用@tool装饰器——定义自定义工具的最简单方式。
2. 使用StructuredTool.from_function类方法——这类似于@tool装饰器，但允许更多配置和同步和异步实现的规范。
3. 通过子类化BaseTool——这是最灵活的方法，它提供了最大程度的控制，但需要更多的工作量和代码。@tool或StructuredTool.from_function类方法对于大多数用例应该足够了。提示：如果工具具有精心选择的名称、描述和JSON模式，模型的性能会更好。

装饰器@tool

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

装饰器@tool应用完整代码

# 从langchain_core库的tools模块导入tool装饰器，用于将函数转换为工具
from langchain_core.tools import tool


# 使用tool装饰器，将下方函数转换为LangChain可用的工具
@tool
# 定义一个名为multiply_numbers的函数，接收两个整数类型的参数，返回值也是整数
def multiply_numbers(num1: int, num2: int) -> int:
    # """此方法用于将两个整数相乘并返回结果。"""
    # 返回两个输入参数相乘的结果
    return num1 * num2


# 打印multiply_numbers工具的名称
print(multiply_numbers.name)
# 打印multiply_numbers工具的描述信息
print(multiply_numbers.description)
# 打印multiply_numbers工具的参数信息
print(multiply_numbers.args)

装饰器@tool应用运行结果

multiply_numbers
此方法用于将两个整数相乘并返回结果。
{'num1': {'title': 'Num1', 'type': 'integer'}, 'num2': {'title': 'Num2', 'type': 'integer'}}

进程已结束，退出代码为 0

我们使用@tool装饰器还可以应用于异步方法中。

装饰器@tool应用异步方法完整代码

# 从langchain_core库的tools模块导入tool装饰器，用于将函数转换为工具
from langchain_core.tools import tool


# 使用tool装饰器，将下方函数转换为LangChain可用的工具
@tool
# 定义一个名为multiply_numbers的异步函数，接收两个整数类型的参数，返回值也是整数
async def multiply_numbers(num1: int, num2: int) -> int:
    """此方法用于将两个整数相乘并返回结果。"""
    # 返回两个输入参数相乘的结果
    return num1 * num2


# 打印multiply_numbers工具的名称
print(multiply_numbers.name)
# 打印multiply_numbers工具的描述信息
print(multiply_numbers.description)
# 打印multiply_numbers工具的参数信息
print(multiply_numbers.args)

装饰器@tool应用异步方法运行结果

multiply_numbers
此方法用于将两个整数相乘并返回结果。
{'num1': {'title': 'Num1', 'type': 'integer'}, 'num2': {'title': 'Num2', 'type': 'integer'}}

进程已结束，退出代码为 0

我们还可以将装饰器@tool定义工具名称，输入参数验证，并设置该方法是否立刻返回结果等。

装饰器@tool多参数应用完整代码

# 从langchain_core库的tools模块导入tool装饰器，用于将函数转换为工具
from langchain_core.tools import tool
# 从pydantic库导入BaseModel和Field，用于定义数据模型和字段元数据
from pydantic import BaseModel, Field


# 定义一个继承自BaseModel的数据模型类，用于规范乘法工具的输入参数
class CalculatorInput(BaseModel):
    # 定义第一个整数类型的字段num1，并添加描述信息
    num1: int = Field(description="first number")
    # 定义第二个整数类型的字段num2，并添加描述信息
    num2: int = Field(description="second number")


# 使用tool装饰器将函数转换为工具，指定工具名称、参数模式和返回方式
@tool("multiplication-tool", args_schema=CalculatorInput, return_direct=True)
# 定义一个名为multiply_numbers的函数，接收两个整数参数，返回它们的乘积
def multiply_numbers(num1: int, num2: int) -> int:
    # 函数的文档字符串，描述函数的功能是将两个数相乘
    """此方法用于将两个整数相乘并返回结果。"""
    # 返回两个输入参数相乘的结果
    return num1 * num2


# 打印乘法工具的名称
print(multiply_numbers.name)
# 打印乘法工具的描述信息
print(multiply_numbers.description)
# 打印乘法工具的参数信息
print(multiply_numbers.args)
# 打印乘法工具是否直接返回结果的标志
print(multiply_numbers.return_direct)

装饰器@tool多参数应用运行结果

multiplication-tool
此方法用于将两个整数相乘并返回结果。
{'num1': {'description': 'first number', 'title': 'Num1', 'type': 'integer'}, 'num2': {'description': 'second number', 'title': 'Num2', 'type': 'integer'}}
True

进程已结束，退出代码为 0

博主笔记：当我们使用@tool这个装饰器(在LangChain框架中用于将普通方法转换为工具方法)来定义一个工具方法(也就是被@tool装饰的方法)的时候，这个@tool装饰器会把被装饰方法所写的文档字符串(通常是在方法定义内部，使用三引号"""括起来的那部分文本内容)当作是这个工具的描述信息。如果没有该描述信息，则代码运行中会报异常。

结构化工具

StructuredTool。在LangChain里，工具是可以被大语言模型调用的外部功能单元。StructuredTool是一种特殊类型的工具，它允许开发者以结构化的方式定义工具的输入和输出，借助类型注解和pydantic模型来明确工具的参数和返回值的类型与结构。

关于LangChain中的工具定义方式，StructuredTool.from_function类方法和@tool装饰器是常用的两种。相比之下，StructuredTool.from_function类方法提供了比@tool装饰器更多的可配置性，而无需太多额外的代码。

完整代码

# 从langchain_core库的tools模块导入StructuredTool类，用于创建结构化工具
from langchain_core.tools import StructuredTool
# 导入asyncio库，用于支持异步编程
import asyncio


# 定义一个同步函数，用于将两个整数相乘并返回结果
def multiply_numbers(num1: int, num2: int) -> int:
    # 函数文档字符串，描述函数功能
    """此方法用于将两个整数相乘并返回结果。"""
    # 返回两个整数相乘的结果
    return num1 * num2


# 定义一个异步函数，用于将两个整数相乘并返回结果
async def async_multiply_numbers(num1: int, num2: int) -> int:
    # 函数文档字符串，描述函数功能
    """此方法用于将两个整数相乘并返回结果。"""
    # 返回两个整数相乘的结果
    return num1 * num2


# 定义一个异步函数，用于测试工具的调用
async def main():
    # 使用StructuredTool的from_function方法创建一个工具实例，同时指定同步和异步函数
    calculator = StructuredTool.from_function(func=multiply_numbers, coroutine=async_multiply_numbers)
    # 同步调用工具，传入参数并打印结果
    print(calculator.invoke({"num1": 2, "num2": 3}))
    # 异步调用工具，传入参数并等待结果返回后打印
    print(await calculator.ainvoke({"num1": 4, "num2": 5}))


# 运行异步函数
asyncio.run(main())

运行结果

6
20

进程已结束，退出代码为 0

与装饰器@tool一样，StructuredTool.from_function()方法也可以设置多个参数。

多参数完整代码

# 从langchain_core库的tools模块导入StructuredTool类，用于创建结构化工具
from langchain_core.tools import StructuredTool
# 从pydantic库导入BaseModel和Field，用于定义数据模型和字段元数据
from pydantic import BaseModel, Field
# 导入asyncio库，用于支持异步编程
import asyncio


# 定义一个继承自BaseModel的数据模型类，用于规范工具的输入参数
class CalculatorInput(BaseModel):
    # 定义第一个整数类型的字段 num1，并添加描述信息
    num1: int = Field(description="first number")
    # 定义第二个整数类型的字段 num2，并添加描述信息
    num2: int = Field(description="second number")


# 定义一个同步函数，用于将两个整数相乘并返回结果
def multiply_numbers(num1: int, num2: int) -> int:
    # 函数的文档字符串，描述函数的功能是将两个数相乘
    """此方法用于将两个整数相乘并返回结果。"""
    # 返回两个输入参数相乘的结果
    return num1 * num2


# 定义一个异步函数，用于测试工具的调用和输出工具信息
async def main():
    # 使用StructuredTool的from_function方法创建一个工具实例
    calculator = StructuredTool.from_function(
        # 指定同步函数为multiply_numbers
        func=multiply_numbers,
        # 指定工具的名称为"Calculator"
        name="Calculator",
        # 指定工具的描述信息为"multiply numbers"
        description="multiply numbers",
        # 指定工具的输入参数验证模式为CalculatorInput
        args_schema=CalculatorInput,
        # 设置工具调用后直接返回结果
        return_direct=True,
        # coroutine= async_addition
        # coroutine= ... <- 如果需要，也可以指定异步方法
    )
    # 同步调用工具，传入参数并打印结果
    print(calculator.invoke({"num1": 2, "num2": 3}))
    # 异步调用工具，传入参数并等待结果返回后打印
    print(await calculator.ainvoke({"num1": 4, "num2": 5}))
    # 打印工具的名称
    print(calculator.name)
    # 打印工具的描述信息
    print(calculator.description)
    # 打印工具的参数信息
    print(calculator.args)


# 运行异步函数
asyncio.run(main())

多参数运行结果

6
20
Calculator
multiply numbers
{'num1': {'description': 'first number', 'title': 'Num1', 'type': 'integer'}, 'num2': {'description': 'second number', 'title': 'Num2', 'type': 'integer'}}

进程已结束，退出代码为 0

工具错误处理

如果我们正在使用带有工具的智能体，我们还需要一个错误处理策略，以便智能可以从错误中恢复并继续执行。

一个简单的策略是在工具内部抛出ToolException，并使用handle_tool_error指定一个错误处理程序。当指定了错误处理程序时，异常将被捕获，错误处理程序将决定从工具返回哪个输出。

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

注意，仅仅抛出ToolException是不会生效的。我们需要首先设置工具的handle_tool_error，因为其默认值是False。

完整代码

# 从langchain_core库的tools模块导入ToolException和StructuredTool类
from langchain_core.tools import ToolException, StructuredTool


# 定义一个函数，用于获取给定城市的天气
def get_weather(city: str) -> int:
    # 函数文档字符串，描述函数功能
    """获取给定城市的天气。"""
    # 抛出一个ToolException异常，模拟获取天气时找不到城市的错误
    raise ToolException(f"错误：没有名为{city}的城市。")


# 使用StructuredTool的from_function方法创建一个工具实例，设置handle_tool_error为True来开启错误处理
get_weather_tool = StructuredTool.from_function(
    # 指定工具对应的函数为get_weather
    func=get_weather,
    # 设置handle_tool_error为True，开启默认的错误处理
    handle_tool_error=True,
)
# 调用工具，传入城市名称为"unknown"，并将结果赋值给result
result = get_weather_tool.invoke({"city": "unknown"})
# 打印工具调用的结果
print(result)

# 再次使用StructuredTool的from_function方法创建一个工具实例，这次将handle_tool_error设置为字符串
get_weather_tool = StructuredTool.from_function(
    # 指定工具对应的函数为get_weather
    func=get_weather,
    # 设置handle_tool_error为一个字符串，作为错误发生时的返回信息
    handle_tool_error="没找到这座城市",
)
# 调用工具，传入城市名称为"unknown"，并将结果赋值给result
result = get_weather_tool.invoke({"city": "unknown"})
# 打印工具调用的结果
print(result)


# 定义一个错误处理函数，接收ToolException类型的参数并返回一个字符串
def tool_handle_error(error: ToolException) -> str:
    # 返回包含错误信息的字符串
    return f"工具执行期间发生以下错误：`{error.args[0]}`"


# 又一次使用StructuredTool的from_function方法创建一个工具实例，将handle_tool_error设置为自定义的错误处理函数
get_weather_tool = StructuredTool.from_function(
    # 指定工具对应的函数为get_weather
    func=get_weather,
    # 设置handle_tool_error为自定义的错误处理函数tool_handle_error
    handle_tool_error=tool_handle_error,
)

# 调用工具，传入城市名称为"unknown"，并将结果赋值给result
result = get_weather_tool.invoke({"city": "unknown"})
# 打印工具调用的结果
print(result)

运行结果

错误：没有名为unknown的城市。
没找到这个城市
工具执行期间发生以下错误：`错误：没有名为unknown的城市。`

进程已结束，退出代码为 0

从上面的结果我们看到：

第一次，我们将StructuredTool.from_function方法里的handle_tool_error设置为True，那么在代码运行后直接输出的是异常捕获的错误提示，即：raise ToolException(f"错误：没有名为{city}的城市。")。

第二次，我们将StructuredTool.from_function方法里的handle_tool_error设置为字符串："没找到这座城市"，那么在代码运行后直接将这段字符串打印出来，即：“没有找到这座城市”。

第三次，我们将StructuredTool.from_function方法里的handle_tool_error设置为一个错误处理函数tool_handle_error(error: ToolException)，那么在代码运行后直接将tool_handle_error方法里面的错误信息打印了出来并且加上get_weather方法里的异常错误提示。即："工具执行期间发生以下错误：`错误：没有名为unknown的城市。`"。

调用内置工具包和拓展工具

工具

1. 工具是智能体、链或聊天模型/大语言模型(LLM)用来与世界交互的接口。一个工具由以下组件组成：
2. 工具的名称
3. 工具的功能描述
4. 工具输入的JSON模式
5. 要调用的函数
6. 工具的结果是否应直接返回给用户(仅对智能体相关)名称、描述和JSON模式作为上下文提供给 LLM，允许LLM适当地确定如何使用工具。给定一组可用工具和提示，LLM可以请求调用一个或多个工具，并提供适当的参数。通常，在设计供聊天模型或LLM使用的工具时，重要的是要牢记以下几点：
   1. 经过微调以进行工具调用的聊天模型将比未经微调的模型更擅长进行工具调用。
   2. 未经微调的模型可能根本无法使用工具，特别是如果工具复杂或需要多次调用工具。
   3. 如果工具具有精心选择的名称、描述和JSON模式，则模型的性能将更好。
   4. 简单的工具通常比更复杂的工具更容易让模型使用。

LangChain拥有大量第三方工具。我们可以使用下面链接访问工具集成查看可用工具列表。Tools | 🦜️🔗 LangChain

下面，我们将结合Python中的wikipedia库，用代码的方式给大家来讲解这部分内容。

我们首先安装wikipedia库，安装命令为：

pip install wikipedia

完整代码

# 从langchain_community库的tools模块导入WikipediaQueryRun类，用于运行维基百科查询工具
from langchain_community.tools import WikipediaQueryRun
# 从langchain_community库的utilities模块导入WikipediaAPIWrapper类，用于包装维基百科API
from langchain_community.utilities import WikipediaAPIWrapper

# 创建一个WikipediaAPIWrapper实例，设置返回的最大结果数为1，文档内容最大字符数为100
api_wrapper = WikipediaAPIWrapper(top_k_results=1, doc_content_chars_max=100)

# 使用创建好的API包装器实例创建一个WikipediaQueryRun工具实例
tool = WikipediaQueryRun(api_wrapper=api_wrapper)

# 调用工具，传入查询关键词"langchain"并打印查询结果
print(tool.invoke({"query": "langchain"}))
# 打印工具的名称
print(f"名称: {tool.name}")
# 打印工具的描述信息
print(f"描述: {tool.description}")
# 打印工具的参数模式信息
print(f"参数模式: {tool.args}")
# 打印工具是否直接返回结果的信息
print(f"直接结果返回吗？: {tool.return_direct}")

运行结果

Page: LangChain
Summary: LangChain is a software framework that helps facilitate the integration of 
名称: wikipedia
描述: A wrapper around Wikipedia. Useful for when you need to answer general questions about people, places, companies, facts, historical events, or other subjects. Input should be a search query.
参数模式: {'query': {'description': 'query to look up on wikipedia', 'title': 'Query', 'type': 'string'}}
直接结果返回吗？: False

进程已结束，退出代码为 0

通过上面完整代码与运行结果我们看到，我们利用LangChain库中的工具和实用程序，通过维基百科API进行对LangChain概念查询，代码运行后展示查询后的相关信息，我们也可以根据自身需要输出打印不同的内容。

自定义默认工具

我们还可以修改内置工具的名称、描述和参数的JSON模式。

在定义参数的JSON模式时，务必确保其输入与函数的输入要求一致，不要对其进行改动。因为一旦改变，函数可能无法正确处理输入，导致工具运行异常。不过，我们可以很方便地为每个输入参数编写自定义描述，以此说明参数的用途、取值限制等信息，这有助于更好地理解和使用该工具，也能让与工具交互的模型更准确地提供合适的输入。

完整代码

# 从langchain_community的tools模块中导入WikipediaQueryRun类，用于运行维基百科查询操作
from langchain_community.tools import WikipediaQueryRun
# 从langchain_community的utilities模块中导入WikipediaAPIWrapper类，用于包装维基百科API以便进行交互
from langchain_community.utilities import WikipediaAPIWrapper
# 从pydantic库中导入BaseModel用于定义数据模型，导入Field用于设置字段的元数据
from pydantic import BaseModel, Field

# 创建一个WikipediaAPIWrapper实例，设置返回的最大结果数为1，文档内容最大字符数为100
api_wrapper = WikipediaAPIWrapper(top_k_results=1, doc_content_chars_max=100)


class WikiInputs(BaseModel):
    """维基百科工具的输入。"""
    # 定义一个名为query的字符串类型字段，用于接收在维基百科中查找的查询内容，有描述信息
    query: str = Field(
        description="要在维基百科中查找的查询内容，应该为三个单词或更少。"
    )


tool = WikipediaQueryRun(
    # 设置工具的名称为“维基工具”
    name="维基工具",
    # 设置工具的描述信息为“在维基百科中查找信息”
    description="在维基百科中查找信息",
    # 指定工具的参数模式为前面定义的WikiInputs类
    args_schema=WikiInputs,
    # 传入之前创建的WikipediaAPIWrapper实例
    api_wrapper=api_wrapper,
    # 设置为直接返回结果
    return_direct=True,
)
# 调用工具的run方法，传入查询关键词“langchain”，并打印查询结果
print(tool.run("langchain"))
# 打印工具的名称
print(f"名称: {tool.name}")
# 打印工具的描述信息
print(f"描述: {tool.description}")
# 打印工具的参数模式
print(f"参数模式: {tool.args}")
# 打印工具是否直接返回结果的信息
print(f"直接结果返回吗？: {tool.return_direct}")

运行结果

Page: LangChain
Summary: LangChain is a software framework that helps facilitate the integration of 
名称: 维基工具
描述: 在维基百科中查找信息
参数模式: {'query': {'description': '要在维基百科中查找的查询内容，应该为三个单词或更少。', 'title': 'Query', 'type': 'string'}}
直接结果返回吗？: True

进程已结束，退出代码为 0

从完整代码我们看到这里定义了WikiInputs类，它继承自BaseModel。query字段使用Field来添加描述信息，这个类实际上就定义了工具输入参数的结构和约束。从JSON模式的角度看，它规定了输入数据应该有一个名为query的字符串类型字段，并且对该字段进行了描述。

当在WikipediaQueryRun中使用args_schema=WikiInputs时：args_schema接收的WikiInputs 类就相当于指定了工具输入的JSON模式，工具在运行时会期望输入的数据符合这个定义的结构和要求，类似于JSON模式对数据的验证和规范作用。虽然代码中没有直接操作JSON字符串或对象，但通过这种方式实现了与JSON模式相关的功能，即定义和验证工具输入数据的格式和属性。

使用内置工具包

LangChain内置工具包是LangChain框架中一系列预先构建好的工具集合，旨在帮助开发者更高效地构建基于语言模型的应用程序，降低开发门槛并提高开发效率。这些工具包涵盖了多个方面，能够满足不同场景下的需求。

CopyFileTool是LangChain内置工具包中的一员，主要用于在文件系统中执行文件复制操作。它允许开发者在语言模型的驱动下，方便地将一个文件从源路径复制到目标路径。这在一些涉及文件处理的应用场景中非常有用，例如自动化数据备份、文件迁移等任务。通过与LangChain的其他组件(如语言模型、智能体等)结合使用，可以实现更加智能和自动化的文件管理流程。

安装或更新langchain-community

运行下面的代码，需要安装或更新langchain-community。安装命令：

pip install -U langchain-community

完整代码

# 从langchain_community.tools模块导入CopyFileTool类，用于文件复制操作
from langchain_community.tools import CopyFileTool
# 导入os模块，用于操作系统相关的功能，如文件路径操作、文件存在性检查等
import os

# 初始化CopyFileTool工具，创建一个CopyFileTool类的实例
copy_file_tool = CopyFileTool()

# 定义源文件路径，这里指定为test_source.txt
source_file_path = "test_source.txt"
# 定义目标文件路径，这里指定为test_target.txt
target_file_path = "test_target.txt"

# 创建一个测试源文件并写入内容，以UTF-8编码打开文件，写入文本后关闭文件
with open(source_file_path, 'w', encoding='utf-8') as file:
    # 向源文件中写入指定的测试内容
    file.write("这是一个测试文件的内容。")

# 调用CopyFileTool工具执行文件复制操作，捕获可能出现的异常
try:
    # 调用copy_file_tool的run方法，传入源文件路径和目标文件路径进行文件复制，并获取复制结果
    result = copy_file_tool.run({"source_path": source_file_path, "destination_path": target_file_path})
    # 打印文件复制的结果信息
    print(f"文件复制结果: {result}")

    # 检查目标文件是否存在并读取其内容
    if os.path.exists(target_file_path):
        # 以UTF-8编码打开目标文件，读取文件内容后关闭文件
        with open(target_file_path, 'r', encoding='utf-8') as file:
            # 读取目标文件的内容
            content = file.read()
            # 打印目标文件的内容
            print(f"目标文件内容: {content}")
    else:
        # 如果目标文件不存在，打印提示信息
        print("目标文件未成功创建。")
# 捕获文件复制过程中出现的任何异常，并打印错误信息
except Exception as e:
    print(f"文件复制过程中出现错误: {e}")

运行结果

文件复制结果: File copied successfully from test_source.txt to test_target.txt.
目标文件内容: 这是一个测试文件的内容。

进程已结束，退出代码为 0

请大家注意，以上代码仅为演示功能用途。在实际应用中，请大家根据具体需求和环境进行适当的调整和扩展。

结束

好了，以上就是本次分享的全部内容了。希望大家能够掌握本次分享的内容，并灵活的去使用LangChain中的自定义工具，结构化工具，以及调用内置工具包和拓展工具。

通过熟练运用这些工具，我们能够更轻松地实现语言模型与其他领域的深度融合，开发出更具创新性和实用性的应用，无论是在智能客服、内容生成、知识图谱构建还是其他相关领域，都能发挥出巨大的潜力。

博主期待大家在实际应用中不断探索和实践，充分挖掘LangChain的价值，创造出更多优秀的成果。

那么本次分享就到这了。最后，博主还是那句话：请大家多去大胆的尝试和使用，成功总是在不断的失败中试验出来的，敢于尝试就已经成功了一半。如果大家对博主分享的内容感兴趣或有帮助，请点赞和关注。大家的点赞和关注是博主持续分享的动力🤭，博主也希望让更多的人学习到新的知识。