⛓️

LangChain Expression Language(LCEL) 応用編

に公開
DSL
LangChain
lcel
tech

LCEL の基礎知識

LCEL とは

LangChain（本記事ではPython 版を前提とします）独自の記法であるLCEL：LangChain Expression Language は、パイプ| の動作を書き換えている点で、一種のドメイン固有言語（DSL）と言える。
そのため若干とっつきにくいが、慣れるとフローを直感的に理解しやすい。

よく見かけるのは、以下のようなRAG の例である。

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

上記のチェーンもよく分からない場合は、LangChain 公式ドキュメント等を読んでおくこと。
以下、RunnableSequence とRunnableParallel は理解している前提とする。

LCEL の立ち位置

LCEL は割とよく出来ており、複雑な処理でも記述可能である。が、LangChain 開発の傾向として、複雑な処理はLangGraph でやってくれ、という意図のようである。
LangChain v0.3 のドキュメントでは、Multiple chains のページ（v0.1 では存在していた）が削除されていたり、RAG のチュートリアルがLangGraph を使うものに置き換わっているのがその証左。

とは言え、多少の分岐にLangGraph を持ち出すのも大げさだと思うし、実行速度もLCEL のチェーンに分があるだろう（LangGraph はノード遷移毎に固有ID を払い出したりするため。実際に実行時間を測ったわけではないが...）。また、わざわざLCEL で複雑な処理が書けないようにdegration するとも考えにくい。

そういうわけで、LCEL でややこしい処理を書きたい需要は残ると思うので、本記事に知見をメモっておく。

LCEL コードの実例

めちゃ簡単な関数をもとに説明する。

基本的な処理

直列関係

これは簡単。@chain のデコレータはチェーン先頭の関数だけで良いが、分かりやすさのため敢えて記載している。

@chain
def append_A(s: str) -> str:
    return s + "A"

@chain
def append_B(s: str) -> str:
    return s + "B"

chain1 = append_A | append_B

append_A で文字列の最後尾にA が、append_B でB が足されるため、実行すると以下のような結果になる。

print(chain1.invoke("test"))
#'testAB'

次で使うので、文字列の最後尾にX, Y を足すチェーンも定義しておく。

@chain
def append_X(s: str) -> str:
    return s + "X"

@chain
def append_Y(s: str) -> str: 
    return s + "Y"

chain2 = append_X | append_Y
print(chain2.invoke("test"))
#'testXY'

並列関係

辞書を使うことで並列処理が記述できる。今回は先程のchain1 とchain2 を並列させ、その出力を連結させてみる。
chain1 とchain2 に同じ文字列test を入力していることに注意。

@chain
def join_strs_in_dict(d: dict) -> str:
    return f"""{d["a"]} and {d["b"]}"""

chain3 = {"a": chain1, "b": chain2} | join_strs_in_dict
print(chain3.invoke("test"))
#'testAB and testXY'

itemgetter()

ここからが本題。chain1 にはtest、chain2 にはTEST のように、並列関係のチェーン群に別々の文字列を入力したかったらどうするのか？

答えは、並列処理への入力を辞書として、個々のチェーンの直前において itemgetter() で欲しい値を取り出せば良い。

from operator import itemgetter

chain4 = {"a": (itemgetter("p") | chain1), "b": (itemgetter("q") | chain2)} | join_strs_in_dict
print(chain4.invoke({"p": "test", "q": "TEST"}))
#'testAB and TESTXY'

RunnablePassthrough().assign()

まずchain1 にtest を入力して途中出力testAB を得る。この文字列をchain2 に入力して出力testABXY を得るとして、この出力（testABXY）と先程の途中出力（testAB）の両方を後段で使いたかったらどうするのか？

答えは、 RunnablePassthrough().assign() で途中出力をチェーンの入力辞書に保存しておき、itemgetter() で欲しい値を取り出せば良い。

from langchain_core.runnables import RunnablePassthrough

chain5 = (
    RunnablePassthrough().assign(a=(itemgetter("p") | chain1)) |
    RunnablePassthrough().assign(b=(itemgetter("a") | chain2)) |
    join_strs_in_dict
)
print(chain5.invoke({"p": "test"}))
#'testAB and testABXY'

Discussion