如果你还在犹豫“多智能体到底是不是伪需求”或者收藏了一堆 LangGraph 资料却始终没有跑通一个完整的示例那么这篇文章就是为你准备的。先说一个判断**LangGraph 不是 LangChain 的简单升级而是把大模型应用从“单次调用”推进到“可编排状态机”的关键基础设施。**在这个体系里多智能体不是噱头而是解决复杂任务拆解、工具调用编排、人工审核介入、长流程状态恢复等真实问题的一套工程化方案。这篇文章会从多智能体架构讲起逐步拆解 LangGraph 的核心组件最后用可复制的代码带你从零构建一个带条件路由、子图和并行分支的多智能体应用。本文不会教你“背概念”而是希望你读完能回答三个问题LangGraph 为什么值得学它的核心组件是怎么配合工作的如果我只想做一个最小可用项目代码该怎么写如果这三个问题正是你关心的建议先收藏再跟着实操。1. 为什么你需要关注 LangGraph 多智能体1.1 单个大模型调用解决不了的问题很多开发者第一次接触大模型应用开发时最先写的是这样的代码from openai import OpenAI client OpenAI() response client.chat.completions.create( modelgpt-4o, messages[{role: user, content: 写一份周报}], ) print(response.choices[0].message.content)这段代码本身没有问题但它只能处理“单次输入、单次输出”的简单请求。一旦业务场景变成下面这样单次调用就撑不住了用户输入一个问题需要先判断该调用哪个领域专家模型。需要从多个数据源检索资料再汇总成一份报告。生成结果之前需要人工审核确认。长任务执行到一半失败重新恢复后要继续之前的进度。这些问题本质上不是“让模型更聪明”能解决的而是需要一套工作流编排机制让多个模型调用、工具调用和人工节点按顺序或条件组合起来。LangGraph 就是为这类场景设计的。1.2 多智能体不是“多个模型聊天”一个常见的误区是多智能体系统就是让几个 AI 角色互相对话谁都能当“智能体”。实际上工程意义上的多智能体系统包含四个核心要素要素说明多个能力边界明确的智能体每个智能体负责一类任务而不是全部任务明确的调度/路由机制决定某个请求应该进入哪个智能体共享或可传递的状态智能体之间需要传递中间结果可恢复的执行流程支持暂停、恢复、回滚和检查点如果只是让两个模型互相聊“你觉得呢”那只是聊天不是多智能体。真正的多智能体应用必须是一个有状态的、可控制的、可观测的工作流。1.3 LangGraph 在这个生态中的定位LangGraph 是由 LangChain 团队推出的框架专门用于构建有状态、可编排的大模型应用。它与 LangChain 的核心区别在于【LangChain】更偏向提供模型调用、Prompt 模板、工具封装等基础能力。【LangGraph】更侧重定义图结构节点Node做什么、边Edge怎么走、状态State怎么流转。说得直白一点LangChain 是积木LangGraph 是图纸。你要建一座房子积木决定你能用什么材料图纸决定房子长什么样。2. LangGraph 核心概念详解在进入代码之前必须先把 LangGraph 的五个核心组件讲清楚。这几个概念贯穿后续所有实战代码。2.1 State全局状态State 是 LangGraph 的灵魂。它表示工作流在任意时刻的“快照”所有节点都从 State 读取输入并把输出写回 State。from typing_extensions import TypedDict class AgentState(TypedDict): messages: list current_step: str result: str你可以把 State 理解为前端里的全局 Store只是这里的 Store 不止存界面状态还存整个工作流的中间数据。2.2 Node节点Node 是工作流里的一个执行单元可以是大模型调用、工具函数、API 请求也可以只是一个普通 Python 函数。def call_model(state: AgentState) - AgentState: # 这里是节点逻辑 return {result: some_result}每个节点本质上就是“输入 State输出 State 增量”的函数。2.3 Edge边Edge 定义节点之间的连接关系。分为普通边和条件边【普通边】上一个节点执行完无条件进入下一个节点。【条件边】根据当前 State 或节点返回值决定进入哪个分支。2.4 Conditional Edge条件路由条件路由是 LangGraph 最强大的能力之一。它允许你根据某个字段的值动态选择下一个节点。from langgraph.graph import END, START, StateGraph def route_by_intent(state: AgentState) - str: if 天气 in state[messages][-1]: return weather_agent return general_agent2.5 Checkpoint检查点Checkpoint 让工作流具备“记忆”和“恢复能力”。你可以把每一步状态保存到内存或数据库即使中间某个节点失败也可以从上一个检查点继续执行。from langgraph.checkpoint.memory import InMemorySaver checkpointer InMemorySaver()3. 环境准备与 LangGraph 安装3.1 版本与环境要求LangGraph 是一个 Python 框架目前同时支持 JavaScript/TypeScript 版本。本文以 Python 为例。建议使用 Python 3.10 及以上版本具体小版本以你本机环境为准。3.2 安装 LangGraph推荐使用 pip 安装pip install langgraph如果需要调用 OpenAI 或其他模型还要安装对应 SDKpip install openai如果你使用的是国产大模型或本地部署模型只要接口兼容 OpenAI 格式也可以直接通过配置 base_url 接入。3.3 确认安装成功python -c import langgraph; print(langgraph.__version__)如果输出版本号说明安装成功。注意LangGraph 版本更新很快API 可能会有细微变化。本文的代码以较新的稳定版本为参考如果你使用的版本过旧或过新请以官方文档为准。4. 从零构建第一个 LangGraph 多智能体应用4.1 业务场景我们构造一个非常典型的多智能体场景一个“客服工单分流系统”。用户提交一句问题系统先做意图识别然后根据意图分配到不同的智能体如果涉及订单问题转给订单处理智能体。如果涉及技术问题转给技术支持智能体。如果意图不明确转给通用助手。这个场景虽然简单但已经包含了多智能体的核心要素多个专用节点、条件路由、共享状态。4.2 定义状态与节点创建一个文件agent_demo.pyfrom typing_extensions import TypedDict from langgraph.graph import END, START, StateGraph class AgentState(TypedDict): user_input: str intent: str final_answer: str def intent_node(state: AgentState) - AgentState: 模拟意图识别实际项目中可以用大模型调用代替 text state[user_input] if 订单 in text or 发货 in text: intent order elif 报错 in text or 无法运行 in text or bug in text: intent tech else: intent general return {intent: intent} def order_agent(state: AgentState) - AgentState: 订单处理智能体 return {final_answer: f【订单客服】收到你的问题{state[user_input]}我们会尽快核查订单状态。} def tech_agent(state: AgentState) - AgentState: 技术支持智能体 return {final_answer: f【技术支持】收到你的问题{state[user_input]}请提供完整报错信息我们将协助排查。} def general_agent(state: AgentState) - AgentState: 通用助手智能体 return {final_answer: f【通用助手】收到你的问题{state[user_input]}建议联系人工客服获得更多帮助。}这里有一点需要特别注意每个节点函数必须返回一个字典字典中的键名要与 State 定义对应。LangGraph 会把这个返回值合并到全局 State 中。4.3 定义条件路由函数条件路由函数是 LangGraph 中决定“下一步去哪”的关键。def route_by_intent(state: AgentState) - str: 根据意图字段决定进入哪个智能体节点 intent state[intent] if intent order: return order_agent elif intent tech: return tech_agent else: return general_agent4.4 构建图并运行# 构建图 builder StateGraph(AgentState) # 添加节点 builder.add_node(intent_node, intent_node) builder.add_node(order_agent, order_agent) builder.add_node(tech_agent, tech_agent) builder.add_node(general_agent, general_agent) # 添加边 builder.add_edge(START, intent_node) builder.add_conditional_edges( intent_node, route_by_intent, { order_agent: order_agent, tech_agent: tech_agent, general_agent: general_agent, }, ) builder.add_edge(order_agent, END) builder.add_edge(tech_agent, END) builder.add_edge(general_agent, END) # 编译图 graph builder.compile() # 运行 result graph.invoke({user_input: 我的订单什么时候发货}) print(意图, result[intent]) print(回答, result[final_answer])4.5 运行与验证python agent_demo.py预期输出意图 order 回答 【订单客服】收到你的问题我的订单什么时候发货我们会尽快核查订单状态。再测试一个技术问题result graph.invoke({user_input: 程序运行时报错怎么办}) print(意图, result[intent]) print(回答, result[final_answer])预期输出意图 tech 回答 【技术支持】收到你的问题程序运行时报错怎么办请提供完整报错信息我们将协助排查。如果看到这样的输出说明第一个多智能体工作流已经成功跑通了。5. 接入真实大模型让意图识别变成模型能力5.1 为什么需要用大模型做意图识别上面示例中意图识别使用的是字符串匹配效果非常有限。真实项目中你应该让大模型来承担意图识别任务这样可以处理更复杂、更模糊的表达。5.2 改造意图识别节点from langchain_openai import ChatOpenAI llm ChatOpenAI( modelgpt-4o-mini, temperature0, base_urlhttps://api.openai.com/v1, # 按你的服务商配置 api_keyyour-api-key, ) def intent_node_with_llm(state: AgentState) - AgentState: prompt f 你是意图识别引擎。用户输入如下 {state[user_input]} 请判断该输入属于哪种意图只能输出三个选项之一 - order订单、物流、售后相关 - tech技术报错、代码问题、使用故障 - general其他情况 输出格式只输出意图单词不要有其他内容。 response llm.invoke(prompt) intent response.content.strip().lower() if intent not in [order, tech, general]: intent general return {intent: intent}注意把temperature设为 0让模型输出更稳定。在 Prompt 中限定输出格式避免模型返回多行解释。即使模型输出了意外内容也要做兜底处理。5.3 接口兼容的模型都可以接入如果你的项目使用的是国产大模型或本地部署模型只要服务商提供了 OpenAI 兼容接口就可以这样配置llm ChatOpenAI( modelyour-model-name, api_keyyour-api-key, base_urlhttps://your-model-endpoint.com/v1, )这里的base_url指向你的模型服务地址即可。LangGraph 不关心底层模型是谁它只关心如何编排这些模型节点的调用关系。6. 深入 Condition Edge分支控制与循环检测6.1 多分支条件路由实际业务中条件路由往往不是“一次判断、一次分流”这么简单而是会出现多级分支、循环重试甚至动态决定是否终止的情况。比如在意图识别之后我们还想加一个“回答质量评估”节点如果评估结果不合格则重新调用模型生成一次答案。def quality_check(state: AgentState) - AgentState: answer state[final_answer] # 简单的质量规则答案长度太短就判为不合格 if len(answer) 10: return {need_retry: True} return {need_retry: False} def route_by_quality(state: AgentState) - str: if state.get(need_retry): return retry_node return END这样图结构就出现了“环”LangGraph 是支持这种循环结构的。但与普通编程不同LangGraph 的循环需要小心处理否则可能出现无限循环。6.2 循环检测与最大步数限制LangGraph 在编译图的时候会做基础的循环检测但不会阻止合法的循环。运行时我们可以通过参数控制最大执行步数config {recursion_limit: 10} result graph.invoke({user_input: 测试}, configconfig)如果工作流执行超过 10 个节点步骤LangGraph 会抛出异常提醒你可能存在无限循环。这是一个非常实用的保护机制。在生产环境中强烈建议显式设置recursion_limit。6.3 一个带重试的完整图示例from typing_extensions import TypedDict from langgraph.graph import END, START, StateGraph class RetryState(TypedDict): user_input: str final_answer: str need_retry: bool def generate_answer(state: RetryState) - RetryState: answer f针对问题「{state[user_input]}」的自动回复 return {final_answer: answer} def quality_check(state: RetryState) - RetryState: answer state[final_answer] if len(answer) 20: return {need_retry: True} return {need_retry: False} def retry_node(state: RetryState) - RetryState: return {final_answer: state[final_answer] 已补充详细说明} def route_by_quality(state: RetryState) - str: if state.get(need_retry): return retry_node return END builder StateGraph(RetryState) builder.add_node(generate_answer, generate_answer) builder.add_node(quality_check, quality_check) builder.add_node(retry_node, retry_node) builder.add_edge(START, generate_answer) builder.add_edge(generate_answer, quality_check) builder.add_conditional_edges(quality_check, route_by_quality, {retry_node: retry_node, END: END}) builder.add_edge(retry_node, quality_check) graph builder.compile() config {recursion_limit: 5} result graph.invoke({user_input: 你好}, configconfig) print(result)这个示例展示了 LangGraph 中非常核心的能力循环 条件跳出 状态更新。这类结构在真实项目中非常常见比如 AI 生成内容后的格式校验、合规检查、安全审查都可以用类似方式实现。7. 子图Subgraph与并行分支7.1 什么是子图当工作流变得复杂时不可能把所有节点都平铺在同一个图里。子图允许你把一部分节点封装成一个独立的 Graph再作为上层图的一个节点使用。打个比方主图是公司的整体流程子图是某个部门内部的详细流程。对外其他部门只看到结果不关心内部细节。7.2 子图实战示例假设我们要做一个“技术问答智能体”其中“日志分析”需要作为独立子图实现因为日志分析内部包含读取日志、提取错误、生成诊断建议三个步骤。from typing_extensions import TypedDict from langgraph.graph import END, START, StateGraph class LogSubGraphState(TypedDict): log_text: str error_summary: str def read_log(state: LogSubGraphState) - LogSubGraphState: return {error_summary: f从日志中提取到关键错误{state[log_text][:20]}} def extract_error(state: LogSubGraphState) - LogSubGraphState: return {error_summary: state[error_summary] 错误类型运行时异常} def generate_diagnosis(state: LogSubGraphState) - LogSubGraphState: return {error_summary: state[error_summary] 建议检查内存配置并重启服务。} log_subgraph_builder StateGraph(LogSubGraphState) log_subgraph_builder.add_node(read_log, read_log) log_subgraph_builder.add_node(extract_error, extract_error) log_subgraph_builder.add_node(generate_diagnosis, generate_diagnosis) log_subgraph_builder.add_edge(START, read_log) log_subgraph_builder.add_edge(read_log, extract_error) log_subgraph_builder.add_edge(extract_error, generate_diagnosis) log_subgraph_builder.add_edge(generate_diagnosis, END) log_subgraph log_subgraph_builder.compile()7.3 把子图接入主图class MainState(TypedDict): question: str log_text: str diagnosis: str def diagnostic_node(state: MainState) - MainState: sub_result log_subgraph.invoke({log_text: state[log_text]}) return {diagnosis: sub_result[error_summary]} def answer_node(state: MainState) - MainState: return {diagnosis: f最终答复{state[diagnosis]}} main_builder StateGraph(MainState) main_builder.add_node(diagnostic_node, diagnostic_node) main_builder.add_node(answer_node, answer_node) main_builder.add_edge(START, diagnostic_node) main_builder.add_edge(diagnostic_node, answer_node) main_builder.add_edge(answer_node, END) main_graph main_builder.compile() result main_graph.invoke({question: 服务挂了怎么办, log_text: OutOfMemoryError at com.example.Main}) print(result[diagnosis])子图的价值在于大型项目可以按业务模块拆分图定义。子图可以被复用比如日志分析子图可以被多个主图调用。调试时只需要关注当前子图范围问题定位更清晰。7.4 并行分支缩短任务耗时LangGraph 支持在一个节点后派出多个并行分支等待所有分支完成后合并结果。比如做市场分析时需要同时抓取竞品信息、用户评价和销售数据from langgraph.graph import END, START, StateGraph class ParallelState(TypedDict): question: str competitor_result: str user_review_result: str sale_result: str final_report: str def competitor_task(state: ParallelState) - ParallelState: return {competitor_result: 竞品信息A 产品主打性价比B 产品主打高端体验} def user_review_task(state: ParallelState) - ParallelState: return {user_review_result: 用户评价对 A 产品好评集中在价格B 产品好评集中在设计} def sale_task(state: ParallelState) - ParallelState: return {sale_result: 销售数据A 产品近一月销量 1000 件B 产品近一月销量 800 件} def merge_report(state: ParallelState) - ParallelState: report ( f竞品{state[competitor_result]}\n f用户{state[user_review_result]}\n f销售{state[sale_result]} ) return {final_report: report} parallel_builder StateGraph(ParallelState) parallel_builder.add_node(competitor_task, competitor_task) parallel_builder.add_node(user_review_task, user_review_task) parallel_builder.add_node(sale_task, sale_task) parallel_builder.add_node(merge_report, merge_report) parallel_builder.add_edge(START, competitor_task) parallel_builder.add_edge(START, user_review_task) parallel_builder.add_edge(START, sale_task) parallel_builder.add_edge(competitor_task, merge_report) parallel_builder.add_edge(user_review_task, merge_report) parallel_builder.add_edge(sale_task, merge_report) parallel_builder.add_edge(merge_report, END) parallel_graph parallel_builder.compile() result parallel_graph.invoke({question: 近期市场分析}) print(result[final_report])并行分支的关键在于多个节点都连接到同一个聚合节点。LangGraph 会等待所有上游分支完成后才执行聚合节点不需要你自己写多线程代码。8. Checkpoint让工作流具备记忆与恢复能力8.1 Checkpoint 解决了什么问题多智能体工作流往往需要多轮交互。比如用户第一句话是“帮我查一下订单”第二句话是“顺便改一下收货地址”。如果没有状态保存机制第二句话就失去了上下文。Checkpoint 解决了这个问题。它把每一步 State 保存到检查点存储中你可以随时恢复到任意历史步骤。8.2 使用 InMemorySaver 实现记忆from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import END, START, StateGraph class ConversationState(TypedDict): user_input: str response: str history: list def respond_node(state: ConversationState) - ConversationState: history state.get(history, []) history.append({user: state[user_input]}) response f这是第 {len(history)} 轮对话你的问题是{state[user_input]} history.append({ai: response}) return {response: response, history: history} checkpointer InMemorySaver() builder StateGraph(ConversationState) builder.add_node(respond_node, respond_node) builder.add_edge(START, respond_node) builder.add_edge(respond_node, END) graph builder.compile(checkpointercheckpointer) # 第一轮对话 result1 graph.invoke( {user_input: 你好}, config{configurable: {thread_id: user-123}} ) print(result1[response]) # 第二轮对话 result2 graph.invoke( {user_input: 帮我查订单}, config{configurable: {thread_id: user-123}} ) print(result2[response])注意thread_id的使用。它相当于一个会话标识同一个thread_id下的多轮调用会共享状态。8.3 生产环境 Checkpoint 选型InMemorySaver只适合测试和单机场景。生产环境中建议使用持久化存储LangGraph 官方提供了 Postgres 等存储方案的适配你也可以把 Checkpoint 数据写入 Redis 或自研存储。核心原则是State 具备可恢复性是生产级工作流的基本要求。9. 将 Checkpoint 内容传入大模型上下文9.1 一个容易被忽略的问题很多人在使用 LangGraph 做对话应用时会直接拿state[messages]作为模型上下文。但如果你使用了 Checkpointmessages 可能已经被序列化或包含了一些内部字段直接传入模型并不合适。更稳妥的做法是从 Checkpoint 中取回历史对话摘要再手动组装成模型需要的 messages 格式。9.2 示例从 Checkpoint 构建模型上下文def build_context_from_history(state: ConversationState) - list[dict]: context [] for item in state.get(history, []): if user in item: context.append({role: user, content: item[user]}) elif ai in item: context.append({role: assistant, content: item[ai]}) return context # 使用方法 def respond_with_context(state: ConversationState) - ConversationState: history state.get(history, []) context build_context_from_history(state) context.append({role: user, content: state[user_input]}) # 此时把 context 传给大模型即可 response f已收到当前上下文共 {len(context)} 条消息。 history.append({user: state[user_input]}) history.append({ai: response}) return {response: response, history: history}要点是不要直接拿内部 State 结构当模型输入要显式组装。这样可以避免字段污染也方便你接入不同的模型接口。10. 常见问题与排查思路问题现象可能原因排查方式解决方案安装失败依赖版本冲突查看 pip 错误信息使用pip check在干净虚拟环境中安装统一 langchain 相关包版本节点返回值没有被更新返回的 dict 键名与 State 定义不一致打印 State 内容确认键名严格使用 State 定义的键名运行时报 KeyError某节点读取了不存在的键在节点函数中打印state在节点函数开头使用state.get(key)而不是state[key]工作流无限循环条件边始终返回同一个节点检查路由函数的返回值和节点映射设置recursion_limit并检查条件逻辑多轮对话不记住上下文没有配置 checkpointer 或 thread_id 不一致检查编译时是否传入 checkpointer编译时传入checkpointer...invoke 时传入一致 thread_id大模型响应格式不符合预期Prompt 约束不足打印模型原始输出在 Prompt 中限定输出格式并增加兜底解析并行分支结果丢失聚合节点读取键名错误打印聚合节点前的 State 内容核对每个分支节点返回的键名是否与 State 定义一致11. LangGraph 多智能体最佳实践建议11.1 图设计原则不要在单个图里堆太多节点。经验阈值是如果主图节点超过 10 个建议拆分子图。每个子图只负责一个领域比如“日志诊断子图”“订单处理子图”“合规审查子图”。11.2 状态设计原则State 字段越少越好。只放需要跨节点共享的数据不要把所有中间变量都塞进 State。比较常见的做法是将中间变量用局部变量保存只把关键结果放入 State。11.3 安全边界大模型生成内容可能包含违规信息生产环境应加入内容审核节点。涉及用户隐私数据时State 中不应保存明文敏感信息必要时应脱敏后再写入 Checkpoint。工具调用节点应遵循最小权限原则不要给智能体过宽的权限边界。11.4 可观测性在关键节点增加日志输出记录节点名、输入 State 摘要、输出结果。LangGraph 生态中有 LangSmith 等观测工具但在没有接入的情况下最简单的做法是def node_with_log(state: State) - State: print(f进入节点node_with_log) print(f输入字段{state.keys()}) result do_something(state) print(f节点输出{result}) return result11.5 从 Java 或其他语言接入 LangGraph很多同学问“Java 怎么使用 LangGraph”。目前 LangGraph 官方有 JavaScript 版本如果你的团队是 Java 技术栈更推荐的方式是把 LangGraph 工作流封装成独立服务通过 HTTP API 暴露给 Java 后端调用。这样既能使用 LangGraph 的编排能力又不会和现有 Java 体系冲突。12. 总结与下一步学习建议LangGraph 多智能体开发的核心不是“让模型变强”而是把大模型调用、工具调用、人工审核、状态恢复等环节用工程化方式组织起来。State、Node、Edge、Conditional Edge、Checkpoint 这五个组件构成了 LangGraph 的最小认知框架本文的示例基本都围绕这五个组件展开。下一步你可以做三件事第一把本文的客服工单分流示例改造成使用真实大模型并替换成你熟悉的数据源和工具调用。这一步能帮你真正理解 State 和 Condition Edge 的配合。第二尝试在一个项目中加入子图和并行分支体会模块化拆分带来的维护收益。第三把内存 Checkpoint 换成持久化方案并接入你自己的日志体系让工作流从“能跑”变成“可运维”。LangGraph 的版本迭代很快新概念也在持续出现但核心的图编排思想不会变。只要把这一套状态机和条件路由的思维方式掌握了不管未来框架怎么升级你都能快速上手。