LangChain与LangGraph两层架构：从积木到编排，落地企业级多Agent智能体流程

最近这半年，我陆续把几个“单 Agent 演示”升级成真正跑在业务里的多 Agent 流程，最大的感受是：单个 Agent 会写的人很多，能把多个 Agent 顺畅编排起来的人很少。问题几乎都出在同一个地方——大家把 LangChain 和 LangGraph 当成同一个层面的东西来用。LangChain 更像是积木层，它负责提供模型调用、工具封装、记忆、检索这些零部件；LangGraph 才是编排层，它负责决定这些积木怎么连、怎么走、什么时候停下等人。这篇内容我按这个分法，把两层架构的定位、核心机制、一套可复用的企业级流程代码，以及我在真实项目里踩过的几个典型的坑，一次性讲清楚。

适合看这篇文章的人：已经有 Python 基础、调用过大模型 API、正在用 LangChain 写业务脚本，但发现流程一旦出现分支、循环、人工审批就不知道怎么组织代码。我会尽量从“为什么需要两层”讲到“具体怎么写”，让你看完能直接往自己的项目里套。

1. 为什么需要“积木层”与“编排层”两层架构

很多开发者接触 LangChain，第一个上手的就是LLMChain或者SequentialChain，写法很直观：第一步让模型做意图识别，第二步把结果传给查询工具，第三步把工具结果再丢给模型生成回复。这种链式调用在固定顺序的场景里没有大问题，但只要业务需求稍微复杂一点，代码结构就会变得很痛苦。

1.1 链式调用的边界在哪里

链式调用的本质是“数据沿着一条直线往前流”，上游的输出是下游的输入。这种模型的优点是简单，缺点也很明显：流程里只要出现一个条件分支，链路就开始失控。

举个例子。一个企业客服系统有三个能力：查订单、查物流、走售后审批。用户问“我的订单什么时候到”，正确路径是“意图识别 → 查物流 → 生成回复”。但如果用户说“这个商品我要退货”，你希望走的是“意图识别 → 查订单 → 判断是否符合退货政策 → 触发人工审批”。这两个流程前缀相同，后面完全分叉。用链式调用写，你只能在链外写一堆if控制逻辑，把不同链路拼来拼去。一旦再加一个“用户中途修改问题”的交互，代码里就会出现大量临时变量和状态标志位，项目维护成本直接上升。

数据流的问题更隐蔽。链式调用里，下游只能拿到上游的返回值，如果想在某个环节同时参考“最初的用户问题”和“中间某一步的工具结果”，就得自己把历史数据一层一层往参数里塞。这个过程极其容易出错，而且没有统一的载体。

真正在企业级系统里，最让我头疼的还不是分支，而是“流程需要停在那里等一个人”。比如退款超过一定金额，需要业务负责人审批。链式调用一次跑到底，根本不会停下来等外部输入。这就不是我写不写得出来的问题，而是这套抽象本身就是错的。

1.2 LangGraph 把流程从“链”改成了“图”

LangGraph 的核心变化是：不再用链条描述流程，而是用有向图来描述。图上的每个节点是一个计算单元，可以是大模型调用、工具执行、或者普通 Python 函数；节点之间用边连接，边上可以附加条件，决定下一步到底走哪个节点。

这个转变背后有一个关键设计：共享状态。LangGraph 里的每个节点都读写同一个状态对象，而不是只依赖上游的返回值。这就好比所有的节点都在一块黑板上写字、看字，而不是像流水线一样只跟前一个工位打交道。状态对象是结构化的，通常是一个TypedDict，里面可以放用户原始问题、中间结果、工具返回、审批状态等等。

图结构把之前链式调用做不了的事情都变成了基础能力：条件路由是图的常规操作；循环是图里的环；人工介入可以靠持久化挂起机制实现。LangGraph 在底层帮我们把执行流程管理好，上层只要定义“有哪些节点、节点之间怎么连、状态怎么更新”，它就能控制流程跑完整个生命周期。

1.3 两层架构到底怎么分工

我的经验是，不要纠结“LangChain 好还是 LangGraph 好”，它们解决的问题不在同一个维度。

LangChain 的真正价值是“组件标准化”。模型有接口封装，工具函数可以转成模型能识别的调用格式，记忆有现成的存储方案，文档检索有整套流程，输出有解析器。这些组件放在一起，就是一座积木仓库。LangGraph 的价值是“流程结构化”。它有节点、边、状态、条件路由、持久化机制，相当于一份施工图加一个项目经理。

如果说以前你写多个 Agent 是在“自己用胶水粘木头”，那引入 LangGraph 之后就相当于有了正式的施工流程，而 LangChain 里的组件就是质量统一的预制板。下面这张表是我做选型时通常会参考的：

先有这个判断，再去看技术细节，就清楚很多了。

2. 积木层 LangChain 的核心组件拆解

LangChain 这个项目历史里，功能模块一直在变，但核心思想其实没变：给 LLM 应用开发提供可复用的标准化组件。我实际用得最多的组件有四类：模型封装、工具封装、记忆模块、检索与输出解析。

2.1 模型调用层的封装逻辑

刚开始我写大模型调用，都是直接用各家的 SDK：填密钥、拼请求体、处理返回结构。后来发现，业务里经常要换模型或者同一个流程要同时用到不同能力的模型，直接写 SDK 的问题就暴露了：每家接口风格还不一样，代码到处是if provider == "xxx"之类的分支。

LangChain 对模型调用做了一层统一封装，把“发请求、收响应”的差异屏蔽掉。你在代码里只需要面向统一的接口写业务逻辑，后面切换服务商或者调整模型版本，改动范围很小。例如：

from langchain.llms import LLMClient # 这里用一个示意接口表达：按实际接入的服务商填写配置 llm = LLMClient(provider="your_provider", model="your_model", temperature=0.1) resp = llm.invoke("请判断这个用户意图：...")

这层封装看起来简单，实际帮我省了非常多零碎的转换代码。更重要的是，LangChain 把“模型输出格式”也做了抽象，让后续接工具调用和输出解析时可以统一处理。

2.2 工具与函数调用的标准化

企业级 Agent 不能只靠模型自带知识，一定要接内部系统。LangChain 里把这类能力设计成“工具”：一个普通函数加上名称、描述、参数说明，框架可以自动把它转成模型可识别的 function calling 格式。

工具定义的重点不是函数本身，而是“描述”和“参数 schema”。描述写得好不好，直接决定了模型在什么场景下会调用这个工具。我的经验是，描述里一定要说清楚“这个工具解决什么问题、什么情况下不应该用”，最好再给一个典型例子。

from langchain.tools import tool @tool def query_order(order_id: str) -> str: """根据订单号查询订单基础信息，包括商品、金额、状态。 仅当用户提供明确订单号时使用；如果用户没有订单号，请先询问。 """ # 伪代码：内部去查订单系统的接口 return f"订单 {order_id}: 金额 299 元，状态 已发货"

工具函数本身可以用任意 Python 代码实现，里面请求内部服务、查数据库都可以。关键是返回值要尽量结构化，方便后面的节点去判断和展示。

2.3 记忆、检索与输出解析等周边能力

记忆这块，LangChain 提供了多种现成的记忆存储方案。实际做企业应用时，我一般不会直接用默认的内存记忆，而是会把它跟流程状态配合使用。多 Agent 场景里，当前这一步需要哪些上下文，应该由编排层的状态来控制，而不是让模型把整个对话历史全记住，那样既费 token，又容易混淆。

检索这块也很重要。企业知识库查询基本都是“先把文档切块，做向量化，再根据用户问题取相关片段”。LangChain 把加载、切分、入库、检索的流程都标准化了。换句话说，我不需要自己维护一套向量检索链路，直接把这些组件当作积木接入流程里的某个节点即可。

输出解析器是经常被低估的组件。模型返回的内容是自然语言，但企业系统往往需要结构化结果，比如{"intent": "refund", "confidence": 0.92}。LangChain 的PydanticOutputParser可以定义一个结构，把模型输出自动解析成目标对象，解析失败还能触发重试。这个能力在编排层里非常有用，因为路由判断要依赖结构化的意图字段，而不是一堆含糊的文字。

3. 编排层 LangGraph 的核心机制解析

如果说 LangChain 的积木解决了“组件从哪来”，LangGraph 解决的就是“流程怎么走”。这块我重点看四个机制：状态图、节点与边、条件路由、持久化与人工介入。

3.1 StateGraph：用共享状态替代链式传参

LangGraph 的入口是一个StateGraph，它接收一个状态类型定义。这个状态是所有节点共享的上下文。用一个TypedDict表示，例如：

from typing import TypedDict, Annotate, Literal class AgentState(TypedDict): user_query: str # 用户原始问题 intent: str # 意图识别结果 order_info: str # 订单查询结果 need_approval: bool # 是否需要人工审批 final_answer: str # 最终回复

这种共享状态的设计思路，类似团队协作时的黑板模式：任何人想了解当前局势，就去看黑板；任何人做完自己的事，就把结果写到黑板上。这样做的直接好处是：某个节点不再需要关心“上一步是谁”，它只需要声明“我需要什么字段”以及“我会写什么字段”。

这里有个容易踩坑的地方：字段更新策略。默认情况下，后写入的字段会覆盖旧值。如果多个节点需要往同一个字段追加内容，就要给字段配置 reducer 函数。比如messages字段通常使用operator.add来把新消息追加到列表里，而不是替换。

3.2 节点、边与条件路由

图的基本组成是节点和边。节点是实际执行动作的函数，边表示流转关系。普通边非常简单：A 执行完进入 B。条件边不一样，它会根据当前状态里的某些字段，动态决定进入哪个节点。这是我批量落地多 Agent 流程时最常用的机制。

举个简化例子。意图识别节点执行完后，根据state["intent"]决定是进入订单查询节点，还是进入物流查询节点，还是直接进入兜底回复节点。

from langgraph.graph import StateGraph def intent_node(state: AgentState) -> dict: # 大模型调用，识别用户意图 intent = classify(state["user_query"]) # "order_query" / "logistics_query" / "other" return {"intent": intent} def order_node(state: AgentState) -> dict: return {"order_info": query_order(state["user_query"])} def logistics_node(state: AgentState) -> dict: return {"order_info": query_logistics(state["user_query"])} def route_by_intent(state: AgentState) -> str: if state["intent"] == "order_query": return "query_order" if state["intent"] == "logistics_query": return "query_logistics" return "fallback" graph = StateGraph(AgentState) graph.add_node("intent", intent_node) graph.add_node("query_order", order_node) graph.add_node("query_logistics", logistics_node) graph.add_node("fallback", fallback_node) graph.set_entry_point("intent") graph.add_conditional_edges("intent", route_by_intent, { "query_order": "query_order", "query_logistics": "query_logistics", "fallback": "fallback", })

条件路由函数本身不负责执行节点，它只负责返回一个字符串，字符串映射到对应的节点名。这种设计非常清晰：判断逻辑集中在一个小函数里，方便单测；节点之间互相不用知道彼此内部实现。

3.3 Checkpoint 持久化与人工介入

企业级流程跟演示级流程最大的差别，是“跑一半要停下来，等外部的人来操作”。LangGraph 用 Checkpoint 机制解决这个问题。每个节点执行完，框架会把当前状态和执行位置保存下来。流程可以中断，比如执行到“人工审批”节点时停下，等待业务人员在后台点“通过”或“拒绝”，之后再从断点恢复执行。

这个能力在没有 Graph 编排框架之前，基本靠自己在数据库里存流程实例状态，然后写各种恢复逻辑。换到 LangGraph 上，框架把这些底层逻辑管理起来，你只需要在状态里增加一个字段（比如approval_status），在节点里调用interrupt语法，流程挂起，外部后续传一个 Command 进来，节点再继续往下走。

这带来的一个架构收益是：所有需要人审的环节都变成了图上的一个普通节点，流程怎么暂停、怎么恢复，跟业务逻辑解耦了。做企业系统的人应该能体会，这一点有多省心。

4. 实操：搭建一个多 Agent 企业内部支持系统

前面讲了不少概念，接下来我把一个真实场景的完整实现思路走一遍。我把它叫做“模拟项目 X”：企业内部智能客服系统，支持查订单、查物流、退换货审批。

4.1 场景设定与整体的流程设计

用户从入口进来，第一件事是意图识别。识别结果分三种：订单查询、物流查询、售后申请。订单和物流直接调内部接口返回结果；售后申请要先查订单信息，然后判断金额，超过一定数额就要走人工审批，审批之前流程必须有意识地停下来。

整体节点设计如下：

1. intent：意图识别节点，输出intent字段。
2. query_order：订单查询节点，调用内部订单工具，把结果写入状态。
3. query_logistics：物流查询节点，调用物流工具。
4. after_sales：售后处理节点，判断是否需要人工审批，需要则挂起。
5. approval_check：人工审批确认节点，模拟操作系统里的审批结果。
6. generate_reply：根据上面所有结果，生成给用户的最终回复。

边的连接逻辑是：入口进intent；根据意图走query_order/query_logistics/after_sales；售后的after_sales根据金额分支到approval_check或generate_reply；最后都汇聚到generate_reply。

4.2 定义状态、积木与节点函数

先定义状态。字段要提前规划好，因为后边所有节点的输入输出都依赖这个结构。

from typing import TypedDict from langgraph.graph import StateGraph, END class SupportState(TypedDict): user_query: str intent: str order_info: str need_approval: bool approval_status: str final_answer: str

然后把之前 LangChain 工具封装的query_order、query_logistics当作积木直接引入。这里我把“意图识别”也写成一个节点函数，内部调用一个固定的 LLM 接口。

def intent_node(state: SupportState) -> dict: prompt = f"判断用户意图，只返回一个词：order_query / logistics_query / after_sales。用户问题：{state['user_query']}" result = llm.invoke(prompt).strip() # 做一层兜底，避免模型输出杂讯 if result not in ("order_query", "logistics_query", "after_sales"): result = "order_query" return {"intent": result}

订单节点和物流节点就是调用工具函数后，把结果写到状态里：

def order_node(state: SupportState) -> dict: return {"order_info": query_order.invoke({"order_id": state["user_query"]})} def logistics_node(state: SupportState) -> dict: return {"order_info": query_logistics.invoke({"order_id": state["user_query"]})}

售后节点稍微复杂一些。它要判断是否需要人工审批：

def after_sales_node(state: SupportState) -> dict: order_info = query_order.invoke({"order_id": state["user_query"]}) # 模拟从订单信息里解析出金额；真实项目里工具应返回结构化字段 amount = extract_amount(order_info) if amount > 500: return {"order_info": order_info, "need_approval": True} return {"order_info": order_info, "need_approval": False}

审批节点里，我把流程挂起，模拟等待人工操作。实际企业级系统中，这里是等一个外部事件：审批后台点按钮，或者 OA 系统回传结果。

def approval_check_node(state: SupportState) -> dict: # 演示：此处流程会中断，由外部传入审批结果 return {"approval_status": "approved"}

最后是生成回复节点，它把状态里已经收集到的信息组装成最终回答。

def generate_reply_node(state: SupportState) -> dict: if state["intent"] == "after_sales" and state.get("approval_status") == "approved": answer = f"您的售后申请已通过审批。订单信息：{state['order_info']}" else: answer = f"查询结果：{state['order_info']}" return {"final_answer": answer}

4.3 把节点连成图并运行

有了积木和节点函数，剩下的就是组装。

builder = StateGraph(SupportState) builder.add_node("intent", intent_node) builder.add_node("query_order", order_node) builder.add_node("query_logistics", logistics_node) builder.add_node("after_sales", after_sales_node) builder.add_node("approval_check", approval_check_node) builder.add_node("generate_reply", generate_reply_node) builder.set_entry_point("intent") # 意图节点之后做条件路由 builder.add_conditional_edges( "intent", route_by_intent, { "query_order": "query_order", "query_logistics": "query_logistics", "after_sales": "after_sales", }, ) # 订单、物流、售后分别汇聚到生成回复 builder.add_edge("query_order", "generate_reply") builder.add_edge("query_logistics", "generate_reply") builder.add_edge("after_sales", "approval_check") builder.add_edge("approval_check", "generate_reply") builder.add_edge("generate_reply", END) app = builder.compile()

运行时，直接 invoke 一个初始状态：

result = app.invoke({"user_query": "我要退掉订单 123456，怎么操作"}) print(result["final_answer"])

如果在审批节点设置了interrupt，invoke 会返回一个挂起结果，你需要通过后台恢复命令继续：

result = app.invoke(Command(resume={"approval_status": "approved"}))

我用这种方法跑过好几个流程。从一开始手工管理状态字典，到后来让框架接管，代码量的下降是次要的，重要的是思维负担减轻了：我只需要想清楚“这个节点进来时要什么、出去时更新什么”，其他都由图和状态来承接。

4.4 关键边界和参数选择

有几个参数需要结合场景反复调。

第一，temperature的设置。意图识别节点我通常设 0，因为要的是稳定的分类结果；生成回复节点可以设到 0.3 左右，让语言自然一点。

第二，工具调用的超时和重试。在企业内部系统中，下游服务偶发超时是常态。我一般会给每个工具调用套一层超时控制，超时就返回一个明确的错误字符串，让模型知道“工具没调通，可以尝试追问用户或换一种表达方式”。不建议让模型无限重试，一定要设置最大重试次数。

第三，是否需要并行。在 LangGraph 里，如果两个节点之间没有依赖关系，可以并行执行。比如既查订单又查物流的场景，可以把两个节点挂在同一个上游之后，框架自动并行，能显著降低响应耗时。如果顺序执行，一个节点要等另一个跑完，整体时延是累加的。

5. 常见问题与排查技巧实录

多 Agent 流程比单 Agent 复杂，主要原因是引入了大量交互边界。接下来这几个问题，是我自己踩过之后觉得特别有代表性的。

5.1 状态字段被覆盖，导致下一步拿到的是旧数据

一个很常见的问题：两个节点都往order_info字段里写结果，但一个写的是订单金额，一个是订单物流，后执行的把先执行的覆盖了。排查起来特别隐蔽，因为代码不报错，只是最终回复内容驴唇不对马嘴。

我现在的做法是：状态字段命名按节点职责来，不要用一个笼统的字段。比如order_amount_info和logistics_info分开；必须合并时，给字段配置 reducer，显式声明追加还是合并。还有一个笨办法很有效：在所有节点入口print一下当前状态，看流转过程，你会发现 90% 的状态问题都能一眼定位。

5.2 条件路由不生效，或者流程陷入死循环

条件路由不生效，绝大多数不是图的问题，而是返回字符串跟映射键不匹配。比如模型返回了"OrderQuery"，你的映射键是"order_query"，结果就会掉进兜底。我一般在条件路由函数里强制做小写化和去空格：

def route_by_intent(state: SupportState) -> str: intent = state["intent"].strip().lower() if intent == "order_query": return "query_order" ...

死循环则通常是“循环边缺少退出条件”。比如让一个节点反复优化方案，但没有设置最大迭代次数，模型就会一直改下去。LangGraph 允许给编译后的应用设置recursion_limit，超限会抛异常。这既是保护，也是排查信号：如果经常触发限制，说明退出条件设计得不对。

5.3 工具调用失败，模型开始“一本正经地胡说”

工具调用失败的场景，经常发生在下游接口参数对不上、返回格式异常、鉴权过期。问题在于：模型拿到一个异常返回后，它不会说我失败了，而是会顺着现有信息编一个合理但错误的回复。这个在企业场景里不可接受。

我现在的标准做法是：工具函数内部捕获所有异常，返回统一结构的错误文本，例如ERROR: timeout when calling order service, please retry。生成回复节点会识别这个前缀，并向用户说明“系统暂时无法查询订单，请稍后再试”。必要的时候，把ERROR状态传到路由节点，走重试或兜底。

5.4 多 Agent 串行导致响应太慢

把多个 Agent 串起来，性能损耗不只是多几次模型调用，更多是等待时间累加。我曾经有一个流程，意图识别 1.5 秒，订单查询 1 秒，生成回复 1.8 秒，总共 4 秒多，业务方直接觉得不可接受。

优化办法有几种。第一，把没有依赖关系的节点并行，LangGraph 原生支持 DAG 并行，收益最直接。第二，给工具结果加缓存：同一个用户、同参数、短时间内的查询结果可以直接复用。第三，减少非必要模型调用：比如售后流程里已经能通过工具拿到明确金额，就别再让模型去“分析提炼”一次。我在一个项目里，靠“砍掉中间一次 LLM 调用”让整体时延降了 40%。

下面是我整理的一个常见问题速查表：

6. 实战沉淀：设计多 Agent 系统的几条经验

我现在的习惯是：无论项目大小，先把“积木层”和“编排层”分开。积木层里只放能复用的能力，比如订单查询工具、物流查询工具、某个业务规则判断函数；编排层里只放流程逻辑，比如哪个节点先执行、什么条件走哪条路、哪里需要停下等人。这个分法带来的直接效果是，业务方改流程时，我基本不用改工具代码，只改图结构，上线风险小了非常多。

另外一个体会是，不管用什么编排框架，设计阶段都要给流程画一张“节点和边”的草图。不用画得多规范，但一定要把“哪些节点有依赖、哪些节点可以并行、哪里会走分支、哪里需要人工”这几个问题答清楚。画不清楚就写代码，后面一定会返工。

如果你刚开始从单 Agent 往多 Agent 流程走，不要一上来就搭一个非常复杂的图。挑一个真实但简单的流程，比如“意图识别 → 一个查询工具 → 生成回复”，先把状态、节点、条件路由这几个基本概念跑通，再逐步加分支和人工审批。框架本身的学习曲线不陡，真正的复杂度永远来自业务场景本身。