本文深入介绍了LangChain Expression Language (LCEL)，这是一种声明式语言，用于构建LangChain中的Chain。文章详细阐述了LCEL的基本构成，包括管道符、RunnableSequence和RunnableParallel等核心概念，并结合实例分析了LCEL在结构化输出和RAG（Retrieval-Augmented Generation）应用中的实践，帮助读者理解其优势和应用场景。

💡 **LCEL 的核心是声明式语言**: LCEL 提供一种声明式的方式来定义 Chain，即由多个可运行单元组合而成的调用链。这种方式使得构建和管理复杂流程更加容易。

🔗 **管道符简化流程构建**: LCEL 使用管道符（|）重载操作符，简化了Chain的构建，例如 `A | B` 等价于 `RunnableSequence([A, B])`，使得代码更具可读性。

🧩 **RunnableSequence 与 RunnableParallel**: RunnableSequence 允许顺序组合多个可运行单元，前一个单元的输出成为下一个单元的输入；而 RunnableParallel 支持并行运行多个单元，并为每个单元提供相同的输入。它们是LCEL的基本组合单元。

📚 **LCEL 在 RAG 应用中的实践**: 文章展示了如何使用 LCEL 构建 RAG 应用，包括加载文档、切分文档、灌库、检索和返回结果等关键环节，展示了LCEL在实际应用中的灵活性和强大功能。

什么是LCEL

LCEL 全称LangChain Expression Language 是LangChain定义的一种声明式语言，其优势在于能够轻松构建不同调用顺序的Chain(由LCEL构建的调用链被称为Chain)

举个例子比如处理结构化输出

from langchain.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.runnables import RunnablePassthroughfrom pydantic import BaseModel, Field, validatorfrom typing import List, Dict, Optionalfrom enum import Enumimport jsonfrom langchain.chat_models import init_chat_model# 输出结构class SortEnum(str, Enum):    data = 'data'    price = 'price'class OrderingEnum(str, Enum):    ascend = 'ascend'    descend = 'descend'        class Semantics(BaseModel):    name: Optional[str] = Field(description="流量包名称", default=None)    price_lower: Optional[int] = Field(description="价格下限", default=None)    price_upper: Optional[int] = Field(description="价格上限", default=None)    data_lower: Optional[int] = Field(description="流量下限", default=None)    data_upper: Optional[int] = Field(description="流量上限", default=None)    sort_by: Optional[SortEnum] = Field(description="按价格或流量排序", default=None)    ordering: Optional[OrderingEnum] = Field(        description="升序或降序排列", default=None)# Prompt 模板prompt = ChatPromptTemplate.from_messages(    [        ("system", "你是一个语义解析器。你的任务是将用户的输入解析成JSON表示。不要回答用户的问题。"),        ("human", "{text}"),    ])# 模型llm = init_chat_model("gpt-4o", model_provider="openai")structured_llm = llm.with_structured_output(Semantics)# LCEL 表达式runnable = (    {"text": RunnablePassthrough()} | prompt | structured_llm)# 直接运行ret = runnable.invoke("不超过100元的流量大的套餐有哪些")print(    json.dumps(        ret.model_dump(),        indent = 4,        ensure_ascii=False    ))

输出：

{    "name": null,    "price_lower": null,    "price_upper": 100,    "data_lower": null,    "data_upper": null,    "sort_by": "data",    "ordering": "descend"

runnable = (    {"text": RunnablePassthrough()} | prompt | structured_llm)

这一部分就是LCEL表达式，不过看上去还是有点疑惑，dict或prompt或structured_llm？python里的或运算符也不是管道符啊，而且一个逻辑运算怎么就能调invoke方法了？

下面我们来说明一下LCEL的基本构成

LCEL的基本构成

首先我们来看管道符是怎么回事。

管道符

这里的管道符，是由LangChain重载过的操作符，它的作用是表示from A to B，如果你觉得这样的写法容易让人混乱，LangChain也支持其他的写法:

chain = A.pipe(B)

而这两个写法都等价于

chain = RunnableSequence([A, B])

而RunnableSequence和RunnableParallel是LCEL中的两个基本组合单元，所谓基本组合单元就是一堆工作单元的组合。

RunnableSequence 和 RunnableParallel

先来解释下什么是Runable，Runable是LCEL的一个基本概念，可以理解为是能够调用，批处理，流处理，转换和组合的工作单元。

RunnableSequence是多个工作单元的组合，可以让我们按顺序组装多个Runable，上一个Runable的输出是下一个Runable的输入，举个例子:

from langchain_core.runnables import RunnableSequence,RunnableLambdadef func1(num):    num +=1    return numdef func2(num):    num+=1    return num        # 函数需要转换为RunnableLambda，不然会触发异常runnable1 = RunnableLambda(func1) runnable2 = RunnableLambda(func2) chain = RunnableSequence(runnable1, runnable2)print(chain.invoke(10))

这种写法就等价于

chain = runnable1|runnable2

实际上，LCEL表达式会被强制转换为

chain = RunnableSequence(runnable1, runnable2)

RunnableParallel，同样是多个工作单元的组合，不同的是RunnableParallel可以支持同时运行多个对象，并为每个可运行对象提供相同的输入

在LCEL中字典会被强制转换为RunnableParallel，我们来看一个例子：

import asynciofrom langchain_core.runnables import RunnableLambdadef add_one(x: int) -> int:    return x + 1def mul_two(x: int) -> int:    return x * 2def mul_three(x: int) -> int:    return x * 3runnable_1 = RunnableLambda(add_one)runnable_2 = RunnableLambda(mul_two)runnable_3 = RunnableLambda(mul_three)sequence = runnable_1 | {     "mul_two": runnable_2,    "mul_three": runnable_3,}async def main():    res = sequence.invoke(1)    print('invoke:',res)    res = await sequence.ainvoke(1)    print('ainvoke:',res)    res = sequence.batch([1, 2, 3])    print('batch:',res)    res = await sequence.abatch([1, 2, 3])    print('abatch:',res)asyncio.run(main())

上面的

sequence = runnable_1 | {     "mul_two": runnable_2,    "mul_three": runnable_3,}

等价于

sequence =RunnableSequence(runnable_1,RunnableParallel(mul_two=runnable_2,mul_three=runnable_3))

RunnableSequence 和 RunnableParallel 是两个最基本的组合单元，其他的组合单元都是这两个组合单元的变体

以上就是LCEL的基本介绍，怎么样,是不是有点复杂？LangChain官方也这么觉得，以下是官方文档原话：

虽然我们已经看到用户在生产中运行具有数百个步骤的链，但我们通常建议使用 LCEL 来执行更简单的编排任务。当应用程序需要复杂的状态管理、分支、周期或多个代理时，我们建议用户利用 LangGraph。

基于LCEL的RAG

下面我们就简单的构建一个RAG应用,使用LCEL实现一下RAG的核心环节

加载文档切分文档灌库检索返回结果

from langchain.chat_models import init_chat_modelfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_openai import OpenAIEmbeddingsfrom langchain_text_splitters import RecursiveCharacterTextSplitterfrom langchain_community.vectorstores import FAISSfrom langchain.chains import RetrievalQAfrom langchain_community.document_loaders import PyMuPDFLoaderfrom langchain.schema.output_parser import StrOutputParserfrom langchain.schema.runnable import RunnablePassthroughimport dotenvdotenv.load_dotenv()# 加载文档loader = PyMuPDFLoader("../data/deepseek-v3-1-4.pdf")pages = loader.load_and_split()# 文档切分text_splitter = RecursiveCharacterTextSplitter(    chunk_size=300,    chunk_overlap=100,    length_function=len,    add_start_index=True,)texts = text_splitter.create_documents(    [page.page_content for page in pages[:4]])# 灌库embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")db = FAISS.from_documents(texts, embeddings)# 检索 top-2 结果retriever = db.as_retriever(search_kwargs={"k": 2})# Prompt模板template = """Answer the question based only on the following context:{context}Question: {question}"""llm = init_chat_model("gpt-4o", model_provider="openai")prompt = ChatPromptTemplate.from_template(template)# Chainrag_chain = (    {"question": RunnablePassthrough(), "context": retriever}    | prompt    | llm    | StrOutputParser())print(rag_chain.invoke("deepseek v3有多少参数"))

输出：

DeepSeek-V3有6710亿（671B）个总参数。