大连网络公司企业_郑州居家办公全员核酸_软文广告素材_神马网站快速排名软件

部分代码：

import json
from langchain_core.messages import ToolMessageclass 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"。
  说人话： 将这个执行工具调用的节点注册到整体的流程图中，这样整个系统在运行时就知道如何找到并调用这些工具了。

---

总结

整个代码的作用可以总结为：

1. 准备工作： 导入 JSON 库和消息封装类。
2. 初始化节点： 定义一个工具节点类，它在初始化时接收一组工具，并根据工具名称建立一个映射字典。
3. 执行工具调用： 当该节点被调用时，它从输入中获取最新的消息，然后依次执行消息中列出的每个工具调用，将工具返回的结果包装成消息。
4. 加入系统流程： 将该节点实例加入到整体的图（工作流）中，使其成为系统的一部分。

实际应用示例：
假设在一个对话系统中，AI生成了一条消息，其中包含一个工具调用指令，例如“用 search_tool 搜索‘Python 教程’”。当这个节点接收到这条消息后，它会：

• 解析出最后一条消息；
• 从消息中的 tool_calls 列表中找到关于 search_tool 的调用；
• 调用 search_tool.invoke({"query": "Python 教程"})；
• 将返回的搜索结果打包成一个 ToolMessage 对象，并返回给系统，供后续处理或直接展示给用户。

这样设计的好处在于：

• 模块化：工具调用被封装在一个节点里，易于管理和扩展。
• 清晰的流程：通过图构建器，可以轻松将不同节点（例如输入处理、工具调用、结果汇总等）组合成一个完整的系统。
• 灵活性：使用字典映射工具名称，支持动态添加和调用各种工具，满足多样化的需求。