LangChainで推奨されるLCEL記法の概要を理解する

こんにちは。AWS CLIが好きな福島です。

はじめに
LangChain Expression Language (LCEL)とは
具体例
解説
イメージ図
コンポーネントとは
- コンポーネントの種類
インターフェース
- 同期メソッド
- 非同期メソッド
LCELのメリット
終わりに

はじめに

今回は今更かも知れませんが、LangChainで推奨されているLCEL記法に関する概要をまとめます。

LangChain Expression Language (LCEL)とは

LangChain Expression Language(LCEL)とは、 LangChain固有のChainを簡単に構築するための宣言型の記法となります。

特徴的なのが、Unixのパイプ演算子のイメージで「|」を使える点です。

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

具体例

LCEL記法の具体例は、以下の通りです。

from langchain_community.chat_models.bedrock import BedrockChat
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
  
prompt = ChatPromptTemplate.from_template("{topic}に関するジョークを言ってください。")
chat = BedrockChat(model_id="anthropic.claude-v2:1", model_kwargs={"temperature": 0.1})
  
## Unix のパイプ演算子のイメージで「|」を使える
chain = prompt | chat | StrOutputParser()
result = chain.invoke({"topic": "アイスクリーム"})
  
print(result)

上記の実行結果は以下の通りです。ジョークの内容には触れないのですが、要はLLMにアイスクリームのジョークを教えてと質問した回答が出力されます。

$ pyhon test.py
 アイスクリームに関するジョークです。
  
Q. アイスクリーム屋さんで一番売れているのは?
A. バニラアイス! なぜなら「平凡なアイス」だから!
  
--- 中略 ---
  
Q. アイスクリームがこぼれたとき、一番困るのは?
A. 床!床が一番冷たくなる

解説

ポイントは、以下の箇所です。ここのprompt、chat、StrOutputParser()のことをコンポーネントと呼びます。コンポーネントについては後述します。

chain = prompt | chat | StrOutputParser()

これは以下のように実行することができます。

chain.invoke({"topic": "アイスクリーム"})

ここから私のイメージで少し解説します。

まず、chainを定義した際は、処理は実行されるのではなく、セットされるイメージです。ドミノで言うと、ドミノを立てている状態です。

UNIXのパイプ演算子と言われるとこの時点で処理が実行されているように思えてしまいますが、あくまでセット(実行可能な状態に)しているイメージであることがポイントかなと思います。

実際にchainの中身をprintすると以下のように表示され、セットされているイメージが伝わるかなと思います。 (実際の出力結果に改行を入れてます。)

first=ChatPromptTemplate(input_variables=['topic'], messages=[HumanMessagePromptTemplate(prompt=PromptTemplate(input_variables=['topic'], template='{topic}に関するジョークを言ってください。'))])
  
middle=[BedrockChat(client=<botocore.client.BedrockRuntime object at 0x108a4cdd0>, region_name='ap-northeast-1', model_id='anthropic.claude-v2:1', model_kwargs={'temperature': 0.1})]
  
last=StrOutputParser()

そして、chain.invokeすることでセットしたchainがドミノが倒れるように実行されます。

chain = prompt | chat | StrOutputParser()
chain.invoke({"topic": "アイスクリーム"})

そのため、上記コードでchain.invoke実行時のイメージとしては、

{"topic": "アイスクリーム"}がpromptに渡さる
promptの出力結果がchatに渡される
chatの出力結果がStrOutputParser()に渡される
StrOutputParser()が出力結果となる

となります。

イメージ図

chainの実行イメージ図を以下に載せておりますので、理解の一助になれば幸いです。

コンポーネントとは

LCELを理解する上で重要なキーワードとして、コンポーネントがあります。 LCELでは、Unixのパイプ演算子のイメージで「|」を使えると説明しましたが、もちろん何でも「|」で連結することはできません。

では、何が連結できるのかと言うと、コンポーネントになります。

コンポーネントの種類

コンポーネントには以下の種類があります。

Component	Input Type	Output Type
Prompt	辞書型	PromptValue
ChatModel	文字列型, ChatMessageのリスト, PromptValue	ChatMessage
LLM	文字列型, ChatMessageのリスト, PromptValue	文字列型
OutputParser	文字列, ChatMessage	OutputParserごとに異なる
Retriever	文字列型	Documentsのリスト
Tool	文字列型, 辞書型, Toolごとに異なる	ツールごとに異なる

参考: https://python.langchain.com/docs/expression_language/interface

インターフェース

各コンポーネントには、共通のインターフェースが用意されております。今のところ私は基本的にinvokeを使い、ストリーミング出力が必要な場合、streamを使うことがほとんどですが、パフォーマンスが求められる場合は、非同期メソッドが必要になってくると思います。

同期メソッド

stream: 応答のチャンクをストリームバックします。
invoke: 入力でチェーンを呼び出します
batch: 入力のリストでチェーンを呼び出します。

非同期メソッド

astream: 応答のチャンクを非同期でストリームバックします
ainvoke: 入力非同期でチェーンを呼び出します
abatch: 入力のリストでチェーンを非同期で呼び出します
astream_log: 最終応答に加えて、発生した中間ステップをストリームバックします。
astream_events:チェーン内で発生するベータlangchain-coreストリームイベント ( 0.1.14 で導入)

参考: https://python.langchain.com/docs/expression_language/interface

LCELのメリット

詳細は、冒頭のLangChainのドキュメントをご確認いただければと思いますが、私が感じたLCELの1番のメリットは、LangChain内の処理が透明化されたことです。

具体的には、塊になっていた処理がコンポーネント化され、共通のインターフェースによりコンポーネントごとの実行結果を手軽に確認できます。

LCELが登場する前は、良くも悪くもLangChainがLLMのアプリで必要な処理を実行してくれ、 LLMのアプリ開発が楽になる反面、どういった処理が行われているのかブラックボックスとなっており、分かりづらかった印象です。

極端な例ですが、ConversationalRetrievalChainというクラスがあります。何をしているか名前から推測できそうですが、どういった処理が行われるのか理解できるでしょうか。

qa = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=retriever,
    memory=memory,
    return_source_documents=True,
    condense_question_llm=llm,
)
  
response = qa({"question": question})

実際に上記のクラスは、以下の処理を行いますが、実行結果としては生成された回答のみが出力されます。オプションによって処理の過程を見ることができますが、それでも少し分かりづらさがあり、理解をするのも苦労しました。 (私の力量不足やRAGに関する知識不足もあったのですが。)

会話履歴を踏まえてquestionを言い換える
retrieverから参考情報を取得する
参考情報を加えた質問をLLMに行う

これがLCELでは、それぞれの処理をコンポーネント化(分割)し、最終的に「|」を使ってChainを組むことで同様の処理を実現します。以下のコードは前処理をだいぶ省いており、それぞれのコンポーネント(変数)に何が定義されているか分からないかと思いますが、直感的に理解できそうではないでしょうか。

final_chain = loaded_memory | standalone_question | retrieved_documents | answer
response = final_chain.invoke({"question": question})

上記のコードの処理内容は以下の通りです。

メモリから会話履歴を取得(loaded_memory)
会話履歴を踏まえて質問を要約(standalone_question)
要約した質問を基にretrieverから参考情報を検索(retrieved_documents)
LLMに検索した情報を含めた質問をして回答を生成(answer)

また実はLCELでも何も考慮しなければ、実行結果としては生成された回答のみが出力されるのですが、コンポーネントは共通のインターフェースを持っているため、chainの定義を少し変更することで各コンポーネントの実行結果を手軽に確認できます。

## loaded_memoryの実行結果を見たい場合
response = loaded_memory.invoke({"question": question})
  
## standalone_questionの実行結果を見たい場合
final_chain = loaded_memory | standalone_question
response = final_chain.invoke({"question": question})
  
## retrieved_documentsの実行結果を見たい場合
final_chain = loaded_memory | standalone_question | retrieved_documents
response = final_chain.invoke({"question": question})

少し理解が深められたら、RunnablePassthrough.assignを使うことで最終的な実行結果に各コンポーネントの実行結果を含めることもできます。この辺りはまた別のブログでまとめたいと思います。

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