企业官网建设流程全解析

1. 先聊聊为什么需要自定义工作流

我最早接触 LangChain 的时候，第一反应是“这玩意儿真方便，Chain 一拼装，就能跑通 LLM 调用、Prompt 管理、输出解析一套流程”。但用多了之后，你会发现标准 Chain 其实是个相当“理想化”的模型——它假设你的业务流程是线性、可预测的：拿到输入，塞给 Prompt，调一次模型，解析结果，完事。

真实世界里的业务逻辑可没这么乖。我在一个模拟项目X里做知识库问答系统时，第一个版本全用内置链拼装。结果遇到一个很现实的问题：用户提问之后，系统要先判断这个问题需不需要查知识库——有些是闲聊，查了反而容易让大模型被噪音干扰；有些则必须走检索增强；还有些问题需要先拆成多个子查询，分别检索，再合并答案。这种带分支、带循环、带条件判断的流程，用默认的顺序链根本写不明白，或者写得极其别扭。

这时候就要上自定义工作流了。用大白话说，它就是让你自己编排“大模型干活”的整套流程：什么时候调用模型、什么时候调工具、什么时候做判断、什么时候提前结束、什么时候重试。所有人都能自己设计流程，只是 LangChain 把这件事的入门门槛降到了一个相当友好的程度。

这篇就围绕 9.6.5 版本里和自定义工作流相关的核心手段展开，包括 Runnable 协议、LCEL 表达式语言、以及如何自己写一个完整可用的自定义工作流。看完之后，你不仅能读懂别人写的工作流代码，还能照着搭一套自己的。

2. 自定义工作流的底层基石：Runnable 协议

2.1 一切皆 Runnable 的设计逻辑

LangChain 进入 0.2 左右开始，底层几乎全面转向了 Runnable 协议。所谓 Runnable，本质上是一个统一的执行接口，它约定任何组件都可以被“调用”“批处理”“流式输出”“异步执行”。只要一个对象实现了这些能力，它就可以跟其他 Runnable 对象自由组合。

这个设计思路其实很像乐高积木：每个积木都是标准接口，不管它是 Prompt、模型、输出解析器还是一个自定义函数，只要按照标准接口拧上去，就能无缝连接。自定义工作流的底层，就是把不同的 Runnable 按照业务逻辑串成一个更大的 Runnable。

在 9.6.5 版本里，你会发现以前习惯用的LLMChain其实已经不再是首选了。官方推荐的方式是直接使用RunnableSequence、RunnableParallel、RunnableLambda这些组合器。这既是好事也是坏事：好事是设计更统一、组合能力更强；坏事是如果不理解 Runnable 协议，看到这些类名可能会一头雾水。

2.2 关键组件速览

先过一遍自定义工作流里最常用的几个 Runnable 组合器：

很多人刚开始用这些组件时会觉得麻烦：我直接写 Python 顺序调用不就行了吗，为什么要绕一圈用 Runnable？这里有个关键区别——直接用 Python 代码描述“流程”，LangChain 是看不到流程的。它无法帮你做流式输出、无法在中间步骤做追踪、无法在并行分支里自动处理异常。而基于 Runnable 的工作流，每一步都是可观测、可插入、可替换的。

我在实际项目中体会最深的一点是：Runnable 协议最有价值的不是“能跑”，而是“可组合、可观测”。调试一个复杂工作流时，你可以在任意一个节点塞一个RunnableLambda来打印中间结果；你可以用with_retry给某一步单独加重试策略；你可以在不改变整体结构的前提下替换某个子流程。这些都是纯 Python 顺序调用很难优雅实现的。

2.3 从 Chain 到 LCEL 的迁移

LCEL（LangChain Expression Language）是 LangChain 提供的一套声明式语法，用管道符|把 Runnable 串起来。你可能会问：这不就是装饰器包装一下吗，有什么稀奇的？其实 LCEL 的精髓不在于语法糖，而在于它让“流程即对象”这件事变得特别自然。

举个直观的例子。传统的LLMChain写法大致是这样：

from langchain.chains import LLMChain from langchain.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个翻译助手"), ("user", "{input}") ]) chain = LLMChain(llm=llm, prompt=prompt) result = chain.invoke({"input": "hello"})

同一件事用 LCEL 写是这样：

from langchain_core.runnables import RunnablePassthrough prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个翻译助手"), ("user", "{input}") ]) model = ChatOpenAI(model="gpt-4o", temperature=0) chain = prompt | model result = chain.invoke({"input": "hello"})

看起来只是少了LLMChain包装，但你现在获得的chain是一个纯粹的 Runnable 对象，天然支持流式、异步、批处理。你可以直接把它嵌入到更大的流程里，也可以对它做batch()调用，或者用流式接口把 token 一个字一个字地推给前端。

从 9.6.5 的实践来看，新代码建议一律用 LCEL 语法。不是说LLMChain不能用了，而是它已经不在核心设计的快车道上，后续扩展能力明显弱于 LCEL。

3. 拆解自定义工作流的五步设计法

3.1 第一步：明确流程节点和分支条件

拿到一个需求，先别急着写代码，先在纸上把流程画出来。不一定是专业的流程图，哪怕画几个框、几条线也行。核心是把“流程节点”和“分支条件”拆清楚。

以智能客服系统为例，我通常这样拆：

1. 接收用户消息。
2. 判断是否为闲聊：是则直接走闲聊回复；否则进入知识库问答链路。
3. 知识库问答链路里，判断是否需要拆分子问题：简单问题直接检索；复杂问题先拆解为多个子问题。
4. 检索出结果后，判断结果置信度是否足够高：高则生成回答；低则走“澄清追问”分支。
5. 所有分支最终汇聚到统一输出格式。

拆完节点，你自然就会知道自己需要的 Runnable 组合器是什么：分支判断那一步用RunnableBranch或者RunnableLambda返回提示词走不同分支；并行拆分用RunnableParallel；汇聚输出用RunnablePassthrough或者自定义RunnableLambda。

3.2 第二步：定义输入输出数据格式

这一步看起来不起眼，但自定义工作流最容易踩坑的地方就是数据结构不一致。LCEL 管道中，上一个节点的输出就是下一个节点的输入。如果你的上一个节点输出的是字符串，而下一个节点需要的是一个包含question字段的字典，那运行时会直接报 KeyError 或者把字符串塞到不该塞的地方。

我自己的经验是：给每个关键节点定义输入输出的预期结构，可以用 Pydantic 模型，也可以用简单的字典约定。例如：

from typing import Dict, Any from pydantic import BaseModel, Field class QuestionIntent(BaseModel): is_chitchat: bool = Field(description="是否为闲聊") sub_questions: list[str] = Field(default_factory=list, description="拆分后的子问题") original_question: str = Field(description="原始问题")

然后从流程最开始就把原始输入包成这个结构，后面的每一个节点都严格按照这个结构的字段名来读取数据。一旦出了异常，调试起来会轻松很多，因为你可以直接打印这个结构，看看数据是在哪个环节被弄丢的。

3.3 第三步：选择适当的组件实现各节点

这一步是技术选型的核心。每一个节点，都要想一想：这个逻辑适合用 Prompt + 模型实现，还是普通 Python 函数实现，或者是检索工具？

举例来说，“判断是否为闲聊”这个节点，有两种实现思路：

• 第一种：写一个RunnableLambda，用一些启发式规则来判断（比如是否含有某些关键词、是否长度过短）。
• 第二种：写一个分类 Prompt，调用小模型判断，输出结果通过with_structured_output解析成结构化格式。

两者各有优劣。规则法快、便宜、可控，但泛化能力差；模型分类法灵活，但会引入额外延迟和费用。我通常在简单场景下用规则，复杂场景下用模型分类。比如判断闲聊可以用规则，判断“是否需要拆分子问题”这种语义性强的任务就用模型。

再比如“检索知识库”这个环节，在 9.6.5 里常用的做法是把检索器包装成RunnableLambda，因为检索器本来就是普通 Python 工具，包装起来之后就能参与 LCEL 的管道组合。

from langchain_core.runnables import RunnableLambda def search_knowledge_base(query: dict) -> dict: retriever = get_retriever() docs = retriever.invoke(query["original_question"]) return {"question": query["original_question"], "docs": docs}

RunnableLambda会把函数的输入和输出自动适配成 Runnable 的协议，调用时输入一个字典，返回的也是一个字典。

3.4 第四步：用 LCEL 把节点串起来

节点都定义好了，串起来是相对简单的一步。核心是遵循“上一步的输出要匹配下一步的输入”这个原则。

举个完整的串接示例：

from langchain_core.runnables import RunnableBranch, RunnablePassthrough, RunnableParallel, RunnableLambda from langchain_core.output_parsers import StrOutputParser # 假设已经定义好了各个节点 def is_chitchat(state: dict) -> bool: # 简单的规则判断 return state["original_question"] in ["你好", "在吗", "你是谁"] def add_chitchat_marker(state: dict) -> dict: state["intent"] = "chitchat" return state def add_qa_marker(state: dict) -> dict: state["intent"] = "qa" return state def decompose_subquestions(state: dict) -> dict: # 调用模型拆分复杂问题, 生成 sub_questions return state def search_all_subquestions(state: dict) -> dict: # 对所有子问题并行检索 return state def generate_answer(state: dict) -> str: # 调用最终生成模型 return answer_str chitchat_chain = RunnableLambda(add_chitchat_marker) | RunnableLambda( lambda state: f"闲聊回复: {state['original_question']}" ) qa_chain = ( RunnableLambda(add_qa_marker) | RunnableLambda(decompose_subquestions) | RunnableLambda(search_all_subquestions) | RunnableLambda(generate_answer) ) router = RunnableBranch( (lambda state: is_chitchat(state), chitchat_chain), qa_chain # 默认分支 ) full_workflow = ( RunnablePassthrough.assign(original_question=lambda x: x["question"]) | router )

这个例子基本展示了 LCEL 自定义工作流的全貌。RunnableBranch接收元组列表，每个元组由一个判断函数和一个分支链组成。判断函数返回 True，就走对应的分支；前面的分支都不命中，就走最后一个默认分支。RunnablePassthrough.assign的作用是在不改变原数据的前提下，往里追加新的字段，非常实用。

3.5 第五步：流式输出、异步调用与重试策略的接入

一个工作流串起来之后，还只是“能跑”的状态。要投入到真实业务里，还需要考虑三个增强能力：

流式输出：如果最终答案是逐 token 生成的，可以用stream()方法替代invoke()。但要注意，如果你的工作流中间存在并行分支或者逻辑判断，那么 LangChain 能够自动处理大部分流式传播，但某些自定义函数如果一次性返回完整内容，流式就没法贯穿。通常的折中方案是：判断和检索步骤保持非流式，最后一步生成答案时用流式。

异步调用：LangChain 从 0.2 版本开始推荐使用langchain-core提供的异步接口。自定义工作流的每个 Runnable 组件都有ainvoke()异步方法。需要注意，如果某个中间步骤是同步阻塞的（比如一个普通的 Python 检索函数），它会阻塞事件循环，因此这类函数要么自己改成异步版本，要么用asyncio.to_thread包装。

重试策略：在 9.6.5 中，可以用with_retry给工作流局部加重试：

from langchain_core.runnables import Runnable optimized_chain = chain.with_retry( stop_after_attempt=3, retry_if_exception_type=(TimeoutError, APIError) )

这个对于调用外部模型、外部 API 的节点特别有价值。模型服务偶尔抖动是常态，局部重试比整个工作流重跑要经济得多。

4. 实际操盘一个多分支检索增强工作流

4.1 需求描述与流程设计

这部分我拿一个真实做过的场景来讲：做一个“行业知识问答助手”，它连接企业内部的文档数据库和外部公开资料库。用户问的问题可能是简单的事实查询，也可能是需要对比分析的综合问题，还可能是根本没有答案的冷门问题。

我把工作流设计成五个主要阶段：

1. 意图识别：判断问题是否属于领域问题，如果不属于则直接礼貌拒绝。
2. 问题分析：判断是简单问题还是复杂问题，简单问题只生成一个检索查询，复杂问题拆成多个子查询。
3. 并行检索：对每个子查询，同时检索内部文档和外部公开资料。
4. 结果融合：合并多个来源的检索结果，去除重复内容，按相关度排序。
5. 答案生成：把融合后的上下文和原始问题一起丢给大模型生成回答。

这个流程最核心的价值在于：它把一个大模型能力拆成了多个可定向优化的模块。意图识别可以单独换模型，并行检索可以单独调参数，结果融合逻辑可以单独优化，而不会牵一发而动全身。

4.2 核心代码实现

下面给出一个简化但可跑的代码示例，使用的类名和接口基于 9.6.5 时代的 LangChain。整体结构会偏工程化，请关注数据流而不是具体实现细节。

import asyncio from typing import Dict, Any from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import ( RunnableLambda, RunnableParallel, RunnablePassthrough, ) from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI # ====== 模型初始化 ====== classifier_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0) generator_llm = ChatOpenAI(model="gpt-4o", temperature=0.3) # ====== 阶段1: 意图识别 ====== intent_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个意图分类器。判断用户问题是否属于'某领域'相关问题，只输出一个词：yes 或 no。"), ("human", "用户问题：{question}"), ]) def clean_intent_output(text: str) -> str: return text.strip().lower() intent_chain = ( intent_prompt | classifier_llm | StrOutputParser() | RunnableLambda(clean_intent_output) ) # ====== 阶段2: 问题分析 ====== analysis_prompt = ChatPromptTemplate.from_messages([ ("system", """ 判断用户问题是否需要拆分成多个子问题来回答。 如果问题涉及多个方面、多个对比对象，或明显是复合问题，就输出一个 JSON 数组，包含 2~3 个子问题。 如果问题简单直接，则输出只含一个元素的 JSON 数组。 只输出 JSON，不要其他文字。 """), ("human", "用户问题：{question}"), ]) def parse_subquestions(text: str) -> list[str]: # 这里用 json.loads 解析，注意防解析失败 try: return list(json.loads(text)) except Exception: return [text] analysis_chain = ( analysis_prompt | classifier_llm | StrOutputParser() | RunnableLambda(parse_subquestions) ) # ====== 阶段3: 并行检索 ====== def search_internal_docs(query: str) -> str: # 实际项目中这里调用内部知识库的检索接口 return f"内部文档关于 {query} 的检索结果片段A" def search_external_docs(query: str) -> str: # 实际项目中这里调用外部的搜索 API 或者网页检索 return f"外部资料关于 {query} 的检索结果片段B" def search_all_subquestions(sub_questions: list[str]) -> list[dict]: results = [] for q in sub_questions: parallel_search = RunnableParallel( internal=RunnableLambda(lambda x: search_internal_docs(x)), external=RunnableLambda(lambda x: search_external_docs(x)), ) # 实际上在 LCEL 中更优雅的写法是构建一个并行链去 batch， # 这里为了可读性直接循环 + 临时 parallel 对象 res = parallel_search.invoke(q) results.append({"sub_question": q, **res}) return results # ====== 阶段4: 结果融合 ====== def merge_results(search_results: list[dict]) -> str: # 简单实现：把所有子问题的检索结果拼在一起，去掉空内容 merged = [] for item in search_results: if item["internal"]: merged.append(f"[内部] {item['internal']}") if item["external"]: merged.append(f"[外部] {item['external']}") return "\n\n".join(merged) merge_chain = RunnableLambda(merge_results) # ====== 阶段5: 答案生成 ====== generation_prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个严谨的知识助手。基于提供的资料回答问题，资料不足时明确说明。"), ("human", "资料：\n{context}\n\n问题：{question}"), ]) generation_chain = ( generation_prompt | generator_llm | StrOutputParser() ) # ====== 串成最终工作流 ====== def route_on_intent(intent: str, state: dict) -> bool: return intent == "yes" def reject_answer(state: dict) -> str: return "抱歉，我只能回答该领域相关问题。" # 注意: 这里为了保证示例结构清晰，先直接用 lambda 组合 # 实际工程中建议用 RunnableBranch 做路由 full_chain = ( RunnablePassthrough.assign(question=lambda x: x["question"]) | RunnableLambda(lambda state: { **state, "intent": intent_chain.invoke({"question": state["question"]}) }) | RunnableBranch( ( lambda state: state["intent"] != "yes", RunnableLambda(lambda state: reject_answer(state)), ), # 主流程 ( RunnablePassthrough.assign( sub_questions=lambda state: analysis_chain.invoke({"question": state["question"]}) ) | RunnableLambda(lambda state: {**state, "search_results": search_all_subquestions(state["sub_questions"])}) | RunnablePassthrough.assign( context=lambda state: merge_results(state["search_results"]) ) | RunnablePassthrough.assign( answer=lambda state: generation_chain.invoke({ "context": state["context"], "question": state["question"] }) ) | RunnableLambda(lambda state: state["answer"]) ) ) )

4.3 关键细节解读

这个示例里有几个值得特别说明的细节。

第一个细节是RunnablePassthrough.assign的用法。它不会覆盖原始输入，而是在输入字典上增加新字段。如果你直接写state["question"] = ...，在某些场景下会碰到不可变字典的问题，而assign方法内部会处理好这层逻辑。在链式调用中，assign几乎是给数据做增量更新的不二选择。

第二个细节是RunnableBranch和普通 Pythonif-else的区别。用 Python 写 if-else 当然更简单，但一旦分支逻辑需要被嵌套进更大的 LCEL 管道，或者需要支持流式/异步调用，原生的if-else就无能为力了。RunnableBranch返回一个 Runnable，意味着整个分支结构也可以被整体复用，这在实际项目里很常见。

第三个细节是不要在 RunnableLambda 内部再去 invoke 一个包含 RunnableLambda 的链。我在早期干过这种事，写出来的代码特别难追踪，一个 Lambda 里嵌套另一个 Lambda，调试的时候根本分不清是哪个环节出了问题。比较好的做法是：把内部链放在外层，通过RunnablePassthrough.assign分阶段执行。

第四个细节是关于lambda state: search_all_subquestions(state["sub_questions"])这种写法。这里的 search_all_subquestions 里面循环调用了多个检索接口，它是一个阻塞函数。在真正的异步场景下，你应该把这个函数改成异步版本，或者把search_internal_docs和search_external_docs都用异步实现，然后通过RunnableParallel来实现真正的并行。

5. 自定义工作流中的常见坑与排查技巧

5.1 输入输出结构不匹配

这是我见过最多的错误。LCEL 管道中每一步的输入输出都是直接传给下一层的，一旦某个RunnableLambda返回的不是字典而是一个字符串，下一层如果还在期待字典，立刻就会出问题。

排查技巧是：在出问题的节点前加一个RunnableLambda(lambda x: print(x) or x)，看一下具体传过来的结构长什么样。不用猜，直接打出来看。另外，尽量从一开始就用统一的字典结构传递状态，比如统一用{states: ..., data: ...}之类的形式，这样能大幅度降低链路中的数据混乱问题。

5.2 在 RunnableLambda 内误用同步阻塞调用

如果你在 FastAPI 服务里用异步接口调用工作流，某个节点内部做了同步的 HTTP 请求，整个事件循环都会被卡住。一位同事在这个问题上卡了很久：调用链本身没问题，但并发一上去响应就变慢，查不到原因。后来定位到是某个RunnableLambda里的同步检索函数阻塞了事件循环。

解决思路有两个。其一，把阻塞函数改成异步版本：

async def async_search_internal_docs(query: str) -> str: # 使用异步 HTTP 客户端去请求检索接口 return await do_async_search(query)

其二，如果依赖的第三方库没有异步版本，就把它丢进线程池：

import asyncio def sync_search(query: str) -> str: return search_internal_docs(query) async def async_search_wrapper(query: str) -> str: return await asyncio.to_thread(sync_search, query)

这两种做法在 9.6.5 里都适用，原则就是：不要在主事件循环里跑阻塞 IO。

5.3 并行分支容易出现重复检索或重复调用

用RunnableParallel并行跑多个子查询时，如果对检索器的调用封装不好，很容易出现多个子问题重复请求同一个接口、返回相同内容的情况。这不仅浪费 token，还会导致最终生成的答案上下文冗余，降低模型对关键信息的注意力。

具体表现是：合并后的 context 里出现了两三段内容差不多的检索结果，模型生成的回答开始在重复信息里打转。

我的处理习惯是：在 merge_results 这一步做一次简单的去重排序。可以按文档标题做去重，也可以按内容哈希做去重。如果是向量检索的结果，通常会带score字段，按分数排序后取前 N 条即可。

5.4 流式输出中断在半路

自定义工作流里只要存在中间解析、中间函数，LangChain 的流式传播就可能断掉。最典型的情况是：前面用StrOutputParser或者自定义的RunnableLambda把模型的 stream token 全部收起来成了一个完整字符串，然后用户体验就变成了“等半天，一次性看到全部答案”。

在实际项目中，流式体验差往往不是 LangChain 的问题，而是工作流设计和流式能力互相妥协的结果。如果你对流式有硬性要求，有两个调整方向：一是把生成环节独立出来，先通过工作流算出检索结果和所有中间信息，再单独调最后一个生成模型做流式输出；二是在前面的步骤中都改用RunnableLambda协议下的生成器函数，把”边生成边传”的模式贯穿到底。后者实现难度高不少，而且对于多分支、多条件判断的流程，收益未必明显。

5.5 重试策略和幂等性

给某个节点加了with_retry之后，要确认这个节点被重复执行时不会产生副作用。比如一个节点往数据库里写日志，重试三次就写三遍日志；一个节点调用扣费接口，重试可能导致重复扣费。这类节点要尽量设计成幂等，或者在重试逻辑里避免副作用。

我自己的习惯是：重试只加在“纯粹的读操作”或者“天然幂等的写操作”上。例如调用模型生成答案、调用检索接口、解析 JSON，这些可以随便重试；涉及状态修改、扣费、通知类的节点，绝不加盲目的全局重试。

6. 进阶：用 RunnableParallel 改造检索效率的实战对比

为了让大家对自定义工作流带来的性能收益有直观感受，我单独把“并行检索”这一节展开讲。很多初学者写检索增强的工作流，下意识会这样写：

docs_internal = retriever_internal.invoke(question) docs_external = retriever_external.invoke(question)

这段代码没什么问题，但它是串行的：内部检索跑完，才开始外部检索。如果每次检索耗时 0.5 秒，两步就是 1 秒。如果是在一个大模型问答接口里，这 1 秒是额外加在用户等待时间里的。

用RunnableParallel改造之后：

parallel_retrieval = RunnableParallel( internal=RunnableLambda(lambda q: retriever_internal.invoke(q)), external=RunnableLambda(lambda q: retriever_external.invoke(q)), )

两个检索任务并发执行，总耗时从“两者之和”变成“两者最大值”。这个优化逻辑极其简单，但收益几乎是白捡的。在一套问答系统里，如果检索环节被拆成 3 个并行检索源，响应时间就能从串行的 1.5 秒压到并行后的 0.5 秒左右。

当然，并行也不是没有代价。它要求各检索源之间互不依赖，且最终结果能在一个合并函数里统一处理。如果你的场景是“上一个检索结果会作为下一个检索的查询条件”，那就必须串行。识别自己场景属于哪种类型，是设计高效工作流的基本功。

7. 关于版本迁移的一点个人观察与建议

9.6.5 这个版本号很具有代表性。看 LangChain 的版本演进，你会发现从 0.1 到 0.2 再到后来，最大的变化就是 Chain 类逐渐退居二线，LCEL 和 Runnable 成为一等公民。这不仅仅是 API 层面的机械替换，而是设计哲学的转换：从“给你一堆预设好的一揽子方案”转向“给你积木，你来设计自己的流程”。

如果你跟我一样，是从LLMChain时代老代码迁移过来的，你会在 9.6.5 上碰到不少兼容性问题，但方向是明确的：所有新代码都应该按 Runnable 协议来写，老代码优先迁移核心链路，边缘功能可以延后。

迁移的时候有几个具体建议：

1. LLMChain调用改成prompt | model | output_parser三件套。
2. 链内嵌入的自定义逻辑，要么抽成RunnableLambda，要么用RunnableBranch组织分支结构。
3. .run()方法全面改成.invoke()，异步方法从.arun()改成.ainvoke()。
4. 输出格式强依赖的，用with_structured_output或者 Pydantic 解析器，别手动写正则解析。

这个改造过程一开始会觉得代码更“碎”了，一个原本 30 行的链可能变成 50 行的装饰器集合。但跑顺之后你会发现，每个节点都可独立测试、独立替换、独立追踪，维护起来反而更轻松。

8. 最后分享一点使用心得

自定义工作流这个东西，第一次接触时容易被那些花哨的组合器吓到，实际上底层逻辑非常朴素：把流程节点拆出来，把数据格式定好，再把节点按顺序装进组合器。真正难的不是 API 本身，而是你能不能像写程序一样，把大模型的能力模块化、流程化。

我自己在这几年的实践里最大的体会是：自定义工作流不是一个技术选项，而是一个思维方式。它在强迫你思考一个 LLM 应用里哪些环节是切割点、哪些环节可以并行、哪些环节有条件分支、哪些数据需要显式传递。如果你把这种思维方式内化了，你就算不写 LCEL，也能用纯 Python 搭出效果不错的工作流，只是可维护性和流式体验会差一些。

我建议你找一个简单的业务场景，哪怕就是“判断用户问题类别，然后走不同提示词”这种 5 分钟能搭完的小流程，亲手把它写成 LCEL 的 Runnable 组合。写完之后再尝试加一个并行检索分支。跑通了这两件事，你对 LangChain 自定义工作流的掌握就算是真正入门了，后面再往上叠多少层分支、并行、重试，都不会觉得心虚。