【LangChain】 LCEL 核心组件详解：RunnableSequence vs RunnableParallel / RunnableMap-深圳市維司達科技有限公司

LangChain LCEL 核心组件详解：RunnableSequence vs RunnableParallel / RunnableMap

发布日期： 2026年5月15日

关键词： LangChain, LCEL, RunnableSequence, RunnableParallel, RunnableMap, 串行执行, 并行执行

一、LCEL 两大核心组件

在 LangChain 的 LCEL（LangChain Expression Language）中，有两个平级的核心编排组件：

组件核心作用执行方式
RunnableSequence 按顺序链式执行多个步骤串行（Sequential）
RunnableParallel / RunnableMap 同时执行多个独立分支并行（Parallel）

它们都是Runnable的子类，地位平等，只是编排策略不同。

二、RunnableSequence：串行执行

2.1 什么是 RunnableSequence？

RunnableSequence 将多个Runnable组件按顺序串联，前一个的输出作为后一个的输入。

2.2 两种等价写法

fromlangchain_core.runnablesimportRunnableSequence,RunnableLambdadefstep1(x):returnx+1defstep2(x):returnx*2defstep3(x):returnf"结果:{x}"# 写法一：管道符（推荐，简洁直观）chain_v1=RunnableLambda(step1)|RunnableLambda(step2)|RunnableLambda(step3)# 写法二：显式构造（完全等价）chain_v2=RunnableSequence(RunnableLambda(step1),RunnableLambda(step2),RunnableLambda(step3))# 执行结果相同print(chain_v1.invoke(5))# "结果: 12"print(chain_v2.invoke(5))# "结果: 12"

2.3 执行流程

输入: 5 │ ▼ [step1] 5 + 1 = 6 │ ▼ [step2] 6 * 2 = 12 │ ▼ [step3] "结果: 12" │ ▼ 输出: "结果: 12" 总耗时 = step1耗时 + step2耗时 + step3耗时

2.4 适用场景

✅ 数据预处理 → 模型调用 → 后处理
✅ 有明确先后依赖的步骤
✅ 需要前一步结果作为下一步输入

三、RunnableParallel / RunnableMap：并行执行

3.1 什么是 RunnableParallel？

RunnableParallel（别名 RunnableMap）将输入同时分发给多个独立分支，各分支同时执行，最后合并结果。

3.2 两种等价写法

fromlangchain_core.runnablesimportRunnableParallel,RunnableMap,RunnableLambdadefadd_one(x):returnx+1defmultiply_by_two(x):returnx*2defto_string(x):returnf"值为:{x}"# 写法一：RunnableParallel（推荐）chain_v1=RunnableParallel({"added":RunnableLambda(add_one),"doubled":RunnableLambda(multiply_by_two),"original":RunnableLambda(to_string),})# 写法二：RunnableMap（完全等价，别名）chain_v2=RunnableMap({"added":RunnableLambda(add_one),"doubled":RunnableLambda(multiply_by_two),"original":RunnableLambda(to_string),})# 执行结果相同print(chain_v1.invoke(5))# {'added': 6, 'doubled': 10, 'original': '值为: 5'}

3.3 执行流程

输入: 5 │ ├──┬──┬──┐ ▼ ▼ ▼ [加1] [乘2] [转字符串] │ │ │ 6 10 "值为: 5" │ │ │ └──┴──┴──┘ ▼ {'added': 6, 'doubled': 10, 'original': '值为: 5'} 总耗时 ≈ max(分支A耗时, 分支B耗时, 分支C耗时)

3.4 适用场景

✅ 多维度同时分析（摘要 + 关键词 + 情感）
✅ 同时调用多个模型对比结果
✅ 同时查询多个数据源
✅ 相互独立的预处理任务

四、平级对比：RunnableSequence vs RunnableParallel

对比维度 RunnableSequence RunnableParallel / RunnableMap
核心作用顺序编排并行编排
执行方式串行，一个接一个并行，同时执行
数据流前一个输出 → 后一个输入同一输入 → 各分支独立处理
输出格式单个值（最后一步的输出）字典{分支名: 分支结果}
语法a \| b \| cRunnableParallel({"x": a, "y": b})
总耗时各步骤耗时之和最慢分支的耗时
依赖关系步骤间有依赖分支间相互独立
底层实现顺序调用invoke()asyncio并发执行
别名无RunnableMap

五、实战：组合使用两大组件

实际项目中，串行和并行往往组合使用：

5.1 场景：智能客服工单分析

需求：收到用户反馈后，先预处理，再并行分析多个维度，最后汇总生成回复建议。

fromlangchain_core.runnablesimport(RunnableSequence,RunnableParallel,RunnableLambda,RunnablePassthrough)fromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplate llm=ChatOpenAI(model="gpt-4o-mini")# ========== 1. RunnableSequence：预处理（串行）==========preprocess=RunnableLambda(lambdax:{"raw":x,"cleaned":x.strip().lower(),"length":len(x.strip())})# ========== 2. RunnableParallel：并行分析（并行）==========analysis=RunnableParallel({# 分支1：意图识别"intent":(ChatPromptTemplate.from_template("判断用户意图（咨询/投诉/建议/其他）：\n{cleaned}")|llm),# 分支2：紧急程度"urgency":(ChatPromptTemplate.from_template("判断紧急程度（高/中/低）：\n{cleaned}")|llm),# 分支3：情感分析"sentiment":(ChatPromptTemplate.from_template("判断情感倾向（正面/负面/中性）：\n{cleaned}")|llm),# 分支4：透传原始数据（用于后续步骤）"meta":RunnableLambda(lambdax:{"length":x["length"],"source":"web"})})# ========== 3. RunnableSequence：汇总生成（串行）==========generate_response=RunnableLambda(lambdax:{"analysis":x,"reply_suggestion":f"根据分析结果：意图-{x['intent'].content}, 紧急度-{x['urgency'].content}, 情感-{x['sentiment'].content}","routing":"转人工"ifx['urgency'].content=="高"else"自动回复"})# ========== 组合：串行 → 并行 → 串行 ==========full_chain=RunnableSequence(preprocess,# 步骤1：预处理（串行）analysis,# 步骤2：多维度分析（并行）generate_response# 步骤3：汇总生成（串行）)# 或者用管道符简写（效果相同）full_chain_pipe=preprocess|analysis|generate_response# 执行result=full_chain.invoke(" 你们的产品质量太差了！我要退货！ ")print(result)

执行流程图：

用户输入 │ ▼ ┌─────────────┐ │ 预处理 │ ← RunnableSequence（串行） │ - 清洗文本 │ │ - 统计长度 │ └─────────────┘ │ ▼ ├──────────────┬──────────────┬──────────────┐ ▼ ▼ ▼ ▼ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │意图识别 │ │紧急程度 │ │情感分析 │ │元数据 │ │ (LLM) │ │ (LLM) │ │ (LLM) │ │ (本地) │ └────────┘ └────────┘ └────────┘ └────────┘ │ │ │ │ └──────────────┴──────────────┴──────────────┘ │ ▼ ┌─────────────┐ │ 结果合并 │ ← RunnableParallel 输出字典 │ {intent, │ │ urgency, │ │ sentiment, │ │ meta} │ └─────────────┘ │ ▼ ┌─────────────┐ │ 汇总生成 │ ← RunnableSequence（串行） │ - 回复建议 │ │ - 路由决策 │ └─────────────┘ │ ▼ 最终输出

六、选择指南：什么时候用哪个？

开始 │ ├─→ 步骤之间是否有依赖？ │ │ │ ├─→ 是 → 用 RunnableSequence（串行） │ │ │ └─→ 否 → 步骤能否同时执行？ │ │ │ ├─→ 能 → 用 RunnableParallel（并行） │ │ │ └─→ 不确定 → 看性能需求 │ │ │ ├─→ 追求低延迟 → 并行 │ │ │ └─→ 逻辑简单 → 串行 │ └─→ 混合场景 → 组合使用：串行 | 并行 | 串行

场景推荐组件
数据清洗 → 向量化 → 检索RunnableSequence
同时生成摘要+关键词+情感RunnableParallel
预处理 → 多模型对比 → 投票决策Sequence → Parallel → Sequence
A/B 测试两个 PromptRunnableParallel
复杂 Agent 的多步骤推理RunnableSequence+ 条件分支

七、底层原理速览

组件核心方法执行机制
RunnableSequence 顺序调用invoke()result = step3.invoke(step2.invoke(step1.invoke(input)))
RunnableParallelasyncio.gather()并发各分支创建独立任务，同时调度，等待全部完成