LangChain的LCEL和Runnable你搞懂了嗎
LangChain的LCEL估計行業內的朋友都聽過，但是LCEL里的RunnablePassthrough、RunnableParallel、RunnableBranch、RunnableLambda又是什么意思？什么場景下用？
一、LCEL的定義和原理
LangChain的核心是Chain，即對多個組件的一系列調用。
LCEL是LangChain 定義的表達式語言，是一種更加高效簡潔的調用一系列組件的方式。
LCEL使用方式就是：以一堆管道符（"|"）串聯所有實現了Runnable接口的組件。
比如這樣：
prompt_tpl = ChatPromptTemplate.from_messages(
[
("system", "{parser_instructions}"),
("human", "列出{cityName}的{viewPointNum}個著名景點。"),
]
)
output_parser = CommaSeparatedListOutputParser()
parser_instructions = output_parser.get_format_instructions()
model = ChatOpenAI(model="gpt-3.5-turbo")
chain = prompt_tpl | model | output_parser
response = chain.invoke(
{"cityName": "南京", "viewPointNum": 3, "parser_instructions": parser_instructions}
)所以LangChain為了讓組件能以LCEL的方式快速簡潔的被調用，計劃將所有組件都實現Runnable接口。比如我們常用的PromptTemplate 、LLMChain 、StructuredOutputParser 等等。
管道符（"|"）在Python里就類似or運算（或運算），比如A|B，就是A.or(B)。
那對應到LangChain的Runnable接口里，這個or運算是怎么實現的呢？一起看到源碼：
LangChain通過or將所有的Runnable串聯起來，在通過invoke去一個個執行，上一個組件的輸出，作為下一個組件的輸入。
LangChain這風格怎么有點像神經網絡呀，不得不說，這個世界到處都是相似的草臺班子。嗨！
總結起來講就是：LangChain的每個組件都實現了Runnable，通過LCEL方式，將多個組件串聯到一起，最后一個個執行每個組件的invoke方法。上一個組件的輸出是下一個組件的輸入。
二、Runnable的含義和應用場景
1.RunnablePassthrough
① 定義
RunnablePassthrough 主要用在鏈中傳遞數據。RunnablePassthrough一般用在鏈的第一個位置，用于接收用戶的輸入。如果處在中間位置，則用于接收上一步的輸出。
② 應用場景
比如，依舊使用上面的例子，接受用戶輸入的城市，如果輸入城市是南京，則替換成北京，其余不變。代碼如下。此處的{}和RunnablePassthrough.assign()是同一個語義。
chain = (
{
"cityName": lambda x: '北京' if x["cityName"] == '南京' else x["cityName"],
"viewPointNum": lambda x: x["viewPointNum"],
"parser_instructions": lambda x: x["parser_instructions"],
}
| prompt_tpl
| model
| output_parser
)2.RunnableParallel
① 定義
RunnableParallel看名字里的Parallel就猜到一二，用于并行執行多個組件。通過RunnableParallel，可以實現部分組件或所有組件并發執行的需求。
② 應用場景
比如，同時要執行兩個任務，一個列出城市著名景點，一個列出城市著名書籍。
prompt_tpl_1 = ChatPromptTemplate.from_messages(
[
("system", "{parser_instructions}"),
("human", "列出{cityName}的{viewPointNum}個著名景點。"),
]
)
prompt_tpl_2 = ChatPromptTemplate.from_messages(
[
("system", "{parser_instructions}"),
("human", "列出關于{cityName}歷史的{viewPointNum}個著名書籍。"),
]
)
output_parser = CommaSeparatedListOutputParser()
parser_instructions = output_parser.get_format_instructions()
model = ChatOpenAI(model="gpt-3.5-turbo")
chain_1 = prompt_tpl_1 | model | output_parser
chain_2 = prompt_tpl_2 | model | output_parser
chain_parallel = RunnableParallel(view_point=chain_1, book=chain_2)
response = chain_parallel.invoke(
{"cityName": "南京", "viewPointNum": 3, "parser_instructions": parser_instructions}
)3.RunnableBranch
① 定義
RunnableBranch主要用于多分支子鏈的場景，為鏈的調用提供了路由功能，這個有點類似于LangChain的路由鏈。我們可以創建多個子鏈，然后根據條件選擇執行某一個子鏈。
② 應用場景
比如，有多個回答問題的鏈，先根據問題找到分類，然后在使用具體的鏈回答問題。
model = ChatOpenAI(model="gpt-3.5-turbo")
output_parser = StrOutputParser()
# 準備2條目的鏈：一條物理鏈，一條數學鏈
# 1. 物理鏈
physics_template = """
你是一位物理學家，擅長回答物理相關的問題，當你不知道問題的答案時，你就回答不知道。
具體問題如下：
{input}
"""
physics_chain = PromptTemplate.from_template(physics_template) | model | output_parser
# 2. 數學鏈
math_template = """
你是一個數學家，擅長回答數學相關的問題，當你不知道問題的答案時，你就回答不知道。
具體問題如下：
{input}
"""
math_chain = PromptTemplate.from_template(math_template) | model | output_parser
# 4. 其他鏈
other_template = """
你是一個AI助手，你會回答一下問題。
具體問題如下：
{input}
"""
other_chain = PromptTemplate.from_template(other_template) | model | output_parser
classify_prompt_template = """
請你對以下問題進行分類，將問題分類為"數學"、"物理"、"其它"，不需要返回多個分類，返回一個即可。
具體問題如下：
{input}
分類結果：
"""
classify_chain = PromptTemplate.from_template(classify_prompt_template) | model | output_parser
answer_chain = RunnableBranch(
(lambda x: "數學" in x["topic"], math_chain),
(lambda x: "物理" in x["topic"], physics_chain),
other_chain
)
final_chain = {"topic": classify_chain, "input": itemgetter("input")} | RunnableLambda(print_info) | answer_chain
# final_chain.invoke({"input":"地球的半徑是多少？"})
final_chain.invoke({"input":"對y=x求導的結果是多少？"})4.RunnableLambda
① 定義
要說牛批還得是RunnableLambda，它可以將Python 函數轉換為 Runnable對象。這種轉換使得任何函數都可以被看作 LCEL 鏈的一部分，我們把自己需要的功能通過自定義函數 + RunnableLambda的方式包裝一下，集成到 LCEL 鏈中，這樣算是可以跟任何外部系統打通了。
② 應用場景
比如，在執行過程中，想在中間插入一段自定義功能（如 打印日志 等），可以通過自定義函數 + RunnableLambda的方式實現。
def print_info(info: str):
print(f"info: {info}")
return info
prompt_tpl_1 = ChatPromptTemplate.from_messages(
[
("system", "{parser_instructions}"),
("human", "列出{cityName}的{viewPointNum}個著名景點。"),
]
)
output_parser = CommaSeparatedListOutputParser()
parser_instructions = output_parser.get_format_instructions()
model = ChatOpenAI(model="gpt-3.5-turbo")
chain_1 = prompt_tpl_1 | model | RunnableLambda(print_info) | output_parser
response = chain_1.invoke(
{"cityName": "南京", "viewPointNum": 3, "parser_instructions": parser_instructions}
)三、總結
本篇主要聊了LangChain的LCEL表達式，以及LangChain鏈的原理，以及常用的幾個Runnable的定義和應用場景，希望對你有幫助。
