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

完整代码示例

下面是一个完整的 StateGraph 实现，展示了如何为聊天机器人添加工具支持。

"""
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()

# 检查关键配置
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:
        self.tools_by_name = {tool.name: tool for tool in tools}

    def __call__(self, inputs: dict):
        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（结束）
    """
    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. 创建图 ====================
def create_graph():
    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):
        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()

# ==================== 5. 运行聊天机器人 ====================
def stream_graph_updates(graph, user_input: str):
    """
    流式处理图更新。
    """
    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()

关键逻辑拆解

1. 工具绑定

tool = TavilySearchResults(max_results=2)
tools = [tool]
llm_with_tools = llm.bind_tools(tools)

这里 bind_tools() 将工具描述注入 LLM 的系统提示中。LLM 会根据用户输入判断是否需要调用工具，如果需要，会输出包含 tool_calls 的特殊消息。

2. 工具节点实现

BasicToolNode 负责实际执行调用。它解析 LLM 的 tool_calls 请求，查找对应工具并执行，然后将结果封装为 ToolMessage 返回给状态。

关键点在于 tool_call_id 必须与请求一致，否则后续的消息关联会出错。

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"
    return END

这个函数决定了图的流向。如果 LLM 决定调用工具，就进入 tools 节点；否则直接结束。这是实现 ReAct 循环控制的关键。

4. 图结构流转

START ──▶ chatbot ──┬──▶ END（无工具调用）
                  │
                  └──▶ tools ──▶ chatbot（循环）
                     （有工具调用）

这是一个典型的循环结构。chatbot 生成响应，如果有工具调用，tools 节点执行并返回结果，然后再次回到 chatbot 进行最终回答生成。

运行效果

执行脚本后，你可以尝试询问实时问题：

python 03 添加工具.py

ReAct 模式详解

执行流程

1. Step 1: chatbot 节点接收用户输入，LLM 判断需要调用工具，输出 AIMessage with tool_calls。
2. Step 2: route_tools 检查发现 tool_calls，返回 "tools"。
3. Step 3: tools 节点调用 TavilySearch，搜索关键词，输出 ToolMessage with results。
4. Step 4: chatbot 节点（循环）接收 ToolMessage，结合搜索结果生成最终回答。
5. Step 5: route_tools 再次检查，无 tool_calls，返回 END。

消息流转

消息历史:
├── 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 工具调用文档