こんにちは。AWS CLIが好きな福島です。
はじめに
今回は、LangChainでchainを組む際に重要なクラスである以下のRunnable〇〇の使い方をご紹介します。
- RunnableParallel
- RunnablePassthrough
- RunnableLambda
上記以外にも、RunnableBranchやRunnableWithMessageHistoryなどRunnable〇〇は存在しますが、 今回はまず抑えておきたい上記3つの解説を行います。
RunnableWithMessageHistoryについては以下のブログに記載しているため、ご興味がありましたら、ご覧ください。
またLCELについても、以下のブログをまとめているため、ご興味ある方がご覧ください。
参考
- https://python.langchain.com/docs/expression_language/how_to/map
- https://python.langchain.com/docs/expression_language/how_to/passthrough
- https://python.langchain.com/docs/expression_language/how_to/functions
RunnableParallel
まずは、RunnableParallelについてです。
モデルに対してあるトピックに関するジョークとポエムを言ってもらうためのプロンプトを作るサンプルコードを基にご説明します。
サンプルコード1
RunnableParallelを使わない場合、以下のような実装になります。
from langchain_core.prompts import ChatPromptTemplate joke_prompt = ChatPromptTemplate.from_template(""" {topic}についてのジョークを言ってください。 """) poem_prompt = ChatPromptTemplate.from_template(""" {topic}についての俳句を言ってください。 """) joke_result = joke_prompt.invoke({"topic": "アイス"}) poem_result = poem_prompt.invoke({"topic": "アイス"}) print(joke_result.messages[0].content) print(poem_result.messages[0].content)
実行結果1
実行結果としては、以下の通りになります。 単純にinvoke時の引数であるアイスがプロンプトテンプレートの{topic}に代入された文章が出力されています。
$ python test.py アイスについてのジョークを言ってください。 アイスについてのポエムを言ってください。 $
サンプルコード２
続いて、RunnableParallelを使った場合のサンプルコードになります。
from langchain_core.runnables import RunnableParallel from langchain_core.prompts import ChatPromptTemplate joke_prompt = ChatPromptTemplate.from_template(""" {topic}についてのジョークを言ってください。 """) poem_prompt = ChatPromptTemplate.from_template(""" {topic}についてのポエムを言ってください。 """) runnable = RunnableParallel( joke=joke_prompt, poem=poem_prompt ) result = runnable.invoke({"topic": "アイス"}) print(result) print(result["joke"].messages[0].content) print(result["poem"].messages[0].content)
ポイントとしては、RunnableParallelを利用することでrunnable.invokeを1回実行するだけで済みます。 内部的には、以下のような処理が実行されています。
joke_prompt.invoke({"topic": "アイス"})
poem_prompt.invoke({"topic": "アイス"})
実行結果は、それぞれ「=」の前に定義した名前のKeyのValueに入ります。
{ "joke": ChatPromptValue(messages=[HumanMessage(content='\nアイスについてのジョークを言ってください。\n')]), "poem":ChatPromptValue(messages=[HumanMessage(content='\nアイスについてのポエムを言ってください。\n')]) }
実行結果2
実行結果は先ほど同様です。 1番上の出力には、resultの結果がそのまま出力されています。 2番目、3番目の出力は、resultの中から抽出した値だけが出力されています。
$ python test.py
{'joke': ChatPromptValue(messages=[HumanMessage(content='\nアイスについてのジョークを言ってください。\n')]), 'poem': ChatPromptValue(messages=[HumanMessage(content='\nアイスについてのポエムを言ってください。\n')])}
アイスについてのジョークを言ってください。
アイスについてのポエムを言ってください。
$
補足: RunnableParallelと同等の書き方①
サンプルコードでは、以下の書き方をしておりますが、同等の書き方が2つあります。 その内の1つはこの段階では説明しづらいため、後述します。
runnable = RunnableParallel(
joke=joke_prompt,
poem=poem_prompt
)
- 同等の書き方①
サンプルコードの書き方と異なり、RunnableParallelの引数を辞書型で定義できます。
runnable = RunnableParallel({
"joke": joke_prompt,
"poem": poem_prompt
})
RunnablePassthrough
続いて、RunnablePassthroughです。 RunnablePassthroughは名前の通り、パススルーしてくれます。
サンプルコード1
以下はRunnablePassthroughを使う必要がない(使っても出力が変わらない)コードですが、 逆にRunnablePassthroughを理解できるのではないでしょうか。
from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough joke_prompt = ChatPromptTemplate.from_template(""" {topic}についてのジョークを言ってください。 """) runnable = RunnablePassthrough() | joke_prompt # joke_prompt.invoke({"topic": "アイス"})と実行結果は同様 joke_result = runnable.invoke({"topic": "アイス"}) print(joke_result.messages[0].content)
実行結果1
$ python test.py アイスについてのジョークを言ってください。 $
解説
では、どういったケースでRunnablePassthroughを使うのかご説明します。
例として、RAGの実装を考えた際、プロンプトに必要な可変な値として、 ユーザーからの質問とコンテキスト(検索システムから取得した回答の参考となる情報)が考えられます。
Chainを組む際、プロンプトに渡す引数は辞書型になるため、以下のような辞書型のデータをプロンプトに渡します。
{ "question": "chain実行時の引数の値(ユーザーからの質問)", "context": "chain実行時の引数の値(ユーザーからの質問)を基に検索された結果" }
つまり、ユーザーからの質問はChainを実行する際の引数の値をそのまま利用したいが、 コンテキストはユーザーからの質問を基に検索した結果(invokeの実行結果)を利用する必要があります。
このような場合、RunnablePassthroughが使えます。
サンプルコード2
from langchain_community.vectorstores import FAISS from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough, RunnableParallel from langchain_community.embeddings import BedrockEmbeddings # ベクターストアの作成 vectorstore = FAISS.from_texts( ["マイケルは株式会社サーバーワークスで働いています。"], BedrockEmbeddings(model_id="amazon.titan-embed-text-v1",) ) # retrieverの定義 retriever = vectorstore.as_retriever() prompt = ChatPromptTemplate.from_template("""次の文脈のみに基づいて質問に答えてください: {context} Question: {question} """) retrieval_chain = ( RunnableParallel({"context": retriever, "question": RunnablePassthrough()}) | prompt ) result = retrieval_chain.invoke("マイケルはどこで働いていますか？") print(result) print(result.messages[0].content)
実行結果
$ python 'test.py' 次の文脈のみに基づいて質問に答えてください: [Document(page_content='マイケルは株式会社サーバーワークスで働いています。')] Question: マイケルはどこで働いていますか？ $
解説
少しコードが増えましたが、RunnableParallelの部分は先ほど解説した通りです。
RunnableParallel({"context": retriever, "question": RunnablePassthrough()})
RunnableParallelの解説の際にはPromptコンポーネントを使っていましたが、今回は、Retrieverコンポーネントです。 コンポーネントが異なっても共通のインターフェースが使えるため、実行イメージは同様です。
retriever.invoke("マイケルはどこで働いていますか？")
出力結果は以下の通りです。
[Document(page_content='マイケルは株式会社サーバーワークスで働いています。')]
そして、RunnablePassthrough()には、invoke時の引数がパススルーされます。 つまり、RunnableParallelの実行結果は以下の通りになります。
{ "context" : "[Document(page_content='マイケルは株式会社サーバーワークスで働いています。')]", "question": "マイケルはどこで働いていますか？" }
補足: RunnableParallelと同等の書き方②
RunnableParallelと同等の書き方の2つ目です。
RunnableParallel({"context": retriever, "question": RunnablePassthrough()})
今回は上記のようなコードを書いていますが、RunnableParallel を別の Runnable で構成する場合、 自動的に型変換されるため、辞書を RunnableParallel クラスでラップする必要がありません。 そのため、以下のように書くことができます。
{"context": retriever, "question": RunnablePassthrough()}
ちなみに、RunnableParallelで解説した以下のコードは、Runnablが構成されていないため、 型変換が行われず、辞書型として認識されエラーになる点は注意が必要です。
RunnableParallel({
"joke": joke_prompt,
"poem": poem_prompt
})
## 辞書型として認識され、エラーになる
{"joke": joke_prompt,"poem": poem_prompt}
RunnableLambda
RunnableLambdaは、Chain実行時にカスタム関数を使うことができるクラスになります。
RunnablePassthrough()を使ったサンプルコードでは実行結果が以下のようになっていましたが、 必要な値としては、page_contentの値だけです。
$ python 'test.py' 次の文脈のみに基づいて質問に答えてください: [Document(page_content='マイケルは株式会社サーバーワークスで働いています。')] Question: マイケルはどこで働いていますか？ $
※RAGを実装する際は、タイトルやURL情報も含まれていることが多いため、 実際には、そういった値も含めてモデルが理解しやすい形式に整形するのが大事になるかと存じます。
サンプルコード
retrieverの検索結果からpage_contentのみを出力するようにRunnableLambdaを使っています。
from langchain_community.vectorstores import FAISS from langchain_core.prompts import ChatPromptTemplate from langchain_core.runnables import RunnablePassthrough, RunnableParallel, RunnableLambda from langchain_community.embeddings import BedrockEmbeddings # ベクターストアの作成 vectorstore = FAISS.from_texts( [ "マイケルは株式会社サーバーワークスで働いています。", "ジャックは株式会社G-genで働いています。" ], BedrockEmbeddings(model_id="amazon.titan-embed-text-v1",) ) # retrieverの定義 retriever = vectorstore.as_retriever() prompt = ChatPromptTemplate.from_template("""次の文脈のみに基づいて質問に答えてください: {context} Question: {question} """) def extract_docs(docs): context = "" for doc in docs: context += doc.page_content + "\n" return context retrieval_chain = ( RunnableParallel( { "context": retriever | RunnableLambda(extract_docs), "question": RunnablePassthrough() } ) | prompt ) result = retrieval_chain.invoke("マイケルはどこで働いていますか？") print(result.messages[0].content)
実行結果
実行結果は以下の通りです。
$ python test.py 次の文脈のみに基づいて質問に答えてください: マイケルは株式会社サーバーワークスで働いています。 ジャックはG-Genで働いています。 Question: マイケルはどこで働いていますか？ $
解説
ポイントはここです。 RunnableParallelの中でもchainを組むことができるため、 retriever と RunnableLambdaでchainを組んでいます。
また、RunnableLambdaの引数には、retrieverの検索結果から必要な値を抽出するextract_docsという関数を指定しています。
"context": retriever | RunnableLambda(extract_docs),
また、extract_docsでは、引数(docs)を1つ受け取ります。
def extract_docs(docs): context = "" for doc in docs: context += doc.page_content + "\n" return context
この引数(docs)には、retriever.invokeの実行結果が含まれるため、以下のようなデータが代入されます。
[Document(page_content='マイケルは株式会社サーバーワークスで働いています。'), Document(page_content='ジャックはG-Genで働いています。')]
そして、extract_docsが実行されることでpage_contentの中の「'マイケルは株式会社サーバーワークスで働いています。'」と 「'ジャックはG-Genで働いています。'」がくっつけられて、最終的にpromptに渡されるデータは以下のようになります。
{ "context" : "マイケルは株式会社サーバーワークスで働いています。\nジャックはG-Genで働いています。", "question": "マイケルはどこで働いていますか？" }
そして最終的には、出力結果が以下の通りになります。
$ python test.py 次の文脈のみに基づいて質問に答えてください: マイケルは株式会社サーバーワークスで働いています。 ジャックはG-Genで働いています。 Question: マイケルはどこで働いていますか？ $
終わりに
今回は、3つのRunnable〇〇をご紹介しました。どなたかのお役に立てれば幸いです。
