LangGraph实战：基于StateGraph构建带记忆的ReAct智能体工作流

1. 项目概述：为什么我们需要 LangGraph？

如果你已经用 LangChain 构建过一些简单的 AI 应用，比如一个问答机器人，你可能会发现一个痛点：当对话稍微复杂一点，需要多步推理、调用工具、或者记住之前的上下文时，代码很快就会变得一团乱麻。各种if-else判断、状态维护的变量散落在各处，调试起来像在走迷宫。这其实就是“工作流”（Workflow）或“智能体”（Agent）的编排问题。

LangGraph 就是为了解决这个问题而生的。你可以把它理解为 LangChain 生态中专门用于构建有状态、多步骤、可循环的 AI 应用的工作流编排框架。它把整个应用流程抽象成一个“图”（Graph），图中的节点代表一个执行步骤（比如调用 LLM、执行工具），边代表步骤之间的流转逻辑。这种抽象让复杂逻辑变得清晰、可维护、且易于可视化。

这次我们要聊的StateGraph和“带记忆的 ReAct 循环”，就是 LangGraph 最核心、最经典的两个概念。StateGraph是构建图的骨架，而 ReAct 循环则是图上最常跑的一个“业务流程”。通过这个组合，我们能轻松构建出像 AutoGPT 那样的、能够自主思考、使用工具、并记住历史的智能体。

简单说，学完这个，你就能告别杂乱无章的 Agent 代码，用一套清晰、强大的框架来构建复杂的 AI 应用。

2. 核心概念拆解：StateGraph 与 ReAct 循环

在深入代码之前，我们必须把几个核心概念掰开揉碎了讲清楚。这就像学武功先扎马步，基础牢了，后面学招式才快。

2.1 什么是 StateGraph（状态图）？

StateGraph是 LangGraph 中用于定义工作流的类。它的核心思想是“状态驱动”。

• 状态（State）： 这是一个贯穿整个工作流的、共享的“记忆体”。它通常是一个字典（dict）或 Pydantic 模型，里面保存了所有步骤都需要访问或修改的数据。比如，用户的输入问题、LLM 的思考过程、工具调用的结果、最终的答案等，都放在状态里。
• 节点（Node）： 节点是工作流中的一个具体执行单元。它本质上是一个函数，这个函数接收当前的“状态”作为输入，执行一些操作（比如调用大模型、查询数据库），然后修改并返回更新后的状态。每个节点只关心自己的任务，不需要知道其他节点。
• 边（Edge）： 边定义了节点之间的流转逻辑。它决定了在一个节点执行完毕后，接下来应该执行哪个节点。边可以是有条件的（根据状态里的某个值决定下一步），也可以是无条件的（直接跳转到下一个节点）。

把这三者结合起来，StateGraph就允许你像搭积木一样，声明式地构建一个由节点和边组成的工作流。运行时，LangGraph 会从一个起始节点开始，根据状态和边的逻辑，在各个节点间穿梭执行，直到到达某个终点。

2.2 什么是 ReAct 循环？

ReAct（Reason + Act）是一个让大模型与外部工具交互的经典范式。它让模型学会“思考-行动”的循环：

1. 推理（Reason）： 模型分析当前情况和目标，思考下一步该做什么。
2. 行动（Act）： 模型根据思考，执行一个具体的动作，比如调用一个搜索工具、一个计算器 API。
3. 观察（Observe）： 获取行动（工具调用）的结果。
4. 循环（Loop）： 将观察到的结果纳入上下文，再次进行推理，决定下一步是继续行动还是结束任务。

这个循环会一直进行，直到模型认为任务已经完成（例如，给出了最终答案）或达到了最大循环次数。带“记忆”的 ReAct，就是指这个循环过程中产生的所有“思考”和“观察”都会被记录下来，作为后续推理的上下文，这样模型就能拥有连贯的“思维链”，避免重复或矛盾的操作。

2.3 二者如何结合？

在 LangGraph 中，一个典型的带记忆的 ReAct 智能体工作流，就是用StateGraph来实现 ReAct 循环的。

• 状态： 保存着用户的原始问题、模型的历史思考（scratchpad）、工具调用历史、最新的观察结果等。
• 节点： 通常至少有两个核心节点：
  • agent节点： 负责“推理”。它读取状态，决定是调用工具还是直接结束，并生成相应的指令或思考。
  • tools节点： 负责“行动”。它执行agent节点指定的工具调用，并将结果写回状态。
• 边： 连接agent和tools节点，形成一个循环。边上的条件逻辑会检查agent节点的输出，如果输出是“调用工具”，就流向tools节点；如果是“结束”，就流向终点。

这样，一个动态的、有状态的 ReAct 循环就在StateGraph中运转起来了。接下来，我们就动手实现它。

3. 环境准备与基础搭建

理论讲得再多，不如一行代码。我们从一个最简单的“带记忆的 ReAct 循环”智能体开始，它能够使用网络搜索工具来回答问题。

3.1 安装依赖

首先，确保你有一个 Python 环境（建议 3.8+）。然后安装必要的包：

pip install langgraph langchain-openai tavily-python

• langgraph: 核心工作流框架。
• langchain-openai: LangChain 对 OpenAI 模型的集成。
• tavily-python: 一个简单好用的搜索 API 工具，我们将用它作为智能体的“眼睛”。你需要去 Tavily 官网注册一个免费账户获取 API Key。

注意： 工具的选择非常灵活。除了 Tavily，你也可以使用 Serper、Google Search API，或者封装一个自定义的函数（比如查询数据库）。这里选用 Tavily 是因为它对于快速原型开发非常友好，返回的结果已经是结构化的摘要。

3.2 定义状态（State）

状态是整个工作流的“共享内存”。我们用一个 TypedDict 来定义它，这样有更好的类型提示。

from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 消息历史，用于记录对话和思考过程。Annotated 是 LangGraph 的语法，add_messages 是一个归约函数，用于自动追加消息。 messages: Annotated[List[dict], add_messages] # 用户最初提出的问题，在整个循环中需要被持续访问。 question: str # 一个字符串，用来累积模型的“思考过程”（scratchpad），这是实现“记忆”的关键。 scratchpad: str

这里的关键是Annotated[List[dict], add_messages]。add_messages是一个“归约器”（reducer），它告诉 LangGraph，当多个节点并发修改messages字段时，应该用“追加”的方式合并，而不是覆盖。这对于维护对话历史至关重要。

scratchpad字段是我们手动管理的记忆，用来存放模型在 ReAct 循环中生成的思考文本，下次推理时会连同对话历史一起喂给模型。

4. 构建智能体工作流

现在我们来一步步搭建这个图。

4.1 初始化模型与工具

from langchain_openai import ChatOpenAI from langchain_community.tools.tavily_search import TavilySearchResults # 1. 初始化大模型。使用 gpt-3.5-turbo 性价比高，足够完成演示。 # 记得设置你的 OPENAI_API_KEY 环境变量。 llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 2. 初始化搜索工具。记得设置你的 TAVILY_API_KEY 环境变量。 # max_results 控制返回几条搜索结果，一般 3-5 条足够。 search_tool = TavilySearchResults(max_results=3) # 将工具包装成 LangChain 可识别的列表。 tools = [search_tool]

4.2 创建工具调用节点

这个节点负责执行具体的行动。它很简单：从状态中取出要执行的动作，调用对应的工具，然后把结果格式化后存回状态。

from langgraph.prebuilt import ToolNode # ToolNode 是 LangGraph 提供的一个预构建节点，专门用于处理工具调用。 # 你只需要把工具列表传给它，它就能自动根据传入的“工具调用”信息来执行。 tool_node = ToolNode(tools)

ToolNode内部会处理复杂的工具调用解析和执行，我们直接使用即可，省去了大量样板代码。

4.3 创建智能体（推理）节点

这是整个循环的大脑。它的任务是：根据当前状态（历史消息、问题、思考草稿），决定下一步该做什么。

from langchain.agents import create_react_agent from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain.tools.render import render_text_description def agent_node(state: AgentState): # 1. 准备提示词（Prompt） # 将工具列表渲染成一段文字描述，让模型知道它有哪些工具可用。 tool_descriptions = render_text_description(tools) # 构建一个 ReAct 风格的提示词模板。 # 这里我们手动构建一个简单的版本，以便理解。实际项目中可以使用 LangChain 的 PromptTemplate。 prompt = f"""你是一个有帮助的AI助手。请使用以下工具来回答问题。如果你已经知道答案，或者工具无法提供更多信息，请直接给出最终答案。 你可以使用的工具： {tool_descriptions} 之前的思考过程： {state.get('scratchpad', '')} 当前问题：{state['question']} 请严格按照以下格式回应： Thought: 我需要思考一下如何解决这个问题。 Action: 要使用的工具名称（必须是以下之一：{', '.join([t.name for t in tools])}） Action Input: 工具的输入参数 ...（这个 Thought/Action/Action Input 循环可以重复多次） 当你有最终答案时，请使用： Final Answer: 你的最终答案在这里。 开始： """ # 2. 调用大模型 # 我们将提示词和对话历史一起发给模型。历史消息提供了上下文。 messages = state['messages'] + [{"role": "user", "content": prompt}] response = llm.invoke(messages) # 3. 解析模型的响应 # 我们使用一个简单的解析器来提取 Thought, Action, Action Input 或 Final Answer。 # 这里为了演示，我们做简化处理。实际中应使用更健壮的解析器（如 ReActSingleInputOutputParser）。 content = response.content # 4. 更新状态 # 将模型的这次响应（思考）追加到消息历史中。 new_messages = state['messages'] + [{"role": "assistant", "content": content}] # 将思考内容也累积到 scratchpad 中，形成记忆。 new_scratchpad = state.get('scratchpad', '') + "\n" + content # 判断是否是最终答案 if "Final Answer:" in content: # 如果是最终答案，返回更新后的状态，这个循环将结束。 return { "messages": new_messages, "scratchpad": new_scratchpad, "question": state["question"] # 问题保持不变 } else: # 如果不是最终答案，说明模型决定调用工具。 # 我们需要从响应中提取出工具调用的信息。 # 这里是一个简化的正则匹配，实际项目请使用 LangChain 的 OutputParser。 import re action_match = re.search(r"Action:\s*(.+)", content) action_input_match = re.search(r"Action Input:\s*(.+)", content) if action_match and action_input_match: action = action_match.group(1).strip() action_input = action_input_match.group(1).strip() # 构造一个 LangChain 能识别的“工具调用请求”格式，并添加到消息中。 # 这样，下一个节点（tool_node）就能识别并执行它。 tool_call_message = { "role": "assistant", "content": "", # 对于工具调用，content 可能为空 "tool_calls": [{ "id": "call_1", # 模拟一个ID "type": "function", "function": { "name": action, "arguments": action_input } }] } new_messages_with_tool_call = new_messages + [tool_call_message] return { "messages": new_messages_with_tool_call, # 消息历史包含了工具调用请求 "scratchpad": new_scratchpad, "question": state["question"] } else: # 如果解析失败，让模型重试或直接结束 raise ValueError(f"无法从模型响应中解析出动作：{content}")

这个agent_node函数是核心中的核心。它展示了 ReAct 循环中“推理”部分的完整逻辑：组织提示词、调用模型、解析输出、更新记忆、并准备工具调用请求。

实操心得： 在实际开发中，强烈建议使用langchain.agents.create_react_agent来创建智能体，并使用ReActSingleInputOutputParser来解析输出。上面手写解析逻辑是为了让你看清底层发生了什么。使用官方组件能极大减少错误处理的工作量。

4.4 组装 StateGraph 并设置路由逻辑

有了节点，我们需要用StateGraph把它们连接起来，并告诉它如何流转。

from langgraph.graph import StateGraph, END # 1. 创建图构建器，并指定我们定义的状态类型 workflow = StateGraph(AgentState) # 2. 添加节点 # 第一个是智能体推理节点 workflow.add_node("agent", agent_node) # 第二个是工具执行节点 workflow.add_node("tools", tool_node) # 3. 设置入口点：从 agent 节点开始 workflow.set_entry_point("agent") # 4. 定义边（路由逻辑） # 这是一个条件边。在 `agent` 节点执行后，根据其输出状态，决定下一步去哪。 def route_after_agent(state: AgentState): # 检查最新的一条消息（应该是agent节点刚添加的） last_message = state['messages'][-1] # 如果这条消息里包含了工具调用请求，说明需要去执行工具 if hasattr(last_message, 'tool_calls') and last_message.tool_calls: return "tools" else: # 否则，说明 agent 给出了最终答案，工作流可以结束了 return END # 将这条条件边从 `agent` 节点引出 workflow.add_conditional_edges( "agent", route_after_agent, # 可选：指定可能的目的地，有助于框架优化 {"tools": "tools", END: END} ) # 5. 从 `tools` 节点出来的边是无条件的：执行完工具后，永远返回 `agent` 节点进行下一轮思考。 workflow.add_edge("tools", "agent") # 6. 编译图 # 编译后得到一个可执行的对象，它优化了内部执行逻辑。 app = workflow.compile()

至此，一个完整的、带记忆的 ReAct 循环智能体工作流就构建完成了。它的结构非常清晰：agent(思考) -> (如果需要工具) ->tools(执行) ->agent(再思考) -> ... ->END。

5. 运行与调试你的第一个智能体

图编译好了，让我们来运行它，看看这个智能体是如何工作的。

5.1 执行工作流

# 定义初始状态 initial_state: AgentState = { "messages": [], # 初始对话历史为空 "question": "2024年巴黎奥运会的吉祥物是什么？它有什么寓意？", "scratchpad": "" # 初始思考草稿为空 } # 运行图 # configurable 参数可以传递一些运行时配置，比如线程、中断等，这里我们先留空。 final_state = app.invoke(initial_state, configurable={}) # 查看最终结果 print("=== 最终对话历史 ===") for msg in final_state['messages']: print(f"{msg['role']}: {msg.get('content', '[Tool Call]')}") print("\n=== 最终答案 ===") # 从最后一条消息中提取 Final Answer final_message = final_state['messages'][-1] if "Final Answer:" in final_message['content']: print(final_message['content'].split("Final Answer:")[-1].strip())

当你运行这段代码时，LangGraph 会启动这个工作流。你会看到它在agent和tools节点间循环：

1. agent看到问题，思考后决定调用tavily_search工具。
2. 状态流转到tools节点，执行搜索，并将搜索结果写回状态。
3. 状态流回agent节点，模型看到搜索结果，进行新一轮思考。它可能觉得信息足够，直接给出最终答案；也可能决定再搜索一次。
4. 直到模型输出Final Answer:，条件边route_after_agent将其导向END，工作流终止。

5.2 可视化你的工作流

LangGraph 一个强大的功能是可视化。你可以将编译好的图导出查看。

# 方法1：在 Notebook 中直接显示 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法生成图片，可以打印文本结构 print(app.get_graph().draw_mermaid()) # 方法2：导出为文件 with open("my_react_agent_graph.png", "wb") as f: f.write(app.get_graph().draw_mermaid_png())

打开生成的图片，你会看到一个清晰的流程图：两个节点（agent,tools），以及它们之间的条件和无条件边。这对于理解复杂工作流和向他人解释你的设计非常有帮助。

5.3 调试与状态追踪

在开发过程中，你肯定需要知道每一步发生了什么。LangGraph 提供了详细的流式输出和中间状态查看。

# 流式输出，可以看到每一步执行了哪个节点，以及输入/输出状态 for event in app.stream(initial_state, configurable={}): for node_name, node_output in event.items(): print(f"--- 节点 [{node_name}] 执行完毕 ---") # node_output 就是该节点执行后的完整状态 last_msg = node_output['messages'][-1] if node_output['messages'] else None if last_msg: print(f"最新消息: {last_msg.get('role')} - {last_msg.get('content', 'N/A')[:200]}...") # 截断显示 print()

通过流式输出，你可以像看日志一样，实时观察智能体的“思考-行动”循环，精准定位问题发生在哪个环节。

6. 常见问题与进阶技巧

构建第一个能跑的智能体只是开始。在实际项目中，你会遇到各种边界情况和性能问题。这里分享一些踩坑后总结的经验。

6.1 如何控制循环，防止无限循环？

这是 ReAct 智能体最常见的问题。模型可能陷入“思考-搜索-再思考-再搜索”的死循环。解决方法有几种：

1. 设置最大迭代次数： 这是最有效的方法。可以在状态中增加一个iteration计数器，每次经过agent节点就加1。在route_after_agent函数中，除了检查工具调用，也检查计数器是否超过阈值（比如10次），如果超过，则强制返回END并返回一个超时提示。

   class AgentState(TypedDict): messages: Annotated[List[dict], add_messages] question: str scratchpad: str iteration: int # 新增迭代计数器 def route_after_agent(state: AgentState): if state['iteration'] > 10: # 添加一条系统超时消息 state['messages'].append({"role": "system", "content": "已达到最大思考次数，强制结束。"}) return END last_message = state['messages'][-1] if hasattr(last_message, 'tool_calls') and last_message.tool_calls: return "tools" else: return END # 在 agent_node 和初始状态中，记得维护 iteration 的值。

2. 优化提示词（Prompt Engineering）： 在提示词中明确要求模型“在得到足够信息后应果断给出最终答案”，并警告“不必要的工具调用会降低效率”。清晰的指令能显著减少无意义循环。

3. 使用“超时”或“看门狗”节点： 在更复杂的图中，可以设计一个专门的节点来监控整个工作流的执行时间或资源消耗，超时则发送中断信号。

6.2 工具调用解析失败怎么办？

我们上面的简化解析器很脆弱。生产环境务必使用 LangChain 提供的ReActSingleInputOutputParser或XMLAgentOutputParser。它们能更稳定地处理模型输出的各种格式。

from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain_core.agents import AgentAction, AgentFinish output_parser = ReActSingleInputOutputParser() def agent_node_with_parser(state: AgentState): # ... 准备提示词，调用模型 ... response = llm.invoke(messages) # 使用官方解析器 parsed_output = output_parser.parse(response.content) if isinstance(parsed_output, AgentFinish): # 模型决定结束 return_state = { "messages": state['messages'] + [{"role": "assistant", "content": parsed_output.return_values['output']}], "scratchpad": state['scratchpad'] + "\n" + response.content, "question": state["question"] } return return_state elif isinstance(parsed_output, AgentAction): # 模型决定调用工具 # parsed_output.tool 是工具名， parsed_output.tool_input 是输入参数 # 构造 tool_calls 消息... # ...

6.3 如何为智能体添加更多工具？

非常简单，只需在初始化tools列表时添加即可。LangChain 有海量的工具集成，从搜索引擎、计算器到代码执行、数据库查询应有尽有。你也可以轻松创建自定义工具：

from langchain.tools import tool @tool def get_weather(city: str) -> str: """根据城市名获取当前天气。""" # 这里调用一个天气API return f"{city}的天气是晴朗，25摄氏度。" # 然后将这个工具添加到列表中 tools = [search_tool, get_weather]

智能体在推理时，会自动从提示词中看到新工具的描述，并学会在合适的时候使用它。

6.4 状态（State）设计有哪些最佳实践？

• 最小化状态： 只把真正需要在节点间共享的数据放入 State。局部变量尽量在节点函数内部解决。
• 使用 Pydantic 模型： 对于复杂状态，使用pydantic.BaseModel代替TypedDict，能获得更好的验证和文档支持。
• 明确归约器（Reducer）： 像add_messages这样的归约器决定了并发写入时的合并策略。对于列表，通常用追加；对于数字，可能用求和。根据字段语义仔细选择。
• 考虑持久化： LangGraph 的状态可以很容易地序列化保存到数据库，从而实现长对话、断点续跑等高级功能。这在构建生产级应用时非常重要。

6.5 这个模式和 LangChain 的 AgentExecutor 有什么区别？

这是一个很好的问题。LangChain 传统的AgentExecutor也是一个封装好的 ReAct 循环执行器。它们的核心逻辑是相似的。主要区别在于：

• 灵活性与控制力：StateGraph提供了更低层、更灵活的控制。你可以自定义任何节点，定义复杂的循环、分支、并行逻辑。而AgentExecutor是一个黑盒，定制它内部的循环逻辑比较困难。
• 可视化与可调试性：StateGraph编译成的图可以可视化，并且流式执行过程清晰可见，调试体验更好。
• 复杂度： 对于简单的 ReAct 智能体，AgentExecutor几行代码就能搞定，更快捷。StateGraph需要更多的设置，但换来的是对复杂工作流的驾驭能力。

简单总结：如果你需要快速构建一个标准 ReAct 智能体，用AgentExecutor。如果你要构建的业务流程包含自定义步骤、复杂路由、子流程、人工审核节点等，LangGraph的StateGraph是你的不二之选。

构建这个带记忆的 ReAct 循环，就像是为你 AI 应用装上了“大脑”和“手脚”。StateGraph是支撑这个身体的“神经系统”，它让一切有序进行。从这个小例子出发，你可以尝试添加更多工具（如计算器、知识库查询）、引入人工审批节点、或者创建并行执行的分支。LangGraph 的世界才刚刚打开，它的能力边界取决于你对业务逻辑的抽象能力。