はじめに
LangChainでは、以下のようにLangChain Expression Language (LCEL) でChainを記述することが推奨されるようになった。LCELでは、Linuxのパイプのように "|"(パイプ文字)を使用する。しかし、通常Pythonでは"|"はビット演算子として働き、論理和が出力される。LCELではどうやって"|"に独自の演算機能を持たせているかが分からなかったため、調べた。
chain = prompt | model | outputparser # この"|"を調べた
chain.invoke("Question.")
[前提知識]演算子のオーバーロード
Pythonでは、__eq__などの特殊メソッドをクラスで独自に宣言することで、演算子の実行内容をユーザーが定義したものに変更できる。
"|"を定義するのは、__or__と__ror__。
- __or__ : "|"の左側にあるときに実行される。A|Bなら、Aの__or__が実行される。
- __ror__: "|"の右側にあるときに実行される。A|Bなら、Bの__or__が実行される。
例
例1:__or__の挙動をチェック
class AとBで__or__を宣言する。Aで"|"の演算が行われると、"A's __or__ method is called"と出力される。Bでは、"B's __or__ method is called"と出力される。
class A:
def __init__(self, value):
self.value = value
def __or__(self, other):
print("A's __or__ method is called")
return self.value | other.value
class B:
def __init__(self, value):
self.value = value
def __or__(self, other):
print("B's __or__ method is called")
return self.value | other.value
objA = A(2)
objB = B(3)
result = objA | objB
出力結果
A's __or__ method is called
Aの "|" が実行されたことがわかる。
ちなみに、
result = objB | objA
print(result)
objAとobjBの順番を逆にしたとき、出力結果は
B's __or__ method is called
となり、Bの"|"が実行されたことがわかる。つまり、"|"の前に置いてあるオブジェクトの__or__が実行される。
例2:__ror__の挙動をチェック
class Bでのみ、__ror__を宣言する。
class A:
def __init__(self, value):
self.value = value
class B:
def __init__(self, value):
self.value = value
def __ror__(self, other):
print("B's __ror__ method is called")
return self.value | other.value
objA = A(2)
objB = B(3)
result = objA | objB
出力結果
B's __ror__ method is called
"|"の後に置いたobjBの__ror__が実行された。
例3:__ror__と__or__の優先順位をチェック
class Aでは__or__を、Bでは__ror__を宣言する。
class A:
def __init__(self, value):
self.value = value
def __or__(self, other):
print("A's __or__ method is called")
return self.value | other.value
class B:
def __init__(self, value):
self.value = value
def __ror__(self, other):
print("B's __ror__ method is called")
return self.value | other.value
objA = A(2)
objB = B(3)
result = objA | objB
出力結果
A's __or__ method is called
"|"の前のクラスで__or__が宣言されていればそちらが実行される。"|"の後のクラスの__ror__は無視される。
[本題]LangChainのソースコードで演算子のオーバーロードを見てみる
LCELで使われる、prompt, model, outputparserなどはすべてRunnableクラスを基底に持っている。そのため、Runnableクラスの__or__と__ror__メソッドを見てみた。
class Runnable(Generic[Input, Output], ABC): # 一部抜粋
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))
def __ror__(
self,
other: Union[
Runnable[Other, Any],
Callable[[Other], Any],
Callable[[Iterator[Other]], Iterator[Any]],
Mapping[str, Union[Runnable[Other, Any], Callable[[Other], Any], Any]],
],
) -> RunnableSerializable[Other, Output]:
"""Compose this runnable with another object to create a RunnableSequence."""
return RunnableSequence(coerce_to_runnable(other), self)
つまり、self(Runnableクラスのオブジェクト)|otherを実行すると、RunnableSequence(self, coerce_to_runnable(other))というようにRunnableSequenceクラスのオブジェクトを生成して、返すようになっている。
other|self(Runnableクラスのオブジェクト)も可能。
ここで使われているcoerce_to_runnableは以下で、RunnableではないPythonの標準のものをRunnable系のものに変換する。ただし、coerce_to_runnableの引数にできるのは、Runnable,callableかdictのみ。それ以外は例外が発生するようになっている。
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に__or__と__ror__が宣言されているために、Runnableのオブジェクト | other(Runnableのオブジェクト, callable, dictのいずれか) や other | Runnableのオブジェクトというように使えるとわかった。この演算によって返ってきたRunnableSequenceに対し、invokeを実行している。
実践 LCELで"|"を使ってみる
実践1:
callable, dictのいずれか | Runnableのオブジェクト を試してみる。
from langchain_core.runnables import RunnableLambda
from operator import itemgetter
# 文字数を返す関数
def length_function(text):
return len(text)
# 以下のchainでは、chainに渡された辞書から、key="foo"で値を取得し次に渡し、length_functionを実行する。
chain = itemgetter("foo")| RunnableLambda(length_function)
# 出力は2。("aa"が2文字だから)
chain.invoke({"foo":"aa"})
ポイントは以下
- callableであるitemgetterを使う。
- length_functionをRunnableにするため、RunnableLambdaでラップする。
ちなみに、以下はエラーが起きた。
# エラー
chain = {"foo":"aa"}| RunnableLambda(length_function)
chain.invoke({"foo":"aa"})
LCELでは、辞書のvalueの部分もrunnableかcallableかdictでないとダメらしい。おそらく再帰的にチェックされる。
どうしても、chainの中に辞書を書きたい場合は、以下を使う。
chain = (lambda x:{"foo":"aa"})| RunnableLambda(length_function)
chain.invoke({"foo":"aa"})
こうすると、dictの部分がまるまるcallableになるため、問題なし。(lambdaの引数xには、invokeの引数が入るが、今回はxを使っていないため、捨てることになる。)