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