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

前言

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

一、核心概念

1.1 什么是工具调用

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

推理（Reasoning）：理解用户需求，判断需要什么信息
行动（Acting）：调用外部工具获取数据
观察（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}

执行流程：

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

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 与请求一致

七、总结

通过本文，你学会了：

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

进阶方向

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

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

前言

一、核心概念

1.1 什么是工具调用

1.2 为什么需要工具调用

1.3 核心组件

二、环境准备

2.1 安装依赖

2.2 配置 API 密钥

三、代码实现

3.1 完整代码

3.2 代码解析

3.2.1 工具绑定

3.2.2 工具节点实现

3.2.3 条件路由

3.2.4 图结构

四、运行效果

4.1 执行步骤

4.2 输出结果

五、ReAct 模式详解

5.1 执行流程图解

5.2 消息流转

六、踩坑记录

七、总结

进阶方向

参考资料

更多推荐文章

相关免费在线工具

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

前言

一、核心概念

1.1 什么是工具调用

1.2 为什么需要工具调用

1.3 核心组件

二、环境准备

2.1 安装依赖

2.2 配置 API 密钥

三、代码实现

3.1 完整代码

3.2 代码解析

3.2.1 工具绑定

3.2.2 工具节点实现

3.2.3 条件路由

3.2.4 图结构

四、运行效果

4.1 执行步骤

4.2 输出结果

五、ReAct 模式详解

5.1 执行流程图解

5.2 消息流转

六、踩坑记录

七、总结

进阶方向

参考资料

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具