## 前言
在前两篇文章中，我们分别学习了 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
> 🔗 转载请注明出处
>
> 💡 如果觉得有帮助，欢迎点赞、收藏、关注！
> 🎯 你的支持是我持续创作的动力！
