贫瘠之地

华北无浪漫，死海扬起帆
多少个夜晚，独自望着天

0%

LangChain 文档学习 No.8 - 工具

发表于 分类于 > >

工具是代理可以用来与世界交互的接口，它们包括：

  • 工具的名称
  • 工具的描述
  • 工具输入 JSON 参数
  • 调用的函数
  • 是否应将工具的结果直接返回给用户

这些信息至关重要，使用这些信息可以用来建立行动系统，LLM 就可以使用名称、描述和 JSON 模式入参作为提示，这样它就知道如何指定要执行的操作，然后要调用的函数就相当于执行该操作

工具的输入越简单，LLM 就越容易使用它。许多代理只使用具有单个字符串输入的工具

需要注意的是名称、描述和 JSON 模式入参（如果使用）都将在提示中被使用。因此这些内容必须清楚并准确地描述应该如何使用该工具；这一点非常重要，如果 LLM 出现了不了解如何使用该工具的情况，则可能需要更改默认名称、描述或 JSON 模式入参

内置工具

LangChain 中内置了很多工具，这里使用 Wikipedia 举例

1
2
api_wrapper = WikipediaAPIWrapper(top_k_results=1, doc_content_chars_max=100)
tool = WikipediaQueryRun(api_wrapper=api_wrapper)

输出一下基本属性

1
2
3
4
5
6
7
8
9
10
11
print(tool.name)
# >> Wikipedia

print(tool.description)
# >> 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.

print(tool.args)
# >> {'query': {'title': 'Query', 'type': 'string'}}

print(tool.return_direct)
# >> False

执行

1
2
3
print(tool.run("China"))
# >> Page: China
# >> Summary: China (Chinese: 中国; pinyin: Zhōngguó), officially the People's Republic of China

LangChain 提供了多种工具，可以在这里查看集成

Tools | 🦜️🔗 Langchain

自定义工具

在构建代理时，需要向它提供一个可以使用的工具列表

除了调用的实际函数外，该工具还包括几个属性：

  • name（str）必填，并且在提供给代理的一组工具中必须是唯一的
  • description（str）选填但是推荐设置，因为它被代理用来确定工具的使用
  • args_schema（Pydantic BaseModel）选填但是推荐设置，可以用于提供更多信息（例如少量示例）或验证预期参数

有多种方法可以定义工具，官方文档中实现了两个类型的工具：

  • 一个虚构的搜索函数，它总是返回字符串 LangChain
  • 一个将两个数字相乘的乘法器函数

两个工具的区别是第一个函数只需要一个输入，而第二个函数需要多个输入；许多 Agent 只处理需要单个输入的函数，因此了解如何处理这些函数很重要

@tool 装饰器

@tool 装饰器是定义自定义工具的最简单方法

默认情况下，装饰器使用函数名称作为工具名称，但可以通过传递字符串作为第一个参数来覆盖此名称；此外装饰器将使用函数的 doc string 作为工具的描述，因此必须提供一个 doc string

1
2
3
4
@tool
def search(query: str) -> str:
    """Look up things online."""
    return "LangChain"
1
2
3
4
5
6
7
8
print(search.name)
# >> search

print(search.description)
# >> search(query: str) -> str - Look up things online.

print(search.args)
# >> {'query': {'title': 'Query', 'type': 'string'}}

还可以通过将工具名称和 JSON 参数传递给工具装饰器来自定义

1
2
3
4
@tool("search-tool", args_schema=SearchInput, return_direct=True)
def search(query: str) -> str:
    """Look up things online."""
    return "LangChain"
1
2
3
4
5
6
7
8
print(search.name)
# >> search-tool

print(search.description)
# >> search-tool(query: str) -> str - Look up things online.

print(search.args)
# >> {'query': {'title': 'Query', 'description': 'should be a search query', 'type': 'string'}}

@Tool 装饰方法的参数

1
2
3
4
5
6
7
def tool(
    *args: Union[str, Callable, Runnable],
    return_direct: bool = False,
    args_schema: Optional[Type[BaseModel]] = None,
    infer_schema: bool = True,
) -> Callable:
  ...

关于 Python 的装饰语法，可以看下方补充内容

BaseTool 子类

使用 BaseTool 子类来显式定义自定义工具

提供了对工具定义的最大控制，但需要做更多的工作

定义入参结构

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

实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
class CustomCalculatorTool(BaseTool):
    name = "Calculator"
    description = "useful for when you need to answer questions about math"
    args_schema: Type[BaseModel] = CalculatorInput
    return_direct: bool = True

    # 需要实现同步执行 _run
    def _run(
            self, a: int, b: int, run_manager: Optional[CallbackManagerForToolRun] = None
    ) -> str:
        """Use the tool."""
        return str(a * b)

    # 需要实现异步执行 _arun
    async def _arun(
            self,
            a: int,
            b: int,
            run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> str:
        """Use the tool asynchronously."""
        raise NotImplementedError("Calculator does not support async")
1
2
3
4
5
6
7
8
9
10
11
12
multiply = CustomCalculatorTool()
print(multiply.name)
# >> Calculator

print(multiply.description)
# >> useful for when you need to answer questions about math

print(multiply.args)
# >> {'a': {'title': 'A', 'description': 'first number', 'type': 'integer'}, 'b': {'title': 'B', 'description': 'second number', 'type': 'integer'}}

print(multiply.return_direct)
# >> True

执行

1
2
print(multiply.run(tool_input={'a': 3, 'b': 4}))
# >> 12

StructuredTool 数据类

也可以使用 StructuredTool 数据类

这种方法是前两种方法的混合，比继承 BaseTool 类更方便，但比装饰器方式提供了更多的功能

1
2
3
4
5
6
7
8
9
10
11
def multiply(a: int, b: int) -> int:
    """Multiply two numbers."""
    return a * b


multiply = StructuredTool.from_function(
    func=multiply,
    name="Calculator",
    description="multiply numbers",
    # coroutine= ... <- you can specify an async method if desired as well
)
1
2
3
4
5
6
7
8
9
10
11
print(multiply.name)
# >> Calculator

print(multiply.description)
# >> Calculator(a: int, b: int) -> int - multiply numbers

print(multiply.args)
# >> {'a': {'title': 'A', 'type': 'integer'}, 'b': {'title': 'B', 'type': 'integer'}}

print(multiply.return_direct)
# >> False

同时也可以定义结构入参

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 定义入参结构
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


multiply = 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
)
1
2
3
4
5
6
7
8
9
10
11
print(multiply.name)
# >> Calculator

print(multiply.description)
# >> Calculator(a: int, b: int) -> int - multiply numbers

print(multiply.args)
# >> {'a': {'title': 'A', 'description': 'first number', 'type': 'integer'}, 'b': {'title': 'B', 'description': 'second number', 'type': 'integer'}}

print(multiply.return_direct)
# >> True

处理工具错误

当工具遇到错误且未捕获异常时，代理将终止执行

如果希望代理继续执行可以 raise ToolException 并相应地设置 handle_tool_error

  • 当抛出 ToolException 时，代理不会停止工作，而是会根据工具的 handle_tool_error 变量处理异常，处理结果会作为观察返回给代理，并以红色日志进行打印
  • 可以将 handle_tool_error 设置为 True，将其设置为统一字符串值，或将其设置成函数；如果它被设置为一个函数，那么该函数应该以 ToolException 作为参数并返回字符串值
  • 仅 raise ToolException 是无效的，需要首先设置该工具的 handle_tool_error，因为其默认值为 False
1
2
3
4
5
6
7
8
9
10
11
def search_tool_error(s: str):
    raise ToolException("工具出现错误")


search = StructuredTool.from_function(
    func=search_tool_error,
    name="Search_tool_error",
    description="A bad tool",
)

search.run("test")
1
2
3
...
  	raise ToolException("工具出现错误")
langchain_core.tools.ToolException: 工具出现错误

设置 handle_tool_error 为 True

1
2
3
4
5
6
7
8
9
search = StructuredTool.from_function(
    func=search_tool_error,
    name="Search_tool_error",
    description="A bad tool",
    handle_tool_error=True
)

print(search.run("test"))
# >> 工具出现错误

还可以定义一种自定义方式来处理工具错误

1
2
def _handle_error(error: ToolException) -> str:
    return error.args[0] + " 所以我不知道"
1
2
3
4
5
6
7
8
9
search = StructuredTool.from_function(
    func=search_tool_error,
    name="Search_tool_error",
    description="A bad tool",
    handle_tool_error=_handle_error
)

print(search.run("test"))
# >> 工具出现错误 所以我不知道

工具箱

工具箱是设计用于特定任务的工具集合，具有便捷的加载方法

完整列表参考集成文档 https://python.langchain.com/docs/integrations/toolkits/

所有工具箱都公开了一个 get_tools 方法，该方法返回一个工具列表

1
2
3
4
5
6
7
8
# Initialize a toolkit
toolkit = ExampleTookit(...)

# Get list of tools
tools = toolkit.get_tools()

# Create agent
agent = create_agent_method(llm, tools, prompt)

例如 SQL Database 工具箱

1
2
3
4
5
6
7
8
toolkit = SQLDatabaseToolkit(db=db, llm=llm)

# 打印下信息
tools = toolkit.get_tools()
for tool in tools:
    print(tool.name)
    print(tool.description)
    print("--------------")
1
2
3
4
5
6
7
8
9
10
11
12
sql_db_query
Input to this tool is a detailed and correct SQL query, output is a result from the database. If the query is not correct, an error message will be returned. If an error is returned, rewrite the query, check the query, and try again. If you encounter an issue with Unknown column 'xxxx' in 'field list', use sql_db_schema to query the correct table fields.
--------------
sql_db_schema
Input to this tool is a comma-separated list of tables, output is the schema and sample rows for those tables. Be sure that the tables actually exist by calling sql_db_list_tables first! Example Input: table1, table2, table3
--------------
sql_db_list_tables
Input is an empty string, output is a comma separated list of tables in the database.
--------------
sql_db_query_checker
Use this tool to double check if your query is correct before executing it. Always use this tool before executing a query with sql_db_query!
--------------

OpenAI Functions

LangChain 工具也可以作为 OpenAI Fuctions

使用 format_tool_to_openai_function 将工具转换为 Functions

因为我没有对应的 API，这里就直接使用官方文档的内容

1
2
3
4
5
6
7
8
9
10
# LLM
model = ChatOpenAI(model="gpt-3.5-turbo-0613")

# 定义 tool，转换 functions
tools = [MoveFileTool()]
functions = [format_tool_to_openai_function(t) for t in tools]

message = model.predict_messages(
    [HumanMessage(content="move file foo to bar")], functions=functions
)
1
2
3
message

AIMessage(content='', additional_kwargs={'function_call': {'name': 'move_file', 'arguments': '{\n  "source_path": "foo",\n  "destination_path": "bar"\n}'}}, example=False)

补充

装饰器语法

当你在一个函数上方使用装饰器时，它实际上相当于将该函数作为参数传递给对应实现，并将其返回的函数替换原来的函数定义

定义一个结构

1
2
3
4
5
6
7
class Student:
    name: str
    age: int
    grade: int

    def __repr__(self):
        return "name:" + self.name + "\nage:" + str(self.age) + "\ngrade:" + str(self.grade)

实现装饰器方法

创建一个学生对象的细节：

  • name 方法名
  • age 方法调用返回值
  • grade 装饰语法的入参
1
2
3
4
5
6
7
8
9
10
11
12
13
14
def my_decorator_student(grade: int) -> Callable:
    def decorator_func(func):
        age = func()

        def build_student(name: str, age: int, grade: int):
            s = Student()
            s.name = name
            s.age = age
            s.grade = grade
            return s

        return build_student(func.__name__, age, grade)

    return decorator_func

使用

1
2
3
4
5
6
@my_decorator_student(grade=5)
def zhang_san():
    return 20


print(zhang_san)
1
2
3
name:zhang_san
age:20
grade:5

参考

Tools | 🦜️🔗 Langchain