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

## 概要

この記事では、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`](#runnable) | LCEL の中枢をなす抽象基底クラス |
| [`RunnableSerializable`](#runnableserializable) | `Runnable` に加えて `Serializable` を継承した抽象基底クラス |
| [`RunnableLambda`](#runnablelambda) | Python の任意の関数から `Runnable` オブジェクトを作るための `Runnable` クラス |
| [`RunnableGenerator`](#runnablegenerator) | ジェネレータ関数を実行するための `Runnable` クラス |
| [`RunnableSequence`](#runnablesequence) | LCEL におけるチェーンを実装した `Runnable` クラス |
| [`RunnableParallel`](#runnableparallel) | `Runnable` オブジェクトのマッピングを並列に実行し、`dict` 型として出力する `Runnable` クラス |
| [`RunnableMap`](#runnablemap) | `RunnableParallel` のエイリアス |
| [`RunnableEach`](#runnableeach) | 入力のリストを出力のリストに変換するための `Runnable` クラス |
| [`RunnableBinding`](#runnablebinding) | `Runnable` オブジェクトをラップし、追加の機能などを与えるための `Runnable` クラス |
| [`RunnableBranch`](#runnablebranch) | 条件に基づいて実行する `Runnable` オブジェクトを切り替えるためのクラス |
| [`RunnableConfigurableAlternatives`](#runnableconfigurablealternatives) | 動的に設定を変更することができる `Runnable` クラス |
| [`RunnableCongicurableFields`](#runnablecongicurablefields) | `Runnable` オブジェクトの特定の属性を `Configurable` にするためのクラス |
| [`RunnableWithFallbacks`](#runnablewithfallbacks) | 失敗時に別の `Runnable` オブジェクトにフォールバックするための `Runnable` クラス |
| [`RunnableWithMessageHistory`](#runnablewithmessagehistory) | チャットメッセージ履歴を管理するための `Runnable` クラス |
| [`RunnablePassthrough`](#runnablepassthrough) | 入力に何も変化を与えない `Runnable` クラス |
| [`RunnableAssign`](#runnableassign) | 入力に対して新規フィールドを付与して出力する `Runnable` クラス |
| [`RunnablePick`](#runnablepick) | 入力された辞書から指定されたキーを取得するための `Runanble` クラス |
| [`RunnableRetry`](#runnableretry) | 失敗時にリトライするための `Runnable` クラス |
| [`RouterRunnable`](#routerrunnable) | 入力に応じて実行する `Runnable` オブジェクトを切り替える `Runnable` クラス |

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


## はじめに

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

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

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

https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91

しかしながら、`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`ファイルを下記の内容で作成してください：

```bash
OPENAI_API_KEY={YOUR_API_KEY}
```

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

#### `Runnable` クラスの継承者

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

https://github.com/langchain-ai/langchain/tree/f96dd57501131840b713ed7c2e86cbf1ddc2761f/libs/core

:::note warn
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 で利用可能です。

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

```plantuml
@startuml LangChain-Runnable-Hierarchy-Optimized

left to right direction

' base.py group
package "base" {
    class Runnable {
    }

    class RunnableSerializable {
    }
    Runnable <|-- RunnableSerializable

    class RunnableSequence {
    }
    RunnableSerializable <|-- RunnableSequence

    class RunnableParallel {
    }
    RunnableSerializable <|-- RunnableParallel

    class RunnableMap {
    }
    RunnableParallel <|-- RunnableMap

    class RunnableGenerator {
    }
    Runnable <|-- RunnableGenerator

    class RunnableLambda {
    }
    Runnable <|-- RunnableLambda

    class RunnableEachBase {
    }
    RunnableSerializable <|-- RunnableEachBase

    class RunnableEach {
    }
    RunnableEachBase <|-- RunnableEach

    class RunnableBindingBase {
    }
    RunnableSerializable <|-- RunnableBindingBase

    class RunnableBinding {
    }
    RunnableBindingBase <|-- RunnableBinding
}

' Other groups
package "passthrough"{
    class RunnablePassthrough {
    }
    RunnablePassthrough --|>RunnableSerializable

    class RunnableAssign {
    }
    RunnableAssign --|>RunnableSerializable

    class RunnablePick {
    }
    RunnablePick --|>RunnableSerializable
}

package "branch" {
    class RunnableBranch {
    }
    RunnableSerializable <|-- RunnableBranch
}

package "fallbacks" {
    class RunnableWithFallbacks {
    }
    RunnableWithFallbacks --|> RunnableSerializable
}

package "history" {
    class RunnableWithMessageHistory {
    }
    RunnableBindingBase <|-- RunnableWithMessageHistory
}

package "retry" {
    class RunnableRetry {
    }
    RunnableBindingBase <|-- RunnableRetry
}

package "router" {
    class RouterRunnable {
    }
    RunnableSerializable <|-- RouterRunnable
}

package "configurable" {
    class DynamicRunnable {
    }
    RunnableSerializable <|-- DynamicRunnable

    class RunnableConfigurableFields {
    }
    DynamicRunnable <|-- RunnableConfigurableFields

    class RunnableConfigurableAlternatives {
    }
    DynamicRunnable <|-- RunnableConfigurableAlternatives
}

@enduml
```

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

## 各クラスの詳細/実行例

### [base.py](https://github.com/langchain-ai/langchain/blob/f96dd57501131840b713ed7c2e86cbf1ddc2761f/libs/core/langchain_core/runnables/base.py)

#### [`Runnable`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.base.Runnable.html)

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

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

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

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

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

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

https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91

#### [`RunnableSerializable`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.base.RunnableSerializable.html)

`Runnable` に加えて、[`Serializable`](https://api.python.langchain.com/en/stable/load/langchain_core.load.serializable.Serializable.html) も継承した**抽象基底クラス**です。

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

```python
>>> 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` は抽象基底クラス（抽象メソッドを持っている）のため、直接インスタンス化できないことにはご注意ください：

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

#### [`RunnableLambda`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.base.RunnableLambda.html)

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

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

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

https://python.langchain.com/docs/expression_language/how_to/functions

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

```python
>>> 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`](https://api.python.langchain.com/en/stable/_modules/langchain_core/runnables/base.html#RunnableGenerator)

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

ジェネレータ関数とはイテレータを作成するためのツールで下記のようなものです（詳細は　[Python の公式ドキュメント](https://docs.python.org/ja/3/tutorial/classes.html#generators) をご覧ください）：

```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` を使ってストリーミングを行う際に、チャンクごとに長さを出力に付与したい場合は下記のように実装できます：

```python
>>> 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`の入力になる）ため、チャンクごとに適用するという操作がなくなることにはご注意ください。

```python
>>> (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`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.base.RunnableSequence.html)

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

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

```python
>>> 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`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.base.RunnableParallel.html)

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

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

```python
>>> 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`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.base.RunnableMap.html)

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

#### [`RunnableEach`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.base.RunnableEach.html)

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

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

```python
>>> 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`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.base.RunnableBinding.html)

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

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

- `Runnable.bind`: `invoke` や `batch` などのキーワード引数をバインドする `RunnableBinding` オブジェクトを作成

  - [https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#bind](https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#bind)

- `Runnable.with_config`: `invoke` や `batch` などにおける `config` をバインドする `RunnableBinding` オブジェクトを作成

  - [https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#with_config](https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#with_config)

- `Runnable.with_listeners`: `Runnable` オブジェクトの **実行開始前** **実行開始後** **実行エラー時** に呼ばれる callback をバインドする `RunnableBinding` オブジェクトを作成

  - [https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#with_listeners](https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#with_listeners)

- `Runnable.with_types`: `Runnable` オブジェクトの入力型と出力型をバインドした `RunnableBinding` オブジェクトを作成

  - [https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#with_types](https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#with_types)

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

```python
>>> 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')
```

:::note info
[`RunnableBinding`の公式ドキュメント](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.base.RunnableBinding.html)では、`RunnableWithFallbacks` が `RunnableBinding` もしくは `RunnableBindingBase` を継承しているような記述がありますが、おそらく誤りかと思います。

```python
>>> 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](https://github.com/langchain-ai/langchain/blob/f96dd57501131840b713ed7c2e86cbf1ddc2761f/libs/core/langchain_core/runnables/branch.py)

#### [`RunnableBranch`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.branch.RunnableBranch.html)

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

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

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

```python
>>> 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!'
```

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

```python
>>> 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`の公式ドキュメントも合わせて参考にすると勉強になると思います。

https://python.langchain.com/docs/expression_language/how_to/routing

### [langchain_core/runnables/configurable.py](https://github.com/langchain-ai/langchain/blob/f96dd57501131840b713ed7c2e86cbf1ddc2761f/libs/core/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 テストなどで役に立つかもしれません。

```python
>>> 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` が発生します。

```python
>>> 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` 属性を確認すればよいです：

```python
>>> 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`の公式ドキュメントも併せてご参照ください。

https://python.langchain.com/docs/expression_language/how_to/configure#configurable-alternatives

<!--
TODO: `prepare_keys` の使い方
-->

#### `RunnableCongicurableFields`

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

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

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

```python
>>> 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()` の中に含まれていなければいけないことには注意が必要です。

```python
>>> 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`の公式ドキュメントも併せてご参照ください。

https://python.langchain.com/docs/expression_language/how_to/configure

### [langchain_core/runnables/fallbacks.py](https://github.com/langchain-ai/langchain/blob/f96dd57501131840b713ed7c2e86cbf1ddc2761f/libs/core/langchain_core/runnables/fallbacks.py)

#### [`RunnableWithFallbacks`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.fallbacks.RunnableWithFallbacks.html)

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

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

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

https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#with_fallbacks

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

```python
>>> 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` します。

```python
>>> # 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`の公式ドキュメントも併せてご参照ください。

https://python.langchain.com/docs/expression_language/how_to/fallbacks

### [langchain_core/runnables/history.py](https://github.com/langchain-ai/langchain/blob/f96dd57501131840b713ed7c2e86cbf1ddc2761f/libs/core/langchain_core/runnables/history.py)

#### [`RunnableWithMessageHistory`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.history.RunnableWithMessageHistory.html)

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

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

- 入力
  - `BaseMessage` の シークエンス
  - `BaseMessage` のシークエンスを取るキーを持つ辞書
  - 最新のメッセージを文字列 または `BaseMessage` のシークエンスとして取得するキー及び履歴メッセージを取得する別のキーを含む辞書

- 出力
  - `AIMessage` のコンテンツとして扱うことができる文字列
  - `BaseMessage` のシークエンス
  - `BaeeMessage` のシークエンスを含むキーを持つ辞書

```python
>>> 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`の公式ドキュメントの方が詳しいサンプルが記載されていましたのでご参考にしてください。

https://python.langchain.com/docs/expression_language/how_to/message_history

### [langchain_core/runnables/passthrough.py](https://github.com/langchain-ai/langchain/blob/f96dd57501131840b713ed7c2e86cbf1ddc2761f/libs/core/langchain_core/runnables/passthrough.py)

#### [`RunnablePassthrough`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.passthrough.RunnablePassthrough.html)

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

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

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

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

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

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

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

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

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

https://python.langchain.com/docs/expression_language/how_to/passthrough

#### [`RunnableAssign`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.passthrough.RunnableAssign.html)

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

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

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

```python
>>> 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` を使った例が下記のコードです。

```python
>>> 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` 変数を用います。

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

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

[こちらの記事](https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#assign)でも別の例を用意していますのでご確認ください。

#### [`RunnablePick`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.passthrough.RunnablePick.html)

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

```python
>>> 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` を作成し `|` と結合することも可能です。詳細は下記の記事をご参考にしてください。

https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#pick

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

### [langchain_core/runnables/retry.py](https://github.com/langchain-ai/langchain/blob/f96dd57501131840b713ed7c2e86cbf1ddc2761f/libs/core/langchain_core/runnables/retry.py)

#### [`RunnableRetry`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.retry.RunnableRetry.html)

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

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

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

```python
>>> 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` を直接インスタンス化したコードです：

```python
>>> 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` を使った例です。

```python
>>> 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` については[別記事](https://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91#with_retry)でもコード例を載せていますのでご参照ください。

### [langchain_core/runnables/router.py](https://github.com/langchain-ai/langchain/blob/f96dd57501131840b713ed7c2e86cbf1ddc2761f/libs/core/langchain_core/runnables/router.py)

#### [`RouterRunnable`](https://api.python.langchain.com/en/stable/runnables/langchain_core.runnables.router.RouterRunnable.html)

`RouterRunnable` は入力として与えられる [`Mapping`](https://docs.python.org/3.3/library/collections.abc.html#collections.abc.Mapping) オブジェクトの `key` キーに対応するバリューに応じて実行する `Runnable` オブジェクトを切り替える `Runnable` クラスです。

入力として与える[`Mapping`](https://docs.python.org/3.3/library/collections.abc.html#collections.abc.Mapping)オブジェクトは `key` および `input` のキーを持っていなければなりません。

```python
>>> 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://qiita.com/hmasdev/items/6414fd7a63dc47a9ef91

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

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

[^1]: [https://python.langchain.com/docs/get_started/introduction](https://python.langchain.com/docs/get_started/introduction)