LangGraph 工具调用实战：实现 ReAct 搜索机器人

前言

在 LangGraph 的进阶领域中，工具调用（Tool Calling）允许 AI 理解用户需求并调用外部工具获取数据。本文将带你掌握 ReAct（Reasoning + Acting）模式的完整实现，让 AI 能够主动调用外部工具获取实时信息。

---

一、核心概念

1.1 什么是工具调用

工具调用（Tool Calling）是 LLM 的重要能力，它允许 AI：

1. 推理（Reasoning）：理解用户需求，判断需要什么信息
2. 行动（Acting）：调用外部工具获取数据
3. 观察（Observation）：整合工具结果生成回答

这就是经典的 ReAct 模式。

1.2 为什么需要工具调用

场景	纯 LLM	带工具调用的 LLM
实时信息	❌ 知识截止，无法回答	✅ 调用搜索工具获取
数学计算	❌ 容易出错	✅ 调用计算器精确计算
数据库查询	❌ 无法访问	✅ 调用 SQL 工具查询
API 调用	❌ 无法执行	✅ 调用 API 工具操作

工具调用让 AI 从'纸上谈兵'变为'实干家'。

1.3 核心组件

┌─────────────────────────────────────────────────────────┐
│                 工具调用架构                           │
├─────────────────────────────────────────────────────────┤
│  1. 工具定义       │  TavilySearch、Calculator 等      │
│  2. 工具绑定       │  llm.bind_tools(tools)            │
│  3. 工具节点       │  BasicToolNode 执行工具调用       │
│  4. 条件路由       │  route_tools 判断是否需要工具     │
│  5. 循环执行       │  chatbot ↔ tools 形成 ReAct 循环  │
└─────────────────────────────────────────────────────────┘

---

二、环境准备

2.1 安装依赖

pip install langgraph langchain langchain-openai langchain-community pydantic python-dotenv typing-extensions

2.2 配置 API 密钥

创建 .env 文件：

# API 密钥
SILICONFLOW_API_KEY=your_siliconflow_key
# Tavily 搜索引擎 API 密钥
TAVILY_API_KEY=your_tavily_key

获取 Tavily API Key：访问 Tavily 注册获取免费 API Key。

---

三、代码实现

3.1 完整代码

"""
LangGraph 教程 - 为聊天机器人添加工具
本示例演示如何为 StateGraph 聊天机器人添加网页搜索工具。
当聊天机器人无法凭记忆回答问题时，可以使用工具查找相关信息。
"""

# 过滤警告信息
import warnings
warnings.filterwarnings("ignore", category=DeprecationWarning)
warnings.filterwarnings("ignore", category=UserWarning)

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.messages import ToolMessage
from langchain_community.tools.tavily_search import TavilySearchResults
from dotenv import load_dotenv
import json
import os

# 加载环境变量
load_dotenv()

# 检查 Tavily API Key 是否设置
if not os.getenv("TAVILY_API_KEY"):
    raise ValueError("TAVILY_API_KEY 未设置，请在 .env 文件中配置")


# ==================== 1. 定义状态 ====================
class State(TypedDict):
    """
    定义图的状态结构。
    messages: 消息列表，使用 add_messages reducer 函数
              确保新消息追加到列表，而不是覆盖
    """
    messages: Annotated[list, add_messages]


# ==================== 2. 定义工具节点 ====================
class BasicToolNode:
    """
    工具节点：运行 LLM 请求的工具。
    检查状态中的最新消息，如果消息包含 tool_calls，
    则调用相应的工具。
    """
    def __init__(self, tools: list) -> None:
        """
        初始化工具节点。
        Args:
            tools: 工具列表
        """
        self.tools_by_name = {tool.name: tool for tool in tools}

    def __call__(self, inputs: dict):
        """
        执行工具调用。
        Args:
            inputs: 包含消息列表的字典
        Returns:
            包含工具执行结果的消息字典
        """
        if messages := inputs.get("messages", []):
            message = messages[-1]
        else:
            raise ValueError("No message found in input")
        outputs = []
        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, ensure_ascii=False),
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"],
                )
            )
        return {"messages": outputs}


# ==================== 3. 定义路由函数 ====================
def route_tools(state: State):
    """
    条件边路由函数。
    检查聊天机器人输出中是否包含 tool_calls。
    - 如果有工具调用，路由到 "tools" 节点
    - 如果没有，路由到 END（结束）
    Args:
        state: 当前图状态
    Returns:
        下一个节点的名称或 END
    """
    if isinstance(state, list):
        ai_message = state[-1]
    elif messages := state.get("messages", []):
        ai_message = messages[-1]
    else:
        raise ValueError(f"No messages found in input state to tool_edge: {state}")
    # 检查是否有工具调用
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"
    return END


# ==================== 4. 搜索工具 ====================
# 使用 Tavily 真实搜索工具
# 需要安装：pip install langchain-community
# Tavily API Key 已在上面的代码中设置


# ==================== 5. 创建图 ====================
def create_graph():
    """
    创建并编译 StateGraph。
    Returns:
        编译后的图对象
    """
    # 创建图构建器
    graph_builder = StateGraph(State)
    # 初始化模型
    llm = ChatOpenAI(
        model="Qwen/Qwen3-Next-80B-A3B-Instruct",
        openai_api_key=os.getenv("SILICONFLOW_API_KEY"),
        openai_api_base="https://api.siliconflow.cn/v1",
        temperature=0.7
    )
    # 创建 Tavily 搜索工具
    tool = TavilySearchResults(max_results=2)
    tools = [tool]
    # 绑定工具到 LLM
    llm_with_tools = llm.bind_tools(tools)
    # 定义聊天机器人节点
    def chatbot(state: State):
        """
        聊天机器人节点。
        Args:
            state: 当前状态
        Returns:
            包含 LLM 响应的字典
        """
        return {"messages": [llm_with_tools.invoke(state["messages"])]}
    # 添加节点
    graph_builder.add_node("chatbot", chatbot)
    # 创建工具节点并添加
    tool_node = BasicToolNode(tools=tools)
    graph_builder.add_node("tools", tool_node)
    # 添加边
    graph_builder.add_edge(START, "chatbot")
    # 添加条件边：从 chatbot 到 tools 或 END
    graph_builder.add_conditional_edges(
        "chatbot",
        route_tools,
        {"tools": "tools", END: END}
    )
    # 添加边：从 tools 回到 chatbot（形成循环）
    graph_builder.add_edge("tools", "chatbot")
    # 编译图
    return graph_builder.compile()


# ==================== 6. 运行聊天机器人 ====================
def stream_graph_updates(graph, user_input: str):
    """
    流式处理图更新。
    Args:
        graph: 编译后的图对象
        user_input: 用户输入的消息
    """
    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
        for value in event.values():
            if "messages" in value:
                last_message = value["messages"][-1]
                # 只打印 AI 消息，不打印工具消息
                if hasattr(last_message, "content") and last_message.content:
                    if not isinstance(last_message, ToolMessage):
                        print("助手:", last_message.content)


def main():
    """主函数 - 运行交互式聊天机器人。"""
    print("🤖 LangGraph 工具增强聊天机器人已启动！")
    print("=" * 50)
    print("提示：")
    print("  - 输入 'quit'、'exit' 或 'q' 退出对话")
    print("  - 聊天机器人可以使用搜索工具回答实时问题\n")
    # 创建图
    graph = create_graph()
    while True:
        try:
            # 获取用户输入
            user_input = input("用户：")
            # 检查退出命令
            if user_input.lower() in ["quit", "exit", "q"]:
                print("\n👋 再见！")
                break
            # 处理用户输入并获取响应
            stream_graph_updates(graph, user_input)
            print()  # 空行分隔对话
        except KeyboardInterrupt:
            print("\n\n👋 再见！")
            break
        except Exception as e:
            print(f"发生错误：{e}")
            break


if __name__ == "__main__":
    main()

3.2 代码解析

3.2.1 工具绑定

# 创建 Tavily 搜索工具
tool = TavilySearchResults(max_results=2)
tools = [tool]
# 绑定工具到 LLM
llm_with_tools = llm.bind_tools(tools)

关键点：

• bind_tools() 将工具信息注入 LLM 的系统提示
• LLM 根据用户输入判断是否需要调用工具
• 如果需要，LLM 输出包含 tool_calls 的特殊消息

3.2.2 工具节点实现

class BasicToolNode:
    def __init__(self, tools: list) -> None:
        # 创建工具名称到工具对象的映射
        self.tools_by_name = {tool.name: tool for tool in tools}
    def __call__(self, inputs: dict):
        message = inputs.get("messages", [])[-1]
        outputs = []
        for tool_call in message.tool_calls:
            # 调用对应工具
            tool_result = self.tools_by_name[tool_call["name"]].invoke(
                tool_call["args"]
            )
            # 创建 ToolMessage
            outputs.append(
                ToolMessage(
                    content=json.dumps(tool_result),
                    name=tool_call["name"],
                    tool_call_id=tool_call["id"],
                )
            )
        return {"messages": outputs}

执行流程：

1. 解析 LLM 的 tool_calls 请求
2. 查找并执行对应工具
3. 将结果封装为 ToolMessage
4. 返回更新后的状态

3.2.3 条件路由

def route_tools(state: State):
    """条件边路由函数。"""
    ai_message = state.get("messages", [])[-1]
    # 检查是否有工具调用
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"  # 有工具调用，路由到 tools 节点
    return END  # 没有工具调用，结束

作用：根据当前状态动态决定下一步执行哪个节点。

3.2.4 图结构

# 添加节点
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
# 添加边
graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges(
    "chatbot",
    route_tools,
    {"tools": "tools", END: END}
)
graph_builder.add_edge("tools", "chatbot")  # 形成循环

图结构：

START ──▶ chatbot ──┬──▶ END（无工具调用）
                  │
                  └──▶ tools ──▶ chatbot（循环）
                     （有工具调用）

---

四、运行效果

4.1 执行步骤

python 03 添加工具.py

4.2 输出结果

(此处省略图片展示)

---

五、ReAct 模式详解

5.1 执行流程图解

用户："今天重庆天气怎么样？"
    │
    ▼
┌─────────────────────────────────────┐
│  Step 1: chatbot 节点               │
│  LLM 判断需要调用工具               │
│  输出：AIMessage with tool_calls    │
└─────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────┐
│  Step 2: route_tools 检查           │
│  发现 tool_calls                    │
│  返回："tools"                      │
└─────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────┐
│  Step 3: tools 节点                 │
│  调用 TavilySearch                  │
│  搜索："重庆今天天气"               │
│  输出：ToolMessage with results     │
└─────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────┐
│  Step 4: chatbot 节点（循环）       │
│  LLM 接收 ToolMessage               │
│  生成最终回答                       │
└─────────────────────────────────────┘
    │
    ▼
┌─────────────────────────────────────┐
│  Step 5: route_tools 检查           │
│  无 tool_calls                      │
│  返回：END                          │
└─────────────────────────────────────┘
    │
    ▼
输出："根据搜索结果，重庆今天天气晴朗..."

5.2 消息流转

消息历史:
├── UserMessage: "今天重庆天气怎么样？"
├── AIMessage: tool_calls（调用意图）
├── ToolMessage: 搜索结果
└── AIMessage: 最终回答

---

六、踩坑记录

问题	原因	解决方案
API Key 泄露风险	硬编码在代码中	使用 .env 文件存储
工具调用不触发	LLM 未正确绑定工具	确认 bind_tools() 调用
ToolMessage 格式错误	content 不是 JSON	使用 json.dumps() 序列化
无限循环	忘记返回 END	检查 route_tools 逻辑
工具结果丢失	tool_call_id 不匹配	确保 ID 与请求一致

---

七、总结

通过本文，你学会了：

1. ✅ 工具绑定：使用 bind_tools() 让 LLM 知道可用工具
2. ✅ 工具节点：实现 BasicToolNode 执行工具调用
3. ✅ 条件路由：使用 add_conditional_edges() 动态控制流程
4. ✅ ReAct 模式：理解推理 - 行动 - 观察的完整循环
5. ✅ 消息类型：掌握 AIMessage、ToolMessage 的使用

进阶方向

• 多工具支持：添加计算器、数据库查询等工具
• 工具选择：让 LLM 智能选择最合适的工具
• 错误处理：添加工具调用失败的容错机制
• 人机协作：在关键步骤添加人工确认

---

参考资料

• LangGraph 官方教程 - 添加工具
• Tavily 搜索引擎
• ReAct 论文
• LangChain 工具调用文档