LangChain Expression Language (LCEL) 是一种强大的工具，用于构建复杂的数据处理链，使AI应用开发更加简洁和高效。通过管道符号“|”连接各种组件，LCEL能够轻松实现组件的任意组合，并提供统一的接口规范，方便监控与调试。文章详细介绍了LCEL的优势，包括代码简洁性、灵活性和可维护性。同时，深入解析了Runnable组件的核心方法（invoke, batch, stream, ainvoke），以及RunnableBranch（条件分支）、RunnableLambda（函数转换）和RunnableParallel（并行处理）等高级用法，展示了如何通过LCEL构建更复杂的AI流程，如多语言翻译、并行任务处理和数据重组等，为开发者提供了一种更优化的AI应用开发范式。

💡 LCEL表达式是LangChain框架的核心，它通过管道符号“|”将提示词模板、大语言模型、输出解析器等组件串联起来，形成清晰、简洁的数据处理链，极大地提高了代码的可读性和可维护性，解决了早期JS中回调地狱般嵌套调用的问题。

🚀 Runnable接口是LCEL的基础，它定义了所有可执行组件的标准方法，如invoke（同步执行）、batch（批量执行）、stream（流式执行）和ainvoke（异步执行），使得不同组件之间能够无缝连接和组合，实现灵活的AI应用构建。

⚖️ RunnableBranch允许根据输入条件执行不同的处理逻辑，实现AI应用的条件分支功能。例如，可以根据用户输入判断语言类型，然后调用对应的翻译模型进行处理，增强了AI应用的适应性。

🛠️ RunnableLambda可以将普通函数转换为可执行组件，方便地将其集成到LCEL链中，实现如字符计数等辅助功能。RunnableParallel则支持多个链的并行执行，提升了AI应用的效率，可以同时处理诗歌创作和数学问题解答等任务。

🔄 RunnablePassthrough用于将输入数据原样传递或进行数据重组，特别是其assign方法，允许在链式调用中动态添加新的属性（如检索信息），为后续组件提供更丰富的数据上下文，进一步优化了AI应用的复杂流程设计。

在前面几篇文章中，我们已经掌握了LangChain的核心组件：提示词模板、大语言模型、输出解析器。细心的读者可能发现，在使用这些组件时，经常会看到类似 prompt | llm | parser 这样的链式操作。这就是今天重点介绍的LCEL（LangChain Expression Language）表达式。

在平时开发中，经常需要将多个组件组合起来形成完整的处理流程，将上一个组件的输出作为下一个组件的输入，在不使用LCEL表达式之前，就会写出这种代码：使用 invoke() 进行层层嵌套，这就好比早期 JS 中的回调地狱，结构混乱、难以维护，并且出现错误很难判断是哪一步出了问题。

# 1.构建提示词prompt = ChatPromptTemplate.from_messages([    ("system", "你是一个资深文学家"),    ("human", "请简短赏析{name}这首诗，并给出评价")])# 2.创建模型llm = ChatOpenAI()# 3.创建字符串输出解析器parser = StrOutputParser()# 4.调用模型返回结果result = parser.invoke(    llm.invoke(        prompt.invoke({"name": "江雪"})    ))

而LCEL让这个过程变得简洁直观，通过管道符号进行连接，可以很轻松地构建出功能强大的AI应用。

文中所有示例代码：github.com/wzycoding/l…

一、什么是LCEL表达式

LCEL（LangChain Expression Language）是LangChain框架的表达式语言，它提供了一种声明式的方式来构建复杂的数据处理链。通过LCEL，我们可以使用管道符号 | 将不同的组件连接起来，形成一个完整的数据处理链。

LCEL有以下优点：

1、代码更加简洁：用管道符号连接组件，代码更加简洁易读

2、可任意组合：任意Runnable组件都可以自由组合，构建复杂的处理逻辑

3、统一接口规范：所有Runnable组件都遵循统一的接口规范

4、方便监控与调试：LangChain内置了日志和监控功能，方便调试和优化

下面是使用LCEL表达式的案例：

# 1.构建提示词prompt = ChatPromptTemplate.from_messages([    ("system", "你是一个资深文学家"),    ("human", "请简短赏析{name}这首诗，并给出评价")])# 2.创建模型llm = ChatOpenAI()# 3.创建字符串输出解析器parser = StrOutputParser()# 4.构建链chain = prompt | llm | parser# 5.执行链print(f"输出结果：{chain.invoke({'name': '题西林壁'})}")

显而易见，LCEL写法更加简洁，而且表达了清晰的数据流向：输入经过提示词模板处理，然后将PromptValue传递给大语言模型，最后将大语言模型输出的Message传递给输出解析器，经过输出解析器解析得到最终结果。

二、什么是Runnable组件

在深入LCEL之前，首先需要理解Runnable接口。Runnable是LangChain中所有可执行组件的基础接口，它定义了组件应该具备的标准方法。前面介绍的LangChain组件如提示词模板、模型、输出解析器等，都实现了Runnable接口，这就是为什么这些组件可以使用管道符进行连接的原因。

在Runnable接口中定义了以下核心方法：

invoke(input)：同步执行，处理单个输入，最常用的方法

batch(inputs)：批量执行，处理多个输入，提升处理效率

stream(input)：流式执行，逐步返回结果，经典的使用场景是大模型是一点点输出的，不是一下返回整个结果，可以通过 stream() 方法，进行流式输出

ainvoke(input)：异步执行，用于高并发场景

三、RunnableBranch条件分支

在LangChain中提供了类RunnableBranch来完成LCEL中的条件分支判断，它可以根据输入的不同采用不同的处理逻辑，具体示例如下，在下方示例中程序会根据用户输入中是否包含‘日语’、‘韩语’等关键词，来选择对应的提示词进行处理。根据判断结果，再执行不同的逻辑分支。

import dotenvfrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnableBranchfrom langchain_openai import ChatOpenAI# 读取env配置dotenv.load_dotenv()def judge_language(inputs):    """判断语言种类"""    query = inputs["query"]    if "日语" in query:        return "japanese"    elif "韩语" in query:        return "korean"    else:        return "english"# 1.构建提示词english_prompt = ChatPromptTemplate.from_messages([    ("system", "你是一个英语翻译专家，你叫小英"),    ("human", "{query}")])japanese_prompt = ChatPromptTemplate.from_messages([    ("system", "你是一个日语翻译专家，你叫小日"),    ("human", "{query}")])korean_prompt = ChatPromptTemplate.from_messages([    ("system", "你是一个韩语翻译专家，你叫小韩"),    ("human", "{query}")])# 2.创建模型llm = ChatOpenAI()# 3.创建字符串输出解析器parser = StrOutputParser()# 4.构建链分支结构，默认分支为英语chain = RunnableBranch(    (lambda x: judge_language(x) == "japanese", japanese_prompt | llm | parser),    (lambda x: judge_language(x) == "korean", korean_prompt | llm | parser),    (english_prompt | llm | parser))# 5.执行链print(f"输出结果：{chain.invoke({'query': '请你用韩语翻译这句话：“我爱你”。并且告诉我你叫什么'})}")

执行结果如下，根据执行结果，执行的是韩语分支。

输出结果：“我爱你”用韩语是：“사랑해” (Saranghae)。我叫小韩，很高兴为你服务！😊

四、RunnableLambda函数转换为可执行组件

LangChain还提供了类RunnableLambda，它可以非常方便的将函数转换为可执行组件，如下示例，将字符个数统计函数包装成一个RunnableLambda，并参与链执行。

import dotenvfrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnableLambdafrom langchain_openai import ChatOpenAI# 读取env配置dotenv.load_dotenv()def character_counter(text):    """统计输出字符个数"""    return len(text)# 1.构建提示词prompt = ChatPromptTemplate.from_messages([    ("system", "你是一个资深文学家"),    ("human", "请以{subject}为主题写一首古诗")])# 2.创建模型llm = ChatOpenAI()# 3.创建字符串输出解析器parser = StrOutputParser()# 4.构建链chain = prompt | llm | parser | RunnableLambda(character_counter)# 5.执行链print(f"输出结果：{chain.invoke({'subject': '大雪'})}")

执行结果：

输出结果：67

五、RunnableParallel并行处理

在某些需求中，为了提高执行效率，可能会有两个链并行执行的情况，比如同时进行古诗创作和解答数学题。RunnableParallel能让多个链并行处理，最终同时返回结果。

5.1 并行处理

RunnableParallel基础用法示例如下，RunnableParallel中需要传入一个字典结构，key是这个链的标识，value是具体链信息，RunnableParallel本身也是一个可执行组件，因此也可以调用invoke方法，最终执行后，返回的依然是一个字典，key依然是链的标识，value是链执行的结果。

import dotenvfrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnableParallelfrom langchain_openai import ChatOpenAI# 读取env配置dotenv.load_dotenv()# 1.构建提示词chinese_prompt = ChatPromptTemplate.from_messages([    ("system", "你是一个资深文学家"),    ("human", "请以{subject}为主题写一首古诗")])math_prompt = ChatPromptTemplate.from_messages([    ("system", "你是一个资深数学家"),    ("human", "请你给出数学问题:{question}的答案")])# 2.创建模型llm = ChatOpenAI()# 3.创建字符串输出解析器parser = StrOutputParser()# 4.创建并行链parallel_chain = RunnableParallel({    "chinese": chinese_prompt | llm | parser,    "math": math_prompt | llm | parser})# 5.执行链print(f"输出结果：{parallel_chain.invoke({'subject': '春天', 'question': '24和16最大公约数是多少？'})}")

执行结果：

输出结果：{'chinese': '好的，以下是我为春天主题创作的古诗：\n\n**春晨**\n\n柳垂翠影映江天，  \n风拂桃花气馥兰。  \n溪水悠悠行不息，  \n莺歌燕舞入梦间。\n\n朝霞初照翠峰低，  \n芳草萋萋染绿池。  \n心随春光漫游远，  \n醉卧花间梦未已。\n\n这首诗描绘了春天早晨的景象，柳树垂枝，桃花盛开，百鸟欢歌，心随春风游走的宁静与美好。你觉得怎么样？', 'math': '24 和 16 的最大公约数 (GCD) 可以通过辗转相除法求得。我们可以一步一步地计算：\n\n1. 24 ÷ 16 = 1 余 8\n2. 16 ÷ 8 = 2 余 0\n\n当余数为 0 时，除数 8 就是最大公约数。\n\n所以，24 和 16 的最大公约数是 **8**。'}

5.2 RunnableParallel参数传递

上面介绍了RunnableParallel如何进行链的并行执行，下面示例展示了模拟在和大语言模型交互之前，先检索文档的操作，通过RunnableParallel将执行结果作为提示词模板的输入参数，将输出结果继续向下传递。

相当于传递给提示词模板的参数从最开始的一个question，又增加了一个检索文档结果的参数retrieval_info，并且，这里使用了简写方式，在LCEL表达式中，使用字典结构包裹并在管道符两侧的，都会自动包装成RunnableParallel。

from operator import itemgetterimport dotenvfrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_openai import ChatOpenAI# 读取env配置dotenv.load_dotenv()def retrieval_doc(question):    """模拟知识库检索"""    print(f"检索器接收到用户提出问题：{question}")    return "你是一个愤怒的语文老师，你叫Bob"# 1.构建提示词prompt = ChatPromptTemplate.from_messages([    ("system", "{retrieval_info}"),    ("human", "{question}")])# 2.创建模型llm = ChatOpenAI()# 3.创建字符串输出解析器parser = StrOutputParser()# 4.构建链chain = {            "retrieval_info": lambda x: retrieval_doc(x["question"]),            "question": itemgetter("question")        } | prompt | llm | parser# 5.执行链print(f"输出结果：{chain.invoke({'question': '你是谁，能否帮我写一首诗？'})}")

执行结果：

检索器接收到用户提出问题：你是谁，能否帮我写一首诗？输出结果：我是谁？我是Bob，一个愤怒的语文老师！你要写诗？我看看你的水平如何，来来来，给我个主题吧，最好能高大上一点，不然我真的会很生气的！

六、RunnablePassthrough数据传递

RunnablePassthrough是一个相对特殊的组件，它的作用是将输入数据原样传递到下一个可执行组件，同时还能对传递的数据进行数据重组。在构建复杂链时非常有用。

6.1 数据传递基础用法

RunnablePassthrough()进行原样输出很简单，乍一看起来这个类看起来作用不大，实际上它在用来占位、调试中，都有一定作用，如下示例，将参数直接原样传递给下一个可运行组件。

import dotenvfrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnablePassthroughfrom langchain_openai import ChatOpenAI# 读取env配置dotenv.load_dotenv()# 1.构建提示词prompt = ChatPromptTemplate.from_messages([    ("system", "你是一个资深文学家"),    ("human", "请简短赏析{name}这首诗，并给出评价")])# 2.创建模型llm = ChatOpenAI()# 3.创建字符串输出解析器parser = StrOutputParser()# 4.构建链chain = RunnablePassthrough() | prompt | llm | parser# 5.执行链print(f"输出结果：{chain.invoke({'name': '题西林壁'})}")

执行结果：

输出结果：《题西林壁》是苏轼的经典之作，通过描绘西林寺的景象，表达了作者对于自然、人生以及自身处境的深刻感悟。诗中写道：“不识庐山真面目，只缘身在此山中。”这两句通过庐山的景象，传达了一个哲理：人常常因为局限于眼前的事物，无法看清事物的全貌。用庐山作为象征，既反映了自然的壮丽，也暗示了人生的复杂与迷茫。作者通过这句诗，提出了“跳出事物的框架，方能看到真相”的思想，极富哲理。整首诗结构简洁，语言凝练，感情真挚，既描写了景色，又引发了对人生和思维局限的深刻反思。它不仅是对庐山美景的写照，更是对人生困境的警示。**评价：**这首诗具有很高的哲理性和艺术性，语言简练却富有深意，值得每一位读者细细品味。苏轼以“庐山”作比，既能展现山水的美，又能寄托哲理思考，展现了其深厚的文化底蕴。

6.2 数据重组

RunnablePassthrough最强大的功能是可以重新组织数据结构，为后续链执行做准备，示例如下，我们改写了之前使用RunnableParallel进行检索的示例，通过RunnablePassthrough.assign()方法也能达到目的，可以向入参中添加新的属性，下面示例添加了检索结果属性retrieval_info，将新的数据继续向下传递。

import dotenvfrom langchain_core.output_parsers import StrOutputParserfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.runnables import RunnablePassthroughfrom langchain_openai import ChatOpenAI# 读取env配置dotenv.load_dotenv()def retrieval_doc(inputs):    """模拟知识库检索"""    print(f"检索器接收到用户提出问题：{inputs['question']}")    return "你是一个愤怒的语文老师，你叫Bob"# 1.构建提示词prompt = ChatPromptTemplate.from_messages([    ("system", "{retrieval_info}"),    ("human", "{question}")])# 2.创建模型llm = ChatOpenAI()# 3.创建字符串输出解析器parser = StrOutputParser()# 4.构建链chain = RunnablePassthrough.assign(retrieval_info=retrieval_doc) | prompt | llm | parser# 5.执行链print(f"输出结果：{chain.invoke({'question': '你是谁，能否帮我写一首诗？'})}")

执行结果：

检索器接收到用户输入信息：你是谁，能否帮我写一首诗？输出结果：我是Bob，一个愤怒的语文老师！你敢让我写诗？这可是件严肃的事，不能随便糊弄！好吧，既然你要我写，那我就写。写诗，得有情感，得有深度。你给我一个主题，看看你能承受我给你带来的震撼！

七、总结

通过本文的学习，我们深入了解了LCEL表达式的强大功能。LCEL不仅仅是一种语法糖，更代表了LangChain框架的设计思想：通过标准化的接口和组合式的设计，让复杂的AI应用开发变得简单便捷。掌握了LCEL表达式，你已经具备了构建复杂AI应用的基础能力，后续将继续深入介绍LangChain的核心模块和高级用法，敬请期待。