A Day in the Life

LangChain の LangChain Expression Language (LCEL) を理解する

LCEL は LangChain の chain を簡単に構築するための方法です。2023 年後半から開発が盛んに進んでおり、現在(2024 年1月)は LangChain のコードを記述するには、基本 LCEL を使って書く(以前の書き方もできますが)ことが推奨されています。LCEL のメリットについてはオフィシャルドキュメントの LCELを参考すると良いでしょう。

しかしながら、LCEL を書き始めると、オフィシャルドキュメント通りに書けば動くけど、ちょっとでもアレンジして書こうとするうまく動かなくなったりします。これは一重に LCEL の挙動を理解していなからなのですが、オフィシャルドキュメントやチュートリアルでは、LLM+RAG のコードなど、LCEL 使うとこんなにシンプルに書けるんだ、というコードは多くのっているのですが、LCEL の挙動についてはあまりのっておらず、のっていても ChatGPT やテンプレートと組み合わせた時の処理などになってしまい、「そもそも LCEL の挙動について知りたい」が解らなくて困りました。

ので、LangChain 0.1.0 を使って、純粋に LCEL の挙動の基本だけを説明し、ステップバイステップで理解が進むようなノートブック記事を書いてみました。Colab のノートブックへは、以下からアクセスができます。

LCEL の基本

LangChain の LCEL の基本的な考え方は単純です。オブジェクトが入力値を受け取り、その出力値を次に渡していくだけです。普通の実装と変わりませんね。

ではまず、値を二倍する関数を定義してみましょう。

def double(x):
    return x * 2

double(2)
4

続いて、引数の値を標準出力に出力して、引数の値をそのまま返す関数を定義してみましょう。

def tap_print(x):
    """
    引数の値を標準出力に出力して、引数の値をそのまま返す
    """
    print(f"tap_print: {x}")
    return x

先ほど作った二つの関数を繋げて実行してみましょう。 double に引数を与えて実行し、その結果を tap_print に渡して実行します。

tap_print(double(2))
tap_print: 4

4

思った結果になりましたね。

続いてこれらの関数を LCEL の実行クラスである、Runnable のサブクラスの RunnableLambda へ変換してみましょう。

from langchain_core.runnables import RunnableLambda

r_double = RunnableLambda(double)

変換後は、Runnable のインターフェイスを使えます。では Runnable を実行する invoke を呼び出してみましょう。

r_double.invoke(2)
4

なお、@chain というデコレータで、同様に RunnableLambda 関数の定義も可能です。後に chain という変数名が出てきて紛らわしいので、ここでは chain_decorator という名前で import してます。

from langchain_core.runnables import chain as chain_decorator

@chain_decorator
def r_double(x):
    return x * 2

r_double.invoke(2)  # r_double は RunnableLambda になるので、invoke で実行できる
4

同様に、tap_print の方も Runnable 化します。

r_tap_print = RunnableLambda(tap_print)
r_tap_print.invoke(2)
tap_print: 2

2

続いてやっとお待ちかね、LCEL の中核である、| を使って繋げて実行してみましょう。

chain = r_double | r_tap_print
chain.invoke(2)
tap_print: 4

4

やった！r_double の結果を r_tap_print が表示しつつ値を返すような、思った通りの結果になりましたね。 さて、この chain とは何者なのでしょう？

chain.__class__
langchain_core.runnables.base.RunnableSequence

そう、chain は RunnableSequence、つまり直列実行する Runnable が実体でした。| のシンタックスシュガーを使わないで書いで、同等の実装を書いてみましょう。

from langchain_core.runnables import RunnableSequence

chain = RunnableSequence(r_double, r_tap_print)
chain.invoke(2)
tap_print: 4

4

先ほどと同じ結果になりましたね。

ではこの Runnable がどう実行されているのか、実行の流れ(実行グラフ)を表示してみましょう。

chain.get_graph().print_ascii()
  +----------------+   
  | r_double_input |   
  +----------------+   
          *            
          *            
          *            
+------------------+   
| Lambda(r_double) |   
+------------------+   
          *            
          *            
          *            
+-------------------+  
| Lambda(tap_print) |  
+-------------------+  
          *            
          *            
          *            
+------------------+   
| tap_print_output |   
+------------------+

double が入力を受け取り、それを次の tap_print に渡し、最終的に tap_print_output が出力されるようですね。

では続いて、このコードを見てましょう。

chain = r_double | tap_print  # tap_print は RunnableLambda ではない!
chain.invoke(2)
tap_print: 4

4

んんん〜〜？tap_print が Runnable ではないのに、うまく chain が作れて実行できてますね。なぜ、tap_print は Runnnable ではないただの関数なのに、うまく実行できるのでしょうか。それは Runnable オブジェクトは、python のビット演算子 | の挙動を利用し、左辺もしくは右辺どちらかのオブジェクトが Runnable なら、自動で Runnable に変換されるからなのです。実際の Runnable のコードを見てみましょう。

    def __or__(
        self,
        other: Union[
            Runnable[Any, Other],
            Callable[[Any], Other],
            Callable[[Iterator[Any]], Iterator[Other]],
            Mapping[str, Union[Runnable[Any, Other], Callable[[Any], Other], Any]],
        ],
    ) -> RunnableSerializable[Input, Other]:
        """Compose this runnable with another object to create a RunnableSequence."""
        return RunnableSequence(self, coerce_to_runnable(other))

    # 同様に __ror__ も定義されている

この実装では、RunnableSequence にくるんで返すことと、| 演算子で渡ってくるもう片方の引数(先ほどの例では tap_print が該当)を coerce_to_runnable 関数に通して変換して返していますね。

coerce_to_runnable 関数も重要なので、実装を見てみましょう。

def coerce_to_runnable(thing: RunnableLike) -> Runnable[Input, Output]:
    """Coerce a runnable-like object into a Runnable.

    Args:
        thing: A runnable-like object.

    Returns:
        A Runnable.
    """
    if isinstance(thing, Runnable):
        return thing
    elif inspect.isasyncgenfunction(thing) or inspect.isgeneratorfunction(thing):
        return RunnableGenerator(thing)
    elif callable(thing):
        return RunnableLambda(cast(Callable[[Input], Output], thing))
    elif isinstance(thing, dict):
        return cast(Runnable[Input, Output], RunnableParallel(thing))
    else:
        raise TypeError(
            f"Expected a Runnable, callable or dict."
            f"Instead got an unsupported type: {type(thing)}"
        )

この関数が行う変換は、Runnable なら何もしない、generator っぽいなら RunnableGenerator 化する、関数などの呼び出し可能なオブジェクト(Callbale)なら RunnableLambda 化する、辞書なら RunnableParallel 化 (後に出てきます)する、そうでないなら例外、という挙動になっています。

これらから、LCEL において、| に繋ぐどちらかのコードが Runnable なら、型変換によってもう片方も Runnable に変換されて、それらを繋いだ RunnableSequence が返されることが理解できたのではないでしょうか。

では、次のコードを実行してみましょう。

chain = double | r_tap_print  # double は　RunnableLambda ではない！

chain.invoke(2)
tap_print: 4

4

今度は先ほどと反対で、double は Runnable ではないですが、r_tap_print.__ror__ によって RunnableSequence に変換され、実行することができました。

  • Runnable は invoke で入力値を受け取って処理を行い、出力値を返すが基本動作
  • Runnable を | で繋ぐことで、RunnableSequence で直列実行できるコードが生成される
  • | のどちらかが Runnable でなかった場合、Runnable への変換が自動で行われる

を、今まで説明してきました。こう見ると Runnable は単純でわかりやすい感じがしますね。

RunnableParallel 化される dict 構文

続いて、私も当初非常に混乱した、| と dict を使った構文です。このような実装を考えてみます。

  • 数値を引数に渡す
    • original_value には、最初の値を保持する
    • double_value にその数値を二倍に計算した値を入れる
  • それらの結果を tap_print する

これらが動く実装を書いてみましょう。

chain = {
    "original_value": lambda x: x,
    "double_value": double,
} | r_tap_print

chain.invoke(2)
tap_print: {'original_value': 2, 'double_value': 4}

{'original_value': 2, 'double_value': 4}

なんかよくわからないけど動きましたね！だだこの挙動を理解せずに雰囲気で使っていると、徐々に混乱してくきます。実際に私がそうでした。

これが何をやっているのか、挙動を確認しましょう。実行グラフを表示します。

chain.get_graph().print_ascii()
+--------------------------------------------+    
| Parallel<original_value,double_value>Input |    
+--------------------------------------------+    
               **              **                 
            ***                  ***              
          **                        **            
+-------------+               +----------------+  
| Lambda(...) |               | Lambda(double) |  
+-------------+               +----------------+  
               **              **                 
                 ***        ***                   
                    **    **                      
+---------------------------------------------+   
| Parallel<original_value,double_value>Output |   
+---------------------------------------------+   
                        *                         
                        *                         
                        *                         
             +-------------------+                
             | Lambda(tap_print) |                
             +-------------------+                
                        *                         
                        *                         
                        *                         
              +------------------+                
              | tap_print_output |                
              +------------------+

いきなり Parallel で並列実行処理になって、分岐して並列実行した結果を集約して、その後 tap_print に渡していますね。

そう、ここが混乱ポイントなのですが、 | で dict を繋ぐと coerce_to_runnable 関数が呼び出され RunnableParallel へ自動で変換され、dict の value を並列実行します。なお RunnableParallel は、dict を渡すとその辞書の value をこれまた自動で Runnable へと変換して、それらを並列実行し、key の値として結果を返します。

では先ほど出てきたcoerce_to_runnable関数を使って、実際に|で型変換された結果をみてみましょう。

from langchain_core.runnables.base import coerce_to_runnable

parallel = coerce_to_runnable(
    {
        "original_value": lambda x: x,
        "double_value": double,
    }
)

parallel.invoke(2)
{'original_value': 2, 'double_value': 4}

parallel.__class__
langchain_core.runnables.base.RunnableParallel

同一の挙動をするコードを宣言的に書くとこうなります。

from langchain_core.runnables import RunnableParallel

parallel = RunnableParallel(
    {
        "original_value": coerce_to_runnable(lambda x: x),
        "double_value": coerce_to_runnable(double),
    }
)
parallel.invoke(2)
{'original_value': 2, 'double_value': 4}

というわけで、LCEL において、| で dict を繋げると、RunnableParallel で並列実行処理して値を返すコードになるのでした。

invoke と dict

続いて、invoke に dict を与えて処理してみましょう。なお先ほどの RunnableParallel に変換される dict とは全く処理が異なり、ふつーの dict を引数とした呼び出しなので注意しましょう。

data = {
    "input_value": 2,
    "input_do_nothing": 100,
}
chain = r_double | r_tap_print
try:
    chain.invoke(data)
except Exception as e:
    print("Error:", e)
Error: unsupported operand type(s) for *: 'dict' and 'int'

r_double は int を期待してますが、dict が渡ってきてしまったので、うまく処理ができません。chain.invoke(data['input_value']) と書けば良いのでは？はその通りなのですが、もし r_double が chain の途中にあって、そいつに dict が渡ってきたら、うまく扱えませんね。

そんな時は input_value だけを取り出す関数を挟みます。

data = {
    "input_value": 2,
    "input_do_nothing": 100,
}
chain = (lambda x: x["input_value"]) | r_double | r_tap_print
chain.invoke(data)
tap_print: 4

4

うまく動きましたね。

では data の input_do_nothing は何もしないで次に値を渡し、input_value の計算結果は double_value として次に渡す、大元の input_value も知りたいのでこれも次に渡す、そんなコードを書いてみましょう。

data = {
    "input_value": 2,
    "input_do_nothing": 100,
}
chain = {
    "double_value": (lambda x: x["input_value"]) | r_double,
    "input_value": lambda x: x["input_value"],
    "input_do_nothing": lambda x: x["input_do_nothing"],
} | r_tap_print
chain.invoke(data)
tap_print: {'double_value': 4, 'input_value': 2, 'input_do_nothing': 100}

{'double_value': 4, 'input_value': 2, 'input_do_nothing': 100}

うまく r_tap_print に意図した値が渡っていますね。

この chain の最初の辞書定義では、次の | 演算子で r_tap_print.__ror__ がよび出され、coerce_to_runnable に渡されることで自動的に RunnableParallel に変換されることで、意図した挙動でうまく動いています。

ただ、めちゃくちゃ冗長ですね。data の key がもっとたくさんあって、全部を後続の Runnable の入力値として値を渡したい時なんかとても困りそうです。そんな時のために RunnablePassthrough が用意されています。RunnablePassthrough を使って書き直してみましょう。

from langchain_core.runnables import RunnablePassthrough

data = {
    "input_value": 2,
    "input_do_nothing": 100,
}
chain = (
    RunnablePassthrough().assign(
        double_value=(lambda x: x["input_value"]) | r_double,
    )
    | r_tap_print
)
chain.invoke(data)
tap_print: {'input_value': 2, 'input_do_nothing': 100, 'double_value': 4}

{'input_value': 2, 'input_do_nothing': 100, 'double_value': 4}

ついでに 3 倍にする結果も入れてみましょう。

chain = (
    RunnablePassthrough().assign(
        double_value=(lambda x: x["input_value"]) | r_double,
        triple_value=lambda x: x["input_value"] * 3,  # 暗黙的に RunnableLambda に変換される
    )
    | r_tap_print
)
chain.invoke(data)
tap_print: {'input_value': 2, 'input_do_nothing': 100, 'double_value': 4, 'triple_value': 6}

{'input_value': 2,
 'input_do_nothing': 100,
 'double_value': 4,
 'triple_value': 6}

このように、入力値の dict の一部のみを変更したり、key を追加したりして、その後の runnable に渡すには、RunnablePassthrough はとても便利ですね。

なお invoke の引数に dict 型を渡すことと、| で dict を繋げて LCEL を書くこと(≒ RunnableParallel 化する)は、全く別の意図・挙動をするので、混乱しないように注意してください。

他にも、RunnablePassthrough は、入力値として渡ってきた dict ではない値を dict にするのに宣言的に使われたりします。

value_format = "value is {value}, double value is {double_value}"

def template(data):
    return value_format.format(**data)

r_template = RunnableLambda(template)

chain = (
    {
        "value": RunnablePassthrough(),
        "double_value": RunnablePassthrough() | double,
    }  # r_double でなく double でも、自動で RunnableLambda に型変換される
    | r_template
    | r_tap_print
)
chain.invoke(100)
tap_print: value is 100, double value is 200

'value is 100, double value is 200'

なぜこのような挙動になるか、もう理解できますね！

次のステップ

ここまで理解できたら、LangChain の LCEL のサンプルコードを読んだり、誰かが書いた LCEL のコードも大抵は理解できるようになったはずです。振り返ってみると、Runnable への自動的な変換がどのように行われるかと、| の前後に dict を書くとそれが RunnableParallel になることを押さえておけば、実装で困ることはだいぶ減りそうです。

また、実際になんでこうなるの？を知りたい場合は、Runnable のソースコードを読んだ方が理解が早い場合もあるので、その場合はコードを読むのがおすすめです。

この記事やノートブックが、どなたかの LCEL の理解の手助けになれば幸いです。

記事の一覧 ＞

関連するかもエントリー

Keras を使わずに TensorFlow 2 を使い素朴な全層結合ニューラルネットワークを作る
TensorFlow では、高レベルAPIであるKerasを使うことで、簡単にニューラルネットワークのモデル作成～訓練、その他NNで行いたい様々なことを実現できる。しかしながら、自分のようなNN初心者にとっては何をやってるか解らないで使ってしまっていたため、簡単な順伝播型のNNを...
TensorFlow では、高レベルAPIであるKerasを使うことで、簡単にニューラルネットワークのモデル作成～訓練、その他NNで行いたい...

書いている人

Yuichi Tateno (舘野 祐一)。ソフトウェアエンジニア。Kaggle Competitions Master。インターネット上では id:secondlife(セコン)、@hotchpotch のニックネームを利用している。

おたよりを送る

おたよりをおまちしてます。お気軽に送りください。
お返事はメールアドレス入力があればメールで、メールアドレスがなければ日記上で行うことがあります。

https://secon.dev/entry/2024/01/11/100000/

このサイトは reCAPTCHA と Google のプライバシーポリシーおよび利用規約が適用され、保護されています。