AzureからGPTを使ってみる｜Runnable編

LangchainのRunnableとは

LangChainのRunnableは、LangChainフレームワークにおける汎用的なコンポーネントで、さまざまな種類のタスクやプロセスを実行するために使用されます。具体的には、LangChainの実行可能オブジェクト（Runnable Object）は、特定のタスクや操作をカプセル化し、これを呼び出して実行することができます。複数のRunnableをチェーン（連鎖）させることで、複雑な処理パイプラインを構築できます。各Runnableは、前のRunnableの出力を受け取り、それを処理して次のRunnableに渡します。
今回は、その中でも、
・RunnableLambda
・RunnableBranch
・RouterRunnable
を、使ってみます。
RunnableWithMessageHistoryについては、こちらの記事を見てください。

RunnableLambda

Pythonのラムダ関数や任意の小さな関数を迅速にRunnableに変換するために使用されます。これにより、複雑なクラスを定義せずに簡単な処理を実行可能な形式で使用できます。
以下の例のように使用できます。

#langchain=0.2.7
#langchain_core=0.2.12
from langchain_core.runnables import RunnableLambda
runnable = RunnableLambda(lambda name: f'Hello! {name}')
runnable.invoke('PON吉') 

'Hello! PON吉'

from langchain_core.runnables import RunnableLambda
class Lambda_test:
    def __init__(self, x):
        self.x = x
    def hello_name(self):
        return f"hello {self.x}"
runnable = RunnableLambda(Lambda_test.hello_name)
test = Lambda_test('PON吉')
runnable.invoke(test)

'hello PON吉'

from langchain_core.runnables import RunnableLambda
def add_one(x: int) -> int:
    return x + 1

runnable_add_one = RunnableLambda(add_one)

result = runnable_add_one.invoke(3)
print(result)  # 出力: 4

result = runnable_add_one.map().invoke([1, 2, 3])
print(result)　#出力：[2, 3, 4]

4
[2, 3, 4]

RunnableBranch

入力データに基づいて異なる処理を動的に選択し、実行することができます。条件に応じて異なるRunnableを実行するため、柔軟な処理フローを構築するのに役立ちます。

from langchain_core.runnables import RunnableBranch

# 分岐処理の定義
branch = RunnableBranch(
    (lambda x: isinstance(x, int), lambda x: f"Processed integer: {x * 2}"),
    (lambda x: isinstance(x, str), lambda x: f"Processed string: {x.upper()}"),
    lambda x: "goodbye",
)

# 入力データをパイプラインに渡して実行
input_data1 = 42
output1 = branch.invoke(input_data1)

input_data2 = "hello"
output2 = branch.invoke(input_data2)

print(output1)  
print(output2)  

Processed integer: 84
Processed string: HELLO

注意点として先に与えた条件が優先されます。

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')
    )
output1 = branch.invoke(0.5)
output2 = branch.invoke(2)

print(output1)  
print(output2)  

0.5 is positive!
2 is positive!

RouterRunnable

RunnableBranchと同じように、条件によって分岐させる。
例は、現在の時間で出力する挨拶を変化させる。

from datetime import datetime
from langchain_core.runnables.router import RouterRunnable
from langchain_core.runnables import RunnableLambda

# RunnableLambdaで関数をラッピング
morning_runnable = RunnableLambda(lambda input: f"Good morning! {input}")
afternoon_runnable = RunnableLambda(lambda input: f"Good afternoon! {input}")
evening_runnable = RunnableLambda(lambda input: f"Good evening!{input}")

# 現在の時間に基づいてキーを選択する関数を定義
def select_time_based_key():
    current_hour = datetime.now().hour
    print(current_hour)
    if current_hour < 12:
        return 'morning'
    elif 12 <= current_hour < 18:
        return 'afternoon'
    else:
        return 'evening'

# RouterRunnableを作成
router = RouterRunnable(
    runnables={
        'morning': morning_runnable,
        'afternoon': afternoon_runnable,
        'evening': evening_runnable
    }
)

# 挨拶を実行する関数を定義
def greet(input_text):
    key = select_time_based_key()
    return router.invoke({'key': key, 'input': input_text})

# テストの挨拶メッセージを設定
test_input = "Hope you have a great day!"

# 挨拶を実行し、結果を取得
result = greet(test_input)

print(result)

15
Good afternoon! Hope you have a great day!

他にも、

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}')}
    )

output1 = router.invoke({'key': 'x', 'input': 'hello'})
output2 = router.invoke({'key': 'y', 'input': 'こんにちは'})

print(output1)  
print(output2)  

called x with input hello
called y with input こんにちは

おまけ

runnableをパイプやRunnableSequenceでつなぐこともできる。

from langchain_core.runnables import RunnableLambda
from langchain_core.runnables import RunnableSequence

add_two = RunnableLambda(lambda x: x+2)
double = RunnableLambda(lambda x: f"Processed{x*2}")

chain1 = add_two|double
chain2 = RunnableSequence(add_two, double)

output1 = chain1.invoke(48)
output2 = chain2.invoke(48)

print("outputt1:" + output1)  
print("outputt2:" + output2)  

outputt1:Processed100
outputt2:Processed100