はじめに
LangChainの会話履歴を保存するMemory機能の1つであるConversationTokenBufferMemoryを検証してみました。LangChainのConversationTokenBufferMemoryの挙動を確認したい方におすすめです。
開発環境
- Windows 11
- Python 3.11.5
- dotenv
- LangChain
- Azure OpenAI Embeddings
- CharacterTextSplitter
- FAISS
- PyPDFLoader
実装
事前準備
今回は以下の記事で事前にベクトルストアを作成し、そこから回答を生成する方式をとります。
私はランニングを行っていることもあり、以下のランニング時計の取扱説明書のPDFをベクトル化しました。
必要なパッケージのインストール
以下のコマンドで必要なパッケージをインストールします。
pip install langchain openai python-dotenv faiss-cpu PyPDF2 tiktoken
環境変数の設定と依存関係の読み込み
.envファイルを作成し、Azure Open AIの環境変数の設定をします。
ご自身の環境に合わせて設定してください。
# APIキー
OPENAI_API_KEY = "XXXXX"
# エンドポイント
AZURE_OPENAI_ENDPOINT = "XXXXX"
# 使用するOpenAI APIのバージョン
OPENAI_API_VERSION = "XXXXX"
必要なライブラリをインポートします。
# 環境変数を.envファイルから読み込む
from dotenv import load_dotenv
load_dotenv()
from langchain.chat_models import AzureChatOpenAI
from langchain.embeddings import AzureOpenAIEmbeddings
from langchain.vectorstores import FAISS
from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationTokenBufferMemory
使用するモデルの準備
Azure Chat OpenAIとAzure OpenAI Embeddingsを定義します。
llm = AzureChatOpenAI(
temperature=0,
azure_deployment="gpt-35-turbo-16k",
)
embeddings = AzureOpenAIEmbeddings(
azure_deployment="text-embedding-ada-002"
)
FAISSベクトルストアの読み込み
ローカルに保存しているベクトルストアの読み込みを行い、as_retriever()メソッドで検索機能を作成します。
vectorstore = FAISS.load_local("./vectorstore", embeddings)
retriever = vectorstore.as_retriever()
ConversationTokenBufferMemoryの定義
以下のようにConversationTokenBufferMemoryを定義します。ConversationTokenBufferMemoryは指定したトークン数だけ前の直近の会話をそのまま保持します。
要約にllmを使います。保持する直近の会話履歴のトークン数をmax_token_limitで指定します。会話履歴と出力のキーを指定しています。
memory = ConversationTokenBufferMemory(
llm=llm,
max_token_limit=300,
memory_key="chat_history",
return_messages=True,
output_key="answer",
)
chainを作成し、実行
最後にchainを作成し、実行します。質問は3問準備しており、2問目は1問目を詳しく説明する質問、3問目はそれまでの会話をもとに回答できるかを検証する質問に設定しました。
chain = ConversationalRetrievalChain.from_llm(
llm=llm, retriever=retriever, memory=memory
)
results = chain({"question": "心拍変動は計測可能ですか？"})
print("あなた：", results["question"])
print("ボット：", results["answer"])
print("会話履歴：", results["chat_history"])
results = chain({"question": "もっと詳しく説明してください。"})
print("あなた：", results["question"])
print("ボット：", results["answer"])
print("会話履歴：", results["chat_history"])
results = chain({"question": "血中酸素濃度は？"})
print("あなた：", results["question"])
print("ボット：", results["answer"])
print("会話履歴：", results["chat_history"])
これまでのPythonコードを上から順に記述し終わったら、Pythonファイルを実行します。
実行結果は以下のようになりました。
あなた： 心拍変動は計測可能ですか？
ボット： はい、デバイスは心拍変動データを計測することができます。心拍変動データはストレスレベルの測定に使用されます。
会話履歴： [HumanMessage(content='心拍変動は計測可能ですか？'), AIMessage(content='はい、デバイスは心拍変動データを計測 することができます。心拍変動データはストレスレベルの測定に使用されます。')]
あなた： もっと詳しく説明してください。
ボット： 心拍変動データは、非活動時の心拍数の変動パターンを基にしてストレスレベルを測定するために使用されます。心拍変動 は、心臓の自律神経系の活動を反映しており、ストレスやリラックスの状態によって変動します。デバイスは心拍変動データを解析し、ストレスレベルを算出します。ストレスレベルの数値が低い場合、それはストレスが少ないことを示しています。
会話履歴： [HumanMessage(content='心拍変動は計測可能ですか？'), AIMessage(content='はい、デバイスは心拍変動データを計測 することができます。心拍変動データはストレスレベルの測定に使用されます。'), HumanMessage(content='もっと詳しく説明してください。'), AIMessage(content='心拍変動データは、非活動時の心拍数の変動パターンを基にしてストレスレベルを測定するために 使用されます。心拍変動は、心臓の自律神経系の活動を反映しており、ストレスやリラックスの状態によって変動します。デバイスは心拍変動データを解析し、ストレスレベルを算出します。ストレスレベルの数値が低い場合、それはストレスが少ないことを示しています。')]
あなた： 血中酸素濃度は？
ボット： はい、Garminデバイスは血中酸素濃度を計測する機能があります。デバイスには光学式センサーが搭載されており、血中酸 素濃度を測定することができます。
会話履歴： [HumanMessage(content='もっと詳しく説明してください。'), AIMessage(content='心拍変動データは、非活動時の心拍 数の変動パターンを基にしてストレスレベルを測定するために使用されます。心拍変動は、心臓の自律神経系の活動を反映しており、ストレスやリラックスの状態によって変動します。デバイスは心拍変動データを解析し、ストレスレベルを算出します。ストレスレベルの数値が低い場合、それはストレスが少ないことを示しています。'), HumanMessage(content='血中酸素濃度は？'), AIMessage(content='はい、Garminデバイスは血中酸素濃度を計測する機能があります。デバイスには光学式センサーが搭載されており、血中酸素 濃度を測定することができます。')]
2問目終了時の会話履歴には1問目の内容が保持されていますが、3問目終了時の会話履歴には1問目の内容は保持されていないことが分かります。max_token_limitｎより前の会話履歴は保持されないことには注意が必要です。
おわりに
LangChainの会話履歴を保存するmemory機能の1つであるConversationTokenBufferMemoryを検証してみました。文書検索に会話履歴を導入する際の選択肢の1つになります。ぜひ、試してみてください。
最後までお読みいただき、ありがとうございました！
以下のXで情報発信してます！
参考文献