☰
LangChain与LangGraph深度对比:从线性管道到状态机式LLM应用编排
2026/10/17 1:05:38 网站建设 项目流程

做 LLM 应用开发的人,早晚会撞上 LangChain 这个名字。我第一次接触它的时候其实挺抗拒的——明明自己写几十行代码就能调通模型接口,为什么要引入一整套抽象层?直到手里的需求从"调一次模型"变成"连续多轮、带工具、带记忆、要重试、要人工介入"的复杂流程,我才意识到问题没那么简单。这篇内容就基于我自己的入门经历,把 LangChain 和它下一代的应用编排工具 LangGraph 放在一起拆开讲,适合刚拿到模型 API Key、准备从零搭一个带知识库或带工具的对话应用的开发者。文章不追求面面俱到,只讲清楚两个框架各自解决什么问题、核心概念怎么串起来、以及我实际踩过的坑。

1. 先搞清楚一件事:LangChain 到底在解决什么问题

1.1 没有框架的时候,你自己写的是什么

很多人的起点都是这样一段代码:拿到模型服务的 API Key,用 requests 或者 openai SDK 直接调接口,传一个 system prompt 和 user message,拿到字符串后自己切分、格式化、塞进下一轮请求。

我最初就是这么干的,而且坦率说,在"只调一次模型"的场景下,这种方式没有任何问题。真正让你难受的是需求开始膨胀之后:

  • 提示词从一段变成十段,不同业务场景要拼装不同模板;
  • 模型的返回从"直接给文本"变成"给 JSON",你需要自己写解析逻辑,还要处理解析失败;
  • 对话历史要自己维护,超长时要截断、要摘要;
  • 每次调用要记录 token 消耗、要加日志、要支持流式输出;
  • 某个环节失败要自动重试,重试时提示词不能原样重发。

这些问题单个拎出来都不难,但全部堆在一起,你的业务代码里就会塞满"和业务无关的胶水逻辑"。我当时维护的代码里,光是为了把对话历史拼成模型要求的格式,就有三个版本,因为不同接口对历史消息的字段要求不一样。LangChain 最初的价值就在这里:它把"模型调用"这个动作周围所有反复出现的脏活,统一成了可组合的抽象。

1.2 LangChain 的四件套抽象

LangChain 早期最核心的抽象可以概括成四件套:Model、Prompt、OutputParser、Chain。

Model 是最底层的封装。不管是文本生成模型、对话模型还是向量化模型,LangChain 都给你一个统一接口。你换模型服务商时,业务代码不需要大改,只改构造模型的那一行。Prompt 负责提示词管理,包括模板渲染、变量注入、少样本示例的组装。OutputParser 把模型输出的字符串转成结构化对象,比如 JSON、Python dict、带格式的列表,解析失败还能触发一次"带错误信息重新生成"的补偿逻辑。Chain 则是把这些串起来的最小执行单元。

这四个概念单独看都很朴素,但组合起来已经能覆盖大部分简单应用:用户输入 → 填充模板 → 模型生成 → 解析结果 → 返回。我最早用 LLMChain 写了一个关键词提取的小工具,输入一段文本模板里写"提取以下文本中的关键词,以 JSON 数组返回",OutputParser 负责把结果解析成 list。整个过程大概二十行代码,比我自己手写解析要稳得多。

1.3 LCEL 管道语法:为什么它成了现在的标准写法

如果你现在打开 LangChain 官方文档,会发现大量|符号,这是后来的 LCEL(LangChain Expression Language)语法。它解决的是 Chain 写法过于啰嗦的问题。

早期写法是创建一个 Chain 对象,往里塞各个组件,再调用.run()或.invoke()。LCEL 则直接把组件用管道符连接:

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser # model 这里先用占位,后面会讲具体怎么接服务商 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个乐于助人的助手。"), ("user", "{input}") ]) chain = prompt | model | StrOutputParser() result = chain.invoke({"input": "你好,请介绍一下你自己"})

这个|语法不是花架子。每个被|连接的组件都实现了统一的 Runnable 协议,协议规定了四组核心方法:invoke同步调用、batch批量调用、stream流式输出,以及对应的异步版本ainvoke、abatch、astream。只要组件实现了这个协议,你就可以像拼乐高一样自由组合,并且天然获得流式、异步、批量、重试、回调这些能力。

我后来才理解为什么 LangChain 要把这套协议做成第一公民:在真实业务里,流式输出几乎是对话应用的标配,异步是服务端性能的关键,回调是日志和监控的基础。这些能力如果每个项目都自己实现一遍,成本极高。LCEL 的价值就是把它们内置到协议层,你只要按规矩组合,能力就自动有了。

2. LangGraph 不是新框架,而是换了一套写应用的方式

2.1 从"线性管线"到"状态机",这个转变很关键

LangChain 的 Chain 无论怎么组合,本质上是一条单向的、无环的管线:输入进去,按固定顺序流经各个组件,最后输出。这在流程确定的情况下没问题,但真实应用经常需要"回头路"。

举一个我自己的例子。写一个带知识库的问答机器人,第一版流程是固定的:问题进来 → 检索知识库 → 把检索结果拼进提示词 → 模型生成答案。但很快发现一个问题:模型生成的答案经常出现"编造",因为它没有依据来源。我想加一个"自我校验"环节:生成答案后,让模型判断答案是否严格基于检索到的资料,如果不是,就重新检索更相关的资料再生成一遍。这个"重新检索"就构成了一个循环,在 Chain 的模型里做不到——你必须自己在外部写 while 循环,把状态变量传来传去。

LangGraph 把这种场景变成了它的核心能力。它的运行模型是状态机:定义一个全局状态,若干节点分别对状态做修改,节点之间用边连接,边可以是普通的,也可以是条件跳转。因为状态可以被反复修改,节点之间可以形成环,所以"走回头路"是天然支持的。

2.2 核心概念拆解:State、Node、Edge

LangGraph 的三个核心概念,每一个都对应一个明确的问题。

State 是全局数据容器,通常用 TypedDict 定义。它解决的是"数据在节点之间怎么传递"的问题。比如一个检索问答的状态可以这样定义:

from typing import TypedDict, List class QAState(TypedDict): question: str retrieved_docs: List[str] answer: str attempts: int

Node 就是一个普通的 Python 函数,接收当前状态,返回对状态的增量修改。之所以强调"增量",是因为每个节点不需要返回完整状态,只需要返回它改动的字段。比如检索节点只需要返回{"retrieved_docs": docs}。

Edge 决定执行顺序。add_edge表示无条件连接,add_conditional_edge表示根据某个条件判断函数的结果走不同的分支。这个条件判断函数和节点函数一样,也是接收状态、返回一个路由标识。

我最初觉得这套东西很抽象,后来用一个小例子就通了:把状态想象成一张正在被多人协作编辑的表格,每个节点是一个编辑者,他只改自己负责的列,边是工作表上设定的流程,条件边就是"如果某列的值大于多少,就转到另一个流程分支"。

2.3 Checkpointer 与记忆:多轮对话真正需要的状态

LangGraph 引入 Checkpointer 之后,我才体会到什么叫"有状态的应用"。State 只解决了单轮执行内的数据传递,而 Checkpointer 解决的是"执行结束后状态如何保存和恢复"。

实际效果是这样的:你把一个图编译到 Checkpointer 上,调用时带上thread_id,那么这一轮执行产生的最终状态会被保存下来。下一轮调用同一个thread_id时,图从头开始执行,但初始状态不再是空的,而是上一轮结束时的状态。多轮对话的历史、用户的偏好、还没完成的步骤,都可以这样续上。

from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() graph = workflow.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "user-123"}} result = graph.invoke({"question": "我的订单状态是什么?"}, config) # 下一轮继续用同一个 thread_id,图会带着上一轮的状态执行 result2 = graph.invoke({"question": "那我能改地址吗?"}, config)

Checkpointer 还带来了另一个能力:Human-in-the-loop。在节点之间插入一个中断,让图执行到某一步时暂停,等人工确认后再继续。这个我之前在 Chain 模型里完全没法干净地实现,只能把整条管线拆成两段接口,前端先调第一段、人工确认后再调第二段,状态的一致性全靠自己维护,非常容易出错。

3. 从零实现一个客服机器人 Demo:两种写法对照

3.1 先用 LCEL 写一版:能跑但僵硬

为了说清楚两个框架的差别,我拿一个读者反馈机器人当例子。需求很简单:用户提问 → 判断是"产品咨询"还是"售后问题" → 检索对应知识库 → 生成回复。

LCEL 版本长这样:

from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI model = ChatOpenAI(model="gpt-4o-mini", temperature=0) classify_prompt = ChatPromptTemplate.from_messages([ ("system", "判断用户问题属于 [产品咨询, 售后问题] 中的哪一类,只输出类别名。"), ("user", "{question}") ]) def route_by_type(question: str): """根据分类结果选择知识库检索策略,这里简化为返回一个标签""" category = (classify_prompt | model | StrOutputParser()).invoke({"question": question}) return category.strip() answer_prompt = ChatPromptTemplate.from_messages([ ("system", "你是客服助手。请根据以下资料回答用户问题,资料中没有的信息不要编造。\n\n{context}"), ("user", "{question}") ]) def build_chain(question: str): category = route_by_type(question) # 伪代码:根据 category 去不同的知识库检索 context = f"检索到与 {category} 相关的资料若干条..." return (answer_prompt | model | StrOutputParser()).invoke({ "context": context, "question": question }) print(build_chain("你们的耳机支持降噪吗?"))

能跑,但问题很明显:分类、检索、回答是三个独立步骤,控制流全写在业务函数里,状态靠函数参数和返回值手工传递。如果后续要加"回答质量不够好就重新检索"的逻辑,只能再嵌套一层 if-else。代码会越来越像意大利面条。

3.2 再用 LangGraph 重写:加一个"重新检索"的循环

用 LangGraph 重写时,我把同样的流程建模成三个节点,外加一条条件边。状态里多了一个retry_count字段,用于控制最多重新检索几次。

from typing import TypedDict, List from langgraph.graph import StateGraph, START, END class BotState(TypedDict): question: str category: str context: str answer: str retry_count: int def classify_node(state: BotState) -> dict: category = (classify_prompt | model | StrOutputParser()).invoke({"question": state["question"]}) return {"category": category.strip(), "retry_count": state.get("retry_count", 0)} def retrieve_node(state: BotState) -> dict: # 伪代码:根据 category 检索知识库,这里演示一个弱检索结果 context = f"基于 {state['category']} 检索到的资料(第 {state['retry_count'] + 1} 次)..." return {"context": context} def answer_node(state: BotState) -> dict: answer = (answer_prompt | model | StrOutputParser()).invoke({ "context": state["context"], "question": state["question"] }) return {"answer": answer} def need_retry(state: BotState) -> str: # 简化策略:答案太短或 retry_count 没超过上限就回去重查 if len(state["answer"]) < 20 and state["retry_count"] < 2: return "retry" return "end" builder = StateGraph(BotState) builder.add_node("classify", classify_node) builder.add_node("retrieve", retrieve_node) builder.add_node("answer", answer_node) builder.add_edge(START, "classify") builder.add_edge("classify", "retrieve") builder.add_edge("retrieve", "answer") builder.add_conditional_edges("answer", need_retry, {"retry": "retrieve", "end": END}) graph = builder.compile() result = graph.invoke({"question": "你们的耳机支持降噪吗?", "retry_count": 0}) print(result["answer"])

这段代码里最关键的是最后那一条条件边:answer节点执行完后,need_retry函数检查答案长度和重试次数,如果判定需要重试,执行流会跳回retrieve节点重新检索。这就是在 Chain 里做不到的循环。注意retrieve节点返回的上下文里带上了retry_count,这样下一轮检索时可以换检索词或换策略,不至于原地打转。

3.3 运行效果与几个容易被忽略的细节

跑这个 Demo 时,我第一次感受到 State 增量更新的好处:answer_node只需要返回它改动的answer字段,其他字段会被 LangGraph 自动合并进全局状态,不需要手工维护"上下文传参链"。

另一个细节是条件边的路由函数不能修改状态,它只能读取状态并返回路由标识。一开始我把重试次数的自增逻辑放在need_retry里,结果发现每次执行流经过条件边时状态并没有变化,重试次数永远不增长,陷入死循环。正确做法是在某个节点里修改retry_count,再让条件边去读它。LangGraph 对"谁可以改状态、什么时候可以改"有明确约束,理解这个约束之后,图的执行逻辑会清晰很多。

还有一点,流式输出在 LangGraph 里不是默认行为。如果想在对话应用里逐字输出,需要调用graph.stream()并监听消息事件,或者给节点里的model打开streaming=True。我最初为省事直接invoke,结果前端一直白屏等完整结果,用户反馈体验很差。这个后面踩坑部分会再展开。

4. 入门阶段我踩过的坑,按频率排序

4.1 包名混乱:langchain / langchain-community / langchain-openai 怎么分

这是 L langChain 入门第一坑,而且几乎每个人都会遇到。打开 PyPI 你会发现一堆相关包:langchain、langchain-core、langchain-community、langchain-openai、langgraph等等。早期版本是"一个大包装所有",后来拆分成核心包和集成包,模型服务商的适配代码都搬到了独立的包。

我建议的安装策略是明确的:核心依赖装langchain和langgraph,用到了哪家模型就装对应的集成包,比如langchain-openai、langchain-anthropic。上面代码里的ChatOpenAI就来自langchain-openai,如果你只装了langchain主包,导入会直接报错。

pip install langchain langgraph langchain-openai

注意langchain-community这个包,它收纳了大量社区维护的第三方集成,功能多但维护质量参差不齐,很多类连官方文档都查不到完整说明。能用官方集成包解决的,就尽量不要依赖 community 包,否则升级一次主版本就可能坏一片。

4.2 API 过山车:昨天能用的代码今天报错

LangChain 系列框架的版本迭代速度在所有 Python 库里都算快的,而且经常是破坏性变更。我写过几段"官方文档里的示例代码",一个月后再跑就被告知方法已废弃或签名变了。

最典型的是早期LLMChain那套写法,现在官方已经推荐直接用 LCEL 管道代替;ConversationBufferMemory这类记忆模块的用法也几经调整,在 LangGraph 出现后官方更是直接建议用 Checkpointer 管理对话状态。我自己经历过一次最无语的情况:某次升级后,原本从langchain主包导入的一个工具函数被移到了langchain_core,但langchain_core的版本要求和主包不完全同步,两头不同版本互相拉扯,最后只能靠锁版本解决。

所以我的建议很朴素:第一,不要追最新版,选一个稳定版本后锁定它,项目里放requirements.txt或pyproject.toml写明版本范围;第二,报错信息看不懂时,先去搜对应版本的官方文档,别信网上的旧教程;第三,升级前先看 release notes 里的 breaking changes 列表。

4.3 State 更新里的 reducer 魔法

LangGraph 的 State 默认是"覆盖式更新":每个节点返回的字段值,直接替换掉状态里的旧值。但有些字段是"追加式"的,比如多轮对话的消息列表、检索结果的累积列表,这时候如果你直接返回一个新列表,旧内容就丢了。

LangGraph 的解决方案是 reducer,一个用来合并新旧值的函数。最常见的内置 reducer 是add_messages,它会把新消息追加到已有消息列表后面,并且能处理消息 ID 去重:

from typing import Annotated from langgraph.graph.message import add_messages class ChatState(TypedDict): messages: Annotated[list, add_messages]

我第一次没加这个注解,直接返回{"messages": [new_msg]},结果发现多轮对话历史只剩下最后一轮,前面几轮全被覆盖了。排查了很久才发现是 reducer 的问题。这个知识点很小,但不知道的话,写多轮对话应用会莫名其妙丢上下文。其他自定义 reducer 也遵循同样的思路:它是一个接收"旧值 + 新值"、返回"合并后值"的函数,完全由你控制。

4.4 序列化问题与递归上限

LangGraph 的 Checkpointer 要把状态保存下来,这就要求状态里的所有值都必须是可序列化的。我在状态里放过一个打开的数据库连接对象,编译时没问题,一跑invoke就报序列化错误。后来老老实实把连接对象改成连接字符串,在节点内部再创建真实连接。记住一条原则:状态里只放"数据",不放"资源"。连接、句柄、客户端对象这些,要么在节点内部创建,要么通过依赖注入的方式传入。

另一个隐蔽的问题是递归上限。LangGraph 默认限制一次执行的步数,防止你的图写成一个死循环然后无限跑下去。如果图里确实需要很多次循环(比如很长的多智能体协作),需要显式提高recursion_limit:

result = graph.invoke(initial_state, config={"recursion_limit": 50})

但如果你的图正常运行就频繁撞上限,先别急着调大它,这往往是路由条件写错了的征兆。我就在条件边里写反过条件,导致节点在两条边之间来回跳,表面上看是"递归不够用",实际上是逻辑错误。

4.5 回调、日志和 Token 统计:入门最容易忽略

最后一个坑不是代码层面,而是工程层面。LLM 应用和普通接口最大的区别是:一次用户请求背后可能有多次模型调用,每次调用都花钱、都耗时。如果不在框架层做统一监控,出了问题根本不知道是哪一步在"烧钱"。

LangChain 提供回调机制来解这个问题。你可以在链或图编译时传入回调处理器,在每次模型调用开始、结束时触发事件,从而记录 token 用量、耗时、完整输入输出。我后来做了一个很简单的实践:把所有回调日志写到结构化文件里,每次跑完一轮对话,都能看到"分类调用花了多少 token,生成调用花了多少 token"。不夸张地说,这个监控能力接入前后,我对整个框架的理解完全不一样了——它让你看到每一次模型调用,而不是把整个图当成一个黑盒。

5. 选型建议:什么时候用 Chain,什么时候上 Graph

5.1 判断标准:先问自己三个问题

新人最常问我的问题是:学 LangGraph 是不是就不用学 LangChain 了?我的看法是,这两个不是替代关系,LangGraph 甚至需要依赖 LangChain 的模型、提示词组件来搭建节点内部逻辑。真正的选型问题是"我的应用需要哪种控制流"。

我一般建议问自己三个问题:

第一,流程是否完全固定、线性?如果用户输入进来之后,执行路径永远是"A 做完做 B,B 做完做 C",且没有分支、没有循环,那 LCEL 管道就够了,简单直接。

第二,是否存在需要根据中间结果改变路径的情况?比如 RAG 里检索结果不好要换关键词重查,比如多智能体场景里要根据任务类型把请求转给不同专家节点。只要出现这种"执行到一半要做决策"的需求,LangGraph 就是更自然的选择。

第三,是否需要人工介入、跨会话记忆、长流程暂停恢复?这三个能力基本是 LangGraph 的独有优势,靠 Chain 自己实现会非常痛苦。

5.2 一张表看懂两者的差异

对比维度Chain / LCELLangGraph
控制流线性、单向、无环有向图,支持循环和条件跳转
状态管理靠函数参数和返回值手工传递全局 State,节点增量更新
多轮记忆需要额外接记忆模块Checkpointer 原生支持
人工介入难以干净实现支持中断、等待、恢复
适用场景固定的单轮或多轮模板化流程复杂决策、循环、多智能体协作
心智负担低较高,但收益明显

这张表不是要说明谁比谁高级。我个人的观点是:如果你的应用连一个循环都没有,硬上 LangGraph 反而是过度设计。图的概念本身有学习成本,调试时你要跟踪状态在节点间的流转,比看一段线性的管道代码要费力。反过来,如果业务里已经出现两处以上需要"根据结果决定下一步"的逻辑,就别再拿 Chain 硬撑了,早点切到 LangGraph,代码反而会变清爽。

5.3 我自己现在的使用习惯

折腾了几个月之后,我现在的习惯是:默认用 LCEL 写单次调用或简单固定管线,一旦出现分支、循环、多轮状态、人工复核这些需求,就直接把对应的那一段迁移到 LangGraph。迁移过程其实不复杂——LCEL 管道可以整体作为一个节点放进图里,两者天然兼容。

如果只让我给新入门的人一句建议,那就是:别把这两个框架当成本体论问题来纠结,它们是工具。先用一个明确的小需求把 LCEL 跑通,理解 Model、Prompt、Parser 的配合方式,然后带着一个"有循环"的真实需求去学 LangGraph,你会比单纯看文档快得多。我自己就是从"给客服机器人加快重试"这个小需求开始上手 LangGraph 的,从认概念到写出第一个可运行的图,大概只花了一个晚上。这个上手顺序,比一开始就想把文档从头啃完要有效得多,建议你也试试。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询