做 LangChain 项目做到一定阶段，你一定会撞上同一个天花板：拿 Prompt 模板拼 LLM 调用，写出一串串 Chain，但业务流程稍微复杂一点——要判断、要循环、要分支、要把多个任务并行跑——链式写法就开始处处别扭。这篇内容是我把 LangChain 从“链”玩到“工作流”的一路踩坑和总结，围绕自定义工作流展开，先说清楚为什么普通 Chain 撑不住复杂业务，再给出一个能直接抄的内容质检场景完整实现，最后整理高频问题的排查实录。适合已经会用基础 Chain、正准备把多步骤业务迁到 LangChain 上的开发者参考，刚上手的新人也能按第 3 节的步骤走通。
1. 为什么普通 Chain 不够用：工作流设计的真实痛点
1.1 链式调用的边界，一碰就破
LangChain 的简单模式很爽：prompt 模板、大模型、输出解析器，用管道符串起来，一条线跑到底。我早期做问答机器人，几乎就是这个套路，需求一多就出问题。真正的转折点是一次内容质检业务：稿件进来，先评审、再决定要不要返修、返修后还要复评，最后归档并提取摘要和标签。这个流程里有判断、有循环、有并行，用prompt | llm | parser这种链式写法根本描述不了，硬写只能把 if/else 塞进 RunnableLambda，所有状态靠外部全局变量撑着，跑两轮就乱套。
这不是 LangChain 的问题，是抽象层级的问题。“链”描述的是单一路径，而业务天然是一张图。你把图压成链，等于把路由、循环、回退全部手工摊平，代码又丑又脆，还丢掉框架自带的流式输出、回调、追踪这些红利。所以我先抛一个核心观点：线性任务用链，非线性任务请直接设计成工作流，别在链里硬凑。判断标准很简单——如果你的流程图画出来有箭头回头（循环）、有菱形分支（判断）、有多条并行线（并行），那它就该是一张图，不是一根管子。
1.2 一个让我放弃纯 Chain 的真实案例
我第一个版本怎么写的？评审节点一个 RunnableLambda，里面调 LLM；返修节点又一个 RunnableLambda；外层用 while 循环控制“最多改三次”。问题出现在第二次迭代：某次评审模型返回的分数解析失败，整个循环直接抛异常，但这时候稿件已经被改过一遍，外部状态和实际执行进度完全对不上。更麻烦的是并发场景，两个请求同时进来，共享的全局状态互相覆盖，我分不清这条稿件是谁的。这个案例给我的教训很直接：工作流引擎的价值，不在于帮你省几行代码，而在于把状态、流程、边界这三件事显式管起来。LangGraph 这类图引擎做的就是把状态收进一个明确的 State 结构里，节点之间只能通过它传递数据，流程则由边来定义，谁再想绕过状态偷偷改全局变量，就算想写也写不出来。
1.3 哪些场景才算“必须自定义工作流”
结合我接触过的项目，下面几类场景基本绕不开工作流这套东西：
- 多轮智能体循环：大模型决定调用什么工具，调用完把结果拿回来看要不要再调，这是典型的回路。
- 条件路由与分支：先判断意图，再分派给不同子流程，甚至同一个任务要并行跑多个子任务后汇总。
- 人工介入节点：自动化流程走到某一步需要人工确认，确认通过才继续，这种“暂停-恢复”需要持久化状态。
- 复杂重试与回退：某个节点失败后要走另一条兜底路径，成功率统计、重试次数、回退策略组合在一起，本质上就是状态机。
反过来，如果任务还是“输入固定、顺序固定、输出固定”，比如单纯的翻译摘要、关键词抽取、单轮问答，那用 LCEL 组合链就够了，上 LangGraph 属于杀鸡用牛刀。记住这个判断逻辑，后面选型就不会纠结。
2. 核心概念拆解：LCEL、Runnable 与 LangGraph 怎么选
2.1 LCEL 声明式组合：能写链，别急着上图
LCEL（LangChain Expression Language）是我最先推荐所有人掌握的东西，它是 LangChain 对“链”的声明式表达。核心就是管道符号：
from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser prompt = ChatPromptTemplate.from_template("把下面内容翻译成英文：{text}") summary_chain = prompt | llm | StrOutputParser()这段代码里，prompt、llm、StrOutputParser都是 Runnable，管道符把它们的输入输出串起来。好处有两个：一是懒加载，链构造的时候不会真调 LLM，只有invoke才执行；二是原生支持流式、异步、批量调用，chain.stream()能边出字边吐，做流式对话体验很好。LCEL 里还有两个重要的组件，RunnablePassthrough用于透传或者顺手算个字段，RunnableParallel用于并行执行多个链再合并结果。
2.2 Runnable 协议：理解一切调用的最小公约数
LangChain 里所有能被链式拼接的东西都实现了 Runnable 协议，协议只要求四个方法：invoke（同步调用）、ainvoke（异步）、stream（流式）、batch（批量）。这个设计非常关键，意味着你可以把自定义函数包装进链里，只要它接受一个 dict 或消息对象、返回一个 dict 或字符串。工作中最常见的做法是RunnableLambda包业务函数，在链里插入规则判断、做数据清洗、接第三方接口，框架完全不干涉你的实现。理解了 Runnable 协议，你会发现 LangGraph 里的每个节点本质上也还是 Runnable，从链到工作流不是换了一套框架，而是同一套抽象的不同层级。
2.3 LangGraph：把工作流当成一张有环图
LangGraph 是 LangChain 生态里专门做图式编排的方案，核心概念只有五个：State（状态）、Node（节点）、Edge（边）、Conditional Edge（条件边）、Checkpointer（检查点）。State 是一个显式定义的数据结构，通常用 TypedDict 或 Pydantic 模型声明；Node 是一个函数，输入当前 State，返回一个 dict，返回的 key 会更新到 State 上；Edge 决定节点间的走向；条件边则根据当前 State 的内容动态选择下一个节点。最有价值的是它允许环的存在，也就是说节点 A 可以回到节点 B，这意味着循环、重试、迭代在图上天然成立，而 LCEL 里做循环只能靠外部 while 或者递归，绕得很。
2.4 选型对照：什么时候用哪种
| 维度 | LCEL 链式 | LangGraph 工作流 |
|---|---|---|
| 适用场景 | 线性流水线、固定顺序、单路线任务 | 分支、循环、并行、人工介入、多智能体 |
| 状态管理 | 隐式由运行时传递，改动要改链结构 | 显式 State，节点按需返回增量字段 |
| 条件逻辑 | 写进 RunnableLambda，藏在代码里 | 条件边显式声明，流程图即代码 |
| 循环重试 | 不原生支持，靠外部控制 | 原生支持环，且带递归上限保护 |
| 可视化与调试 | 链路简单，打印结构即可 | 支持图结构查看，定位更直观 |
| 学习成本 | 低 | 中，需要理解状态机思维 |
我的选型口诀：能链不图，一有分支循环就立刻上图。两者是互补关系，LangGraph 的节点内部仍然可以用 LCEL，完全不是二选一。
3. 从零搭建内容质检工作流：状态、节点、路由与并行
3.1 需求拆解与状态设计
为了让步骤可复现，我用一个虚构的“内容质检工作流”做例子。业务规则如下：稿件进入评审节点，模型打分；得分低于阈值则进入返修节点，最多改三次；改完重新评审；得分合格后进入归档节点，同时并行提取摘要和标签；所有结果汇总输出。注意，设计工作流的第一步不是写代码，是画流程和列状态。我习惯先把需要跨节点共享的数据列成一张清单，再把这些字段写进 State。
from typing import TypedDict class ReviewState(TypedDict): article: str # 原始稿件 score: float # 最近一次评审得分 comments: str # 评审意见 attempts: int # 已修改次数 fixed_article: str # 返修后的稿件 summary: str # 最终摘要 tags: list[str] # 最终标签这里每个字段的设计都有理由：article不放在节点局部变量里，因为归档节点和摘要节点都要读原始内容；attempts必须放进状态，否则返修节点不知道这是第几次重试；fixed_article单独放一份，避免把修改稿回写到article导致评审节点看到的历史信息被覆盖。很多新手翻车就翻在这里：一个字段既当输入又当输出，轮次一多状态就糊了。我的建议是“输入字段和输出字段分开”，宁可多两个字段，也别图省事复用。
3.2 节点实现：评审、修改、归档
节点本质就是函数，签名为(state) -> dict。评审节点负责调大模型，我强烈建议用结构化输出而不是让模型返回一大段文本再手工解析。手工解析在测试环境能跑，一上线就被各种换行、多余标点搞得稀碎，踩过太多次了。用with_structured_output配合 Pydantic 定义结果结构，稳得多。
from pydantic import BaseModel, Field class ReviewResult(BaseModel): score: float = Field(description="0~1 之间的质量分") must_fix: bool = Field(description="是否需要返修") comments: str = Field(description="具体修改意见") review_llm = llm.with_structured_output(ReviewResult) def review_node(state: ReviewState) -> dict: target = state.get("fixed_article") or state["article"] result = review_llm.invoke(f"请评审以下稿件：\n{target}") return { "score": result.score, "comments": result.comments, "attempts": state["attempts"] + 1, } def fix_node(state: ReviewState) -> dict: repaired = fix_llm.invoke( f"根据意见修改稿件：{state['comments']}\n稿件：{state['article']}" ) return {"fixed_article": repaired.content}注意review_node里的一个小细节：优先评审fixed_article，如果不存在才评审原始article，这样返修后的稿件才会被复评，否则循环会失去意义。fix_node只动了fixed_article一个字段，因为返修节点不需要知道分数和意见之外的东西，节点返回的 dict 越精简，状态越不容易被污染。
3.3 条件路由与循环控制
路由和循环靠条件边实现。定义路由函数，返回值是目标节点名；再用add_conditional_edges把它挂到评审节点后面：
from langgraph.graph import StateGraph, START, END MAX_ATTEMPTS = 3 def route_after_review(state: ReviewState) -> str: if state["score"] >= 0.6: return "archive" if state["attempts"] >= MAX_ATTEMPTS: return "reject" return "fix" builder = StateGraph(ReviewState) builder.add_node("review", review_node) builder.add_node("fix", fix_node) builder.add_node("archive", archive_node) builder.add_edge(START, "review") builder.add_conditional_edges( "review", route_after_review, {"fix": "fix", "archive": "archive", "reject": END}, ) builder.add_edge("fix", "review") # 形成回路：改完回去复评 builder.add_edge("archive", "summarize") builder.add_edge("archive", "tag") builder.add_edge("summarize", END) builder.add_edge("tag", END) graph = builder.compile()这里最关键的一行是builder.add_edge("fix", "review")，它画出了整个工作流唯一的环。有了环之后，伤害控制就特别重要，我一般做三层保险：一是路由函数里判断attempts上限；二是在compile()时设置执行上限；三是给每个节点内部做异常兜底，防止模型调用挂掉导致流程卡死。
3.4 并行分支与收口
archive节点之后接两个并行分支summarize和tag，这在 LangGraph 里就是两条边同时从 archive 指出去。它们的执行顺序不受彼此影响，各自产出后都汇到 END。并行带来的好处是响应时间从“串行之和”变成“串行之和中最慢的那个”，代价是同时消耗两份 token，所以不是所有任务都值得并行。我通常只在“耗时差异大”且“彼此无依赖”时并行，比如摘要和标签提取都依赖原文但不依赖彼此，这就是典型的并行场景。
def summarize_node(state: ReviewState) -> dict: text = state.get("fixed_article") or state["article"] return {"summary": summarizer_llm.invoke(f"生成摘要：{text}").content} def tag_node(state: ReviewState) -> dict: text = state.get("fixed_article") or state["article"] return {"tags": tagger_llm.invoke(f"提取标签：{text}").content.split(",")}这种写法有个隐含约定：并行节点的返回 key 不能重叠，summary和tags各管各的。如果两个并行节点都试图更新同一个字段，LangGraph 会按节点执行顺序做覆盖更新，结果就不可控了。这点我踩过坑，记得排查时先看是不是并行分支写重了字段。
3.5 加记忆：持久化的两种姿势
工作流本身是无状态的，每次invoke都从 START 开始。但遇到人工介入、多轮追问、断点续跑这些场景，就必须引入 Checkpointer。LangGraph 提供了专门的检查点机制，最简单的内存版等于给工作流加了快照：
from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() graph = builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "case-001"}} result = graph.invoke({"article": "..."}, config=config)thread_id相当于会话标识，同一线程的多次调用会共享历史状态。比如内容审核时先跑候选意见，人工点了同意，再用同一个thread_id触发后续归档，工作流会从保存的位置继续而不是重头跑。内存版只适合开发和单机演示，生产环境建议换用持久化存储，具体选哪种看你们基础设施，重点是理解thread_id就是会话的钥匙，不同用户、不同任务都要有自己的 id。
4. 高频问题与排查实录：状态污染、死循环与预算失控
4.1 状态被污染：能改状态的只有节点
我排查过的案例里，八成问题出在状态上。典型症状是某个字段的值变得很怪，比如attempts变成字符串、score凭空消失。常见原因：节点返回了 State 里没声明的 key，或者有人把不相关的数据塞进了节点返回值。LangGraph 的机制是节点返回的 dict 会按 key 合并进 State，如果你返回的 key 在 schema 里不存在，新版本会抛异常或者静默忽略，两种表现都让人懵。我的习惯是节点返回前先自查：这个 key 在 TypedDict 里声明过吗？这个节点真的需要改这个字段吗？只有节点能改 State，外部代码想改只能通过invoke传入初始状态。
4.2 死循环：设定迭代上限只是第一步
回路设计出来后，最怕路由函数条件恒真。有一次我把 threshold 写成>= 0.6，构成死循环的条件我完全没意识到，图一直跑，直到进程超时。LangGraph 有个recursion_limit保护，默认 25 步，超过会抛错，但这只是兜底，不能指望它。正确做法是三层防线：路由条件里带计数、节点内部也要带计数、编译时配置一个比默认值更合理的步数限制。我通常把重试次数和业务阈值都放进 State，让它们在 Lambda 里能被直接看到，而不是写死在路由函数闭包里。
4.3 Token 消耗失控：预算先行
并行节点越多、循环次数越多，token 花得越快。我见过有人把返修上限设成 5，结果每条稿件最多烧掉基础成本的 4 倍。控制手段不复杂：给循环节点设更小的上限；不同节点用不同规格的模型，评审用高精度模型，摘要用便宜模型；并行分支里尽量只保留必需的任务。还有一点容易忽略，Streaming 输出不会省 token，省的是首字延迟，真正省钱的是把输出长度限制好，max_tokens该设就设，别让模型自由发挥。
4.4 失败重试与兜底策略
大模型调用没有不失败的，频率高的就是超时和限流。节点内部的try/except是第一层，但光捕获不够，要设计回退。我常用的模式：主模型失败后切成备选模型，再不行就返回一个“业务兜底结果”，比如评审失败直接标记为需人工介入而不是默认通过，这个选择很重要，业务宁愿卡住也不要给用户错误结论。另外 LLM 输出解析失败很常见，用with_structured_output能解决大半，剩下的靠重试再跑一轮。最后建议给关键节点加上详细日志，把输入裁剪到可读长度后打出来，这样排查时能看到模型到底收到了什么。
4.5 问题速查表
| 症状 | 可能原因 | 处理办法 |
|---|---|---|
| 字段值变成字符串 | 节点返回了未声明类型的数据 | 节点返回值严格对齐 State schema |
| 工作流无限执行 | 条件边判断恒真，没考虑计数 | 路由函数加 attempts 上限 + recursion_limit |
| 同一用户连续两次结果串了 | 没有用 thread_id 隔离会话 | 每个会话分配独立 thread_id |
| 输出格式偶尔解析失败 | 直接让模型裸输出 JSON | 改用 with_structured_output / 工具调用 |
| 并行节点返回互相覆盖 | 两个分支写了同一个字段 | 拆分字段或改串行 |
| 模型调用频繁超时 | 单次调用无重试 | 节点内层重试 + 备选模型兜底 |
5. 收尾实操体会：三个让我少踩坑的习惯
5.1 状态字典是给图看的，不是给你存业务的
早期我什么中间结果都想塞进 State，字段越堆越多，图执行时心智负担巨大。后来我才想明白，状态字典是节点之间通信的协议，不是业务数据库。凡是只有单一节点需要的临时数据，就放节点局部变量，别动不动加字段，状态越小，图越好维护。
5.2 先画图再写代码，图就是架构文档
进了 LangGraph 之后，我写代码前必须先把流程图画出来，哪怕只是手绘。这个习惯帮我在动工前就发现过两次设计硬伤：一次是忘了画返回复评的边，另一次是并行分支的收口条件没想清楚。画图用的容器工具都是外部的，方向对了就好，重点是流程必须能自洽：每个可能的分支都有出口，每条边都有业务含义。
5.3 能跑不算完，要能观测
工作流最难受的是它跑起来像个黑盒，你不知道当前停在哪一步。后来我给每个节点都加了一个轻量的日志函数，记录节点名、耗时、输入输出的关键字段摘要，再配合graph.get_state(config)查看当前线程状态。调试一次死循环复现问题时，就是靠日志定位到路由函数里阈值方向写反了。这个投入很值，比对着报错猜半天强太多。
5.4 最后分享一个实用小技巧
把所有节点的逻辑做成“纯函数”，也就是输入 State、输出 dict，不要在节点函数里读写文件、访问数据库、改全局变量。我有一阵子不遵守这个，把缓存直接写在了节点里，结果并发一上来全是脏数据。改成纯函数之后，每个节点都可以单独测、单独复用，甚至可以直接搬到另一套工作流里，这大概是我在 LangChain 自定义工作流上最值得的一条经验。
