在前面幾篇文章中，我們已經掌握了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應用。

文中所有示例代碼：https://github.com/wzycoding/langchain-study

一、什么是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 dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableBranch
from 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 dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda
from 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 dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel
from 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 itemgetter

import dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from 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 dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from 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 dotenv
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from 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的核心模塊和高級用法，敬請期待。