LangChain LCEL 核心模式与组合

一、概念速查

LCEL 核心组件

组件	类型	作用	类比
Runnable	基类	所有可执行单元的抽象接口	函数签名
RunnableSequence	组合器	链式调用，前一个输出是后一个输入	管道 `
RunnableParallel	组合器	多个 Runnable 并行执行，合并输出	Promise.all
RunnablePassthrough	工具	透传输入，或调用函数后合并到输出	中间件
RunnableLambda	适配器	把任意 Python 函数包装为 Runnable	函数包装
RunnableBranch	路由	根据条件选择不同执行路径	if/else
RunnableMap	工具	对输入做字段映射和转换	map

各组件核心方法

# Runnable 接口三方法
invoke(input, config)          # 同步执行
ainvoke(input, config)         # 异步执行
batch(inputs, config)          # 批量执行
stream(input, config)          # 流式输出
astream(input, config)         # 异步流式

代码示例：从一个 Prompt 到完整 Chain

from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

prompt = ChatPromptTemplate.from_template("用一句话解释{concept}")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

chain = prompt | model | parser
print(chain.invoke({"concept": "LCEL"}))
# 输出类似：LCEL 是 LangChain 的声明式组合语法，用 | 运算符将组件串联成可执行的 Runnable 链。

并行执行与字段映射

from langchain_core.runnables import RunnableParallel

def extract_keywords(x: str) -> dict:
    return {"keywords": x[:20]}

chain = RunnableParallel(
    explanation=prompt | model | parser,
    metadata=RunnablePassthrough() | RunnableLambda(extract_keywords)
)

result = chain.invoke({"concept": "RunnableParallel"})
print(result["explanation"])  # LLM 输出
print(result["keywords"])     # "RunnableParallel"

二、底层原理

Runnable 协议架构

LCEL 的核心抽象是一个统一的 Runnable 协议。任何对象只要实现了 invoke/ainvoke/batch/stream 四个方法，就可以通过 | 运算符组合。运算符重载的实质是构造一个 RunnableSequence，内部维护了一个有序的子节点列表，执行时按序传递输出。

这种设计解决了传统回调式框架的两个根本问题：一是链中每个环节都可以单独测试（任何 Runnable 都可以独立 invoke），二是组合关系与执行逻辑解耦，运行时可以通过中间件拦截任意节点的输入输出。

RunnableSequence 的执行模型

class RunnableSequence:
    def __init__(self, *steps):
        self.steps = steps
    
    def invoke(self, input, config=None):
        for step in self.steps:
            input = step.invoke(input, config)
        return input

每一步接收上一步的输出作为输入。对于 ChatPromptTemplate，输入是 dict，输出是 ChatPromptValue。对于 ChatOpenAI，输入是 ChatPromptValue，输出是 ChatGeneration。对于 StrOutputParser，输入是 ChatGeneration，输出是 str。LCEL 自动处理这些类型转换，开发者不需要手动适配。

RunnableParallel 的 fork-join 模式

RunnableParallel 本质上是一个 fork-join 原语。输入被广播到所有分支，每个分支独立执行，结果以字段名聚合为 dict。这在需要同时做多件事的场景中极其有用，例如同时生成文本解释和提取元数据。

流式处理的实现机制

for chunk in chain.stream({"concept": "流式处理"}):
    print(chunk, end="", flush=True)

流式的底层依赖生成器协议。每个 Runnable 节点可选择实现 stream 方法返回迭代器。RunnableSequence 在流式模式下会逐级传递生成器，最后一个节点产出流式 token。StrOutputParser 是流式友好的——它逐个 yield 字符，而不是等全部生成再返回。

三、架构设计原则

1. 组合优于继承

所有行为通过 Runnable 组合实现，而不是通过子类化扩展。新增功能意味着接入新的 Runnable，而不是修改现有类的继承树。

2. 每个 Runnable 可独立测试

assert isinstance(prompt.invoke({"concept": "test"}), ChatPromptValue)
assert isinstance(model.invoke(prompt.invoke({"concept": "test"})), AIMessage)

这条原则直接来源于 Runnable 协议的统一性。如果某段 LCEL 链出错了，你可以从中间截取任意节点独立验证，而不是从头跑到尾 debug。

3. 类型安全优先于运行时容错

LCEL 在 __or__ 运算符中不做类型检查——类型不匹配会在 invoke 时报错，而不是在构造时静默吞掉。这鼓励开发者在组合阶段就发现接口不匹配，而不是留到生产运行时爆。

4. Config 透传与中间件能力

每个 Runnable 的 invoke 方法都接受一个 config 字典，框架用它透传回调、标签、元数据。这意味着你可以在不修改任何 Runnable 实现的前提下，通过 config 注入日志、追踪、限流等横切关注点。

5. 延迟执行优于即时求值

LCEL 链在构造时只是建立计算图，不会执行任何实际计算。prompt | model | parser 只是构造了一个 RunnableSequence 对象，直到调用 invoke 才真正触发执行。这让链可以被缓存、序列化、动态修改后再执行。