跳到主要内容LangGraph工具调用实战：手把手教你实现ReAct搜索机器人 | 极客日志Python
LangGraph工具调用实战：手把手教你实现ReAct搜索机器人
\## 前言 在前两篇文章中，我们分别学习了 LangGraph 的快速入门和 StateGraph 基础。将带你进入 LangGraph 的进阶领域——\*\*工具调用（Tool Calling）\*\*。通过为聊天机器人添加 Tavily 搜索引擎，你将掌握 ReAct（Reasoning + Acting）模式的完整实现，让 AI 能够主动调用外部工具获取实时信息。 \--- \## 一…
墨染流年24K 浏览 ## 前言
在前两篇文章中，我们分别学习了 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/
微信扫一扫，关注极客日志
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
相关免费在线工具
- curl 转代码
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
- Base64 字符串编码/解码
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
- Base64 文件转换器
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
- Markdown转HTML
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
- HTML转Markdown
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online
- JSON 压缩
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online
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
if not os.getenv("TAVILY_API_KEY"):
raise ValueError("TAVILY_API_KEY 未设置，请在 .env 文件中配置")
# ==================== 1. 定义状态 ====================
messages: 消息列表，使用 add_messages reducer 函数
messages: Annotated[list, add_messages]
# ==================== 2. 定义工具节点 ====================
检查状态中的最新消息，如果消息包含 tool_calls，
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", []):
raise ValueError("No message found in input")
for tool_call in message.tool_calls:
tool_result = self.tools_by_name[tool_call["name"]].invoke(
content=json.dumps(tool_result, ensure_ascii=False),
tool_call_id=tool_call["id"],
return {"messages": outputs}
# ==================== 3. 定义路由函数 ====================
def route_tools(state: State):
检查聊天机器人输出中是否包含 tool_calls。
if isinstance(state, list):
elif messages := state.get("messages", []):
ai_message = messages[-1]
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:
# ==================== 4. 搜索工具 ====================
# 需要安装: pip install langchain-community
# Tavily API Key 已在上面的代码中设置
# ==================== 5. 创建图 ====================
graph_builder = StateGraph(State)
model="Qwen/Qwen3-Next-80B-A3B-Instruct",
openai_api_key=os.getenv("SILICONFLOW_API_KEY"),
tool = TavilySearchResults(max_results=2)
llm_with_tools = llm.bind_tools(tools)
def chatbot(state: State):
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(
{"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):
for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
for value in event.values():
last_message = value["messages"][-1]
if hasattr(last_message, "content") and last_message.content:
if not isinstance(last_message, ToolMessage):
print("助手:", last_message.content)
print("🤖 LangGraph 工具增强聊天机器人已启动！")
print(" - 输入 'quit'、'exit' 或 'q' 退出对话")
print(" - 聊天机器人可以使用搜索工具回答实时问题\n")
user_input = input("用户: ")
if user_input.lower() in ["quit", "exit", "q"]:
stream_graph_updates(graph, user_input)
except KeyboardInterrupt:
if __name__ == "__main__":
tool = TavilySearchResults(max_results=2)
llm_with_tools = llm.bind_tools(tools)
- `bind_tools()` 将工具信息注入 LLM 的系统提示
- 如果需要，LLM 输出包含 `tool_calls` 的特殊消息
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]
for tool_call in message.tool_calls:
tool_result = self.tools_by_name[tool_call["name"]].invoke(
content=json.dumps(tool_result),
tool_call_id=tool_call["id"],
return {"messages": outputs}
1. 解析 LLM 的 `tool_calls` 请求
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 节点
**作用**：根据当前状态动态决定下一步执行哪个节点。
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("tools", tool_node)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_conditional_edges(
{"tools": "tools", END: END}
graph_builder.add_edge("tools", "chatbot") # 形成循环
START ──▶ chatbot ──┬──▶ END（无工具调用）
└──▶ tools ──▶ chatbot（循环）
┌─────────────────────────────────────┐
│ 输出: AIMessage with tool_calls │
└─────────────────────────────────────┘
┌─────────────────────────────────────┐
│ Step 2: route_tools 检查 │
└─────────────────────────────────────┘
┌─────────────────────────────────────┐
│ 输出: ToolMessage with results │
└─────────────────────────────────────┘
┌─────────────────────────────────────┐
│ Step 4: chatbot 节点（循环） │
└─────────────────────────────────────┘
┌─────────────────────────────────────┐
│ Step 5: route_tools 检查 │
└─────────────────────────────────────┘
├── UserMessage: "今天重庆天气怎么样？"
├── AIMessage: tool_calls（调用意图）
|------|------|----------|
| 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 智能选择最合适的工具
> 📌 本文首发于 ZEEKLOG，作者：码上AI_123
