如何利用 LangChain 的 LCEL 语法优化多步 RAG 链的并发执行效率

LCEL (LangChain Expression Language) 最核心的价值不在于那几个管道符 |，而在于它内置的异步并行处理机制。很多同学写 RAG 链的时候习惯用传统的 Python 顺序调用，导致检索（Retrieval）和文档处理（Processing）成了串行瓶颈，整个响应延迟（Latency）直接翻倍。

在处理多步 RAG（比如需要同时从向量库、知识图谱和 Web 搜索三个渠道取数）时，直接用 RunnableParallel 能瞬间把执行时间从「三者之和」压缩到「最慢的一个」。

实战配置技巧

不要在 chain 外部写 asyncio.gather，直接把并行逻辑写在 LCEL 声明里。以下是一个典型的并发检索配置：

from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 定义三个独立的检索分支
retriever_vector = vector_store.as_retriever()
retriever_graph = graph_store.as_retriever()
retriever_web = web_search_tool

# 使用 RunnableParallel 构建并发层
# 这里的 key 会直接传递给下游的 Prompt
parallel_retrieval = RunnableParallel({
    "vector_docs": retriever_vector,
    "graph_docs": retriever_graph,
    "web_docs": retriever_web,
    "original_query": RunnablePassthrough()
})

# 最终链条：并发检索 -> 格式化合并 -> LLM -> 解析
rag_chain = (
    parallel_retrieval 
    | prompt 
    | llm 
    | StrOutputParser()
)

# 必须使用 ainvoke 才能触发真正的异步并发
result = await rag_chain.ainvoke("什么是LCEL的并行机制？")

踩过的坑与避坑指南

1. 忘记使用 ainvoke：这是最常见的坑。如果你用 .invoke()，LCEL 内部虽然定义了并行，但在同步环境下依然是顺序执行的，完全没起到加速作用。必须搭配 async/await。

2. 令牌（Tokens）爆炸：并发检索回来的上下文量是叠加的。如果三个分支都返回 5 个文档，Prompt 瞬间会被撑爆。建议在每个 retriever 后面接一个自定义的 RunnableLambda 进行长度过滤或重排序（Rerank）。

3. 依赖传递失效：在 RunnableParallel 中，如果下游 Prompt 需要原始问题，记得加上 "original_query": RunnablePassthrough()，否则并行分支执行完后，输入值会被替换为并行结果字典，导致 LLM 拿不到用户最初问了什么。

效率提升量化

在我的一个法律文档问答项目中，通过将「向量检索」和「关键词索引检索」改为 RunnableParallel 并行执行，端到端响应时间（TTFT）从 4.2s 降低到了 2.1s 左右，基本省去了其中一个检索器的等待时间。