第 08 期 | 构建标准的 ReAct Agent 架构

通过一个循环边实现 "思考-观察-决策"（ReAct）经典循环模型。

各位极客们，欢迎回到我们的《LangGraph 多智能体专家课》。我是你们的老朋友。

在前面的 7 期课程中，我们为「AI 万能内容创作机构 (AI Content Agency)」打下了坚实的基础：我们定义了状态机，玩转了基本的节点流转，甚至让大模型学会了简单的结构化输出。但是，如果你仔细审视我们之前的架构，你会发现一个致命的问题——它们太“线性”了，太“乖”了。

想象一下，我们机构里的 Researcher（研究员）如果是一个纯线性思维的家伙：你让他去查“2024年苹果发布会的新品”，他如果脑子里没有这个知识，他会怎么做？之前的他会直接开始“胡说八道”（幻觉）。

真正的高级研究员是怎么工作的？ 他会先去 Google 搜索（Act/行动），看看搜索结果的标题（Observe/观察），在脑子里想“嗯，这个链接看起来有戏，我得点进去看看”（Reason/思考），然后再调用浏览器工具读取网页（Act/行动），发现内容不够，再换个关键词搜……直到他觉得收集的信息足够了，才会向你汇报。

这就是今天我们要啃下的硬骨头：ReAct (Reason + Act) 架构。

今天，我们将利用 LangGraph 的循环边（Cyclic Edges），为我们的 AI 机构注入真正的“思考与探索”能力，彻底重构我们的 Researcher 智能体。准备好咖啡，我们发车！

🎯 本期学习目标

1. 吃透 ReAct 底层逻辑：理解“思考-行动-观察”循环是如何打破大模型能力天花板的。
2. 掌握 LangGraph 循环图构建：学会使用条件边（Conditional Edges）实现不可预知的动态路由。
3. 实战重构 Researcher：为我们的 AI Content Agency 打造一个能自主调用工具、自主决定何时结束调查的超级研究员。
4. 状态与上下文管理：理解在无限循环中，MessagesState 是如何像记忆一样累积线索的。

📖 原理解析

1. 什么是 ReAct？

ReAct 是普林斯顿大学和 Google 在 2022 年提出的一种范式（论文《ReAct: Synergizing Reasoning and Acting in Language Models》）。在这之前，大模型要么纯思考（Chain of Thought），要么纯行动（直接调用 API）。

ReAct 把两者结合了：让大模型在采取行动之前先“大声思考”，在看到行动结果之后再“评估现状”。

对于我们的 Researcher 来说，它的内心 OS 是这样的：

• Thought (思考)：老板让我查 LangGraph 的最新特性。我得先用搜索引擎查一下。
• Action (行动)：调用 search_tool("LangGraph new features 2024")
• Observation (观察)：工具返回了 5 条搜索结果。
• Thought (思考)：第 2 条结果看起来最相关，我需要读取它的详细内容。
• Action (行动)：调用 scrape_web_tool("url_to_doc")
• Observation (观察)：...网页文本...
• Thought (思考)：信息足够了，我可以开始总结并结束任务了。

2. 在 LangGraph 中如何映射 ReAct？

在 LangGraph 中，我们不需要写死这个循环。我们只需要定义两个核心节点和一个条件边：

1. Agent Node (大模型节点)：负责思考并决定是否调用工具。
2. Tool Node (工具节点)：负责执行工具并返回结果。
3. Conditional Edge (条件边)：连接大模型节点。如果大模型决定调用工具，就走向 Tool Node；如果大模型直接输出了最终答案，就走向 END。

这种架构的美妙之处在于：循环的次数是由大模型自己决定的。它觉得查清楚了，图就结束；觉得没查清楚，图就继续转。

3. 核心架构图解 (Mermaid)

下面是我们今天要为 Researcher 构建的 ReAct 工作流图。请仔细看图中的循环回路：

graph TD
    classDef start_end fill:#f96,stroke:#333,stroke-width:2px;
    classDef agent fill:#69b3a2,stroke:#333,stroke-width:2px;
    classDef tool fill:#ff9999,stroke:#333,stroke-width:2px;
    classDef condition fill:#e1d5e7,stroke:#333,stroke-width:2px;

    START((START)):::start_end
    END((END)):::start_end

    subgraph ReAct_Loop [ReAct 核心循环 (Researcher)]
        Agent[🤖 Agent Node
思考并决定下一步]:::agent
        Condition{💡 路由判断
有Tool Call吗?}:::condition
        Tools[🛠️ Tool Node
执行工具并记录观察]:::tool
    end

    START --> Agent
    Agent --> Condition
    
    Condition -- "Yes (去行动)" --> Tools
    Tools -- "返回观察结果" --> Agent
    
    Condition -- "No (任务完成)" --> END

讲师犀利点评：看到那条从 Tools 指回 Agent 的线了吗？这就是 LangGraph 比传统 LangChain Chain 强大的地方。Chain 是单向的，而 Graph 是图灵完备的，有了循环，Agent 才有了“生命”。

💻 实战代码演练

现在，让我们回到「AI Content Agency」项目。我们的 Planner（策划）刚刚下达了一个任务：“调查目前市面上最火的 AI 编程助手工具，并给出一份简短的对比报告”。

我们的 Researcher 需要接单了。

1. 准备工作与工具定义

首先，我们要给 Researcher 配备“手和眼”（Tools）。为了演示，我们使用模拟的搜索和网页读取工具。

import json
from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# ==========================================
# 1. 定义工具 (Tools) - Researcher的武器库
# ==========================================

@tool
def search_web(query: str) -> str:
    """当需要搜索互联网获取最新信息时使用此工具。"""
    print(f"   [Tool 执行] 🔍 正在搜索: {query}")
    # 模拟搜索引擎返回结果
    if "AI 编程助手" in query:
        return json.dumps([
            {"title": "Cursor: 彻底改变编程方式", "url": "https://cursor.sh/about"},
            {"title": "GitHub Copilot 最新特性", "url": "https://github.com/copilot"}
        ])
    return "未找到相关信息。"

@tool
def scrape_web(url: str) -> str:
    """当需要读取特定网页的详细内容时使用此工具。"""
    print(f"   [Tool 执行] 📄 正在抓取网页: {url}")
    # 模拟网页抓取
    if "cursor" in url.lower():
        return "Cursor 是一款基于 VS Code 分支的 AI 编辑器，深度集成了 Claude 3.5 Sonnet，支持多文件上下文重构。"
    elif "copilot" in url.lower():
        return "GitHub Copilot 集成了 OpenAI 的模型，优势在于与 GitHub 生态的无缝对接以及企业级安全合规。"
    return "网页内容无法读取。"

# 工具列表
tools = [search_web, scrape_web]

2. 定义状态与 Agent 节点

在 ReAct 中，状态（State）极其重要。我们需要记录整个“思考-行动-观察”的历史。LangGraph 提供的 add_messages reducer 是最完美的解决方案。

# ==========================================
# 2. 定义 State (状态)
# ==========================================
class AgentState(TypedDict):
    # 使用 add_messages 确保消息是追加的，而不是覆盖的
    # 这就是 ReAct 能记住前几轮搜索结果的秘密所在！
    messages: Annotated[Sequence[BaseMessage], add_messages]

# ==========================================
# 3. 定义大模型与节点绑定
# ==========================================
# 初始化 LLM (假设你已经配置了 OPENAI_API_KEY)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 关键一步：将工具绑定到大模型
# 这相当于告诉大模型："如果你需要，你可以随时调用这些函数"
llm_with_tools = llm.bind_tools(tools)

def researcher_agent_node(state: AgentState):
    """
    Agent 节点：负责阅读当前上下文（包含之前的观察），
    思考并决定是输出最终答案，还是继续调用工具。
    """
    print("\n[Agent 思考中...] 🧠 正在分析当前情报...")
    messages = state["messages"]
    
    # 注入系统提示词，赋予 Researcher 人设
    sys_msg = SystemMessage(content="""
    你是 AI Content Agency 的高级研究员 (Senior Researcher)。
    你的目标是通过工具收集准确的信息。
    请遵循以下原则：
    1. 不要瞎编！必须基于工具返回的事实。
    2. 如果信息不够，请继续搜索或读取网页。
    3. 当你收集到足够的信息时，请直接输出最终的调查报告。
    """)
    
    # 调用大模型
    response = llm_with_tools.invoke([sys_msg] + messages)
    
    # 返回新的消息（会被追加到状态中）
    return {"messages": [response]}

3. 构建工具节点与条件路由

这是本期最核心的代码。我们要处理大模型返回的 tool_calls，执行它们，并把结果作为 ToolMessage 塞回给大模型。

(注：LangGraph 官方提供了一个现成的 ToolNode，但为了让你彻底懂底层逻辑，我们手写一个极简版的工具执行节点。)

# ==========================================
# 4. 定义工具执行节点 (Tool Node)
# ==========================================
def tool_execution_node(state: AgentState):
    """
    工具节点：专门负责拦截大模型的 tool_calls，执行真实的 Python 函数，
    并将结果封装为 ToolMessage 返回。
    """
    messages = state["messages"]
    # 拿到大模型最新的一条消息
    last_message = messages[-1]
    
    tool_responses = []
    # 遍历大模型要求调用的所有工具
    for tool_call in last_message.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        
        # 找到对应的工具函数并执行
        tool_instance = next(t for t in tools if t.name == tool_name)
        try:
            result = tool_instance.invoke(tool_args)
        except Exception as e:
            result = f"工具执行出错: {str(e)}"
            
        # 必须封装为 ToolMessage，并带上 tool_call_id
        # 这样大模型才知道这个结果对应的是它哪一次的调用
        tool_responses.append(
            ToolMessage(
                content=str(result),
                name=tool_name,
                tool_call_id=tool_call["id"]
            )
        )
        
    return {"messages": tool_responses}

# ==========================================
# 5. 定义条件路由 (Conditional Edge)
# ==========================================
def should_continue(state: AgentState) -> str:
    """
    路由逻辑：决定图是继续循环还是结束。
    """
    messages = state["messages"]
    last_message = messages[-1]
    
    # 如果大模型返回的消息中包含了 tool_calls，说明它想行动
    if last_message.tool_calls:
        print("   [路由决策] 🔀 发现工具调用，前往 Tools 节点。")
        return "continue"
    
    # 否则，说明大模型认为任务完成了，输出了普通文本
    print("   [路由决策] 🏁 任务完成，准备输出报告。")
    return "end"

4. 组装并运行我们的 Agency Researcher

把节点和边拼装起来！

# ==========================================
# 6. 构建 LangGraph
# ==========================================
workflow = StateGraph(AgentState)

# 添加节点
workflow.add_node("agent", researcher_agent_node)
workflow.add_node("tools", tool_execution_node)

# 设置起点：总是先进入 agent 节点思考
workflow.add_edge(START, "agent")

# 添加条件边：从 agent 出发，根据 should_continue 的返回值决定去向
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools", # 如果返回 continue，去 tools 节点
        "end": END           # 如果返回 end，结束图
    }
)

# 添加循环边：工具执行完后，必须回到 agent 节点进行"观察和下一步思考"
workflow.add_edge("tools", "agent")

# 编译图
researcher_app = workflow.compile()

# ==========================================
# 7. 模拟运行 (Demo)
# ==========================================
if __name__ == "__main__":
    from langchain_core.messages import SystemMessage # 补全上文依赖
    print("=== AI Content Agency: Researcher 启动 ===")
    
    # Planner 下达的任务
    initial_task = "请调查目前市面上最火的 AI 编程助手工具(Cursor和Copilot)，并给出一份简短的对比报告。"
    
    inputs = {"messages": [HumanMessage(content=initial_task)]}
    
    # stream 方法可以让我们看到每一步的执行过程
    for output in researcher_app.stream(inputs, stream_mode="updates"):
        # 打印当前执行的节点
        for node_name, state_update in output.items():
            pass # 详细日志已经在节点内部 print 了
            
    # 打印最终结果
    final_state = researcher_app.get_state(inputs) # 获取最终状态
    # (注意：上面的 get_state 用法在某些版本中需要 thread_id，
    # 最简单的获取最终消息的方式是直接从最后一次流输出中拿)
    print("\n==================================")
    print("📊 Researcher 最终报告:")
    print("==================================")
    # 取出状态中最后一条消息的内容
    print(output['agent']['messages'][-1].content)

🧠 运行过程脑补（大模型的内心戏）

当你运行这段代码时，你会看到控制台疯狂输出：

1. [Agent 思考中...] -> 大模型觉得需要查资料。
2. [路由决策] 🔀 发现工具调用... -> 触发条件边。
3. [Tool 执行] 🔍 正在搜索... -> 执行 search_web。
4. 返回 Agent，[Agent 思考中...] -> 看到搜索结果只有 URL，决定深挖。
5. [路由决策] 🔀 发现工具调用...
6. [Tool 执行] 📄 正在抓取... -> 执行 scrape_web。
7. 返回 Agent，[Agent 思考中...] -> 信息收集完毕，开始写报告。
8. [路由决策] 🏁 任务完成... -> 走向 END。

这就是真正的智能！它不再是死板的流程，而是一个有自主探索能力的 Agent。

坑与避坑指南

作为你们的导师，我不能只教你们怎么跑通 Demo，还得告诉你们在真实生产环境中，这个架构会怎么把你们“坑”死。

💣 坑一：死循环（Infinite Loop）

现象：大模型陷入了迷思。比如它搜索不到某个东西，它就换个词搜，还是搜不到，它就一直搜……你的 API 账单会在一夜之间爆炸。 避坑指南： 在 LangGraph 中，编译图的时候必须设置递归限制（Recursion Limit）。默认是 25，但建议显式设置。

# 限制图最多流转 10 步
researcher_app.invoke(inputs, {"recursion_limit": 10})

如果达到限制，LangGraph 会抛出 GraphRecursionError，你可以在外层捕获并优雅降级。

💣 坑二：工具报错导致图崩溃

现象：scrape_web 遇到反爬虫拦截，抛出 TimeoutError，整个 Graph 直接报错退出。 避坑指南： 永远不要让工具的 Error 抛到图的层级！ 就像我们在 tool_execution_node 里写的 try...except 一样，要把错误信息转成字符串，包装在 ToolMessage 里返回给大模型。 大模型非常聪明，如果它看到 ToolMessage 里说 "Error: 403 Forbidden"，它会自己想办法（比如换个网站，或者在报告里说“该网站无法访问”）。把异常变成观察（Observation），这是 ReAct 架构的精髓。

💣 坑三：上下文窗口爆炸

现象：Researcher 查了 10 个网页，每个网页 5000 字，messages 列表越来越长，最后直接超出 GPT-4 的 128k 上下文限制，报错 TokenLimitExceeded。 避坑指南： 在真实项目中，我们不能无限 add_messages。我们需要在循环中引入状态裁剪（State Trimming）或总结节点（Summarize Node）。当 len(messages) 超过一定阈值时，触发一个内部的小 LLM 把前面的无用网页内容压缩成摘要。 (剧透：这部分高阶玩法，我们会在第 12 期《长上下文与记忆管理》中专门展开！)

📝 本期小结

今天，我们完成了一次从“线性执行”到“动态循环”的认知飞跃：

1. ReAct 架构不是什么神秘的魔法，它就是 大模型节点 + 工具节点 + 循环条件边。
2. 我们为「AI Content Agency」成功打造了 Researcher 角色。它不再是一个只会背诵训练数据的书呆子，而是一个会使用搜索引擎、会阅读网页、会自主决定何时交差的真正研究员。
3. 我们掌握了 LangGraph 中 messages 状态的累加机制，这是大模型能够在多次循环中保持记忆的关键。

下期预告： 现在的 Researcher 虽然能干，但它是一个人在战斗。在我们的 Agency 中，Planner（主管）如何把任务交给 Researcher，Researcher 做完后又如何把报告丢给 Writer（主笔）？ 在第 09 期《多 Agent 协作：SubGraph（子图）与网络化组织》中，我们将把今天写的 Researcher 作为一个子节点，接入到我们庞大的机构网络中。

极客们，把今天的代码跑起来，试着给你的 Researcher 加一个“天气查询”工具，看看它能不能告诉你明天要不要带伞。我们下期见！