LangGraph 实战：赋予 AI 手脚 —— 构建 ReAct Agent

LangGraph 实战（四）：赋予 AI 手脚 —— 构建 ReAct Agent

1. 什么是 ReAct？为什么它很重要？

在上一篇中，我们写了一个“作家-审稿人”的循环。那个循环虽然看起来很智能，但其实非常死板：作家必须把稿子给审稿人，审稿人必须打分。AI 没有任何选择权。

ReAct (Reasoning + Acting) 是一种让 AI 变得真正智能的架构。它的运作模式非常像人类解决复杂问题时的过程：

1. Reason (思考)：用户问我“今天上海的天气适合穿短袖吗？” -> 我需要先知道今天上海的气温。
2. Act (行动)：调用“天气查询工具”去查上海的气温。
3. Observe (观察)：工具返回“上海今日气温 15度”。
4. Reason (再思考)：15度有点冷，不适合穿短袖。
5. Final Answer (回答)：建议穿长袖或外套，不建议短袖。

LangGraph 是实现这种“思考-行动-观察”循环的最佳框架。

2. 核心难点：状态记忆 (Memory)

在之前的例子中，我们的 State 是一个简单的字典（draft, revision_number）。但在 Agent 场景下，核心状态是对话历史（Message History）。

为什么？因为 Agent 需要记住：

1. 用户刚开始问了什么。
2. 它自己决定调用了什么工具。
3. 工具返回了什么结果。

如果不记住这些，Agent 拿到工具结果后就会一脸懵逼：“我为什么要查天气来着？”

LangGraph 提供了一个关键的Reducer（归约器）函数：add_messages。它的作用是：当新的消息进来时，不要覆盖旧消息，而是追加到列表后面。

3. 环境准备

我们需要安装 LangGraph 的预构建组件库，它里面包含了一些现成的 ToolNode，能帮我们省很多代码。

# 安装 LangGraph 及其预构建库
pip install langgraph 

# 核心库
pip install langchain langchain-openai

4. 实战：构建一个能查“股票”和“算数”的 Agent

我们将构建一个 Agent，它拥有两个工具：

1. get_stock_price: 一个模拟的查询股票价格的工具。
2. multiply: 一个乘法计算器。

目标：用户问“苹果股票价格乘以 100 是多少？”，Agent 需要先查股价，再算乘法，最后回答。

第一步：定义工具 (Define Tools)

工具本质上就是 Python 函数，但需要用 @tool 装饰器标记，这样 LLM 才能理解这个函数是干嘛的。

from langchain_core.tools import tool

# 1. 定义乘法工具
@tool
def multiply(a: int, b: int) -> int:
    """计算两个数的乘积。"""
    return a * b

# 2. 定义查股价工具 (模拟)
@tool
def get_stock_price(symbol: str) -> float:
    """查询股票的当前价格。输入必须是股票代码（如 AAPL）。"""
    print(f"\n--- [工具调用] 正在查询 {symbol} 的价格... ---")
    # 这里我们模拟一下，直接返回假数据
    if "AAPL" in symbol.upper():
        return 150.0
    elif "GOOG" in symbol.upper():
        return 2800.0
    else:
        return 100.0

# 把工具放在一个列表里
tools = [multiply, get_stock_price]

# 看看 LangChain 自动生成的工具描述
# LLM 就是靠这个 JSON 描述来决定是否调用工具的
print(f"工具描述预览: {get_stock_price.args}")

第二步：定义状态 (Define State)

这次我们使用 Annotated 和 add_messages。这是 LangGraph 处理对话的标准姿势。

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    # messages 是一个列表。
    # add_messages 的意思是：如果某个节点返回了 {'messages': [新消息]}，
    # LangGraph 会自动把 [新消息] append 到原来的列表中，而不是替换它。
    messages: Annotated[list, add_messages]

第三步：绑定工具到 LLM

我们需要告诉 GPT-3.5：“你有这些工具可用”。

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# bind_tools 会把 Python 函数转换成 OpenAI API 支持的 tool definitions
llm_with_tools = llm.bind_tools(tools)

第四步：定义节点 (Nodes)

我们需要两个核心节点：

1. Chatbot Node: 调用 LLM 进行思考。
2. Tool Node: 如果 LLM 决定用工具，这个节点负责真正执行工具函数。

from langgraph.prebuilt import ToolNode
from langchain_core.messages import HumanMessage

# 1. 定义思考节点 (Agent)
def chatbot_node(state: State):
    # 获取当前的对话历史
    messages = state["messages"]
    # 调用绑定了工具的 LLM
    response = llm_with_tools.invoke(messages)
    # 返回结果。得益于 add_messages，这个 response 会被追加到历史中
    return {"messages": [response]}

# 2. 定义工具执行节点
# LangGraph 提供了一个现成的 ToolNode，我们直接用，不用自己写执行逻辑
tool_node = ToolNode(tools)

第五步：定义条件边 (Conditional Edge)

这是 ReAct 的灵魂。 当 chatbot_node 运行完，我们需要检查：LLM 是想说话（直接回答），还是想干活（调用工具）？

def tools_condition(state: State):
    messages = state["messages"]
    last_message = messages[-1]
    
    # 检查最后一条消息是否包含 tool_calls
    if last_message.tool_calls:
        return "tools" # 指向工具节点
    else:
        return "end"   # 结束

第六步：构建图 (Build Graph)

from langgraph.graph import StateGraph, END

# 初始化图
graph = StateGraph(State)

# 添加节点
graph.add_node("agent", chatbot_node)
graph.add_node("tools", tool_node)

# 设置入口
graph.set_entry_point("agent")

# 添加条件边
# 从 agent 出发，根据 tools_condition 的结果，决定去 tools 还是 END
graph.add_conditional_edges(
    "agent",
    tools_condition,
    {
        "tools": "tools",
        "end": END
    }
)

# 添加普通边
# 工具执行完，必须把结果拿回来给 Agent 看，让 Agent 继续思考
# 这就形成了一个环：Agent -> Tools -> Agent
graph.add_edge("tools", "agent")

# 编译
app = graph.compile()

第七步：运行与深度观察

我们问一个需要两步操作的问题。

print("--- 开始运行 ReAct Agent ---")

# 用户提问
user_input = "AAPL 的股票价格乘以 100 是多少？"
initial_state = {"messages": [HumanMessage(content=user_input)]}

# 运行
# 我们使用 stream 模式，这样可以看到每一步发生了什么
for event in app.stream(initial_state):
    for key, value in event.items():
        print(f"\n[Node: {key}] 刚刚运行结束")
        # 打印最新产生的消息
        last_msg = value["messages"][-1]
        print(f"输出内容: {last_msg}")
        if hasattr(last_msg, 'tool_calls') and last_msg.tool_calls:
            print(f"触发工具调用: {last_msg.tool_calls}")

5. 深度解析：刚刚发生了什么？

如果你运行了上面的代码，你会看到一个精彩的“接力跑”：

1. START -> Agent: Agent 接收到问题“AAPL 股价乘以 100”。它思考后发现自己不知道股价，于是生成了一个 AIMessage，里面包含 tool_calls=[{'name': 'get_stock_price', 'args': {'symbol': 'AAPL'}}]。
2. Agent -> Condition: 条件边检测到 tool_calls 存在，路由指向 Tools。
3. Tools 节点: ToolNode 看到系统里有个请求要查 get_stock_price，它自动运行了我们的 python 函数，拿到了 150.0。然后它生成了一个 ToolMessage（这是关键，专门用来放工具结果的消息类型）。
4. Tools -> Agent: 带着 ToolMessage（内容是 "150.0"），流程回到了 Agent。
5. Agent 再次思考: 现在它的内存里有：
   • 用户：查 AAPL 乘 100。
   • 自己：我要查 AAPL。
   • 工具：结果是 150.0。
   • Agent 思考：好了，我知道股价是 150，现在我要算 150 * 100。
   • Agent 生成新的 AIMessage，包含 tool_calls=[{'name': 'multiply', 'args': {'a': 150, 'b': 100}}]。
6. Agent -> Condition -> Tools: 再次进入工具节点，执行乘法，得到 15000。
7. Tools -> Agent: 结果 15000 返回给 Agent。
8. Agent 最终思考: 结合所有信息，生成自然语言：“AAPL 的股价是 150，乘以 100 等于 15000。” 这一次，它没有生成 tool_calls。
9. Agent -> Condition -> END: 任务结束。

6. 总结

在这一篇中，我们构建了一个真正的智能体。

与之前的 Chain 相比，最大的区别在于：我们没有硬编码执行顺序。 我们没有写代码说“先查股票，再算乘法”。我们只是给了工具。如果用户问“10 乘以 10 是多少”，Agent 就会跳过查股票，直接调乘法。如果用户问“你好”，Agent 就会直接回答，一个工具都不调。

这就是 ReAct 的魅力：通用性。

但是，现在的 Agent 还有一个致命弱点：它没有长时记忆。如果你关掉程序再打开，它就不记得刚才聊了什么。在下一篇（也是最后一篇），我们将探讨 LangGraph 的持久化（Persistence）和人工介入（Human-in-the-loop），这对于生产环境的应用至关重要。