LangGraph工具调用实战：手把手教你实现ReAct搜索机器人

## 前言

在前两篇文章中，我们分别学习了 LangGraph 的快速入门和 StateGraph 基础。本文将带你进入 LangGraph 的进阶领域——**工具调用（Tool Calling）**。通过为聊天机器人添加 Tavily 搜索引擎，你将掌握 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 安装依赖

```bash

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

```

### 2.2 配置 API 密钥

创建 `.env` 文件：

```env

# 硅基流动平台 API 密钥

SILICONFLOW_API_KEY=your_siliconflow_key

# Tavily 搜索引擎 API 密钥

TAVILY_API_KEY=your_tavily_key

```

**获取 Tavily API Key**：访问 [Tavily](https://tavily.com/) 注册获取免费 API Key。

---

## 三、代码实现

### 3.1 完整代码

```python

"""

LangGraph 教程 - 为聊天机器人添加工具

本示例演示如何为 StateGraph 聊天机器人添加网页搜索工具。

当聊天机器人无法凭记忆回答问题时，可以使用工具查找相关信息。

官方教程地址：https://langchain-ai.github.io/langgraph/tutorials/introduction/

"""

# 过滤警告信息

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 工具绑定

```python

# 创建 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 工具节点实现

```python

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 条件路由

```python

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 图结构

```python

# 添加节点

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 执行步骤

```bash

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 官方教程 - 添加工具](https://langchain-ai.github.io/langgraph/tutorials/introduction/)

- [Tavily 搜索引擎](https://tavily.com/)

- [ReAct 论文](https://arxiv.org/abs/2210.03629)

- [LangChain 工具调用文档](https://python.langchain.com/docs/modules/agents/tools/)

---

> 📌 本文首发于 ZEEKLOG，作者：码上AI_123

> 🔗 转载请注明出处

>

> 💡 如果觉得有帮助，欢迎点赞、收藏、关注！

> 🎯 你的支持是我持续创作的动力！