LangGraph实战：从单智能体到多智能体，构建有状态工作流

上周在帮一个团队做技术选型他们想用大模型做一套自动化流程把客服、质检、工单、报表几个环节串起来。一开始他们觉得不就是调用几个 API 吗用 LangChain 把工具链起来不就行了结果真跑起来问题全来了状态丢了、循环卡住了、分支判断逻辑写成一团乱麻、一个节点出错整个流程全停。折腾了两周流程图画了又改代码越写越复杂。这其实是一个典型的误区把“多智能体”或“复杂工作流”简单理解成“把几个模型调用按顺序排好”。顺序执行当然简单但现实中的任务尤其是涉及决策、循环、条件分支和人机交互的任务本质是一个有状态、可中断、可回溯的图。你需要的不只是一个链而是一个能清晰定义节点、边、状态流转和异常处理的运行时框架。这也是为什么 LangGraph 的出现对很多想深入应用大模型的人来说是一个关键转折点。它不是一个替代 LangChain 的工具而是一个补上了 LangChain 在复杂流程编排上短板的专用框架。很多人一上来就纠结“学 LangChain 还是学 LangGraph”这其实问错了。真正的问题是你面对的任务是简单的线性管道还是复杂的、带状态的工作流如果你需要处理“根据中间结果决定下一步”、“在循环中积累信息”、“等待外部输入后再继续”这类场景那么直接上手 LangGraph可能比用 LangChain 硬拗要省力得多也清晰得多。这篇文章我就从一个实践者的角度带你彻底吃透 LangGraph。我们不只讲“怎么用”更要讲清楚为什么是它、它解决了什么根本问题以及从单智能体到多智能体实战真正要避开的坑在哪里。1. 先想清楚你的任务真的需要 LangGraph 吗在动手写第一行代码之前我们先做一个关键判断。LangGraph 的核心价值是管理有状态的、图结构的工作流。如果你的需求满足以下任何一个特征那么 LangGraph 很可能就是你的菜任务有明确的“状态”概念整个流程的运行依赖于一个不断更新的上下文比如对话历史、已收集的信息、临时决策。流程不是一条直线需要根据某些条件if/else走不同的分支。流程中包含循环需要反复执行某个或某组操作直到满足退出条件比如不断追问用户直到信息收集完整。需要“暂停”并等待外部输入比如等待用户回复、等待一个异步 API 回调、等待人工审核。多个“智能体”或工具需要协作它们之间有清晰的交互协议和状态传递而不是简单的接力赛。反过来如果你的任务纯粹是输入 A - 模型处理 - 输出 B中间没有决策、没有循环、没有状态依赖那么一个简单的 LangChain Chain 或甚至直接调用模型 API 就足够了。引入 LangGraph 只会增加不必要的复杂度。一个常见的误判把“多步骤”等同于“需要图”。很多任务步骤虽多但只是线性串联上一步的输出直接作为下一步的输入中间没有决策岔路。这种是“链”不是“图”。图的核心在于路由Routing——根据状态决定下一步走向哪里。所以在决定使用 LangGraph 前先在白板上画出你的工作流。如果画出来是一条带分叉、合并或循环的流程图那么恭喜你找对工具了。2. 理解 LangGraph 的核心心智模型把工作流画成一张图LangGraph 的官方定义是“用于构建有状态、多智能体应用程序的库”。这个定义有点抽象。我更愿意把它理解为一个让你用“图”的思维来设计和运行程序的框架。它的核心组件非常简洁对应着图论的基本概念2.1 节点Node就是你的一个“操作单元”一个节点可以是一个函数它接收当前的“状态”做一些事情然后返回更新后的“状态”。它可以做什么调用大模型、执行工具、运行计算逻辑、访问数据库。关键理解节点是无副作用的纯函数吗不完全是但它应该专注于处理输入状态并产生输出状态。与外界数据库、API的交互是允许的但 LangGraph 关心的是这些交互如何影响“图的状态”。# 一个简单的节点示例调用大模型生成回复 def llm_node(state): messages state[messages] # 假设有一个 call_llm 函数 response call_llm(messages) # 更新状态将AI回复加入历史 state[messages].append({role: assistant, content: response}) return state2.2 边Edge决定流程怎么走边定义了节点之间的连接关系。但 LangGraph 的边是“有条件”的这是它比简单链强大得多的地方。条件边Conditional Edge根据当前状态的值决定下一步去哪个节点。这是实现分支if/else的核心。普通边无条件地从一个节点指向下一个节点。边的条件判断通常由一个特殊的“路由器”函数来决定。2.3 状态State整个图的“记忆中枢”这是 LangGraph 的灵魂。状态是一个共享的数据结构在所有节点之间传递和修改。你可以把它想象成流程的“共享白板”。它通常是一个字典或 Pydantic 模型包含流程所需的所有信息如用户输入、对话历史、中间结果、循环计数器、标志位等。关键机制状态缩减器当多个节点可能并发修改状态时你需要定义如何合并这些修改。LangGraph 使用“缩减器”来声明。例如对于messages列表缩减器可能是“追加”append。2.4 图Graph把节点和边组装起来你定义节点用边把它们连起来就构成了一张图。LangGraph 提供了两种主要的图StateGraph最常用、最灵活。你显式地定义状态结构、节点和边。MessageGraph一个更高级的抽象专为基于“消息”的智能体对话设计。它内置了常见的模式用起来更简单但定制性稍弱。把以上组合起来你创建一个StateGraph定义好状态结构添加节点然后用边包括条件边把节点连接起来。最后将它编译成一个可执行的CompiledGraph。运行这个图时你提供一个初始状态它就会按照你定义的路径节点-边-节点...执行下去直到到达某个终点。这个心智模型——节点做具体工作边负责路由状态负责记忆——是理解 LangGraph 所有高级特性的基础。3. 从零构建你的第一个 LangGraph 智能体一个决策助手理论讲多了容易晕我们直接上手构建一个简单的“旅行规划助手”智能体。这个智能体会接收用户关于旅行的问题。判断用户是想获取“信息”还是进行“规划”。如果是“信息”就调用搜索工具。如果是“规划”就进入一个多轮对话循环收集目的地、预算、时间等信息。收集完信息后生成一份规划建议。这个流程包含了条件分支和循环正好用上 LangGraph。3.1 第一步定义状态状态是我们设计的蓝图。我们用一个 Pydantic 模型来定义这样类型更安全。from typing import List, Dict, Any, Optional, Literal from typing_extensions import TypedDict from pydantic import BaseModel, Field class AgentState(TypedDict): 图的状态定义 # 用户输入的问题 input: str # 对话消息历史 messages: List[Dict[str, str]] # 当前流程阶段classify, get_info, collect_planning_info, generate_plan stage: Literal[classify, get_info, collect_planning_info, generate_plan] # 收集到的规划信息 planning_info: Dict[str, Any] # 最终输出 output: Optional[str]3.2 第二步创建图和状态图from langgraph.graph import StateGraph, END # 创建状态图并传入我们定义的状态结构 workflow StateGraph(AgentState)3.3 第三步定义节点函数每个节点都是一个接收状态、返回更新后状态的函数。# 节点1分类节点 - 判断用户意图 def classify_intent(state: AgentState) - AgentState: user_input state[input] # 这里简化处理实际应用中应该调用一个分类模型或LLM if 多少钱 in user_input or 天气 in user_input or 景点 in user_input: state[stage] get_info else: state[stage] collect_planning_info state[messages].append({role: assistant, content: f已识别意图进入阶段{state[stage]}}) return state # 节点2信息查询节点 def get_travel_info(state: AgentState) - AgentState: # 模拟调用一个搜索工具 query state[input] # 假设 search_tool 是一个返回信息的函数 info f关于{query}的信息这里是模拟的搜索结果。 state[output] info state[messages].append({role: assistant, content: info}) state[stage] generate_plan # 信息查询完毕可以结束了 return state # 节点3信息收集节点可能多次执行 def collect_planning_info(state: AgentState) - AgentState: messages state[messages] # 简单模拟如果是第一次进入询问目的地 if destination not in state[planning_info]: question 请问您的旅行目的地是哪里 state[messages].append({role: assistant, content: question}) # 注意这里我们模拟用户回答。真实场景中图会在这里暂停等待外部输入。 # LangGraph 支持“中断”机制这里为了演示先模拟。 state[planning_info][destination] 模拟目的地巴黎 state[messages].append({role: user, content: 巴黎}) elif budget not in state[planning_info]: question 您的预算是多少 state[messages].append({role: assistant, content: question}) state[planning_info][budget] 模拟预算10000元 state[messages].append({role: user, content: 10000元}) else: # 信息已收集完 state[stage] generate_plan return state # 节点4生成规划节点 def generate_travel_plan(state: AgentState) - AgentState: info state[planning_info] plan f为您生成{info.get(destination)}的旅行规划预算{info.get(budget)}。这里是详细计划... state[output] plan state[messages].append({role: assistant, content: plan}) return state3.4 第四步将节点添加到图中workflow.add_node(classify, classify_intent) workflow.add_node(get_info, get_travel_info) workflow.add_node(collect_info, collect_planning_info) workflow.add_node(generate_plan, generate_travel_plan)3.5 第五步定义边和路由逻辑这是最关键的一步决定了流程的走向。# 设置入口点从 classify 节点开始 workflow.set_entry_point(classify) # 从 classify 节点出来的边根据 state[‘stage’] 的值决定下一步 def route_after_classify(state: AgentState) - str: # 这个函数返回下一个节点的名字 if state[stage] get_info: return get_info else: # “collect_planning_info” return “collect_info” # 添加一条条件边 workflow.add_conditional_edges( “classify”, # 源节点 route_after_classify, # 路由函数 { # 路由函数返回值到目标节点的映射 “get_info”: “get_info”, “collect_info”: “collect_info” } ) # 从 get_info 节点出来后直接结束 workflow.add_edge(“get_info”, END) # 从 collect_info 节点出来的边如果 stage 变成了 ‘generate_plan’就去生成规划否则继续收集 def route_after_collect(state: AgentState) - str: if state[“stage”] “generate_plan”: return “generate_plan” else: return “collect_info” # 继续循环收集 workflow.add_conditional_edges(“collect_info”, route_after_collect, { “generate_plan”: “generate_plan”, “collect_info”: “collect_info” }) # 从 generate_plan 节点出来后结束 workflow.add_edge(“generate_plan”, END)3.6 第六步编译并运行图# 编译图得到一个可执行对象 app workflow.compile() # 准备初始状态 initial_state: AgentState { “input”: “我想去巴黎旅游怎么规划”, “messages”: [{“role”: “user”, “content”: “我想去巴黎旅游怎么规划”}], “stage”: “classify”, “planning_info”: {}, “output”: None } # 运行图 final_state app.invoke(initial_state) print(“最终输出:”, final_state[“output”]) print(“完整消息历史:”, final_state[“messages”])运行这个流程你会看到图如何根据classify节点的结果走向collect_info节点并在该节点内循环模拟多轮对话直到信息收集完毕再跳转到generate_plan节点最后结束。这个例子虽然简单但已经包含了 LangGraph 最核心的要素。你看到了状态如何流动、条件边如何实现分支、以及节点如何通过修改stage字段来驱动循环。4. 进阶实战构建多智能体协作系统单智能体展示了 LangGraph 管理状态和流程的能力。而它的真正威力在于优雅地编排多个智能体协同工作。多智能体不是简单地把几个 LangChain Agent 串起来而是需要解决职责划分、通信协议、冲突消解、统一状态管理。我们设计一个“技术问答系统”包含三个智能体分析员Analyzer分析用户问题拆解子任务并决定派发给哪个专家。代码专家Code Expert负责解决编程、语法、调试类问题。概念专家Concept Expert负责解释技术概念、原理、架构类问题。它们在一个 LangGraph 的调度下工作。4.1 设计多智能体状态状态需要容纳多个智能体的交互上下文。class MultiAgentState(TypedDict): user_query: str # 所有智能体的对话历史可以是分线程的这里简化用全局 messages: List[Dict] # 当前活跃的智能体名称 current_agent: str # 分析员对问题的分析结果 analysis: Dict[str, Any] # 各个专家给出的答案 answers: Dict[str, str] # key: 专家名 value: 答案 # 最终整合后的答案 final_answer: Optional[str]4.2 实现智能体节点每个智能体都是一个独立的节点它们有清晰的输入输出契约。def analyzer_agent(state: MultiAgentState) - MultiAgentState: query state[“user_query”] # 模拟分析过程判断问题类型分配专家 if “error” in query.lower() or “debug” in query.lower() or “code” in query.lower(): state[“analysis”] {“type”: “code_problem”, “assigned_expert”: “code_expert”} else: state[“analysis”] {“type”: “concept_explanation”, “assigned_expert”: “concept_expert”} state[“current_agent”] state[“analysis”][“assigned_expert”] state[“messages”].append({“role”: “analyzer”, “content”: f”分析完成分配给了 {state[‘current_agent’]}”}) return state def code_expert_agent(state: MultiAgentState) - MultiAgentState: query state[“user_query”] # 模拟代码专家工作 answer f”[代码专家] 对于问题 ‘{query}’建议检查语法示例代码print(‘Hello’)。“ state[“answers”][“code_expert”] answer state[“messages”].append({“role”: “code_expert”, “content”: answer}) state[“current_agent”] “synthesizer” # 假设下一个节点是合成器 return state def concept_expert_agent(state: MultiAgentState) - MultiAgentState: query state[“user_query”] # 模拟概念专家工作 answer f”[概念专家] ‘{query}’ 的核心原理是…“ state[“answers”][“concept_expert”] answer state[“messages”].append({“role”: “concept_expert”, “content”: answer}) state[“current_agent”] “synthesizer” return state def synthesizer_agent(state: MultiAgentState) - MultiAgentState: # 合成器汇总各个专家的答案形成最终回复 all_answers “\n”.join(state[“answers”].values()) final_answer f”综合解答\n{all_answers}“ state[“final_answer”] final_answer state[“messages”].append({“role”: “synthesizer”, “content”: “答案已整合。”}) return state4.3 编排多智能体图图的编排逻辑体现了智能体的协作协议。multi_agent_workflow StateGraph(MultiAgentState) # 添加节点 multi_agent_workflow.add_node(“analyzer”, analyzer_agent) multi_agent_workflow.add_node(“code_expert”, code_expert_agent) multi_agent_workflow.add_node(“concept_expert”, concept_expert_agent) multi_agent_workflow.add_node(“synthesizer”, synthesizer_agent) # 设置入口 multi_agent_workflow.set_entry_point(“analyzer”) # 路由逻辑分析员根据分析结果派发 def router_after_analyze(state: MultiAgentState) - str: return state[“analysis”].get(“assigned_expert”, “code_expert”) # 默认派给代码专家 multi_agent_workflow.add_conditional_edges(“analyzer”, router_after_analyze, { “code_expert”: “code_expert”, “concept_expert”: “concept_expert” }) # 专家完成后都去合成器 multi_agent_workflow.add_edge(“code_expert”, “synthesizer”) multi_agent_workflow.add_edge(“concept_expert”, “synthesizer”) # 合成器完成后结束 multi_agent_workflow.add_edge(“synthesizer”, END) # 编译图 multi_agent_app multi_agent_workflow.compile()运行这个多智能体图你会清晰地看到控制流分析员 - (代码专家 或 概念专家) - 合成器 - 结束。每个智能体只关心自己的任务状态的流转和路由由 LangGraph 框架负责。这种架构的扩展性很强要新增一个“文档专家”只需要添加一个节点并在分析员的路由逻辑里增加一个分支即可。5. 避坑指南从 Demo 到生产环境的关键跨越能用 LangGraph 跑通一个 Demo 不难但要让它在生产环境中稳定、可靠、可维护地运行需要注意以下几个关键点这些往往是新手最容易踩坑的地方。5.1 状态设计的陷阱避免过度复杂和冲突状态是共享的设计不好就会变成“共享灾难”。保持扁平结构清晰不要嵌套过深的字典。使用 Pydantic 模型强制类型和结构。明确字段的归属和更新规则哪个节点可以修改哪个字段是覆盖、追加还是合并对于列表或字典务必在创建图时通过StateGraph(…, config{“recursion_limit”: 50})或自定义缩减器来定义合并行为否则在多路径下可能出错。为状态设计版本如果流程可能迭代考虑在状态中加入一个version字段便于后续兼容性处理。5.2 错误处理与持久化流程不能“断”在 Demo 中我们假设所有节点都成功。现实中API 会失败代码会有 Bug。节点级 Try-Catch在每个节点函数内部进行细致的异常捕获。发生错误时不要直接崩溃而是将错误信息写入状态例如state[“error”] str(e)并修改状态让流程能路由到一个“错误处理节点”。设置检查点对于长时运行流程需要定期将状态持久化保存到数据库或文件。LangGraph 本身不提供自动持久化你需要在关键节点后手动调用持久化函数。或者利用 LangGraph 的“中断”和“检查点”机制。这涉及到更高级的checkpointer配置它允许你暂停图执行并将状态保存下来后续可以从断点恢复。这是实现“人工介入”或“长时工作流”的核心。超时控制对于调用外部 API 的节点务必设置超时。可以在节点逻辑内实现或者在图的外部调用层如app.invoke(…, timeout30)控制。5.3 调试与监控看清流程的每一步当图变得复杂调试不能只靠print。善用app.get_graph().draw_mermaid()这个功能能生成 Mermaid 图表可视化你的工作流。在调整图结构时非常有用。结构化日志在每个节点的开始和结束将关键状态变更、节点名、时间戳记录到结构化日志系统如 JSON 格式文件或日志服务。这比打印整个状态更清晰。追踪执行路径app.invoke()返回的最终状态包含了所有中间状态吗默认不包含。你可以通过配置或使用app.stream()来获取每一步的状态流这对于调试和审计至关重要。5.4 性能与扩展性当图变得庞大时避免阻塞操作如果一个节点需要执行长时间同步操作如处理大文件会阻塞整个图。考虑将其改为异步节点如果 LangGraph 支持或者将该操作剥离为异步任务节点只负责触发和轮询结果。节点的纯度与副作用尽量让节点函数保持“相对纯净”即输出由输入状态决定。将严重的副作用如发送邮件、写入核心数据库放在图的最末端或特定节点并做好事务和重试。图的复杂度避免创建具有数百个节点和复杂条件边的图这会使理解和调试变得极其困难。尝试将大图拆分为多个子图通过“子图节点”进行组合。LangGraph 支持将已编译的图作为一个节点嵌入到另一个图中。5.5 与 LangChain 的整合不要非此即彼很多人问用了 LangGraph 还要用 LangChain 吗答案是可以而且经常一起用。LangChain 作为“工具包”LangChain 提供了丰富的集成模型、向量库、工具。你可以在 LangGraph 的节点函数中自由地使用 LangChain 的LLMChain、Tool、Retriever等组件。把 LangChain 看作乐高积木LangGraph 是搭建积木的图纸和控制器。典型模式在 LangGraph 节点内使用 LangChain 的AgentExecutor来运行一个具备工具调用能力的智能体。这样你既拥有了 LangGraph 的流程控制能力又利用了 LangChain 的生态集成。6. 总结LangGraph 带来的范式转变回顾开头的那个团队遇到的问题本质上是在用“链式思维”解决“图式问题”。LangGraph 提供的不仅仅是一套新的 API更是一种新的范式它要求我们显式地设计状态迫使你在编码前就想清楚流程中需要记忆和传递哪些信息。这本身就是一种架构设计。显式地定义流程拓扑用代码画出流程图让业务逻辑的路径一目了然不再是隐藏在复杂的if-else和循环变量中。将“决策”与“执行”分离路由逻辑边独立于节点业务逻辑使得变更流程走向变得非常容易只需修改路由函数而无需深入每个节点。拥抱中断和持久化为构建需要人工审核、等待外部事件、或运行时间极长的复杂业务系统提供了可能。所以学习 LangGraph 的最终目的不是多掌握一个库而是掌握这种“以状态为中心、以图为蓝图”的复杂系统编排思想。当你下次再面对一个涉及多步骤、多分支、多角色、长时运行的业务场景时不妨先问自己这个流程我能用一张图画出来吗如果能那么 LangGraph 就是你最趁手的武器。从今天开始忘掉简单的“链”尝试用“图”的视角来构建你的下一个智能应用。先从定义一个清晰的State开始。