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循环拆解成几个明确的节点节点1Reasoning 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(contentreasoning_text) state[‘messages‘].append(new_ai_msg) state[‘current_reasoning‘] reasoning_text state[‘next_action‘] action state[‘next_action_input‘] action_input return state节点2Action 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(contentstate[‘last_observation‘], tool_call_id“simulated_id”) state[‘messages‘].append(tool_msg) return state节点3Check 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_turns10): 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(checkpointercheckpointer) # 第一次运行传入一个线程IDthread_id用于标识这次会话 config {“configurable”: {“thread_id”: “user_session_123”}} initial_state {“messages”: [HumanMessage(content“帮我规划北京旅行”)]} # 流式执行 for event in app.stream(initial_state, configconfig, 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, configconfig, 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的LCELLangChain Expression Language来构建可序列化的链并将其作为一个节点。一个重要的心智转变从“写Agent逻辑”到“设计工作流”。使用LangGraph后你的主要工作不再是编写控制循环的代码而是定义状态模型思考你的Agent需要记住什么。设计节点将复杂的任务拆解成一个个单一职责的步骤。绘制边用流程图思维连接这些步骤定义好正常流程、异常流程和分支判断。 这种转变能让你更专注于业务逻辑本身而不是控制流的细节从而构建出更健壮、更易维护的智能体应用。LangGraph不是一颗银弹它引入了一定的学习成本和设计复杂度。但对于需要清晰流程控制、复杂状态管理、长时运行或高可观测性的Agent场景它提供的抽象能力和工程化支持是传统Agent架构难以比拟的。它让AI Agent的开发从“脚本”走向了“系统”。