LangChain并行流程实战：用RunnableParallel把问答链延迟从4.5秒降到1.7秒

做AI开发有一段时间的人，基本都会撞上同一个坎：单条链越写越长，每个节点都要干等上一个节点跑完，一问一答动不动就是七八秒，用户体验直接拉胯。我第一次意识到必须上并行流程，是在做一个多路检索问答助手的时候——同一个问题要同时查知识库、抽关键词、生成精炼问法，三个动作明明互不依赖，却老老实实排队跑完。那次之后我认真把LangChain的并行相关组件从头捋了一遍：Runnable、RunnableParallel、map、batch，这些东西拆开看都不难，组合起来就是标题里说的"并行流程"这条主线。这篇文章写给已经能跑通一条简单链、想把请求延迟再压一压的人，从头到尾用一个实际示例讲透，代码可以直接抄走改改就用。

1. 为什么要把流程拆成并行

1.1 从一次慢请求说起

假设现在有一个问答链，用户提问之后要做三件事：把问题改写成更规范的问句、在文档库里检索相关片段、给问题抽一组关键词。如果按顺序写，链路就是 A 等 B、B 等 C，总耗时等于三次模型调用和检索耗时的累加。假设每次调用平均1.5秒，三次就是4.5秒，用户视角就是"转圈圈转好久，到底行不行"。

但如果这三个动作互相没有依赖，它们其实可以同时发起。并行之后的总耗时不再是相加，而是约等于最慢那一个动作的耗时，同样是1.5秒级别的服务，整体响应直接砍到1.5秒左右。这个差距不是优化了几个百分点，而是把三分之二以上的等待时间直接抹掉了。

列一张对比表更直观：

这张表背后是一个很朴素的道理：并行本身不减少工作量，它减少的是"等待的浪费"。多个任务都在等模型返回的时候，串行是大家排队，并行是大家同时排队，谁先回来谁先等着，总时长由最慢的那个人决定。

1.2 并行流程的适用边界

也不是所有地方都适合并行，判断标准很简单：任务之间有没有数据依赖。我在项目里总结的适用场景是这几类：

• 多路检索：同一问题去多个不同的数据源或向量库查询，各自独立返回；
• 多角度分析：对同一段文本同时做摘要、关键词抽取、情感判断、实体识别；
• 多模型交叉验证：让多个模型分别给出答案，再合并做对比或投票；
• 独立的前置处理：比如同时拿到用户问题、上下文时间、历史记录，再统一组装提示词。

不适合并行的场景也要心里有数。后一步需要前一步输出的流程，比如"先生成提纲再按提纲写正文"，这是典型串行依赖，强行并行等于白做。还有共享可变状态的流程，并行分支之间如果互相读写同一个变量，结果会变得完全不可预期。最后一种情况是上游服务并发能力极差、限流极严格，这时候并行前必须先做并发控制，否则第一批请求就把配额打满了，后面全在报错。

我的习惯是：动手写代码前先画一画数据流图，把每个节点的输入来源标出来。如果两个节点的输入都只来自同一个原始输入、彼此之间没有连线，它们就是天然的并行候选。这一步花五分钟，比盲目套模板有用得多。

2. 并行流程的核心概念：Runnable 与 RunnableParallel

2.1 Runnable 统一协议

LangChain里几乎所有可执行对象都实现了Runnable协议。所谓"实现协议"，就是它统一提供了几个方法：invoke（单次调用）、batch（批量调用）、stream（流式输出），以及各自的异步版本ainvoke、abatch、astream。只要对象实现了这套方法，它就能参与LCEL表达式（也就是LangChain Expression Language）的组合。

为什么要强调这个？因为并行流程本质上就是把这些Runnable对象组织成"一横一纵"的结构：横向是每个分支本身，纵向是分支之间的汇合点。LCEL管道操作符 | 可以把多个Runnable串起来，形成一个串行链；RunnableParallel则把多个Runnable并联起来。串并联结构和电路图是一个道理：串联是电流依次流过每个元件，并联是电流同时流过多个支路。

一个Runnable只要实现了invoke/batch/stream，就能参与这种组合。所以普通Python函数也能并行跑，只要用RunnableLambda包一层，或者用RunnablePassthrough.assign在流程里注入计算结果。这套抽象最舒服的地方在于：你不必关心某个分支内部到底调了什么，只要它对外是Runnable，就能塞进并行容器。

2.2 创建并行流程的三种常用姿势

第一种是直接传字典给LCEL表达式。在LCEL里，一个普通字典就可以声明并行流程：

python复制parallel_flow = {
    "summary": summary_chain,
    "keywords": keywords_chain,
    "docs": retrieval_chain
}

这种方式最直观，等于声明"我要并行跑三路，结果放进以summary、keywords、docs为键的字典里"。字典的key就是输出字典的key，所以命名要认真一点，别用a、b、c这种，后面看代码会疯。

第二种是显式创建RunnableParallel实例：

python复制from langchain_core.runnables import RunnableParallel

parallel_flow = RunnableParallel(
    summary=summary_chain,
    keywords=keywords_chain,
    docs=retrieval_chain
)

两种写法在功能上等价。早期版本里这个类叫RunnableMap，后来改名成了RunnableParallel，老代码里看到RunnableMap不用慌，逻辑完全一致，只是命名更符合直觉了。

第三种是用.map()方法，它处理的是"列表进、列表出"的场景：把某个Runnable包装成可以一次处理多条输入的形式。注意.map()和RunnableParallel解决的不是同一个问题：RunnableParallel是对"同一个输入做多个分支处理"，.map()是对"多个输入做同一个处理"。两者可以组合使用，比如先对一批文档做预处理，再把预处理结果并联给多个分析分支。

2.3 并行之后的结果如何合并

RunnableParallel的输出是一个字典，key是每个分支的名字，value是对应分支的输出。这个字典可以直接作为下一个Runnable的输入，也可以用RunnablePassthrough.assign往里面继续扩展字段。

python复制final_chain = {
    "qa": qa_chain,
    "mood": mood_chain,
} | RunnablePassthrough.assign(
    merged=lambda x: merge(x["qa"], x["mood"])
)

这里的lambda只是示例，真实项目里我建议把合并逻辑写成一个独立函数再传进去，别在管道里堆一大坨匿名函数，后期调试会想哭。

还有一个细节容易被忽略：RunnableParallel在派发输入时，是把原始输入复制给每个分支。也就是说每个分支拿到的都是完整的输入字典，而不是被切分好的子集。如果某个分支只用其中一个字段，其他字段对它就是冗余，不影响正确性，但会影响传输效率。节点之间传大体积文档时要注意，复制和序列化开销会随输入体量上升。

异步场景下使用方式也一致，只是方法名换成异步版本：

python复制result = await parallel_flow.ainvoke({"question": "..."})
results = await parallel_flow.abatch([{"question": "..."}, {"question": "..."}])

需要注意的是，ainvoke和abatch必须在异步环境中调用，要么在async函数里用await，要么用asyncio.run包裹。我见过不少人在同步代码里直接写await，然后报语法错误，一脸懵。

3. 实操记录：一个多路并行问答助手

3.1 场景设计与数据准备

我实际做过一个技术问答助手，输入是一个问题，输出是一份结构化的回答。核心思路是：单条链太慢，拆成三个互不依赖的分支并行跑，最后汇总成一个答案。

三个分支分别是：

• 分支A：生成"问题精炼版"，把用户口语化的问题改写成更规范的问句；
• 分支B：抽取关键词，方便后续做标签聚合；
• 分支C：从文档库检索相关片段，作为回答的事实依据。

这三个分支都只依赖原始问题，互相之间没有数据交换，是标准的并行结构。

先定义好各分支的链条：

python复制from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnableLambda
from operator import itemgetter

# 模型实例按你实际接入的服务替换
llm = get_llm_instance()

refine_prompt = ChatPromptTemplate.from_template(
    "把下面的用户问题改写成更规范的技术问句，只输出改写结果：\n{question}"
)
keywords_prompt = ChatPromptTemplate.from_template(
    "提取下面问题中的关键词，用中文逗号分隔：\n{question}"
)

refine_chain = refine_prompt | llm | StrOutputParser()
keywords_chain = keywords_prompt | llm | StrOutputParser()

# 从输入字典里取出 question 字段，再交给检索器
retriever_chain = itemgetter("question") | vectorstore.as_retriever(
    search_kwargs={"k": 3}
)
format_docs = RunnableLambda(
    lambda docs: "\n\n".join(doc.page_content for doc in docs)
)

这里vectorstore假设你已经构建好了。format_docs是为了把检索返回的Document对象转成纯文本，否则后续拼提示词时会出现一堆对象字符串，模型根本看不懂。

3.2 串行版本先跑一遍

为了对比效果，我先写一个串行版本，逻辑就是先精炼问题，再用精炼结果抽关键词，最后用精炼结果去检索。用普通Python函数表示更直观：

python复制import time

def serial_invoke(question: str):
    start = time.perf_counter()

    refined = refine_chain.invoke({"question": question})
    keywords = keywords_chain.invoke({"question": refined})
    docs = format_docs.invoke(retriever_chain.invoke({"question": refined}))

    elapsed = time.perf_counter() - start
    return {
        "refined": refined,
        "keywords": keywords,
        "contexts": docs,
        "elapsed": round(elapsed, 2)
    }

这里有个很关键的依赖：keywords_chain和retriever_chain用的都是refined的结果，而不是原始question。也就是说第二个和第三个动作都要等refine_chain跑完才能开始，没有任何重叠空间。

实测跑下来的耗时大约4.2秒，基本就是三次调用的耗时相加。这个数字符合预期，但也非常难看。

3.3 改用 RunnableParallel 并联

把串行结构改造成并行结构，核心是把三个分支放进同一个并行容器，同时保留原始question字段传给后续提示词。

python复制pass_question = RunnableLambda(lambda x: x["question"])

parallel_flow = RunnableParallel(
    question=pass_question,
    refined=refine_chain,
    keywords=keywords_chain,
    contexts=retriever_chain | format_docs
)

merged_prompt = ChatPromptTemplate.from_template("""
根据以下材料回答问题。
问题：{question}
精炼问法：{refined}
关键词：{keywords}
参考资料：
{contexts}
回答要求：先给出结论，再补充依据。
""")

full_chain = parallel_flow | merged_prompt | llm | StrOutputParser()

注意这里我把question单独用RunnableLambda取出来了。原因是RunnableParallel的输出字典里不会自动包含原始输入，而merged_prompt需要用到question字段，所以必须显式传一个分支去透传它。

第二次实测，整个流程耗时大约1.7秒。不是某个分支变快了，而是三个动作的等待时间重叠了。第一次跑通这个数字的时候，我是真的觉得LCEL这套抽象值了。

3.4 批量场景用 map 与 batch

除了"一个输入多路处理"，还有"多个输入同时处理"的批量场景。比如要给10篇文档各生成一句话摘要，用链的.batch()一次性传入列表：

python复制summaries = summary_chain.batch(
    [{"text": doc} for doc in docs]
)

.batch()内部会并行执行，返回一个与输入顺序对应的list。顺序是对应的这件事很重要，批量结果的顺序由输入列表的顺序决定，所以后面做zip、拼接都不会错位。

.map()则是把单条链升级为"处理列表"的链：

python复制list_summary_chain = summary_chain.map()
result = list_summary_chain.invoke([{"text": doc1}, {"text": doc2}])

两者区别在于：.batch()是链层面的一次性批量调用；.map()则是构造了一个新链，之后可以被继续组合。需要把"批量处理"作为一个环节嵌入更大流程时用.map()，只是临时跑一批数据用.batch()更省事。

批量并行同样要警惕上游限流。我有一次一次性batch几十条请求，结果直接被限流刷了一脸，整批返回全是错误。解决办法要么拆成小批次循环，要么在模型调用层加自定义重试策略。后面会细说。

4. 并行流程与路由、回退机制的联动

4.1 用 RunnableBranch 做前置判断

真实场景里，流程往往不是一条直线：先判断问题类型，再决定走哪条并行子链。RunnableBranch就是干这个的，它按条件路由，满足某个条件的走对应分支，都不满足就走兜底分支。

python复制from langchain_core.runnables import RunnableBranch

def is_technical(question: str) -> bool:
    return "报错" in question or "安装" in question

tech_flow = RunnableParallel(
    debug=collect_logs_chain,
    docs=retriever_chain
)
common_flow = RunnableParallel(
    docs=retriever_chain,
    summary=summary_chain
)

router = RunnableBranch(
    (is_technical, tech_flow),
    common_flow
)

这种结构下，路由判断是串行的，但一旦命中分支，分支内部又是并行的。整体延迟控制得不错，代码结构一眼就能看出来"先分叉、再并联"的意图。

4.2 用 fallbacks 给并行分支兜底

并行流程有一个天生的风险：任何一个分支抛异常，整个invoke都会抛异常。三路并行里有一路模型超时，另外两路的成果全部作废。这个问题在串行链上也存在，但并行放大了失败概率——分支越多，整体失败的可能性越高。

解决办法是给敏感分支单独加fallback：

python复制safe_refine_chain = refine_chain.with_fallbacks(
    [simple_refine_chain],
    exception_key="error"
)

with_fallbacks接收一个列表，列表里的备用链会按顺序尝试，当前一个抛异常时自动切换到下一个。exception_key这个参数可以把异常信息透传到备用链的输入里，方便备用链根据异常做不同处理。

我在实际项目里的原则是：调用外部模型的网络型分支必加fallback，本地检索、本地计算的分支反而不太加。因为网络调用失败是常态，本地逻辑失败通常意味着代码写错了，加fallback反而掩盖问题。

4.3 把并行结果喂给下游上下文

并行分支的结果合并之后，经常要经过一轮"上下文组装"再交给最终模型。这一步的关键是格式化。我一般会写一个辅助函数：

python复制def format_contexts(contexts):
    return "\n".join(
        f"[{i+1}] {c}" for i, c in enumerate(contexts)
    )

把检索片段、关键词列表转成干净规整的字符串，避免直接把复杂对象塞进提示词。这一步看着不起眼，却是并行流程里最容易出乱子的地方：并行出来的是结构化字典，提示词模板里必须字段名对得上，少一个键，最终模型就拿不到那部分信息。我在code review时经常看到这类错误，模板里写的是{contexts}，并行分支key写的是context，运行时不报错，但生成的内容明显缺了一段。

5. 高频问题与避坑实录

5.1 问题速查表

5.2 实战心得与调试技巧

第一，并行不是免费的。它节省的是时间，但会放大调用频率、占用更多连接、带来更多失败点。上线前一定要给整体流程加超时和重试，不然一个慢分支把整个请求拖死的场景一定会遇到。我一般会给"调用外部模型"的子链统一包一个超时时间，超过预期就直接走fallback。

第二，调试并行流程时，先别急着看整体输出。把parallel_flow单独invoke一次，用同样的输入，观察它到底输出一个什么样的字典。字典的结构就是后续所有流程的契约，契约对了，后面的问题少一半。我的做法是在流程里临时打印输出结构，确认每个字段的类型，再往下接prompt模板。

第三，异步别混着用。我踩过的坑是：在一个async环境里，某个分支内部还在用同步的invoke，结果整个事件循环被卡住，请求全军覆没。当时的解决办法是统一改成ainvoke、abatch；另一个办法是保持全同步，靠LangChain内部的线程池实现并行。反正一个流程里不要混用两套执行模型，混用就是给自己埋雷。

第四，善用可视化工具。chain.get_graph().print_ascii()能把LCEL的执行结构图打印出来，节点和依赖关系清清楚楚。并行分支在图上会呈现为并列节点，一眼就能看出哪几条链是并联的、哪几步有依赖。这个命令对review别人写的复杂链特别有用，比对着代码猜结构快多了。

第五，控制分支数量。别把并行分支堆到十几路，超出上游服务的并发能力之后，只会互相拖累。我的经验是三到五路并行收益最明显，响应时间接近最慢分支；到了七八路以上，整体收益曲线就开始平了，而限流和报错开始变多。

最后分享一个小技巧：给每个分支加一段统一格式的时间日志，记录"分支名-耗时-是否成功"。跑完一轮请求后按耗时排序，瓶颈藏在哪一目了然。并行流程一旦养成这个监控习惯，就不再是一个模糊的"加速手段"，而是你可以精确控制、按需调节的延迟工具。我后来做性能优化，基本都是从这份日志入手，而不是靠猜。