実行例で理解する `Runnable` の継承者たち in `langchain`

Last updated at 2024-03-09Posted at 2024-03-03

概要

この記事では、LLM（Large Language Models）を活用したアプリケーション開発のためのフレームワーク、langchainにおけるRunnableクラスを継承しているクラスについて紹介します。langchain は OpenAI や Gemini などの LLM を容易に統合し、RAG（Retrieval-Augmented Generation）やエージェント作成を行うことが可能なフレームワークです。特に、Runnable クラスは langchain のLangChain Expression Language (LCEL) で中心的な役割を果たす抽象基底クラスであり、Runnable クラスを継承することで多くの機能が実装されています。

本記事では、Runnableクラスを継承している具体的なクラス群に焦点を当て、それらの役割や使い方について詳しく解説します。これにより、langchainを使用した開発において、より高度なコーディング能力を身につけることを目指します。

本記事で触れている Runnable クラスのファミリー

クラス名	概要
`Runnable`	LCEL の中枢をなす抽象基底クラス
`RunnableSerializable`	`Runnable` に加えて `Serializable` を継承した抽象基底クラス
`RunnableLambda`	Python の任意の関数から `Runnable` オブジェクトを作るための `Runnable` クラス
`RunnableGenerator`	ジェネレータ関数を実行するための `Runnable` クラス
`RunnableSequence`	LCEL におけるチェーンを実装した `Runnable` クラス
`RunnableParallel`	`Runnable` オブジェクトのマッピングを並列に実行し、`dict` 型として出力する `Runnable` クラス
`RunnableMap`	`RunnableParallel` のエイリアス
`RunnableEach`	入力のリストを出力のリストに変換するための `Runnable` クラス
`RunnableBinding`	`Runnable` オブジェクトをラップし、追加の機能などを与えるための `Runnable` クラス
`RunnableBranch`	条件に基づいて実行する `Runnable` オブジェクトを切り替えるためのクラス
`RunnableConfigurableAlternatives`	動的に設定を変更することができる `Runnable` クラス
`RunnableCongicurableFields`	`Runnable` オブジェクトの特定の属性を `Configurable` にするためのクラス
`RunnableWithFallbacks`	失敗時に別の `Runnable` オブジェクトにフォールバックするための `Runnable` クラス
`RunnableWithMessageHistory`	チャットメッセージ履歴を管理するための `Runnable` クラス
`RunnablePassthrough`	入力に何も変化を与えない `Runnable` クラス
`RunnableAssign`	入力に対して新規フィールドを付与して出力する `Runnable` クラス
`RunnablePick`	入力された辞書から指定されたキーを取得するための `Runanble` クラス
`RunnableRetry`	失敗時にリトライするための `Runnable` クラス
`RouterRunnable`	入力に応じて実行する `Runnable` オブジェクトを切り替える `Runnable` クラス

本記事で提供されているコードは、特定の環境や条件下での動作を示すものであり、全ての環境やケースで同様に機能するとは限りません。また、時間の経過とともにソフトウェアの更新や互換性の問題が生じる可能性があるため、掲載されているコードが最新の状態であるとは限りません。本コードの使用は、読者の責任において行ってください。実行する前に、コードをよく理解し、必要に応じて環境に適合させることを推奨します。また、可能な限りバックアップを取ることを忘れないでください。また、執筆にあたり生成AIも活用しておりますのでご了承ください。本記事の内容やコードに関する質問や改善提案があれば、コメント欄やソーシャルメディアを通じてお知らせください。

はじめに

LLM が流行って一年程度。技術の進歩の早さに焦りを感じながらもインプットしたものを気が向くままにアウトプットしていきます。

langchain¹ は言語モデルを活用したアプリケーション開発のためのフレームワークの１つです。langchain を使えば OpenAI や Gemini など一般的によく知られている LLM などを使って RAG やエージェント作成を簡単に使うことができます。

以前の記事で langchain の Runnable クラスについて深堀いたしました。

しかしながら、Runnable クラスはあくまでlangchain の LangChain Expression Language (LCEL) で中心的な役割を果たす抽象基底クラスで、直接利用することはできません。

そこで、Runnableのより実用的な部分に踏み入れるため、本記事では Runnable クラスを継承しているクラスを見ていきます（※langchain_core で定義されている一部のクラスに限ります）。

本記事を通して langchain への理解を深めて LCEL のコーディング能力を向上させましょう！

本書の対象読者

langchain を触ったことあるけど、まだよくわからない
- Runnable クラスという言葉を聴いて、どのようなメソッドが定義されてるか思いつくかどうかが一つの判断基準です

本書の実行環境

本書の Python コードの実行例などは下記の環境に基づいています：

Windows 11

Python 3.10.11

requirements.txt

langchain==0.1.10
langchain-openai==0.0.8
python-dotenv==1.0.1

実行する際には OpenAI の API キーを取得の上、.envファイルを下記の内容で作成してください：

OPENAI_API_KEY={YOUR_API_KEY}

OpenAI API の利用には費用が発生します。詳細は https://openai.com/pricing をご確認ください。

`Runnable` クラスの継承者

langchain_core.runnables の下には Runnable クラスとそれを継承した最も基本的なクラスが定義・実装されています：

2024/03/02 時点（Commit Hash: f96dd57501131840b713ed7c2e86cbf1ddc2761f）のリンクなのでご注意ください

langchain_core/runnables/base.py
- Runnable (ABC)
- RunnableSerializable (ABC)
- RunnableSequence
- RunnableParallel
- RunnableMap
- RunnableGenerator
- RunnableLambda
- RunnableEachBase
- RunnableEach
- RunnableBindingBase
- RunnableBinding
langchain_core/runnables/branch.py
- RunnableBranch
langchain_core/runnables/configurable.py
- DynamicRunnable(ABC)
- RunnableCongicurableFields
- RunnableConfigurableAlternatives
langchain_core/runnables/fallbacks.py
- RunnableWithFallbacks
langchain_core/runnables/history.py
- RunnableWithMessageHistory
langchain_core/runnables/passthrough.py
- RunnablePassthrough
- RunnableAssign
- RunnablePick
langchain_core/runnables/retry.py
- RunnableRetry
langchain_core/runnables/router.py
- RouterRunnable

これらすべては Runnable クラスを継承したクラスとなっており、langchain の LCEL で利用可能です。

これらの継承関係は下記の図のようになっています：

以降、これらの各クラスを見ていきましょう

各クラスの詳細/実行例

base.py

`Runnable`

Runnable クラスは LCEL の中枢をなす抽象基底クラスです。

このクラスを継承して多くのクラスの実装がなされています。

特に重要なのが Runnable プロトコルと呼ばれる下記のようなメソッドが実装されている点です。

invoke/ainvoke : 単一の入力を出力に変換
batch/abatch : 複数の入力を出力に効率的に変換
stream/astream : 単一の入力からの出力を生成時にストリーム.
astream_log: 入力からの出力と選択された中間結果をストリーム

これらを標準的なインターフェースとしてカスタムチェーンなどを容易に呼び出すことが可能になっています。

また、Runnable クラスではその他の便利なメソッドも数多く実装されています。
詳細は下記の記事をご覧ください：

`RunnableSerializable`

Runnable に加えて、Serializable も継承した抽象基底クラスです。

Serializable を継承しているため、JSON にシリアル化することが可能です。

>>> from langchain_core.runnables import RunnableSerializable, RunnablePassthrough
>>> runnable = RunnablePassthrough()  
>>> isinstance(runnable, RunnableSerializable)
True
>>> runnable.to_json()
{'lc': 1, 'type': 'constructor', 'id': ['langchain', 'schema', 'runnable', 'RunnablePassthrough'], 'kwargs': {'func': None, 'afunc': None, 'input_type': None}, 'name': 'RunnablePassthrough', 'graph': {'nodes': [{'id': 0, 'type': 'schema', 'data': {'title': 'RunnablePassthroughInput'}}, {'id': 1, 'type': 'runnable', 'data': {'id': ['langchain', 'schema', 'runnable', 'RunnablePassthrough'], 'name': 'RunnablePassthrough'}}, {'id': 2, 'type': 'schema', 'data': {'title': 'RunnablePassthroughOutput'}}], 'edges': [{'source': 0, 'target': 1}, {'source': 1, 'target': 2}]}}

ただし、RunnableSerializable は抽象基底クラス（抽象メソッドを持っている）のため、直接インスタンス化できないことにはご注意ください：

>>> RunnableSerializable()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: Can't instantiate abstract class RunnableSerializable with abstract method invoke

`RunnableLambda`

Runnable クラスは Python の任意の関数から Runnable オジェクトを作るためのクラスです。

>>> from langchain_core.runnables import RunnableLambda
>>> runnable = RunnableLambda(lambda name: f'Hello! {name}')
>>> runnable.invoke('Alice') 
'Hello! Alice'

なお、下記のlangchain の公式ドキュメントにもサンプルコードが記載されているため合わせてご参照ください。

Tips ですが、入力されるインスタンスのメソッドを使用する際には下記のようなやり方も可能です：

>>> class Dummy:
...     def __init__(self, x):
...         self.x = x
...     def hello_x(self):
...         return f'Hello {self.x}'
...
>>> runnable = RunnableLambda(Dummy.hello_x)
>>> dummy = Dummy('Bob')
>>> runnable.invoke(dummy)
'Hello Bob'

`RunnableGenerator`

RunnableGenerator クラスはジェネレータ関数を実行するための Runnable クラスです。

ジェネレータ関数とはイテレータを作成するためのツールで下記のようなものです（詳細は　Python の公式ドキュメントをご覧ください）：

>>> def gen():
...     yield "good morning"
...     yield "good afternoon"
...     yield "good evening"
...
>>> gen()
<generator object gen at 0x00000236003EE7A0>
>>> list(gen()) 
['good morning', 'good afternoon', 'good evening']

RunnableGenerator クラスを使用するとストリーミング機能を保持しながら、カスタム出力パーサーなどのカスタム動作を実装することができます。
つまり、チェーンの前のステップからストリームされた出力チャンクをすぐに利用できるようになります。

例えば、ChatOpenAI を使ってストリーミングを行う際に、チャンクごとに長さを出力に付与したい場合は下記のように実装できます：

>>> from dotenv import load_dotenv
>>> from langchain_core.runnables import RunnableGenerator
>>> from langchain_openai import ChatOpenAI
>>> load_dotenv()
True
>>> chat = ChatOpenAI()
>>> chat.invoke('hello')  # test
AIMessage(content='Hello! How can I assist you today?')
>>> list(chat.stream('hello'))  # test
[AIMessageChunk(content=''), AIMessageChunk(content='Hello'), AIMessageChunk(content='!'), AIMessageChunk(content=' How'), AIMessageChunk(content=' can'), AIMessageChunk(content=' I'), AIMessageChunk(content=' assist'), AIMessageChunk(content=' you'), AIMessageChunk(content=' today'), AIMessageChunk(content='?'), AIMessageChunk(content='')]
>>> 
>>> # create a function to add their length to chunks
>>> def add_chunk_langth_generator(iterator):
...     '''generator of a tuple of chank length and chank.'''
...     for ai_message in iterator:
...         print('Received: ', ai_message)  # for debug
...         yield (len(ai_message.content), ai_message)
...
>>> runnable_generator = RunnableGenerator(add_chunk_langth_generator)
>>> list((chat | runnable_generator).stream('hello'))
Received:  content=''
Received:  content='Hello'
Received:  content='!'
Received:  content=' How'
Received:  content=' can'
Received:  content=' I'
Received:  content=' assist'
Received:  content=' you'
Received:  content=' today'
Received:  content='?'
Received:  content=''
[(0, AIMessageChunk(content='')), (5, AIMessageChunk(content='Hello')), (1, AIMessageChunk(content='!')), (4, AIMessageChunk(content=' How')), (4, AIMessageChunk(content=' can')), (2, AIMessageChunk(content=' I')), (7, AIMessageChunk(content=' assist')), (4, AIMessageChunk(content=' you')), (6, AIMessageChunk(content=' today')), (1, AIMessageChunk(content='?')), (0, AIMessageChunk(content=''))]

ただし、invoke を使う場合はチェーンの要素すべてに invoke が適用される（例えば上記の例では chat.invoke('hello')の実行結果がrunnable_generator.invokeの入力になる）ため、チャンクごとに適用するという操作がなくなることにはご注意ください。

>>> (chat | runnable_generator).invoke('hello') 
Received:  content='Hello! How can I assist you today?'
(34, AIMessage(content='Hello! How can I assist you today?'))

`RunnableSequence`

RunnableSequence クラスは LCEL におけるチェーンを実装したクラスで、Runnableオブジェクトからなる一連のシークエンスのためのクラスです。

RunnableSequence はそれ単体でインスタンス化するというよりもむしろ | 演算子で Runnableオブジェクトを繋げることでインスタンス化することが一般的です。

>>> from langchain_core.runnables import RunnableLambda, RunnableSequence
>>> r1 = RunnableLambda(lambda x: x)
>>> r2 = RunnableLambda(lambda x: (x, 1))
>>> r3 = RunnableLambda(lambda x: [x, 10])
>>> seq = r1 | r2 | r3
>>> isinstance(seq, RunnableSequence)
True
>>> isinstance(r1, RunnableSequence)
False
>>> seq.invoke(-1)
[(-1, 1), 10]

`RunnableParallel`

RunnableParallel クラスは Runnable オブジェクトのマッピングを並列に実行し、dict 型として出力する Runnable クラスです。インスタンス化はキーワード引数で行います。

以下は RunnableParallel で GPT-4 と GPT-3.5 を同時に実行する例です。

>>> from langchain_openai import ChatOpenAI
>>> from dotenv import load_dotenv
>>> from langchain_core.runnables import RunnableParallel
>>> load_dotenv()
True
>>> gpt35 = ChatOpenAI(model='gpt-3.5-turbo')
>>> gpt4 = ChatOpenAI(model='gpt-4-turbo-preview')
>>> gpt4.invoke('hello')  # test
AIMessage(content='Hello! How can I assist you today?')
>>> gpt35.invoke('hello')  # test
AIMessage(content='Hello! How can I assist you today?')
>>> 
>>> runnable = RunnableParallel(gpt4=gpt4, gpt35=gpt35)
>>> runnable.invoke('hello')
{'gpt4': AIMessage(content='Hello! How can I help you today?'), 'gpt35': AIMessage(content='Hello! How can I assist you today?')}
>>> runnable.invoke('Suppose, x=3. Calculate x^3') 
{'gpt4': AIMessage(content='To calculate \\(x^3\\) when \\(x = 3\\), you simply raise 3 to the power of 3. \n\n\\[ x^3 = 3^3 = 3 \\times 3 \\times 3 = 27 \\]'), 'gpt35': AIMessage(content='x^3 = 3^3 = 27')}

`RunnableMap`

RunnableMap は RunnableParallel のエイリアスです。

`RunnableEach`

RunnableEach は Runnable オブジェクトの実行を拡張し、入力のリストを出力のリストに変換するためのクラスです。
直接インスタンス化することもできますが、Runnable.map で作成することができます。Runnable.map についてはこちらも合わせてご参照ください。

以下、直接インスタンス化した場合と Runnable.map を使用して RunnableEach を作成した例です。

>>> from dotenv import load_dotenv
>>> from langchain_openai import ChatOpenAI
>>> from langchain_core.runnables import RunnableEach
>>> load_dotenv()
True
>>> chat = ChatOpenAI(model='gpt-3.5-turbo')
>>> each = RunnableEach(bound=chat)
>>> # Case: directly instantiate RunnableEach
>>> each.invoke(['hello', 'good morning', 'good evening'])
[AIMessage(content='Hello! How can I assist you today?'), AIMessage(content='Good morning! How can I assist you today?'), AIMessage(content='Good evening! How can I assist you today?')]
>>> # Case: use .map to instantiate RunnableEach
>>> chat.map().invoke(['hello', 'good morning', 'good evening'])
[AIMessage(content='Hello! How can I assist you today?'), AIMessage(content='Good morning! How can I assist you today?'), AIMessage(content='Good evening! How can I assist you today?')]
>>> 
>>> isinstance(chat.map(), RunnableEach)
True

`RunnableBinding`

RunnableBinding は Runnable オブジェクトをラップし、追加の機能などを与えるためのクラスです。

直接インスタンス化することも可能ですが、それは推奨されておらず、基本的には

Runnable.bind: invoke や batch などのキーワード引数をバインドする RunnableBinding オブジェクトを作成
- https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#bind
Runnable.with_config: invoke や batch などにおける config をバインドする RunnableBinding オブジェクトを作成
- https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#with_config
Runnable.with_listeners: Runnable オブジェクトの 実行開始前 実行開始後 実行エラー時 に呼ばれる callback をバインドする RunnableBinding オブジェクトを作成
- https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#with_listeners
Runnable.with_types: Runnable オブジェクトの入力型と出力型をバインドした RunnableBinding オブジェクトを作成
- https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#with_types

とともにインスタンス化することが一般的です。

>>> from dotenv import load_dotenv
>>> from langchain_openai import ChatOpenAI
>>> load_dotenv()
True
>>> chat = ChatOpenAI()
>>> # No kwargs for invoke
>>> chat.invoke('Explain the word "friend"')
AIMessage(content='A friend is a person with whom one has a bond of mutual affection, trust, and support. Friends are people who share common interests, values, and experiences, and who offer emotional support, companionship, and encouragement. Friends are often considered to be like family members, and are trusted confidants with whom one can share personal thoughts and feelings. Friends are important for social connection, emotional well-being, and personal growth.')
>>> # stop kwargs for invoke
>>> chat.invoke('Explain the word "friend"', stop=['end'])
AIMessage(content='A fri')
>>> # Wrap with `.bind`
>>> chat.bind(stop=['end']).invoke('Explain the word "friend"')  
AIMessage(content='A fri')

RunnableBindingの公式ドキュメントでは、RunnableWithFallbacks が RunnableBinding もしくは RunnableBindingBase を継承しているような記述がありますが、おそらく誤りかと思います。

>>> from langchain_core.runnables import RunnablePassthrough
>>> chat.with_fallbacks([RunnablePassthrough()])  
RunnableWithFallbacks(runnable=ChatOpenAI(client=<openai.resources.chat.completions.Completions object at 0x000002132C10F430>, async_client=<openai.resources.chat.completions.AsyncCompletions object at 0x000002132C128910>, openai_api_key=SecretStr('**********'), openai_proxy=''), fallbacks=[RunnablePassthrough()])
>>> from langchain_core.runnables.base import RunnableBinding, RunnableBindingBase
>>> isinstance(chat.with_fallbacks([RunnablePassthrough()]), RunnableBinding) 
False
>>> isinstance(chat.with_fallbacks([RunnablePassthrough()]), RunnableBindingBase)  
False

langchain_core/runnables/branch.py

`RunnableBranch`

RunnableBranch クラスは条件に基づいて実行する Runnable オブジェクトを切り替えるためのクラスです。

インスタンス化する際には、条件式を表す Runnable(もしくは Callable) と条件を満たした場合に実行される Runnable(もしくは Callable)のペア、および、デフォルトで（どの条件にも当てはまらなかった場合に）実行されるRunnable(もしくは Callable)を与えます。

Python の match-case や他の言語の switch-case の構文に似た働きをします。

>>> from langchain_core.runnables import RunnableBranch
>>> branch = RunnableBranch((lambda x: x>0, lambda x: f'{x} is positive!'), (lambda x: x<0, lambda x: f'{x} is negative!'), lambda x: f'{x} is Zero!')
>>> branch.invoke(1)
'1 is positive!'
>>> branch.invoke(-1)
'-1 is negative!'
>>> branch.invoke(0)
'0 is Zero!'

また、位置引数として先に与えた条件が優先されることには注意が必要です。

>>> branch = RunnableBranch((lambda x: x>0, lambda x: f'{x} is positive!'), (lambda x: x>1, lambda x: f'{x} is greater than 1!'), lambda x: 'Not Implemented')
'Not Implemented'
>>> branch.invoke(0.5)
'0.5 is positive!'
>>> branch.invoke(2)
'2 is positive!'

下記の langchainの公式ドキュメントも合わせて参考にすると勉強になると思います。

langchain_core/runnables/configurable.py

`RunnableConfigurableAlternatives`

RunnableConfigurableAlternatives クラスは DynamicRunnable 抽象基底クラスを継承ししており、動的に設定を変更することができるクラスです。

直接インスタンス化することも可能ですが、RunnableSerializable.configure_alternatives を用いてインスタンス化します。

引数は下記のとおりです：

which (ConfigurableField): (ConfigurableField(id='mode') のように変更可能なフィールドの ID を指定
default_key (str): configure_alternatives メソッドを呼び出す元のインスタンを選択する場合のキー（Defaults to "default"）
prefix_keys (bool): 元の Runnable の (Defaults to False)
kwargs(Union[Runnable[Input, Output], Callable[[], Runnable[Input, Output]]]): 切替用のキー及び切替先の Runnable オブジェクト等

そして with_config とともに用いることによりモデルの切り替えやプロンプトの切り替えなどを行うことができるようになります。

プロダクト・開発・テストの切り替えや A/B テストなどで役に立つかもしれません。

>>> from dotenv import load_dotenv
>>> from langchain_core.runnables import ConfigurableField, RunnableLambda
>>> from langchain_core.runnables.configurable import RunnableConfigurableAlternatives
>>> from langchain_openai import ChatOpenAI
>>> load_dotenv()
True
>>> gpt4 = ChatOpenAI(model='gpt-4-turbo-preview')
>>> chain = gpt4 | RunnableLambda(lambda ai_msg: ai_msg.content)
>>> chain_mock = RunnableLambda(lambda x: 'hello')
>>> chain_mock2 = RunnableLambda(lambda x: 'こんにちは')
>>> # Add a new config "mode" with key in ['default', 'mock', 'mock2']
>>> chain_configured = chain.configurable_alternatives(ConfigurableField(id='mode'), default_key='default', mock=chain_mock, mock2=chain_mock2)
>>> isinstance(chain_configured, RunnableConfigurableAlternatives)
True
>>> # switch model demo
>>> chain_configured.invoke('hello')
'Hello! How can I assist you today?'
>>> chain_configured.with_config(configurable=dict(mode='default')).invoke('')
'Hello! How can I assist you today? If you have any questions or need information, feel free to ask.'
>>> chain_configured.with_config(configurable=dict(mode='mock')).invoke('')
'hello'
>>> chain_configured.with_config(configurable=dict(mode='mock2')).invoke('')
'こんにちは'

もし、with_config で configurable_alternatives で指定していないキーを選択すると ValueError が発生します。

>>> chain_configured.with_config(configurable=dict(mode='chain_mock')).invoke('')
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "D:\qiita\runnables-in-langchain\venv\lib\site-packages\langchain_core\runnables\base.py", line 4069, in invoke
    return self.bound.invoke(
  File "D:\qiita\runnables-in-langchain\venv\lib\site-packages\langchain_core\runnables\configurable.py", line 94, in invoke
    runnable, config = self._prepare(config)
  File "D:\qiita\runnables-in-langchain\venv\lib\site-packages\langchain_core\runnables\configurable.py", line 475, in _prepare
    raise ValueError(f"Unknown alternative: {which}")
ValueError: Unknown alternative: chain_mock

もし、RunnableConfigurableAlternatives で設定されている ID やキーとキーに対応するオブジェクトを確認したいときは config_specs や alternatives 属性を確認すればよいです：

>>> chain_configured.config_specs
[ConfigurableFieldSpec(id='mode', annotation=<enum 'mode'>, name=None, description=None, default='default', is_shared=False, dependencies=None)]
>>> chain_configured.alternatives
{'mock': RunnableLambda(...), 'mock2': RunnableLambda(...)}

下記のlangchainの公式ドキュメントも併せてご参照ください。

`RunnableCongicurableFields`

RunnableConfigurableFields は Runnable オブジェクトの特定の属性を Configurable にするためのクラスです。

直接インスタンス化することも可能ですが、RunnableSerializable.configurable_fields を用いてインスタンス化します。

以下、max_tokens を Configurable な属性に変換し、with_config を用いて動的に max_tokens を 1 に指定するコードです。

>>> from langchain_core.runnables import ConfigurableField
>>> from langchain_core.runnables.configurable import RunnableConfigurableFields
>>> from langchain_openai import ChatOpenAI
>>> from dotenv import load_dotenv
>>> load_dotenv() 
True
>>> chat = ChatOpenAI()
>>> # Test
>>> chat.invoke('Create a poem') 
AIMessage(content="In the quiet of the night,\nI hear the whispers of my soul,\nTelling me to hold on tight,\nAnd never let go.\n\nThe stars above shine bright,\nGuiding me through the dark,\nI feel their warmth, their light,\nIgniting a spark.\n\nIn this moment of stillness,\nI find peace within,\nA sense of calm, of wholeness,\nA new journey to begin.\n\nI embrace the unknown,\nWith courage in my heart,\nI trust in the path I'm shown,\nAnd I know I'll never part.\n\nSo I walk on, with faith,\nAnd a song in my soul,\nFor in the darkness, there's always a wraith,\nTo guide me towards my goal.")>>> chat.with_config(configurable=dict(max_tokens=1)).invoke('Create a poem') 
AIMessage(content="In the stillness of the night,\nI dream of a world so bright.\nWhere flowers bloom and birds take flight,\nAnd everything feels just right.\n\nThe stars above twinkle and gleam,\nGuiding me through this beautiful dream.\nThe moon shines down, a comforting beam,\nAs I drift into a peaceful stream.\n\nI see the beauty all around,\nIn every sight and every sound.\nNature's wonders know no bound,\nIn this magical place where I am found.\n\nI close my eyes and let go,\nSurrendering to the gentle flow.\nIn this moment, I feel the glow,\nOf love and light that forever grow.\n\nIn the stillness of the night,\nI am filled with pure delight.\nIn this dream, everything is so right,\nA world of wonder, shining bright.")
>>> # configurable_fields Example
>>> chat.configurable_fields(max_tokens=ConfigurableField(id='max_tokens')).with_config(configurable=dict(max_tokens=1)).invoke('Create a poem')  
AIMessage(content='In')

なお、runnable_serializable.configurable_fields で指定するキーワード引数は runnable_serializable.__fields__.keys() の中に含まれていなければいけないことには注意が必要です。

>>> chat.__fields__.keys()
dict_keys(['name', 'cache', 'verbose', 'callbacks', 'callback_manager', 'tags', 'metadata', 'client', 'async_client', 'model_name', 'temperature', 'model_kwargs', 'openai_api_key', 'openai_api_base', 'openai_organization', 'openai_proxy', 'request_timeout', 'max_retries', 'streaming', 'n', 'max_tokens', 'tiktoken_model_name', 'default_headers', 'default_query', 'http_client'])
>>> 'max_tokens' in chat.__fields__
True
>>> 'min_tokens' in chat.__fields__
False
>>> chat.configurable_fields(min_tokens=ConfigurableField(id='min_tokens'))
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "D:\qiita\runnables-in-langchain\venv\lib\site-packages\langchain_core\runnables\base.py", line 1673, in configurable_fields
    raise ValueError(
ValueError: Configuration key min_tokens not found in client=<openai.resources.chat.completions.Completions object at 0x000002132C127B80> async_client=<openai.resources.chat.completions.AsyncCompletions object at 0x000002132C1247C0> openai_api_key=SecretStr('**********') openai_proxy='': available keys are dict_keys(['name', 'cache', 'verbose', 'callbacks', 'callback_manager', 'tags', 'metadata', 'client', 'async_client', 'model_name', 'temperature', 'model_kwargs', 'openai_api_key', 'openai_api_base', 'openai_organization', 'openai_proxy', 'request_timeout', 'max_retries', 'streaming', 'n', 'max_tokens', 'tiktoken_model_name', 'default_headers', 'default_query', 'http_client'])

下記のlangchainの公式ドキュメントも併せてご参照ください。

langchain_core/runnables/fallbacks.py

`RunnableWithFallbacks`

RunnableWithFallbacks はある Runnable オブジェクトの失敗時に別の Runnable オブジェクトにフォールバックすることができるクラスです。

このフォールバックは LLM のモデルレベルだけでなく、Runnable オブジェクトのレベルで定義することが可能です。

RunnableWithFallbacks は直接インスタンス化可能ですが、Runnable.with_fallbacks を使用してインスタンス化する方が便利です。過去の記事も併せてご参照ください。

以下、with_fallbacks を用いたサンプルコードです。

>>> from langchain_core.runnables import RunnableLambda
KeyboardInterrupt
>>> def raise_exception(*args):
...     raise Exception('Dummy')
...
>>> runnable = RunnableLambda(raise_exception).with_fallbacks([RunnableLambda(lambda x: 'fallback1')])
>>> runnable.invoke('hi')
'fallback1'

なお、RunnableWithFallbacks はフォールバックの内 1 つが成功もしくは全部が失敗するまで順番に試行されます。
すべてのフォールバックが失敗した場合は一番最初の例外を最終的に raise します。

>>> # CASE: 2nd fallback works
>>> runnable_2nd_fallback = RunnableLambda(raise_exception).with_fallbacks([RunnableLambda(raise_exception), RunnableLambda(lambda x: 'fallback2'), RunnableLambda(raise_exception)])
>>> runnable_2nd_fallback.invoke('hello')
'fallback2'
>>> # CASE: All fallbacks raise an Exception
>>> def raise_1(*args):
...     raise Exception('1')
...
>>> def raise_2(*args):
...     raise Exception('2')
...
>>> def raise_3(*args):
...     raise Exception('3')
...
>>> runnable_all_error = RunnableLambda(raise_1).with_fallbacks([RunnableLambda(raise_2), RunnableLambda(raise_3)])
>>> runnable_all_error.invoke('hello') 
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "D:\qiita\runnables-in-langchain\venv\lib\site-packages\langchain_core\runnables\fallbacks.py", line 185, in invoke
    raise first_error
  File "D:\qiita\runnables-in-langchain\venv\lib\site-packages\langchain_core\runnables\fallbacks.py", line 167, in invoke
    output = runnable.invoke(
  File "D:\qiita\runnables-in-langchain\venv\lib\site-packages\langchain_core\runnables\base.py", line 3523, in invoke
    return self._call_with_config(
  File "D:\qiita\runnables-in-langchain\venv\lib\site-packages\langchain_core\runnables\base.py", line 1262, in _call_with_config
    context.run(
  File "D:\qiita\runnables-in-langchain\venv\lib\site-packages\langchain_core\runnables\config.py", line 326, in call_func_with_variable_args
    return func(input, **kwargs)  # type: ignore[call-arg]
  File "D:\qiita\runnables-in-langchain\venv\lib\site-packages\langchain_core\runnables\base.py", line 3397, in _invoke
    output = call_func_with_variable_args(
  File "D:\qiita\runnables-in-langchain\venv\lib\site-packages\langchain_core\runnables\config.py", line 326, in call_func_with_variable_args
    return func(input, **kwargs)  # type: ignore[call-arg]
  File "<stdin>", line 2, in raise_1
Exception: 1

下記のlangchainの公式ドキュメントも併せてご参照ください。

langchain_core/runnables/history.py

`RunnableWithMessageHistory`

RunnableWithMessageHistory は Runnable オブジェクトをラップしてチャットメッセージ履歴を管理するためのクラスです。

下記の入力・出力を持つ Runnable であればどれに対してもラップすることが可能です：

入力
- BaseMessage のシークエンス
- BaseMessage のシークエンスを取るキーを持つ辞書
- 最新のメッセージを文字列または BaseMessage のシークエンスとして取得するキー及び履歴メッセージを取得する別のキーを含む辞書
出力
- AIMessage のコンテンツとして扱うことができる文字列
- BaseMessage のシークエンス
- BaeeMessage のシークエンスを含むキーを持つ辞書

>>> from dotenv import load_dotenv
>>> load_dotenv()
True
>>> from langchain.memory import ChatMessageHistory
>>> from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
>>> from langchain_core.runnables.history import RunnableWithMessageHistory
>>> global_store = {}
>>> def get_session_history(session_id):  
...     if session_id not in global_store:
...         global_store[session_id] = ChatMessageHistory()
...     return global_store[session_id]

>>> prompt = ChatPromptTemplate.from_messages([MessagesPlaceholder(variable_name='history'), ('human', '{input}')])
>>> chat = ChatOpenAI()
>>> # add a memory to a chain with prompt and chat
>>> chat_with_memory = RunnableWithMessageHistory(prompt | chat, get_session_history, input_messages_key='input', history_messages_key='history')
>>> chat_with_memory.invoke({'input': 'echo 1'}, config={'configurable': {'session_id': 'abc123'}}) 
AIMessage(content='1')
>>> chat_with_memory.invoke({'input': 'add 1 to the previous number'}, config={'configurable': {'session_id': 'abc123'}})
AIMessage(content='2')
>>> chat_with_memory.invoke({'input': 'add 1 to the previous number'}, config={'configurable': {'session_id': 'abc123'}})
AIMessage(content='3')
>>> chat_with_memory.invoke({'input': 'add 1 to the previous number'}, config={'configurable': {'session_id': 'abc123'}})
AIMessage(content='4')
>>> global_store
{'abc123': ChatMessageHistory(messages=[HumanMessage(content='echo 1'), AIMessage(content='1'), HumanMessage(content='add 1 to the previous number'), AIMessage(content='2'), HumanMessage(content='add 1 to the previous number'), AIMessage(content='3'), HumanMessage(content='add 1 to the previous number'), AIMessage(content='4')])}

下記のlangchainの公式ドキュメントの方が詳しいサンプルが記載されていましたのでご参考にしてください。

langchain_core/runnables/passthrough.py

`RunnablePassthrough`

RunnablePassthrough は入力に何も変化を与えない Runnable クラスです。
恒等関数のような役割を果たします。

>>> from langchain_core.runnables import RunnablePassthrough
>>> passthrough = RunnablePassthrough()
>>> passthrough.invoke(1)
1
>>> passthrough.invoke('hello')
'hello'

また、RunnablePassthrough の引数に Calalble を渡すことでコールバックのような役割も果たせます（筆者はどう使えばよいかあまりイメージできていませんのでご注意ください）

>>> runnable = RunnablePassthrough(lambda x: print('Received: ', x))
>>> runnable.invoke('hello') 
Received:  hello
'hello'

RunnablePassthrough ではクラスメソッドとして assign が定義されており、後述の RunnableAssign のファクトリとしても使用することができます。

>>> runnable = RunnablePassthrough.assign(x=RunnablePassthrough(), y=RunnableLambda(lambda dic: dic['x']**2))
>>> runnable.invoke({'x': 2}) 
{'x': {'x': 2}, 'y': 4}

ただし、RunnablePassthrough.assign() への入力タイプは辞書型なのでご注意ください：

>>> RunnablePassthrough.assign().InputType
typing.Dict[str, typing.Any]

下記のlangchainの公式ドキュメントも併せてご参照ください。

`RunnableAssign`

RunnableAssign は入力に対して新規フィールドを付与して出力する Runnable クラスです。

直接インスタンス化することも可能ですが、RunnablePassthrough.assign や Runnable.assign を使ってインスタンス化することが多いです（RunnablePassthrough.assign はクラスメソッド、 Runnable.assign はインスタンスメソッドであることにご注意ください）。

直接インスタンス化して利用した例が下記のコードです。

>>> from langchain_core.runnables import RunnableAssign, RunnableLambda, RunnablePassthrough
>>> assign = RunnableAssign({'x': RunnablePassthrough()})
>>> assign.invoke({'human': 100})
{'human': 100, 'x': {'human': 100}}

このように {'human': 100} という入力に対して、新たに x というキーで RunnablePassthrough() の出力を付与していることが分かります。

また、Runnable.assign を使った例が下記のコードです。

>>> runnable = RunnableLambda(lambda x: {'input': f'Input >> {x}'})
>>> runnable.invoke('hello') 
{'input': 'Input >> hello'}
>>> runnable.assign(output=lambda dic: f'Output >> {dic["input"]}').invoke('hello')
{'input': 'Input >> hello', 'output': 'Output >> Input >> hello'}

念のためですが、RunnableAssign オブジェクトの invoke メソッドが入力された辞書を破壊しないことを確認しておきましょう。上で作成した assign 変数を用います。

>>> dic = {'human': 'hello'}
>>> assign.invoke(dic)
{'human': 'hello', 'x': {'human': 'hello'}}
>>> dic
{'human': 'hello'}

先の RunnablePassthrough のセクションでも触れましたが RunnableAssign への入力タイプは辞書型なのでご注意ください。

こちらの記事でも別の例を用意していますのでご確認ください。

`RunnablePick`

RunnablePick は入力された辞書から指定されたキーを取得するための Runanble クラスです。

>>> from langchain_core.runnables.passthrough import RunnablePick
>>> RunnablePick(['human']).invoke({'human': 'hi', 'extra': 'dummy'})
{'human': 'hi'}
>>> RunnablePick('human').invoke({'human': 'hi', 'extra': 'dummy'})
'hi'

また、Runnable.pick メソッドを用いて RunnablePick を作成し | と結合することも可能です。詳細は下記の記事をご参考にしてください。

なお、RunnablePick クラスも RunnableAssign クラスと同様に入力タイプは辞書型なのでご注意ください。

langchain_core/runnables/retry.py

`RunnableRetry`

RunnableRetry は Runanble オブジェクトをラップし失敗時にリトライするための Runnable クラスです。

直接インスタンス化することも可能ですが、Runnable.with_retry を使用してインスタンス化できます。

デモのために内部にカウンターを持ち、３回目の呼び出し時に初めて成功する Callable オブジェクトを使用します。

>>> class SuccessAfter2:
...     def __init__(self):
...         self.c = 0
...     def __call__(self, *args, **kwargs):
...         if self.c < 2:
...             self.c += 1
...             raise ValueError(f"self.c = {self.c} < 2")
...         return 'success'
...
>>> obj = SuccessAfter2()
>>> obj()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 7, in __call__
ValueError: self.c = 1 < 2
>>> obj()
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "<stdin>", line 7, in __call__
ValueError: self.c = 2 < 2
>>> obj()
'success'

さて、下記は SuccessAfter2 を使用して RunnableRetry を直接インスタンス化したコードです：

>>> from langchain_core.runnables.retry import RunnableRetry 
>>> # Without Retry
>>> runnable = RunnableLambda(SuccessAfter2())
>>> runnable.invoke(None)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  --- 中略 ---
  File "<stdin>", line 7, in __call__
ValueError: self.c = 1 < 2
>>> runnable.invoke(None)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  --- 中略 ---
  File "<stdin>", line 7, in __call__
ValueError: self.c = 2 < 2
>>> runnable.invoke(None)
'success'
>>> # With retry
>>> runnable_with_retry = RunnableRetry(
...     bound = RunnableLambda(SuccessAfter2()),
...     max_attempt_number=3,  # How many times to retry ?
...     retry_exception_types=(ValueError,)  # Which exceptions will be retried?
... )
>>> runnable_with_retry.invoke(None)
'success'

また、下記は Runnable.with_retry を使った例です。

>>> runnable.with_retry(stop_after_attempt=3, retry_if_exception_type=(ValueError,)).invoke(None)
'success'

若干、キーワード引数が異なるので要注意です

max_attempt_number -> stop_after_attempt
retry_exception_types -> retry_if_exception_type

Runanble.with_retry については別記事でもコード例を載せていますのでご参照ください。

langchain_core/runnables/router.py

`RouterRunnable`

RouterRunnable は入力として与えられる Mapping オブジェクトの key キーに対応するバリューに応じて実行する Runnable オブジェクトを切り替える Runnable クラスです。

入力として与えるMappingオブジェクトは key および input のキーを持っていなければなりません。

>>> from langchain_core.runnables.router import RouterRunnable
>>> router  = RouterRunnable(runnables={'x': RunnableLambda(lambda i: f'called x with input {i}'), 'y': RunnableLambda(lambda i: f'called y with input {i}')})
>>> router.invoke({'key': 'x', 'input': 'hello'})
'called x with input hello'
>>> router.invoke({'key': 'y', 'input': 'こんにちは'})
'called y with input こんにちは'

もちろん、同じような機能は RunanbleBranch でも実現できるとは思いますが、
うまく出力の方を定義すれば条件式が不要となるためコードがすっきりするかもしれません。

最後に

ここまで読んでいただきありがとうございました。

勝手な判断ですが「覚えておくべき」「できれば覚えておくべき」「多分知らなくても大丈夫」で分類してみるとこんな感じでしょうか？

覚えておくべき
- Runnable
- RunnableLambda
- RunnableParallel
- RunnableBranch
- RunnableRetry
- RunnableWithFallbacks
- RunnablePassthrough
できれば覚えておくべき
- RunnableSerializable
- RunnableSequence
- RunnableEach
- RunnableBinding
- RunnableGenerator
- RunnableWithMessageHistory
- RunnableAssign
- RunnablePick
多分知らなくても大丈夫
- RunnableConfigurableAlternatives
- RunnableCongicurableFields
- RunnableMap
- RouterRunnable

ただ、これらのクラスを特に意識していなくても、例えば Runnable.with_fallbacks や Runnable.with_retry などでうまくクラスの存在が隠されているのでそこまで身構えなくて大丈夫だと思います。

個人的には RunnableGenerator と RouterRunnable が目からウロコでした。

ぜひ、本記事が langchain を勉強している方のお役に立てれば幸いです。

また、下記の記事では Runnable クラスについて実装されているメソッドについて実行例を以て使用方法を説明しておりますので、合わせてご活用ください。

ご質問やご指摘等ございましたらぜひコメントいただけますと幸いです。
もっとわかりやすい例などありましたら、ぜひご教示ください。

それでは、このあたりで。

https://python.langchain.com/docs/get_started/introduction ↩

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up

実行例で理解する `Runnable` の継承者たち in `langchain`

概要

はじめに

本書の対象読者

本書の実行環境

Runnable クラスの継承者

各クラスの詳細/実行例

RunnableConfigurableAlternatives

RunnableCongicurableFields

最後に

`Runnable` クラスの継承者

`RunnableConfigurableAlternatives`

`RunnableCongicurableFields`