当前位置：首页 > news >正文

LangGraph：从ReAct到工作流编排，构建可控大模型Agent的工程实践

news 2026/8/14 6:33:37

1. 从“工具调用”到“工作流编排”：Agent架构的演进与LangGraph的定位

如果你最近在折腾大模型应用，尤其是想让它帮你自动完成一些复杂的、多步骤的任务，那么“Agent”这个词你一定不陌生。它听起来很酷，仿佛你的代码拥有了一个能自主思考、调用工具、解决问题的智能体。但当你真正上手LangChain的Agent时，可能会遇到一些困惑：为什么我写的Agent总是“跑偏”？为什么它执行几步后就卡住或者开始胡说八道？为什么一个简单的多轮对话任务，用Agent实现起来却感觉异常笨重？

这些问题，恰恰暴露了传统Agent架构（比如LangChain早期版本中基于AgentExecutor的范式）的局限性。它更像一个“单次决策-执行”的循环，缺乏对复杂、有状态、长流程任务的优雅管理能力。而LangGraph的出现，正是为了解决这些问题。它不是要取代LangChain的Agent，而是为Agent提供了一个更强大、更灵活的“骨架”和“神经系统”，让你能够像绘制流程图一样，清晰地定义和控制智能体的整个工作流程。简单来说，LangChain Agent给了你“智能体”的能力，而LangGraph则让你能指挥这个智能体去完成一部“交响乐”，而非只是弹几个单音。

2. 传统Agent架构的“阿喀琉斯之踵”：以ReAct模式为例

要理解LangGraph的价值，我们得先看看它要解决什么问题。在LangChain中，最经典的Agent模式之一是ReAct。ReAct代表“Reasoning + Acting”，即“思考-行动”。其核心思想是让模型在每一步都先“思考”（Reasoning）一下当前状况和下一步该做什么，然后再“行动”（Acting）去执行一个具体的工具调用。

2.1 ReAct的工作流程拆解

一个典型的ReAct Agent工作循环是这样的：

观察：Agent获得当前的用户输入和之前的交互历史。
思考：大语言模型（LLM）基于观察，生成一段“思考”文本。这段文本会分析当前目标、可用工具，并决定下一步是使用某个工具，还是直接给出最终答案。
决策：一个特定的输出解析器（如ReActSingleInputOutputParser）会解析模型的“思考”文本，提取出两个关键信息：action（要执行哪个工具）和action_input（调用该工具的输入参数）。
执行：Agent根据解析出的action，找到对应的工具函数并执行，传入action_input，得到工具的执行结果observation。
循环：将工具执行的结果observation作为新的观察，连同历史记录，再次喂给LLM进行下一轮的“思考”。如此循环，直到模型决定输出最终答案（Final Answer）。

这个过程听起来很合理，但它有几个固有的痛点：

2.2 传统架构的三大核心痛点

痛点一：状态管理混乱且脆弱。在整个循环中，“状态”是分散且隐式的。它可能存在于AgentExecutor的内部变量里，存在于对话历史记录中，也可能存在于你自定义的某个全局对象里。当你需要实现一个包含条件分支（比如“如果查询天气为雨，则推荐室内活动；否则推荐户外活动”）的复杂流程时，管理这些状态并确保它们在每一步正确传递，会变得非常棘手。代码很快就会变成一堆if-else和状态标志位，难以维护和调试。

痛点二：流程控制能力薄弱。传统的AgentExecutor本质上是一个while循环：只要模型不输出Final Answer，它就继续循环。你很难在这个循环中插入复杂的控制逻辑，比如：

暂停与恢复：让Agent执行到某一步后暂停，等待外部人工确认后再继续。
并行执行：同时发起多个不依赖的工具调用（比如同时查询北京和上海的天气）。
循环与迭代：明确指定某个子任务需要重复执行N次，或者直到满足某个条件为止。
优雅失败与重试：当某个工具调用失败时，是重试、换一种方式，还是进入备选流程？

这些需求在AgentExecutor的框架下实现起来非常别扭，往往需要侵入性地修改其内部逻辑。

痛点三：可观测性与调试困难。当你的Agent跑飞了或者卡在一个循环里时，你很难一眼看出它当前处在流程的哪个阶段，历史决策路径是怎样的。调试通常依赖于打印大量的日志，但日志是线性的，难以还原出非线性的、有分支的决策过程。

LangGraph正是为了系统性地解决这些痛点而设计的。它将Agent的工作流程显式化、图化、状态化。

3. LangGraph核心概念四要素：图、节点、边、状态

理解LangGraph，关键在于掌握四个核心概念：图（Graph）、节点（Node）、边（Edge）和状态（State）。你可以把它想象成画一个业务流程图。

3.1 状态：流程的“记忆白板”

在LangGraph中，状态是一个中心化的、定义明确的数据结构。它通常是一个Pydantic模型或一个TypedDict，规定了在整个工作流中流转的所有信息。这是与传统Agent最根本的区别。

假设我们在构建一个“旅行规划Agent”，它的状态可能包含：

from typing import TypedDict, List, Optional from datetime import date class AgentState(TypedDict): # 用户输入 user_query: str # 中间结果 destination: Optional[str] travel_dates: Optional[List[date]] weather_info: Optional[dict] flight_options: Optional[List[dict]] hotel_options: Optional[List[dict]] # 执行日志和LLM调用记录 reasoning_log: List[str] # 最终输出 final_itinerary: Optional[str]

这个AgentState就像一块共享的白板。工作流中的每个步骤（节点）都从这块白板上读取自己需要的信息，处理完后，再把结果写回白板的对应位置。这样，状态管理就从“散兵游勇”变成了“中央集权”，清晰且强类型安全。

3.2 节点与边：构建可执行的流程图

节点就是工作流中的步骤，它是一个普通的Python函数（或可调用对象）。这个函数接收一个state字典，对其进行修改或读取，然后返回更新后的state。

边定义了节点之间的流转关系。它决定了在一个节点执行完毕后，接下来应该执行哪个节点。边可以是有条件的，基于state中的某个值来决定下一步走向（这解决了分支问题）；也可以是无条件的，总是流向某个固定节点。

图则是所有这些节点和边的集合，它定义了整个工作流的拓扑结构。

让我们用代码勾勒一个极度简化的旅行规划流程：

from langgraph.graph import StateGraph, END # 1. 定义构建图的工作流 workflow = StateGraph(AgentState) # 2. 定义节点（步骤） def parse_user_input(state: AgentState): # 从state[‘user_query’]中解析出目的地和日期，写入state # 例如，使用一个LLM或简单的规则进行解析 state[‘destination‘] = “北京” state[‘travel_dates‘] = [date(2024, 10, 1), date(2024, 10, 7)] state[‘reasoning_log‘].append(“已解析用户输入：目的地北京，日期10.1-10.7”) return state def fetch_weather(state: AgentState): # 调用天气查询工具，结果写入state[‘weather_info’] if state[‘destination‘]: state[‘weather_info‘] = call_weather_api(state[‘destination‘], state[‘travel_dates‘]) state[‘reasoning_log‘].append(f“已获取{state[‘destination‘]}的天气信息”) return state def plan_activities(state: AgentState): # 根据天气规划活动 weather = state[‘weather_info‘] activities = [] if weather and weather.get(‘is_rainy‘): activities = [“参观国家博物馆”, “逛王府井商场”, “看话剧”] else: activities = [“游览故宫”, “爬长城”, “颐和园划船”] state[‘activities‘] = activities state[‘reasoning_log‘].append(f“根据天气规划活动：{activities}”) return state def generate_itinerary(state: AgentState): # 汇总所有信息，生成最终行程单 final_text = f”目的地：{state[‘destination‘]}... 活动：{‘, ‘.join(state[‘activities‘])}” state[‘final_itinerary‘] = final_text state[‘reasoning_log‘].append(“已生成最终行程单”) return state # 3. 将节点添加到图中 workflow.add_node(“parse_input”, parse_user_input) workflow.add_node(“get_weather”, fetch_weather) workflow.add_node(“plan”, plan_activities) workflow.add_node(“generate”, generate_itinerary) # 4. 添加边，定义流程 workflow.set_entry_point(“parse_input”) # 设置入口节点 workflow.add_edge(“parse_input”, “get_weather”) # 解析完输入就去查天气 workflow.add_edge(“get_weather”, “plan”) # 查到天气后规划活动 workflow.add_edge(“plan”, “generate”) # 规划完活动生成行程 workflow.add_edge(“generate”, END) # 生成行程后，工作流结束 # 5. 编译图，得到一个可执行的对象 app = workflow.compile()

这个简单的图是一个线性流程：A -> B -> C -> D -> 结束。但这已经比一个黑盒的while循环清晰多了。我们能看到完整的数据流和步骤。

3.3 条件边与循环：实现复杂逻辑

LangGraph的强大之处在于条件边。我们可以让流程“分叉”。

假设我们修改流程：只有在天气查询成功后才规划活动，如果查询失败，则直接向用户请求手动输入天气。

from langgraph.graph import StateGraph, END from langgraph.graph import START def fetch_weather_with_fallback(state: AgentState): try: state[‘weather_info‘] = call_weather_api(state[‘destination‘], state[‘travel_dates‘]) state[‘reasoning_log‘].append(“天气查询成功”) state[‘weather_status‘] = “success” except Exception: state[‘reasoning_log‘].append(“天气查询失败，需要用户补充”) state[‘weather_status‘] = “fail” return state def ask_user_for_weather(state: AgentState): # 这里可以触发一个等待用户输入的逻辑 state[‘reasoning_log‘].append(“正在等待用户输入天气信息...”) # 假设我们模拟用户输入 state[‘weather_info‘] = {“is_rainy”: False} return state def should_plan_activities(state: AgentState) -> str: # 这是一个路由函数，它不修改state，只返回下一个节点的名字 if state.get(‘weather_status‘) == “success”: return “plan” # 去规划活动 else: return “ask_user” # 去询问用户 # 重新构建图 workflow = StateGraph(AgentState) workflow.add_node(“parse_input”, parse_user_input) workflow.add_node(“get_weather”, fetch_weather_with_fallback) workflow.add_node(“ask_user”, ask_user_for_weather) workflow.add_node(“plan”, plan_activities) workflow.add_node(“generate”, generate_itinerary) workflow.set_entry_point(“parse_input”) workflow.add_edge(“parse_input”, “get_weather”) # 关键：从‘get_weather’节点出来的边，由`should_plan_activities`函数决定 workflow.add_conditional_edges( “get_weather”, # 源节点 should_plan_activities, # 路由函数 {“plan”: “plan”, “ask_user”: “ask_user”} # 可能的目的地映射 ) workflow.add_edge(“ask_user”, “plan”) # 用户补充天气后，再去规划 workflow.add_edge(“plan”, “generate”) workflow.add_edge(“generate”, END)

通过add_conditional_edges，我们实现了一个条件分支。should_plan_activities函数像一个交通警察，根据state里的weather_status，决定流程是走向plan节点还是ask_user节点。

循环则是通过将边指向之前的节点来实现的。例如，你可以设置一个“优化行程”节点，如果对生成的结果不满意，就让它指回“规划活动”节点，形成一个循环，直到满足某个条件（比如循环次数或评分阈值）再跳出。

4. 实战：用LangGraph重构一个ReAct风格的多工具查询Agent

现在，让我们把经典的多工具ReAct Agent用LangGraph重新实现一遍，你会立刻感受到其清晰度的提升。假设我们有一个能查询天气和搜索百科的Agent。

4.1 定义状态与工具

首先，定义状态。我们需要记录LLM的“思考”过程、工具调用历史和最终答案。

from typing import TypedDict, List, Optional, Literal from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage import json class ReActState(TypedDict): # 消息历史，用于记录对话和思考 messages: List[BaseMessage] # 当前轮次模型生成的“思考”文本（包含action/action_input） current_reasoning: Optional[str] # 当前轮次解析出的工具调用名称 next_action: Optional[str] # 当前轮次解析出的工具调用参数 next_action_input: Optional[dict] # 工具执行的结果 last_observation: Optional[str] # 流程是否应该结束 should_finish: bool # 最终答案 final_answer: Optional[str] # 模拟两个工具 def search_wikipedia(query: str) -> str: return f”根据维基百科，‘{query}’的相关结果是：...(模拟数据)” def get_current_weather(location: str) -> str: return f”{location}的天气是晴朗，25摄氏度。(模拟数据)” tools = [ { “name”: “SearchWikipedia”, “description”: “用于查询事实性知识、历史事件、人物生平等信息。”, “func”: search_wikipedia }, { “name”: “GetCurrentWeather”, “description”: “用于查询指定城市的当前天气情况。”, “func”: get_current_weather } ] tools_by_name = {tool[“name”]: tool[“func”] for tool in tools}

4.2 构建LangGraph节点

我们将ReAct循环拆解成几个明确的节点：

节点1：Reasoning Node (思考节点)这个节点模拟LLM的“思考”步骤。在实际中，这里会调用LLM。我们这里用硬编码模拟一个简单的决策逻辑。

def reasoning_node(state: ReActState) -> ReActState: messages = state[‘messages‘] # 取最后一条用户消息 last_human_msg = [m for m in messages if isinstance(m, HumanMessage)][-1] query = last_human_msg.content # 模拟LLM的思考过程：分析问题，决定行动 if “天气” in query: reasoning_text = “用户想了解天气。我应该使用GetCurrentWeather工具。” action = “GetCurrentWeather” # 简单地从查询中提取城市名，实际应用需更复杂的解析 action_input = {“location”: “北京”} elif “是谁” in query or “什么是” in query: reasoning_text = “用户询问事实性知识。我应该使用SearchWikipedia工具。” action = “SearchWikipedia” action_input = {“query”: query} else: reasoning_text = “这个问题不需要使用工具，我可以直接回答。” action = “FinalAnswer” action_input = {“answer”: “这是一个通用问题，我的回答是...”} # 更新状态 new_ai_msg = AIMessage(content=reasoning_text) state[‘messages‘].append(new_ai_msg) state[‘current_reasoning‘] = reasoning_text state[‘next_action‘] = action state[‘next_action_input‘] = action_input return state

节点2：Action Node (行动节点)这个节点根据next_action执行对应的工具。

def action_node(state: ReActState) -> ReActState: action = state[‘next_action‘] action_input = state[‘next_action_input‘] if action == “FinalAnswer”: # 如果是最终答案，直接更新状态并结束 state[‘final_answer‘] = action_input[‘answer‘] state[‘should_finish‘] = True return state if action not in tools_by_name: state[‘last_observation‘] = f”错误：未知工具 {action}” state[‘should_finish‘] = True return state # 执行工具 tool_func = tools_by_name[action] try: # 注意：实际调用时需根据工具函数签名解包参数 result = tool_func(**action_input) state[‘last_observation‘] = result except Exception as e: state[‘last_observation‘] = f”工具{action}执行出错：{str(e)}” # 将工具执行结果以ToolMessage形式存入历史 tool_msg = ToolMessage(content=state[‘last_observation‘], tool_call_id=“simulated_id”) state[‘messages‘].append(tool_msg) return state

节点3：Check Finish Node (检查结束节点)这个节点决定流程是否继续。在真正的ReAct中，LLM的思考里会包含“Final Answer”关键词。

def check_finish_node(state: ReActState) -> Literal[“reasoning”, “__end__”]: # 如果已经标记结束，或者已经尝试了太多次（防止无限循环），则结束 if state.get(‘should_finish‘, False): return “__end__” # 如果上一步是最终答案，也结束 if state.get(‘next_action‘) == “FinalAnswer”: return “__end__” # 否则，继续下一轮思考 return “reasoning”

4.3 组装并运行图

现在，我们把节点组装成一个图，它清晰地展示了ReAct的循环逻辑。

from langgraph.graph import StateGraph, START, END # 构建图 workflow = StateGraph(ReActState) workflow.add_node(“reasoning”, reasoning_node) workflow.add_node(“action”, action_node) # `check_finish`是一个特殊的节点，它只做路由判断，不修改状态 workflow.add_node(“check_finish”, check_finish_node) # 设置流程 workflow.set_entry_point(“reasoning”) workflow.add_edge(“reasoning”, “action”) # 思考完就行动 workflow.add_edge(“action”, “check_finish”) # 行动完检查是否结束 # 从检查节点出来的边是条件边，决定回到思考节点还是结束 workflow.add_conditional_edges( “check_finish”, check_finish_node, # 路由函数就是它自己 {“reasoning”: “reasoning”, “__end__”: END} ) app = workflow.compile() # 初始化状态并运行 initial_state: ReActState = { “messages”: [HumanMessage(content=“北京的天气怎么样？”)], “current_reasoning”: None, “next_action”: None, “next_action_input”: None, “last_observation”: None, “should_finish”: False, “final_answer”: None } # 运行图，并设置中断条件（例如最大步数） final_state = None for step, output in app.stream(initial_state, stream_mode=“values”, max_turns=10): node_name = list(output.keys())[0] print(f”步骤执行节点：{node_name}”) print(f”当前推理：{output[node_name].get(‘current_reasoning‘)}”) print(f”下一步行动：{output[node_name].get(‘next_action‘)}”) print(f”观察结果：{output[node_name].get(‘last_observation‘)}”) print(“-” * 20) final_state = output[node_name] if output[node_name].get(‘should_finish‘): break print(f”最终答案：{final_state.get(‘final_answer‘)}”)

运行这个图，你会看到清晰的执行轨迹：

进入reasoning节点：思考“用户问天气，用GetCurrentWeather工具”。
进入action节点：执行天气工具，得到观察结果“北京天气晴朗...”。
进入check_finish节点：判断next_action不是FinalAnswer，且should_finish为False，所以路由回reasoning。
再次进入reasoning节点：基于“北京天气晴朗...”这个新观察，思考“我已经获得了天气信息，现在可以给出最终答案了”，并设置next_action为FinalAnswer。
再次进入action节点：因为是FinalAnswer，所以设置final_answer和should_finish。
再次进入check_finish节点：发现should_finish为True，路由到END，流程结束。

通过这个重构，ReAct的“思考-行动”循环被清晰地具象化为一个有两个主要节点（思考、行动）和一个判断节点的有向图。状态（ReActState）在整个流程中单向、明确地流动。如果你想增加一个“验证答案”的步骤，或者让它在给出最终答案前先总结一下工具调用历史，只需要在图中插入新的节点并调整边的连接即可，模块化程度极高。

5. LangGraph高级特性与工程化实践

掌握了基础概念后，我们来看看LangGraph如何解决更复杂的生产级问题。

5.1 持久化与检查点：实现长时运行与恢复

这是LangGraph相比传统Agent执行器的杀手级特性。你可以将整个工作流的状态持久化到数据库（如Redis、PostgreSQL），并为状态创建一个唯一的checkpoint_id。

from langgraph.checkpoint import MemorySaver # 使用内存检查点管理器（生产环境可用RedisCheckpointer等） checkpointer = MemorySaver() # 在编译图时传入检查点管理器 app = workflow.compile(checkpointer=checkpointer) # 第一次运行，传入一个线程ID（thread_id），用于标识这次会话 config = {“configurable”: {“thread_id”: “user_session_123”}} initial_state = {“messages”: [HumanMessage(content=“帮我规划北京旅行”)]} # 流式执行 for event in app.stream(initial_state, config=config, stream_mode=“values”): print(event) # 假设流程在这里因为某种原因中断了（比如等待用户异步输入） # 我们可以通过相同的thread_id恢复状态 print(“\n— 流程中断，现在恢复 —\n”) # 直接获取当前最新状态 resumed_state = app.get_state(config) print(f”恢复后的状态: {resumed_state.values}”) # 继续执行，传入新的用户输入作为消息 new_message = HumanMessage(content=“我的出行日期是国庆节”) resumed_state[‘messages‘].append(new_message) for event in app.stream(resumed_state, config=config, stream_mode=“values”): print(event)

这个机制使得构建需要等待外部事件（如人工审核、第三方回调）的异步Agent、或者实现聊天机器人对话状态的长期记忆变得非常简单。每个用户的对话就是一个独立的thread，其完整状态都被保存下来。

5.2 子图与模块化：管理复杂工作流

当你的Agent工作流非常复杂时，把所有逻辑塞进一个图里会难以维护。LangGraph支持子图，允许你将一部分节点和边打包成一个独立的、可复用的子工作流。

例如，我们可以把“查询并处理天气信息”这一系列操作封装成一个子图。

from langgraph.graph import StateGraph, END # 定义一个处理天气的子图 def create_weather_subgraph(): from typing import TypedDict class WeatherSubState(TypedDict): destination: str dates: list raw_weather: dict processed_advice: str sub_builder = StateGraph(WeatherSubState) def fetch_raw_weather(state): # 调用API state[‘raw_weather‘] = {“temp”: 25, “condition”: “sunny”} return state def process_advice(state): if state[‘raw_weather‘][‘condition’] == “sunny”: state[‘processed_advice‘] = “天气晴朗，建议户外活动。” else: state[‘processed_advice‘] = “天气不佳，建议室内活动。” return state sub_builder.add_node(“fetch”, fetch_raw_weather) sub_builder.add_node(“process”, process_advice) sub_builder.set_entry_point(“fetch”) sub_builder.add_edge(“fetch”, “process”) sub_builder.add_edge(“process”, END) return sub_builder.compile() # 在主图中，将这个子图作为一个“超级节点”加入 weather_subgraph = create_weather_subgraph() main_workflow.add_node(“weather_processor”, weather_subgraph)

这样，在主图的设计中，你只需要关心“这里需要处理天气”，而不必关心里面具体有几个步骤。这极大地提升了复杂工作流的可读性和可维护性。

5.3 中间件与可观测性：监控与调试

LangGraph的流式执行（app.stream）本身就提供了强大的可观测性，你可以实时看到每个节点的输入输出。此外，你还可以利用中间件来注入监控、日志记录、性能分析等逻辑。

例如，为每个节点的执行添加计时和日志：

from langgraph.graph import StateGraph import time from functools import wraps def log_node_execution(node_func): @wraps(node_func) def wrapper(state): node_name = node_func.__name__ print(f”>>> 开始执行节点: {node_name}”) start_time = time.time() result = node_func(state) elapsed = time.time() - start_time print(f”<<< 节点 {node_name} 执行完毕，耗时 {elapsed:.2f}秒”) # 这里可以记录到监控系统 return result return wrapper # 装饰你的节点函数 @log_node_execution def my_node(state): # ... 节点逻辑 return state

在生产环境中，你可以将日志发送到ELK栈，将耗时和错误信息记录到Prometheus/Grafana，实现对Agent工作流全链路的监控。

6. 避坑指南与性能优化心得

在实际项目中用LangGraph构建复杂Agent，我踩过不少坑，也总结了一些经验。

坑一：状态设计过于臃肿。初期很容易把所有的中间数据都塞进State里，导致State类型复杂，每个节点都需要处理大量字段。最佳实践是保持State的精简，只存放真正需要在节点间流转的核心数据。对于一些中间计算结果，如果只在单个节点内使用，完全可以作为局部变量。

坑二：条件边路由函数过于复杂。路由函数（add_conditional_edges中使用的函数）应该只做简单的判断，返回下一个节点名。不要把复杂的业务逻辑放在路由函数里。如果判断逻辑很复杂，应该设计一个专门的“路由决策”节点，它负责计算并将结果写入State，然后由一条简单的条件边读取这个结果来做路由。

坑三：忽视错误处理和回退机制。在图中，一个节点的失败可能导致整个流程中断。务必在每个可能出错的节点（尤其是调用外部API、工具、LLM的节点）内部做好try-catch，并将错误信息妥善地写入State，并设计好错误处理路径。例如，可以有一个专门的error_handler节点，或者让条件边路由到fallback节点。

性能优化点一：异步节点执行。如果你的节点中有大量的I/O操作（如网络请求、数据库查询），强烈建议使用异步函数（async def）来定义节点，并使用支持异步的图执行器。这可以大幅提升高并发下的吞吐量。LangGraph完全支持异步。

性能优化点二：LLM调用优化。多个节点可能都需要调用LLM。避免在每个节点内部创建新的LLM实例或重复定义Prompt。应该在图外定义好LLM和关键Prompt模板，以依赖注入的方式传给各个节点函数。对于复杂的、多步骤的LLM调用，可以考虑使用LangChain的LCEL（LangChain Expression Language）来构建可序列化的链，并将其作为一个节点。

一个重要的心智转变：从“写Agent逻辑”到“设计工作流”。使用LangGraph后，你的主要工作不再是编写控制循环的代码，而是：