第 06 期 | 原生大招：利用 ToolNode 无缝接入外部技能

更新于 2026/4/14

🎯 本期学习目标

各位架构师们，欢迎回到《LangGraph 多智能体专家课》！这一期，我们要放大招了。在 LangGraph 的世界里，智能体（Agent）的强大，绝不仅仅在于它们能“思考”，更在于它们能“行动”——也就是调用外部工具。之前我们可能还在手搓工具执行节点，但今天，我将带你见识 LangGraph 原生提供的“黑科技”：ToolNode。学完本期，你将：

  1. 理解 ToolNode 的核心价值： 为什么它比你自己写一个工具执行节点更优雅、更健壮、更符合 LangGraph 的设计哲学。
  2. 掌握 ToolNode 的使用姿势： 轻松将 langchain_core.tools.Tool 封装成 LangGraph 中的一等公民。
  3. 为 Agency 注入外部搜索能力： 以我们的 Researcher 智能体为例，无缝接入搜索引擎，让它不再是“井底之蛙”。
  4. 洞察智能体与工具的协作模式： 学习如何设计智能体，使其能灵活判断何时需要调用工具，以及如何解析工具返回的结果。

📖 原理解析

在 AI Content Agency 的宏大叙事中，我们的 Researcher 智能体肩负着“求真务实”的重任。它需要深入互联网的汪洋大海，为 Writer 提供最准确、最前沿的资料。想象一下，如果 Researcher 每次需要搜索都得自己去调用 requests、解析 HTML，那它还怎么有精力去“思考”呢？这不仅效率低下，还极易出错。

这就是 ToolNode 诞生的意义！

什么是 ToolNode？

简单来说，ToolNode 是 LangGraph 提供的一种特殊节点类型，它的核心职责是：接收智能体（或任何上游节点）发出的工具调用请求，执行对应的工具，然后将工具的执行结果返回到图的状态中。

它就像一个专业的“工具执行官”，你告诉它要用什么工具、参数是什么，它就负责帮你搞定一切，然后把结果汇报给你。智能体只需要发出“指令”，而无需关心“执行细节”。

为什么选择 ToolNode？

  1. 解耦与简化： 将工具的执行逻辑从智能体的决策逻辑中分离。智能体只负责“决策”和“指令”，ToolNode 负责“执行”。这让你的图结构更清晰，每个节点的职责更单一。
  2. 标准化接口： ToolNode 能够理解并处理 langchain_core.tools.ToolInvocation 对象。这意味着只要你的 LLM 能够以这种标准格式输出工具调用指令（这在 LangChain/LangGraph 中是标配），ToolNode 就能开箱即用。
  3. 状态管理自动化： ToolNode 会自动将工具的执行结果封装成 ToolMessage，并添加到图的 messages 状态中。这意味着你的智能体可以轻松地从 messages 历史中获取工具的输出，继续进行决策。
  4. 鲁棒性： 官方提供的 ToolNode 经过精心设计和测试，通常比我们自己手搓的工具执行逻辑更健壮，能更好地处理各种边缘情况。

ToolNode 的工作流

我们来看一个 Researcher 智能体如何与 ToolNode 协作的典型流程，通过 Mermaid 图来直观感受一下：

graph TD
    A[Start: User Query] --> B(Planner Agent);
    B --> C{Planner Decision: Need Research?};
    C -- Yes --> D(Researcher Agent);
    D -- Researcher Output: ToolInvocation --> E(ToolNode: Search Tool);
    E -- Tool Result: ToolMessage --> D;
    D -- Researcher Output: Final Answer / Need More Tools --> F{Researcher Decision: Research Done?};
    F -- No --> D;
    F -- Yes --> G(Writer Agent);
    G --> H[End: Content Draft];

    style A fill:#f9f,stroke:#333,stroke-width:2px;
    style H fill:#f9f,stroke:#333,stroke-width:2px;
    style E fill:#ccf,stroke:#333,stroke-width:2px;
    style D fill:#bbf,stroke:#333,stroke-width:2px;
    style B fill:#bbf,stroke:#333,stroke-width:2px;

图解说明：

  1. Planner Agent (B): 接收用户请求，决定是否需要 Researcher 介入。
  2. Researcher Agent (D):
    • 思考阶段： 接收当前状态（包括历史消息），根据其内部逻辑判断是否需要调用外部工具（如搜索）。
    • 输出工具调用： 如果需要，它会生成一个 ToolInvocation 对象，指示需要调用的工具名称和参数。
  3. ToolNode: Search Tool (E):
    • 接收指令： 捕获到 Researcher Agent 输出的 ToolInvocation。
    • 执行工具： 根据 ToolInvocation 中的信息，调用预先绑定的搜索工具（例如 DuckDuckGoSearch）。
    • 返回结果： 将搜索结果封装成 ToolMessage，并将其添加到图的全局状态中。
  4. Researcher Agent (D - 回环): 再次被激活，这一次，它的输入状态中包含了 ToolMessage（也就是搜索结果）。Researcher 可以读取这些结果，进行分析、提炼，然后决定是继续搜索、调用其他工具，还是已经得到了足够的资料，可以输出最终的调研报告。

看到了吗？ToolNode 完美地嵌入了智能体的工作流中，让工具调用变得像呼吸一样自然。它就像是你的 AI Content Agency 的“行政助理”，专门负责处理各种工具调用，让你的核心“专家”们（Planner, Researcher 等）可以专注于自己的专业领域。

核心概念：ToolInvocation 与 ToolMessage

理解 ToolNode，就必须理解 LangChain/LangGraph 中的两个核心消息类型：

  • ToolInvocation (工具调用请求): 这是 Agent 决定使用工具时发出的“意图”。它通常包含 tool (工具名称) 和 tool_input (工具参数)。例如：ToolInvocation(tool="duckduckgo_search", tool_input={"query": "LangGraph ToolNode tutorial"})。
  • ToolMessage (工具执行结果): 这是 ToolNode 执行完工具后，将结果封装成的消息。它包含 content (工具的输出结果) 和 tool_call_id (可选，用于关联到哪个 ToolInvocation)。例如：ToolMessage(content="LangGraph ToolNode is a powerful feature...", tool_call_id="call_abc123")。

ToolNode 的神奇之处在于，它能够识别 ToolInvocation，执行工具，然后生成 ToolMessage。这个 ToolMessage 会被添加到 LangGraph State 中的 messages 列表中，供后续的智能体读取。

💻 实战代码演练 (Agency项目中的具体应用)

好了，理论说得再多，不如撸起袖子干一场！现在，我们将把 ToolNode 真正融入到我们的 AI Content Agency 项目中，为 Researcher 智能体添加强大的搜索能力。

准备工作：定义工具与状态

首先，我们需要一个真实的搜索工具。这里我们使用 DuckDuckGoSearchRun，它是一个简单易用的搜索工具。

# agency_core/state.py (假设我们有一个通用的状态文件)
from typing import List, TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, FunctionMessage, ToolMessage, AIMessage

# 定义我们的Agency的全局状态
class AgentState(TypedDict):
    """
    代表我们的AI内容创作机构的当前状态。
    它包含所有的对话消息以及其他可能需要共享的信息。
    """
    messages: Annotated[Sequence[BaseMessage], lambda x: x + []] # 聊天消息历史，使用Annotated实现append行为
    # 可以在这里添加更多全局状态，例如：
    # research_results: str = "" # 研究结果
    # content_draft: str = "" # 内容草稿
    # current_task: str = "" # 当前任务描述

接下来，定义我们的搜索工具。

# agency_core/tools.py
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.tools import Tool
from langchain_core.pydantic_v1 import BaseModel, Field

# 1. 定义搜索工具的输入Pydantic模型，这有助于LLM理解参数
class SearchInput(BaseModel):
    query: str = Field(description="需要搜索的查询关键词或短语")

# 2. 实例化DuckDuckGo搜索工具
# DuckDuckGoSearchRun默认不需要API key，非常适合演示
duckduckgo_search_tool = DuckDuckGoSearchRun()

# 3. 将其封装成LangChain的Tool对象，并指定输入schema
search_tool = Tool(
    name="duckduckgo_search",
    description="一个用于互联网搜索的工具。当需要获取实时信息、事实核查或查找特定主题的资料时非常有用。",
    func=duckduckgo_search_tool.run,
    args_schema=SearchInput, # 绑定输入Pydantic模型
    # return_direct=False # 默认是False，表示工具执行结果不会直接返回给用户，而是进入AgentState
)

# 我们可以把所有工具放在一个列表中
all_tools = [search_tool]

构建 Researcher 智能体和 ToolNode

现在，我们将构建 Researcher 智能体，并把它与 ToolNode 结合起来。

# agency_core/researcher_agent.py
import operator
from typing import List, Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage, AIMessage, HumanMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode

# 导入我们之前定义的状态和工具
from agency_core.state import AgentState
from agency_core.tools import all_tools, search_tool # 确保导入了search_tool

# 假设你已经设置了OPENAI_API_KEY环境变量
# 或者直接在这里设置：
# os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"

# 1. 定义 Researcher 智能体
class ResearcherAgent:
    def __init__(self, llm: ChatOpenAI, tools: list):
        self.llm = llm
        self.tools = tools
        self.prompt = ChatPromptTemplate.from_messages(
            [
                ("system", "你是一个专业的市场研究员。你的任务是根据用户的请求，利用提供的工具进行准确和深入的互联网搜索。请仔细阅读搜索结果，总结关键信息。如果需要进一步搜索，请继续使用工具。当你认为已经收集到足够的信息来回答请求时，请提供一个清晰、简洁的总结。"),
                MessagesPlaceholder(variable_name="messages"),
                MessagesPlaceholder(variable_name="agent_scratchpad"), # 用于工具调用
            ]
        )
        # 绑定工具到LLM，让LLM知道有哪些工具可用
        self.runnable = self.prompt | self.llm.bind_tools(tools)

    def __call__(self, state: AgentState) -> dict:
        print(f"\n--- Researcher Agent 正在思考 ---")
        messages = state["messages"]
        # 为了让LLM能够看到并决定调用工具，我们需要将过去所有的消息传递给它
        # 包括用户消息、AI消息、以及工具消息
        # agent_scratchpad 用于存储工具调用前的中间思想和动作
        
        # 过滤掉非AIMessage和HumanMessage，确保LLM只关注核心对话和工具结果
        # 但这里我们直接传递所有消息给LLM，因为它需要历史上下文来判断是否需要工具
        
        # 确保LLM能够看到最新的ToolMessage，从而继续处理搜索结果
        response = self.runnable.invoke({"messages": messages, "agent_scratchpad": []})
        
        # LangGraph会自动处理AIMessage中的tool_calls，并将其转换为ToolInvocation
        # 如果LLM决定调用工具，response中会包含tool_calls
        # 如果LLM决定给出最终答案，response就是普通的AIMessage
        
        print(f"Researcher Agent 输出: {response}")
        return {"messages": [response]}

# 2. 定义 Researcher 的决策逻辑：何时停止研究，何时继续或调用工具
def researcher_should_continue(state: AgentState) -> str:
    messages = state["messages"]
    last_message = messages[-1]

    # 如果最后一条消息是AIMessage并且包含tool_calls，说明智能体想调用工具
    if isinstance(last_message, AIMessage) and last_message.tool_calls:
        print("--- Researcher 决定调用工具 ---")
        return "call_tool"
    # 如果是AIMessage但没有tool_calls，说明智能体认为已经完成任务，或者给出了最终答案
    elif isinstance(last_message, AIMessage):
        print("--- Researcher 认为研究完成，准备输出结果 ---")
        return "end_research"
    # 其他情况（例如，意外消息类型），可以根据需要处理
    print("--- Researcher 状态异常或待定，默认结束 ---")
    return "end_research" # 安全起见，如果不是明确的tool_calls，就认为结束

# 3. 构建 LangGraph
def create_research_graph(llm: ChatOpenAI) -> StateGraph:
    workflow = StateGraph(AgentState)

    # 实例化 Researcher 智能体
    researcher_agent_instance = ResearcherAgent(llm, all_tools)

    # 添加 Researcher 节点
    workflow.add_node("researcher", researcher_agent_instance)

    # 添加 ToolNode，传入我们所有可用的工具
    # ToolNode 会自动处理工具调用和结果的封装
    tool_node = ToolNode(all_tools)
    workflow.add_node("call_tool", tool_node)

    # 设置入口点
    workflow.set_entry_point("researcher")

    # 定义边
    # 从 researcher 节点出去，根据 researcher_should_continue 的判断结果进行路由
    workflow.add_conditional_edges(
        "researcher",
        researcher_should_continue,
        {
            "call_tool": "call_tool", # 如果需要调用工具，则路由到 call_tool 节点
            "end_research": END      # 如果研究完成，则结束图的执行
        }
    )

    # 从 call_tool 节点出去，无论工具执行结果如何，都返回给 researcher 节点，让其处理工具输出
    workflow.add_edge("call_tool", "researcher")

    # 编译图
    app = workflow.compile()
    return app

# 4. 运行演示
if __name__ == "__main__":
    import os
    from dotenv import load_dotenv

    load_dotenv() # 加载 .env 文件中的环境变量

    # 确保你的 OPENAI_API_KEY 环境变量已设置
    if not os.getenv("OPENAI_API_KEY"):
        raise ValueError("OPENAI_API_KEY 环境变量未设置。请在 .env 文件中设置或直接在代码中提供。")

    # 实例化 LLM
    llm = ChatOpenAI(model="gpt-4o", temperature=0) # 使用 gpt-4o 确保工具调用能力

    # 创建并编译研究图
    research_graph = create_research_graph(llm)

    print("--- 启动 AI Content Agency 的 Researcher 模块 ---")
    # 模拟一个用户请求
    initial_message = HumanMessage(content="请帮我研究一下 2024 年最热门的 AI 编程框架有哪些？重点关注多智能体框架。")

    # 运行图
    # stream 方法可以让我们看到每一步的输出
    for s in research_graph.stream({"messages": [initial_message]}):
        if "__end__" not in s:
            print(s)
            print("---")
    
    # 打印最终结果
    final_state = research_graph.invoke({"messages": [initial_message]})
    print("\n--- 最终研究结果 ---")
    # 找到最后一个 AIMessage 作为最终答案
    final_answer = next((msg.content for msg in reversed(final_state["messages"]) if isinstance(msg, AIMessage) and not msg.tool_calls), "未找到最终答案。")
    print(final_answer)

    # 另一个示例：一个不需要搜索就能回答的问题 (虽然我们的Researcher被设计为爱搜索)
    print("\n--- 另一个示例：简单问题 ---")
    initial_message_simple = HumanMessage(content="你好，请问你是谁？")
    for s in research_graph.stream({"messages": [initial_message_simple]}):
        if "__end__" not in s:
            print(s)
            print("---")
    final_state_simple = research_graph.invoke({"messages": [initial_message_simple]})
    final_answer_simple = next((msg.content for msg in reversed(final_state_simple["messages"]) if isinstance(msg, AIMessage) and not msg.tool_calls), "未找到最终答案。")
    print(final_answer_simple)

代码解析：

  1. AgentState: 我们的核心状态，主要通过 messages 列表来跟踪对话历史和工具执行结果。Annotated 的用法确保了 messages 每次更新都是追加（append）操作，而不是覆盖。
  2. search_tool: 我们将 DuckDuckGoSearchRun 封装成 LangChain 的 Tool 对象。关键是 args_schema，它使用 Pydantic 模型 SearchInput 定义了工具的输入格式，这对于 LLM 理解如何调用工具至关重要。
  3. ResearcherAgent:
    • 它接收 llm 和 tools。
    • prompt 明确了其作为研究员的角色。
    • self.llm.bind_tools(tools) 是关键！它告诉 LLM，当它需要工具时，这些是它可以调用的工具。LLM 会自动学习如何根据上下文生成 tool_calls。
    • __call__ 方法接收 AgentState，调用 LLM，并将 LLM 的响应（可能是 AIMessage，包含或不包含 tool_calls）更新到状态中。
  4. researcher_should_continue: 这是一个条件路由函数。它检查 AgentState 中最后一条消息。
    • 如果 AIMessage 包含 tool_calls，说明 Researcher 决定使用工具，我们将路由到 call_tool 节点。
    • 如果 AIMessage 不包含 tool_calls，说明 Researcher 已经给出了答案或总结，图的执行可以 END。
  5. create_research_graph:
    • 我们实例化 ResearcherAgent。
    • workflow.add_node("researcher", researcher_agent_instance)：添加 Researcher 节点。
    • tool_node = ToolNode(all_tools)：这是本期的核心！ 我们直接实例化 ToolNode，并将 all_tools 列表传递给它。ToolNode 会自动处理这些工具的执行。
    • workflow.add_node("call_tool", tool_node)：将 ToolNode 添加为图中的一个节点，命名为 call_tool。
    • 条件边 add_conditional_edges: 从 researcher 节点出发，根据 researcher_should_continue 的判断结果，决定是去 call_tool 还是 END。
    • 普通边 add_edge: 从 call_tool 节点执行完毕后，无论结果如何，都将结果（作为 ToolMessage 自动添加到状态中）返回给 researcher 节点。这样，researcher 就可以读取工具结果，并进行下一轮思考。

通过这个设计，我们的 Researcher 智能体现在拥有了强大的互联网搜索能力，而且整个过程是高度自动化和模块化的。这正是 LangGraph ToolNode 的魅力所在！

坑与避坑指南

ToolNode 虽然强大，但使用不当也可能踩坑。作为你的高级导师，我来给你点拨一二：

  1. LLM 工具调用幻觉 (Hallucination)：
    • 坑点： LLM 可能会“想象”出不存在的工具名称，或者生成错误的工具参数格式，导致 ToolNode 无法识别或执行失败。
    • 避坑指南：
      • 明确的 args_schema： 务必为你的 Tool 提供清晰、准确的 args_schema（Pydantic 模型）。这是 LLM 理解工具如何使用的“说明书”。
      • 高质量的 description： 工具的 description 要写得详细、准确、无歧义，告诉 LLM 这个工具是做什么的，以及何时应该使用。
      • 合适的 LLM 模型： 优先选择支持函数调用（Function Calling）能力强且稳定的模型，如 OpenAI 的 gpt-4o, gpt-4-turbo, gpt-3.5-turbo 等。这些模型在处理工具调用方面表现更佳。
      • Prompt Engineering： 在你的 Agent Prompt 中，可以适当地引导 LLM 在需要工具时明确指出，并强调参数的准确性。
  2. 状态管理混乱：
    • 坑点： ToolNode 会自动将 ToolMessage 添加到 messages 状态中。如果你的 Agent 没有正确地处理这些 ToolMessage，或者将其与普通的 AIMessage 混淆，可能导致逻辑错误。
    • 避坑指南：
      • 清晰的 Agent 逻辑： 你的 Agent 应该能够区分 AIMessage (来自其他 Agent 或其自身的决策) 和 ToolMessage (工具执行结果)。在 __call__ 方法中，LLM 在接收到 ToolMessage 后，其后续的 AIMessage 应该基于对工具结果的分析。
      • 使用 MessagesPlaceholder("agent_scratchpad")： 在 Agent 的 Prompt 中使用 MessagesPlaceholder("agent_scratchpad") 可以帮助 LLM 更好地管理工具调用前后的中间思考过程。通常，ToolMessage 会被放置在这里，或者直接在 messages 中，LLM 会自动处理。
  3. 无限循环与死锁：
    • 坑点： 如果你的条件路由函数 should_continue 设计不当，例如 Agent 总是决定调用工具，或者总是返回给自身而没有明确的退出条件，就可能导致无限循环。
    • 避坑指南：
      • 明确的终止条件： 确保你的 should_continue 函数有明确的终止条件（例如 END），并且 Agent 能够在某个阶段判断任务已完成。
      • 任务完成判断： Agent 的 Prompt 应该引导它在收集到足够信息后，直接给出答案，而不是继续调用工具。例如：“当你认为已经收集到足够的信息来回答请求时，请提供一个清晰、简洁的总结，不要再调用工具。”
      • 迭代次数限制： 在实际生产环境中，可以为图的执行设置最大迭代次数，防止无限循环导致资源耗尽。
  4. 工具执行错误与超时：
    • 坑点： 外部工具（如搜索引擎 API）可能会失败、返回空结果或超时。ToolNode 默认情况下会捕获这些错误，并将其作为 ToolMessage 的内容返回。但如果 Agent 没有预期到这些失败情况，可能会导致后续逻辑崩溃。
    • 避坑指南：
      • Agent 的错误处理： 设计你的 Agent，使其能够识别并处理 ToolMessage 中表示错误或空结果的内容。例如，如果搜索结果为空，Agent 应该能够尝试更换关键词再次搜索，或者告知用户无法找到相关信息。
      • 工具层面的重试与超时： 在 Tool 的 func 实现中，可以加入重试逻辑和超时机制，增强工具本身的鲁棒性。
      • LangGraph 的错误处理： LangGraph 提供了 interrupt_before 和 interrupt_after 等机制，可以在关键节点前后暂停或介入，以便进行调试或错误处理。

记住，ToolNode 是一个强大的执行器，但它背后的 Agent 才是真正的“大脑”。确保 Agent 的逻辑足够健壮和智能，才能充分发挥 ToolNode 的威力。

📝 本期小结

恭喜你，各位未来的 AI 架构师们！本期我们深入探索了 LangGraph 的“原生大招”—— ToolNode。我们不仅理解了它如何优雅地将外部工具集成到多智能体工作流中，更重要的是，我们亲手为我们的 AI Content Agency 的 Researcher 智能体插上了“互联网的翅膀”，让它能够通过 DuckDuckGoSearch 进行实时的信息检索。

我们学会了：

  • ToolNode 如何作为智能体与外部工具之间的桥梁，标准化工具执行流程。
  • ToolInvocation 和 ToolMessage 在这个协作模式中的关键作用。
  • 如何将一个 langchain_core.tools.Tool 与 ToolNode 结合，并构建一个能够自主决策是否使用工具的 Researcher 智能体。
  • 以及，那些在实践中可能遇到的“坑”和“避坑指南”，让你少走弯路。

现在，你的 Researcher 不再是一个只能“闭门造车”的理论家，而是一个能够“求真务实”、深入一线获取最新信息的实干家。这对于提升我们 AI Content Agency 的内容质量和时效性至关重要！

在接下来的课程中，我们将继续升级我们的 Agency，引入更多高级功能和更复杂的智能体协作模式。敬请期待！下期再见！