当前位置：首页 > article >正文

BasicToolNode(tools=[search_tool, lookup_policy, query_sqldb])的内部执行逻辑

article 2025/3/6 21:07:24

部分代码：

import json
from langchain_core.messages import ToolMessage


class BasicToolNode:
    """A node that runs the tools requested in the last AIMessage."""

    def __init__(self, tools: list) -> None:
        self.tools_by_name = {tool.name: tool for tool in tools}

    def __call__(self, inputs: dict):
        if messages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raise ValueError("No message found in input")
        outputs = []
        print('\n\n self.tools_by_name\n===',self.tools_by_name)
        print('\nmessage===\n',message)
        for tool_call in message.tool_calls:
            tool_result = self.tools_by_name[tool_call["name"]].invoke(
                tool_call["args"]
            )
            outputs.append(
                ToolMessage(
                    content=json.dumps(tool_result),
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"],
                )
            )
        return {"messages": outputs}


tool_node = BasicToolNode(tools=[search_tool, lookup_policy, query_sqldb])

graph_builder.add_node("tools", tool_node)

下面我们逐行解释这段代码，并在实际场景中应用它们。

1. 导入模块和依赖

import json
from langchain_core.messages import ToolMessage

import json
这行代码导入了 Python 的内置模块 json。其主要作用是将 Python 的数据结构（例如字典）转换成 JSON 格式的字符串，或者反过来将 JSON 字符串转换成 Python 数据。
举例说明： 假如你有一个字典 {"name": "Alice", "age": 30}，使用 json.dumps() 可以将其转换为字符串 '{"name": "Alice", "age": 30}'。
from langchain_core.messages import ToolMessage
这行代码从 langchain_core.messages 模块中导入了 ToolMessage 类。这个类通常用于 封装工具调用后返回的消息，使其具有统一的结构和格式，方便后续处理和传递。
举例说明： 当你调用某个工具（比如搜索工具）后，返回的结果会被包装成一个 ToolMessage 对象，其中包含工具名称、调用ID和返回内容。

2. 定义节点类 BasicToolNode

class BasicToolNode:
    """A node that runs the tools requested in the last AIMessage."""

这里定义了一个名为 BasicToolNode 的类，它的作用是“节点”——在整个工作流（或图）中，它负责执行上一条 AI 消息中所请求的所有工具调用。
注释（docstring）明确说明了这个类的功能：运行最后一条 AI 消息中请求的工具。

3. 初始化方法 `init`

    def __init__(self, tools: list) -> None:
        self.tools_by_name = {tool.name: tool for tool in tools}

def __init__(self, tools: list) -> None:
构造函数接受一个工具列表 tools，这些工具可能是例如搜索工具、数据库查询工具等。
self.tools_by_name = {tool.name: tool for tool in tools}
这行代码利用列表推导式构建了一个字典，将每个工具的名称映射到工具对象本身。
说人话： 这样做的好处是，当后续需要根据工具名称找到对应工具时，可以直接通过字典查找，而不用遍历整个列表。
举例说明：
假设传入的 tools 列表包含三个工具对象，每个对象都有一个 name 属性，如 "search_tool"、"lookup_policy" 和 "query_sqldb"。那么构建后的字典就会是：
```
{
    "search_tool": <search_tool 对象>,
    "lookup_policy": <lookup_policy 对象>,
    "query_sqldb": <query_sqldb 对象>
}
```
当你需要调用 "search_tool" 时，只需执行 self.tools_by_name["search_tool"] 即可快速定位到对应的工具对象。

4. 定义可调用方法 `call`

    def __call__(self, inputs: dict):
        if messages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raise ValueError("No message found in input")

def __call__(self, inputs: dict):
这里定义了 __call__ 方法，使得 BasicToolNode 的实例可以像函数一样被调用，接受一个字典类型的输入。
if messages := inputs.get("messages", []):
这行使用了 Python 3.8 引入的“海象运算符” :=。它尝试从输入字典中取出键 "messages" 对应的值（如果没有则默认返回空列表 \[\]），并将其赋值给变量 messages。
说人话： 如果输入中包含消息列表，就把它取出来；否则返回空列表。
message = messages[-1]
取出消息列表中的最后一条消息。
举例说明： 如果消息列表是 [msg1, msg2, msg3]，这里会选择 msg3，因为我们通常假设最后一条消息包含最新的工具调用指令。
else: raise ValueError("No message found in input")
如果输入中没有 "messages"，程序会抛出一个错误，提示“没有找到消息”。
说人话： 程序要求必须有一条消息，否则无法进行工具调用。

5. 遍历并执行工具调用

        outputs = []
        for tool_call in message.tool_calls:
            tool_result = self.tools_by_name[tool_call["name"]].invoke(
                tool_call["args"]
            )

outputs = []
初始化一个空列表 outputs，用于存储每个工具调用的返回结果。
for tool_call in message.tool_calls:
遍历消息对象中存储的工具调用指令列表。假设消息对象有一个属性 tool_calls，它是一个列表，每个元素都是一个字典，描述了某个工具调用的信息。
tool_result = self.tools_by_name[tool_call["name"]].invoke(tool_call["args"])
- 解析过程：
  - 从 tool_call 字典中获取工具的名称：tool_call["name"]。
  - 利用之前构造的 tools_by_name 字典，找到对应的工具对象。
  - 调用该工具对象的 invoke 方法，并传入工具调用的参数（tool_call["args"]）。
- 说人话： 根据消息里的指令，找到对应的工具并执行操作，然后把执行结果存储在 tool_result 变量中。
- 举例说明：
  假如 tool_call 是：
```
{
    "id": "123",
    "name": "search_tool",
    "args": {"query": "Python 教程"}
}
```
  那么这行代码就会找到名为 "search_tool" 的工具，并执行 search_tool.invoke({"query": "Python 教程"})。假设该调用返回了搜索结果，比如 {"results": ["教程1", "教程2"]}，那么 tool_result 就会是这个字典。

6. 封装工具调用的返回结果

outputs.append(
        ToolMessage(
            content=json.dumps(tool_result),
            name=tool_call["name"],
            tool_call_id=tool_call["id"],
        )
    )

ToolMessage(...)
创建一个 ToolMessage 对象，用于包装工具调用的结果。
- content=json.dumps(tool_result)
  将工具返回的结果 tool_result 转换为 JSON 字符串，方便消息传输和后续处理。
- name=tool_call["name"]
  将工具的名称传递过去，便于标识这个消息是由哪个工具产生的。
- tool_call_id=tool_call["id"]
  保留工具调用的标识符，这样在后续流程中可以追踪到这个调用的上下文。
outputs.append(...)
将创建好的 ToolMessage 对象添加到 outputs 列表中。

说人话： 每个工具调用执行完后，我们将返回的结果包装成一个标准格式的消息，然后存储到一个列表中，方便后续统一返回。

7. 返回结果

        return {"messages": outputs}

这行代码将封装好的工具消息列表放入一个字典中，键名为 "messages"，并作为函数的返回值返回。
说人话： 最后，所有工具执行的结果都会被打包成一个消息列表返回出去。

8. 创建节点实例并加入图中

tool_node = BasicToolNode(tools=[search_tool, lookup_policy, query_sqldb])
graph_builder.add_node("tools", tool_node)

tool_node = BasicToolNode(tools=[search_tool, lookup_policy, query_sqldb])
创建了一个 BasicToolNode 的实例，并传入一个工具列表。这里假设 search_tool、lookup_policy、query_sqldb 是之前已经定义好的工具对象。
举例说明：
- search_tool 是用于互联网搜索的工具；
- lookup_policy 用于查找某些策略或者规则；
- query_sqldb 用于对 SQL 数据库进行查询。
graph_builder.add_node("tools", tool_node)
这里假设存在一个 graph_builder 对象，它负责构建或管理一个工作流图。通过这行代码，我们将创建的 tool_node 节点加入到图中，节点名称为 "tools"。
说人话： 将这个执行工具调用的节点注册到整体的流程图中，这样整个系统在运行时就知道如何找到并调用这些工具了。