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 循环ReActReason Act是一个让大模型与外部工具交互的经典范式。它让模型学会“思考-行动”的循环推理Reason 模型分析当前情况和目标思考下一步该做什么。行动Act 模型根据思考执行一个具体的动作比如调用一个搜索工具、一个计算器 API。观察Observe 获取行动工具调用的结果。循环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-pythonlanggraph: 核心工作流框架。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(modelgpt-3.5-turbo, temperature0) # 2. 初始化搜索工具。记得设置你的 TAVILY_API_KEY 环境变量。 # max_results 控制返回几条搜索结果一般 3-5 条足够。 search_tool TavilySearchResults(max_results3) # 将工具包装成 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(rAction:\s*(.), content) action_input_match re.search(rAction 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节点间循环agent看到问题思考后决定调用tavily_search工具。状态流转到tools节点执行搜索并将搜索结果写回状态。状态流回agent节点模型看到搜索结果进行新一轮思考。它可能觉得信息足够直接给出最终答案也可能决定再搜索一次。直到模型输出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 智能体最常见的问题。模型可能陷入“思考-搜索-再思考-再搜索”的死循环。解决方法有几种设置最大迭代次数 这是最有效的方法。可以在状态中增加一个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 的值。优化提示词Prompt Engineering 在提示词中明确要求模型“在得到足够信息后应果断给出最终答案”并警告“不必要的工具调用会降低效率”。清晰的指令能显著减少无意义循环。使用“超时”或“看门狗”节点 在更复杂的图中可以设计一个专门的节点来监控整个工作流的执行时间或资源消耗超时则发送中断信号。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 的世界才刚刚打开它的能力边界取决于你对业务逻辑的抽象能力。