LangChain の Memory の概要

「LangChain」の「Memory」の概要をまとめました。

・langchain 0.1

1. Memory

「Memory」は、将来の推論や行動に役立つ情報を記録し、必要に応じて利用するコンポーネントです。具体的には、会話履歴を保存して、次に何を話すべきかを決める情報として使うことができます。

2. LangChainの準備

ColabでのLangChainの準備の手順は、次のとおりです。

(1) パッケージのインストール。

# パッケージのインストール
!pip install langchain==0.1.9
!pip install langchain_openai

(2) 環境変数の準備。
以下のコードの <OpenAI_APIのトークン> にはOpenAI APIのトークンを指定します。(有料)

# 環境変数の準備
import os
os.environ["OPENAI_API_KEY"] = "<OpenAI_APIのトークン>"

3. Memory追加前の動作確認

「Memory」追加前の動作確認を行います。

(1) 言語モデルの準備。

from langchain_openai.chat_models import ChatOpenAI

# 言語モデルの準備
chat_model = ChatOpenAI()

(2) 1回目の会話

# 1回目の会話
output = chat_model.invoke("うちの猫の名前は白子です。")
print(output.content)

白子という名前はとてもかわいいですね！白子ちゃんは元気ですか？

(3) 2回目の会話
前回の会話を覚えていないことがわかります。

# 2回目の会話
output = chat_model.invoke("うちの猫の名前を読んでください。")
print(output.content)

申し訳ありませんが、猫の名前を教えていただかないと読むことができません。猫の名前を教えていただけますか？

4. Memory追加後の動作確認

「Memory」を追加して動作確認を行います。

(1) プロンプトテンプレートで会話履歴を追加。
「history」の「MessagesPlaceholder」を追加します。OpenAI APIでは、Humanメッセージになります。

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

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

# Runnableの準備
runnable = prompt_template | chat_model

(2)  RunnableWithMessageHistory で Runnableをラップ。
「RunnableWithMessageHistory」は、「Runnable」に会話履歴を追加することができるコンポーネントです。

from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

store = {}

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

# RunnableWithMessageHistoryの準備
runnable_with_history = RunnableWithMessageHistory(
    runnable,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

「RunnableWithMessageHistory」のパラメータは、次のとおりです。

・input_messages_key : 辞書型で入力する場合の入力のキー
・output_messages_key : 辞書型で出力する場合の出力のキー
・history_messages_key : 辞書型で入力する場合の会話履歴のキー
・history_factory_config : ChatHistoryFactoryに渡すフィールド。詳しくは ConfigurableFieldSpec を参照
・**kwargs : 親クラス RunnableBindingBase の init に渡すパラメータ

get_session_history() はセッションIDごとの会話履歴 (ChatMessageHistory) を取得します。

「ChatMessageHistory」は Messageリストを保持するクラスで、以下のメソッドとプロパティを提供しています。

・add_messages(messages: [Message]) : Messageリストの追加 (同期)
・aadd_messages(messages: [Message]) : Messageリストの追加 (非同期)
・messages : Messageリストの取得 (同期)
・aget_messages() : Messageリストの取得 (非同期)
・clear() : メッセージリストのクリア (同期)
・aclear() : メッセージリストのクリア(非同期)

(3) 1回目の会話。
configでセッションIDを指定します。同じセッションIDは同じ会話履歴(ChatMessageHistory)を利用します。

# 1回目の会話
output = runnable_with_history.invoke(
    {"input": "うちの猫の名前は白子です。"},
    config={"configurable": {"session_id": "123"}},
)
print(output.content)

白子という名前、とても可愛らしいですね！白子ちゃんは元気ですか？

(4) 2回目の会話。
前回の会話を覚えていることがわかります。

# 2回目の会話
output = runnable_with_history.invoke(
    {"input": "うちの猫の名前を読んでください。"},
    config={"configurable": {"session_id": "123"}},
)
print(output.content)

はい、白子（しらこ）さんですね。可愛らしい名前ですね。どんな性格の猫さんなんですか？

(5) 会話履歴の確認。

# 会話履歴の確認
print(store["123"])

Human: うちの猫の名前は白子です。
AI: 白子という名前、とても可愛らしいですね！白子ちゃんは元気ですか？
Human: うちの猫の名前を読んでください。
AI: はい、白子（しらこ）さんですね。可愛らしい名前ですね。どんな性格の猫さんなんですか？

(6) 別のセッションIDでの動作確認。
別のセッションの会話を覚えていることがわかります。

# 別のセッションIDでの動作確認
output = runnable_with_history.invoke(
    {"input": "うちの猫の名前を読んでください。"},
    config={"configurable": {"session_id": "456"}},
)
print(output.content)

申し訳ございませんが、猫の名前がわかりませんので読むことができません。もし猫の名前を教えていただければ、喜んで読ませていただきます。

5. 会話履歴のカスタマイズ

ChatMessageHistoryをカスタマイズして、会話履歴の最大数を指定します。

(1) ChatMessageHistoryを継承してカスタムクラスを実装。

from langchain_core.messages import BaseMessage
from pydantic import Field
from typing import Union, Sequence

# 限定
class LimitedChatMessageHistory(ChatMessageHistory):
    max_messages: int = Field(default=10)

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

    def add_messages(self, messages: Sequence[BaseMessage]) -> None:
        super().add_messages(messages)
        self._limit_messages()

    def _limit_messages(self):
        if len(self.messages) > self.max_messages:
            self.messages = self.messages[-self.max_messages:]

(2) 1回目の会話。

# 1回目の会話
output = runnable_with_history.invoke(
    {"input": "うちの猫の名前は白子です。"},
    config={"configurable": {"session_id": "123"}},
)
print(output.content)

# 履歴の確認
print("\nhistory:\n", store["123"])

可愛い名前ですね！白子ちゃんは元気ですか？

history:
 Human: うちの猫の名前は白子です。
AI: 可愛い名前ですね！白子ちゃんは元気ですか？

(3) 2回目の会話。

# 2回目の会話
output = runnable_with_history.invoke(
    {"input": "うちの猫の名前を読んでください。"},
    config={"configurable": {"session_id": "123"}},
)
print(output.content)

# 履歴の確認
print("\nhistory:\n", store["123"])

しろこ ですね。かわいい名前です！しろこちゃんは元気ですか？

history:
 Human: うちの猫の名前は白子です。
AI: 可愛い名前ですね！白子ちゃんは元気ですか？
Human: うちの猫の名前を読んでください。
AI: しろこ ですね。かわいい名前です！しろこちゃんは元気ですか？

(4) 3回目の会話。
会話履歴が4メッセージ以上増えていないことがわかります。

# 3回目の会話
output = runnable_with_history.invoke(
    {"input": "もう一回、うちの猫の名前を読んでください。"},
    config={"configurable": {"session_id": "123"}},
)
print(output.content)

# 履歴の確認
print("\nhistory:\n", store["123"])

白子 ですね。わかりました、白子ちゃんですね。元気そうですか？

history:
 Human: うちの猫の名前を読んでください。
AI: しろこ ですね。かわいい名前です！しろこちゃんは元気ですか？
Human: もう一回、うちの猫の名前を読んでください。
AI: 白子 ですね。わかりました、白子ちゃんですね。元気そうですか？

OpenAI GPT-4V／ChatGPT／GPTs 人工知能プログラミング実践入門 | 株式会社ボーンデジタル