LangChain LCEL 核心组件详解RunnableSequence vs RunnableParallel / RunnableMap发布日期 2026年5月15日关键词 LangChain, LCEL, RunnableSequence, RunnableParallel, RunnableMap, 串行执行, 并行执行一、LCEL 两大核心组件在 LangChain 的 LCELLangChain Expression Language中有两个平级的核心编排组件组件 核心作用 执行方式RunnableSequence 按顺序链式执行多个步骤 串行SequentialRunnableParallel / RunnableMap 同时执行多个独立分支 并行Parallel它们都是Runnable的子类地位平等只是编排策略不同。二、RunnableSequence串行执行2.1 什么是 RunnableSequenceRunnableSequence 将多个Runnable组件按顺序串联前一个的输出作为后一个的输入。2.2 两种等价写法fromlangchain_core.runnablesimportRunnableSequence,RunnableLambdadefstep1(x):returnx1defstep2(x):returnx*2defstep3(x):returnf结果:{x}# 写法一管道符推荐简洁直观chain_v1RunnableLambda(step1)|RunnableLambda(step2)|RunnableLambda(step3)# 写法二显式构造完全等价chain_v2RunnableSequence(RunnableLambda(step1),RunnableLambda(step2),RunnableLambda(step3))# 执行结果相同print(chain_v1.invoke(5))# 结果: 12print(chain_v2.invoke(5))# 结果: 122.3 执行流程输入: 5 │ ▼ [step1] 5 1 6 │ ▼ [step2] 6 * 2 12 │ ▼ [step3] 结果: 12 │ ▼ 输出: 结果: 12 总耗时 step1耗时 step2耗时 step3耗时2.4 适用场景✅ 数据预处理 → 模型调用 → 后处理✅ 有明确先后依赖的步骤✅ 需要前一步结果作为下一步输入三、RunnableParallel / RunnableMap并行执行3.1 什么是 RunnableParallelRunnableParallel别名 RunnableMap将输入同时分发给多个独立分支各分支同时执行最后合并结果。3.2 两种等价写法fromlangchain_core.runnablesimportRunnableParallel,RunnableMap,RunnableLambdadefadd_one(x):returnx1defmultiply_by_two(x):returnx*2defto_string(x):returnf值为:{x}# 写法一RunnableParallel推荐chain_v1RunnableParallel({added:RunnableLambda(add_one),doubled:RunnableLambda(multiply_by_two),original:RunnableLambda(to_string),})# 写法二RunnableMap完全等价别名chain_v2RunnableMap({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 llmChatOpenAI(modelgpt-4o-mini)# 1. RunnableSequence预处理串行preprocessRunnableLambda(lambdax:{raw:x,cleaned:x.strip().lower(),length:len(x.strip())})# 2. RunnableParallel并行分析并行analysisRunnableParallel({# 分支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_responseRunnableLambda(lambdax:{analysis:x,reply_suggestion:f根据分析结果意图-{x[intent].content}, 紧急度-{x[urgency].content}, 情感-{x[sentiment].content},routing:转人工ifx[urgency].content高else自动回复})# 组合串行 → 并行 → 串行 full_chainRunnableSequence(preprocess,# 步骤1预处理串行analysis,# 步骤2多维度分析并行generate_response# 步骤3汇总生成串行)# 或者用管道符简写效果相同full_chain_pipepreprocess|analysis|generate_response# 执行resultfull_chain.invoke( 你们的产品质量太差了我要退货 )print(result)执行流程图用户输入 │ ▼ ┌─────────────┐ │ 预处理 │ ← RunnableSequence串行 │ - 清洗文本 │ │ - 统计长度 │ └─────────────┘ │ ▼ ├──────────────┬──────────────┬──────────────┐ ▼ ▼ ▼ ▼ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │意图识别 │ │紧急程度 │ │情感分析 │ │元数据 │ │ (LLM) │ │ (LLM) │ │ (LLM) │ │ (本地) │ └────────┘ └────────┘ └────────┘ └────────┘ │ │ │ │ └──────────────┴──────────────┴──────────────┘ │ ▼ ┌─────────────┐ │ 结果合并 │ ← RunnableParallel 输出字典 │ {intent, │ │ urgency, │ │ sentiment, │ │ meta} │ └─────────────┘ │ ▼ ┌─────────────┐ │ 汇总生成 │ ← RunnableSequence串行 │ - 回复建议 │ │ - 路由决策 │ └─────────────┘ │ ▼ 最终输出六、选择指南什么时候用哪个开始 │ ├─→ 步骤之间是否有依赖 │ │ │ ├─→ 是 → 用 RunnableSequence串行 │ │ │ └─→ 否 → 步骤能否同时执行 │ │ │ ├─→ 能 → 用 RunnableParallel并行 │ │ │ └─→ 不确定 → 看性能需求 │ │ │ ├─→ 追求低延迟 → 并行 │ │ │ └─→ 逻辑简单 → 串行 │ └─→ 混合场景 → 组合使用串行 | 并行 | 串行场景 推荐组件数据清洗 → 向量化 → 检索RunnableSequence同时生成摘要关键词情感RunnableParallel预处理 → 多模型对比 → 投票决策Sequence → Parallel → SequenceA/B 测试两个 PromptRunnableParallel复杂 Agent 的多步骤推理RunnableSequence 条件分支七、底层原理速览组件 核心方法 执行机制RunnableSequence 顺序调用invoke()result step3.invoke(step2.invoke(step1.invoke(input)))RunnableParallelasyncio.gather()并发 各分支创建独立任务同时调度等待全部完成八、一句话总结RunnableSequence 是排队RunnableParallel是并发。两者平级互补组合起来能构建高效、灵活的 LLM 应用流水线。参考文档LangChain LCEL 官方文档RunnableSequence APIRunnableParallel API如果这篇文章对你有帮助欢迎点赞收藏有任何问题欢迎留言讨论。