如何利用 LangChain 的 LCEL 语法优化复杂链条的并行执行效率

LCEL (LangChain Expression Language) 最大的价值不在于那个类似 Unix 管道的 | 符号，而在于它原生支持的异步并发处理。很多同学写 LangChain 习惯用传统的 SequentialChain，结果导致每个步骤都得等上一个结束，整个 Pipeline 慢得像蜗牛。

其实只要把同步的逻辑拆解，利用 RunnableParallel（或者简写为字典形式），就能让互不依赖的步骤在同一个线程池里并行跑。

举个实战场景：我最近在做一个“多维度文档分析”工具，需要同时从同一个输入中提取“摘要”、“情感分析”和“关键词”。如果串行执行，得等 LLM 回复三次，耗时 10s+；改成并行后，时间直接压缩到了单次 LLM 响应的时间。

具体的配置写法如下：

from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

model = ChatOpenAI(model="gpt-4o")

# 定义三个独立的任务
summary_chain = ChatPromptTemplate.from_template("总结这段文字: {text}") | model
sentiment_chain = ChatPromptTemplate.from_template("分析这段文字的情感: {text}") | model
keyword_chain = ChatPromptTemplate.from_template("提取这段文字的关键词: {text}") | model

# 核心技巧：使用字典构建并行分支
# 这里的 key 会成为输出字典的 key，value 是对应的 Runnable
map_chain = RunnableParallel({
    "summary": summary_chain,
    "sentiment": sentiment_chain,
    "keywords": keyword_chain,
    "original_text": RunnablePassthrough() # 把原文本透传给下一步
})

# 最后用一个汇总链把结果拼起来
final_chain = map_chain | ChatPromptTemplate.from_template(
    "综合分析结果：摘要是{summary}，情感是{sentiment}，关键词是{keywords}。"
) | model

# 必须用 ainvoke 才能触发真正的异步并行
# result = await final_chain.ainvoke({"text": "这里是长文本..."})

这里面有几个踩过的坑值得注意：

1. 必须调用 ainvoke 而非 invoke
如果你用 invoke，LangChain 内部虽然也会尝试优化，但无法发挥异步 IO 的最大优势。在 FastAPI 或异步框架中，一定要全链路 await。

2. 避免在并行分支中共享状态
LCEL 的并行分支是独立运行的。如果你试图在 summary_chain 里修改一个全局变量，然后在 sentiment_chain 里读取，大概率会因为竞态条件导致结果不可预测。状态传递必须严格通过 RunnablePassthrough 显式传递。

3. 令牌消耗激增
并行执行虽然快，但它瞬间发起了多个请求。如果你的并发量很大，很容易触碰到 OpenAI 的 TPM (Tokens Per Minute) 限制。建议在 ChatOpenAI 初始化时配置 max_retries，防止因为并发过高导致部分分支请求失败而导致整个链条崩溃。

这种结构把复杂的逻辑解耦成了“并行采集 → 串行聚合”，在处理 RAG 检索增强时，把多个向量数据库的查询放进 RunnableParallel 就能显著降低首字响应延迟。