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

"""
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: 最终回答

六、踩坑记录

七、总结

通过本文，你学会了：

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

进阶方向

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

参考资料

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