🐷

【Langchain】LCEL記法でのMemory機能の利用方法まとめ

2024/07/16に公開

チャットボット

LangChainでチャットボットを作るときに必須なのが、会話履歴を保持するMemoryコンポーネントです。ひさびさにチャットボットを作ろうとして、LCEL記法でのMemoryコンポーネントの基本的な利用方法を調べてみたので、まとめておきます。

LangChain LCEL記法でのMemoryコンポーネントの利用方法

LangChainでは、各コンポーネントをパイプで接続するLCEL記法（LangChain Expression Language）での記述を推奨しています。

chain = prompt_template | chat_model

LCEL記法では、プロンプトやLLMモデルなどの各コンポーネントをパイプ|で連結して記述します。各コンポーネントの入出力の形式を意識する必要はありますが、どのようなコンポーネントでChainが構成されているかが一目瞭然です。

LangChainでチャットボットを作成する際には、これまでの会話履歴を保持しておくMemoryコンポーネントが必要になります。LCEL記法でMemoryコンポーネントを利用するには、LCELで定義したChainを、RunnableWithMessageHistoryでラップして利用します。

RunnableWithMessageHistoryの概念図（LangChain Webサイトより）

LangChainのWebサイトに掲載されている概念図がそれを端的に示しています。Your RunnableがLCELで定義したChainです。それをラップする形でRunnableWithMessageHistoryを利用します。

この記事では、具体的なコードでRunnableWithMessageHistoryの使い方やカスタマイズ方法を見ていきます。

なお、詳しくは以下のLangChainのWebサイトも参考にしてください。

基本的なLCEL記法でのMemoryコンポーネントの実装

LCEL記法で簡単なチャットボットを作ってみました。コード全体を以下に示します。

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai.chat_models import ChatOpenAI

# 会話履歴のストア
store = {}

# セッションIDごとの会話履歴の取得
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# プロンプトテンプレートで会話履歴を追加
prompt_template = ChatPromptTemplate.from_messages(
    [
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)

# 応答生成モデル（例としてchat_model）
chat_model = ChatOpenAI(model="gpt-3.5-turbo")

# Runnableの準備
runnable = prompt_template | chat_model

# RunnableをRunnableWithMessageHistoryでラップ
runnable_with_history = RunnableWithMessageHistory(
    runnable=runnable,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# 実際の応答生成の例
def chat_with_bot(session_id: str):
    count = 0
    while True:
        print("---")
        input_message = input(f"[{count}]あなた: ")
        if input_message.lower() == "終了":
            break

        # プロンプトテンプレートに基づいて応答を生成
        response = runnable_with_history.invoke(
            {"input": input_message},
            config={"configurable": {"session_id": session_id}}
        )
        
        print(f"AI: {response.content}")
        count += 1


if __name__ == "__main__":

    # チャットセッションの開始
    session_id = "example_session"
    chat_with_bot(session_id)

以下、コードを抜粋しながらポイントを解説していきます。なお、コードは以下のリポジトリにmemory_runnable1.pyとして置いておきます。

セッション毎の会話履歴を取得する関数

# セッションIDごとの会話履歴の取得
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

get_session_historyは、セッション毎の会話履歴を取得するための関数です。後述するRunnableWithMessageHistoryでchainをラップするときに、この関数を引数の一つとして与えます。

会話履歴の実体はChatMessageHistoryクラスのインスタンスです。ChatMessageHistoryクラスは、以下のように単なるMessageのリストを保持しているだけの簡潔なクラスです。

class ChatMessageHistory(BaseChatMessageHistory, BaseModel):
    """In memory implementation of chat message history.

    Stores messages in an in memory list.
    """

    messages: List[BaseMessage] = Field(default_factory=list)

（以下、省略）

プロンプトテンプレート・LLMモデルとChain定義

次に、チャットボットでは定番のプロンプトテンプレートとLLMモデルの定義、そして、これらをLCEL記法で連結したChainの定義を見ていきます。

以下のコードは、プロンプトテンプレートです。

# プロンプトテンプレートで会話履歴を追加
prompt_template = ChatPromptTemplate.from_messages(
    [
        MessagesPlaceholder(variable_name="history"),
        ("human", "{input}"),
    ]
)

MessagesPlaceholderは会話履歴をプロンプトに挿入するためのものです。ここでvariable_nameで指定されているキーhistoryが、後述するRunnableWithMessageHistoryの設定と関係してきます。("human", "{input}")の部分は、ユーザーの入力がinputとして入ります。

以下のコードは、LLMモデルの定義です。

# 応答生成モデル
chat_model = ChatOpenAI(model="gpt-3.5-turbo")

ここではOpenAIのgpt-3.5-turboを利用していますが、他のモデルでも同様に定義すれば利用できます。

# Runnableの準備
runnable = prompt_template | chat_model

ここまでで定義したプロンプトテンプレートとLLMモデルを連結して、RunnableなChainを作成しておきます。

RunnableWithMessageHistoryによるRunnable Chainのラップ

先ほど作成したRunnableなChainrunnableを、会話履歴を付与するRunnableWithMessageHistoryでラップします。

# RunnableをRunnableWithMessageHistoryでラップ
runnable_with_history = RunnableWithMessageHistory(
    runnable=runnable,
    get_session_history=get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

各引数は以下のとおりです。

runnable: LCEL記法で定義したrunnableなChainを指定
get_session_history: 会話履歴を取得する関数
input_messages_key: ユーザーの現在の入力を示すキー
history_messages_key: 会話履歴を示すキー

runnableには先ほど定義したLCEL記法のChainを、get_session_historyには、こちらも先ほど用意した会話履歴を取得する関数を渡します。

input_messages_keyには、ユーザーの現在の入力を示すキーを渡しますが、これはプロンプトテンプレートで指定した("human", "{input}")のinputに対応します。

history_messages_keyには、会話履歴を示すキーを渡しますが、これはプロンプトテンプレートのMessagesPlaceholder(variable_name="history")の部分のhistoryに対応します。

コードで会話履歴を明示的に追加する部分を記述する必要はありません。RunnableWithMessageHistoryが、get_session_historyに指定された関数で取得した会話履歴のインスタンスに対して、input_messages_keyやhistory_messages_keyに指定したキーを用いて、必要な情報を会話履歴に追加してくれます。

チャットボットの実行

チャットボットはchat_with_bot関数で実際に実行しますが、その中で、先ほどRunnableWithMessageHistoryでラップしたrunnable_with_historyをinvokeしています。

# プロンプトテンプレートに基づいて応答を生成
response = runnable_with_history.invoke(
    {"input": input_message},
    config={"configurable": {"session_id": session_id}}
)

inputにユーザーの入力を、configのsession_idにセッションIDを渡してinvokeします。今回はセッションIDは固定しています。

これだけで会話履歴付きのチャットボットが実現できます。

LangSmithによる動作確認

それでは動作確認をしてみます。

何往復かチャットをしてみた結果が以下のとおりです。

$ python .\memory_runnable1.py 
---
[0]あなた: こんにちは。うちの犬の名前はポチです。
AI: こんにちは！ポチちゃん、かわいい名前ですね。元気な犬ですか？
---
[1]あなた: はい、とても元気な犬です。毎日散歩に連れていきます。
AI: それは素晴らしいですね！散歩が楽しみでしょうね。ポチちゃんと一緒に過ごす時間はきっととても幸せな時間ですね。ポチちゃんにたくさんの愛情を注いであげてくださいね。
---
[2]あなた: うちの犬の名前がわかりますか？    
AI: はい、ポチという名前の犬ですね。かわいい名前ですね。どんな犬ですか？

[2]の「うちの犬の名前がわかりますか？」に対して、「はい、ポチという名前の犬ですね。」と答えていますので、会話の履歴が利用されていることがわかります。

これをLangSmithでも確認してみます。LangSmithを利用するには、コードに以下を追加します。LangChainのAPIキーが必要となりますので、取得してください。

# Langchain LangSmith
unique_id = uuid4().hex[0:8]
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = f"Tracing Walkthrough - {unique_id}"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = <LANGCHAIN_API_KEY>

LangSmithの画面を見てみると、以下のように1ターンの会話で一つのログが表示されています。

LangSmithのログ表示画面

ここで、先ほどの会話の最後の部分（うちの犬の名前がわかりますか？の問い）を詳しく見てみましょう。

会話履歴が含まれるLangSmithのログ

insert_historyの部分を見てみると、2ターン分（4つ）の会話履歴が取得されていることがわかります。これがLLMに渡されているため、犬の名前を答えることができているわけです。

ということで、会話履歴のMemoryコンポーネントをRunnableWithMessageHistoryで実装したチャットボットの動作を確認することができました。

会話履歴のカスタマイズ

先ほどの例では、会話履歴が無限に増えていきます。会話履歴は、すべてプロンプトとしてLLMに渡されるため、どんどんプロンプトが肥大化していきます。プロンプトが肥大化するとトークン数の上限やコストの問題もありますし、また、LLMが生成する回答の精度にも影響してきます。

そこで、会話履歴の上限を決めて、直近の履歴のみを保存するようにコードを修正してみます。

コード全体は、以下のリポジトリのmemory_runnable2.pyをご覧ください。

`ChatMessageHistory`クラスの修正

会話履歴を決められた上限値で制限するには、会話履歴を保持するクラスChatMessageHistoryを継承してカスタマイズします。ここでは、LimitedChatMessageHistoryクラスを作成します。

# 保持する会話履歴数
DEFAULT_MAX_MESSAGES = 4

# 会話履歴数をmax_messagesに制限するLimitedChatMessageHistoryクラス
class LimitedChatMessageHistory(ChatMessageHistory):

    # 会話履歴の保持数
    max_messages: int = DEFAULT_MAX_MESSAGES

    def __init__(self, max_messages=DEFAULT_MAX_MESSAGES):
        super().__init__()
        self.max_messages = max_messages

    def add_message(self, message):
        super().add_message(message)
        # 会話履歴数を制限
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]

    def get_messages(self):
        return self.messages

max_messagesで指定した数の会話履歴のみを保持するようにしています。具体的には、会話履歴を追加するadd_message関数をオーバーライドして、上限となるmax_messagesを超えたら、古いものから削除するようにしています。

なお、ユーザーの入力、AIの応答は、それぞれ1メッセージと数えます。そのため、例えばmax_messages=4と設定すると、直近の2ターン分の会話のみが保存されることになります。

そして、get_session_history関数で、作成したLimitedChatMessageHistoryクラスを、元のChatMessageHistoryクラスの代わりに利用します。

# セッションIDごとの会話履歴の取得
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        # 会話履歴上限を制限するLimitedChatMessageHistoryを利用
        store[session_id] = LimitedChatMessageHistory()
    return store[session_id]

修正箇所は以上ですが、会話履歴をコンソールに表示しながら実行するために、以下のように実行部分の関数chat_with_botを修正します。

# 実際の応答生成の例
def chat_with_bot(session_id: str):
    count = 0
    while True:
        print("---")
        input_message = input(f"[{count}]あなた: ")
        if input_message.lower() == "終了":
            break

        # プロンプトテンプレートに基づいて応答を生成
        response = runnable_with_history.invoke(
            {"input": input_message},
            config={"configurable": {"session_id": session_id}}
        )
        
        print(f"AI: {response.content}")
        print(f"\n会話履歴:\n{store[session_id]}")
        count += 1

動作確認

ということで、動作確認をしてみます。上記のコードのとおり、会話履歴は4、すなわち、2ターン分のみ保持することにします。

以下、実行結果です。会話のターン毎に会話履歴を表示しています。長いので適宜省略しています。

$ python .\memory_runnable2.py 
---
[0]あなた: こんにちは。わたしの名前はひさです。キキという名前の猫を飼っています。
AI: こんにちは、ひささん。キキちゃんという名前の猫を飼っているんですね。キキちゃんは元気ですか？可愛い名前ですね。
キキちゃんと一緒に過ごす時間は楽しいですか？どんな性格の猫なんですか？話してくれてありがとうございます。

会話履歴:
Human: こんにちは。わたしの名前はひさです。キキという名前の猫を飼っています。
AI: こんにちは、ひささん。キキちゃんという名前の猫を飼っているんですね。キキちゃんは元気ですか？可愛い名前ですね。 
キキちゃんと一緒に過ごす時間は楽しいですか？どんな性格の猫なんですか？話してくれてありがとうございます。
---
（省略）
---
[3]あなた: ところで、私の名前を知っていますか？
AI: 申し訳ありませんが、私はユーザーの名前や個人情報を保存や記憶する機能は持っていません。プライバシーを尊重し、セッションごとに新しい情報のみを使用しています。ですので、お名前を教えていただいても、次回の会話では忘れてしまいます 。何か質問やお話がありましたら、お気軽にどうぞ！

会話履歴:
AI: キキちゃんはボール遊びが好きなんですね！ボール遊びは猫にとって楽しい運動になりますし、猫の本能を引き出す遊びでもありますね。キキちゃんがボールを追いかける姿はきっととても可愛いですね！一緒に遊んでいるときのキキちゃんの表情やしぐさはどんな感じですか？キキちゃんとのボール遊び、またたくさん楽しい時間を過ごしてくださいね。
Human: ところで、私の名前を知っていますか？
AI: 申し訳ありませんが、私はユーザーの名前や個人情報を保存や記憶する機能は持っていません。プライバシーを尊重し、セッションごとに新しい情報のみを使用しています。ですので、お名前を教えていただいても、次回の会話では忘れてしまいます 
。何か質問やお話がありましたら、お気軽にどうぞ！
---
[4]あなた: うちの猫の名前を呼んでもらえますか？
AI: もちろんです！キキちゃん、元気ですか？ 一緒に遊ぶのが楽しみですね。キキちゃんはとても可愛らしい名前ですね。どんな性格の猫ちゃんなんでしょうか？キキちゃんのために何か特別なおもちゃやご褒美を用意されていますか？キキちゃんとの楽しい日々が続きますように！

（省略）

[3]の部分で私の名前を聞いていますが、すでに会話履歴から消えてしまっているので、AIは「申し訳ありませんが、私はユーザーの名前や個人情報を保存や記憶する機能は持っていません。」と答えています。

一方、ここまでの会話で話題にしている猫のキキの名前については、次の[4]の質問「うちの猫の名前を呼んでもらえますか？」に対して、「もちろんです！キキちゃん、元気ですか？」と正確に答えています。

猫のキキの名前をAIに教えたのは最初の[0]の会話で、すでに会話履歴から消えてしまっています。ただ、直前までキキを話題にした会話を継続していますので、その部分の履歴を参照して、猫の名前を答えているのです。

まとめ

LangChainのLCEL記法でのMemoryコンポーネントの実装として、RunnableWithMessageHistoryを試してみました。

会話履歴を保持するクラスとしてChatMessageHistoryまたはそれを継承したクラスを利用する
会話履歴を取得する関数get_session_historyを実装し、上記の会話履歴を保持するクラスを返すようにする
プロンプトやLLMのコンポーネントを連結した runnable chain をRunnableWithMessageHistoryでラップする

RunnableWithMessageHistoryを利用すれば、これだけで会話履歴を保持するチャットボットを実現できます。

どのくらいの会話履歴を保持しておくかは、用途によります。生成AIを用いてチャットボットを実装する場合には、何らかのナレッジを読み込ませて、いわゆるRAGとして動作させることが多いと思います。その場合、ユーザーのプロンプトにナレッジから取得した文章が含まれるため、入力トークン数がかなり多くなります。

一般的なチャットボットでは、それほど長い会話はしないかもしれませんが、取得したナレッジの情報量が多い場合には、会話履歴をそのまま保存せず、LLMを用いて要約したうえで保存するといった手法もあります。次は、RAGのチャットボットで試してみたいと思います。