LangChainの新しいチャット履歴管理RunnableWithMessageHistory
こんにちはmakokonです。
毎日チャットしてますか。
makokonは、毎日自作チャットアプリで、毎日マブダチのgpt-4XXXたちと友情を深めています。
そんな中で、langchainを用いたチャット履歴の変更があったようなワーニングが出ていたので、確認してみたという話。
すでにいろいろ試した記事はあるようですけど、自分で試さないとわからなくなりますので。
いつものチャットにエラーが出た。
pythonライブラリを更新して、今日もいつもの楽しいおしゃべりをしていると、なにか不吉なメッセージが、定番のように使っていたlangchainのクラスConversationChainが非推奨になって、新しいクラスを使ってくださいとのことらしい。
警告の内容
LangChainDeprecationWarning: The class ConversationChain was deprecated in LangChain 0.2.7 and will be removed in 1.0.
`ConversationChain`クラスは、LangChainのバージョン0.2.7で非推奨となりました。
これにより、将来的にこのクラスは削除される予定です。
代替手段
新しく推奨されているクラス:
`RunnableWithMessageHistory`詳細なドキュメント:
RunnableWithMessageHistory Documentation
いつものチャットプログラム(最小規模にしてあります)
これは最小限のメモリ管理付きAIチャットで、基本的な部分は全部これで使っていたのですが、これが将来使えないとなると面倒だなあという感じ。
basechat.py
#ごくごく基本的なAICHAT import openai
# ChatOpenAI GPT
from langchain_openai.chat_models import ChatOpenAI
from langchain.chains import ConversationChain
# Memory
from langchain.memory import ConversationBufferWindowMemory
# json
"""
import key
# 環境変数にAPIキーを設定 現在システムでは設定済み
os.environ["OPENAI_API_KEY"] = key.OPEN_API_KEY
"""
#prompt template for chat
from langchain.prompts.chat import (
ChatPromptTemplate,
MessagesPlaceholder,
SystemMessagePromptTemplate,
HumanMessagePromptTemplate,
)
# chatgpt llm memory set
chat = ChatOpenAI(model_name="gpt-4o-mini",temperature=0.7)
# Memory の作成と参照の獲得 今回はターン制のメリにする
memory = ConversationBufferWindowMemory(k=8, return_messages=True)
# message template for system message
template = "あなたは優秀なアシスタントです。"
# チャットプロンプトテンプレート、
prompt = ChatPromptTemplate.from_messages([
SystemMessagePromptTemplate.from_template(template),
MessagesPlaceholder(variable_name="history"),
HumanMessagePromptTemplate.from_template("{input}")
])
# 初期化時に、使用するチャットモデル、メモリオブジェクト、プロンプトテンプレートを指定します
conversation = ConversationChain(llm=chat, memory=memory, prompt=prompt)
user = ""
while user != "exit":
user = input("何かお話しましょう。")
#print(user)
ai = conversation.predict(input=user)
print(ai)
#end phase
print("チャットを終了します")
LangChainの新しいチャット履歴管理RunnableWithMessageHistory
公式ページの説明抜粋
概要
RunnableWithMessageHistoryは、他のRunnable(処理の実行単位)にチャットメッセージ履歴を紐づけるための仕組みです。チャットメッセージ履歴とは
過去の会話を時系列で辿れるメッセージの記録です。RunnableWithMessageHistoryの役割
別のRunnableをラップし、そのRunnableに対するチャットメッセージ履歴の読み書き、更新を管理します。
ラップされたRunnableの入力と出力の形式を定義します。RunnableWithMessageHistoryの使用方法
- 必ずチャットメッセージ履歴ファクトリの設定を含む設定情報が必要です。
- デフォルトでは、ラップされたRunnableは "session_id" という文字列型の設定パラメータを1つ受け取ります。このパラメータは、指定されたsession_idに一致するチャットメッセージ履歴を新規作成または検索するために使用されます。
例: with_history.invoke(…, config={"configurable": {"session_id": "bar"}})
history_factory_config パラメータに ConfigurableFieldSpec オブジェクトのリストを渡すことで、設定をカスタマイズできます。
パラメータ
get_session_history: 新しいBaseChatMessageHistoryを返す関数。セッションIDを受け取り、対応する履歴を返します。
input_messages_key: ラップされたRunnableが入力として辞書を受け取る場合、メッセージを含むキーを指定します。
output_messages_key: ラップされたRunnableが出力として辞書を返す場合、メッセージを含むキーを指定します。
history_messages_key: ラップされたRunnableが入力として辞書を受け取り、履歴メッセージに別のキーを期待する場合に指定します。
history_factory_config: チャット履歴ファクトリに渡す設定フィールド。詳細はConfigurableFieldSpecを参照してください。
掲載サンプルコードの実行確認
念の為、各種ライブラリは最新に更新しておきましょう
pip install --upgrade langchain
pip install --upgrade langchain-core
pip install --upgrade langchain-community
pip install --upgrade langchain-openai
pip install --upgrade openai
公式ページのサンプルコードです。
basechat01.py
from operator import itemgetter
from typing import List
from langchain_openai.chat_models import ChatOpenAI
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.documents import Document
from langchain_core.messages import BaseMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import (
RunnableLambda,
ConfigurableFieldSpec,
RunnablePassthrough,
)
from langchain_core.runnables.history import RunnableWithMessageHistory
class InMemoryHistory(BaseChatMessageHistory, BaseModel):
"""In memory implementation of chat message history."""
messages: List[BaseMessage] = Field(default_factory=list)
def add_messages(self, messages: List[BaseMessage]) -> None:
"""Add a list of messages to the store"""
self.messages.extend(messages)
def clear(self) -> None:
self.messages = []
# Here we use a global variable to store the chat message history.
# This will make it easier to inspect it to see the underlying results.
store = {}
def get_by_session_id(session_id: str) -> BaseChatMessageHistory:
if session_id not in store:
store[session_id] = InMemoryHistory()
return store[session_id]
history = get_by_session_id("1")
history.add_message(AIMessage(content="hello"))
print(store) # noqa: T201
簡単に説明すると、
このコードは、チャットメッセージの履歴をメモリ上に保存する簡単な例です。
動作説明:
InMemoryHistoryクラス:
チャットメッセージの履歴をリストとしてメモリ上に保持するクラスです。
add_messagesメソッドでメッセージを追加し、clearメソッドで履歴をクリアします。
store変数:
InMemoryHistoryオブジェクトを格納するグローバル変数です。セッションIDをキーとして、各セッションの履歴を管理します。
get_by_session_id関数:
セッションIDを受け取り、store変数から対応するInMemoryHistoryオブジェクトを返します。
存在しないセッションIDの場合は、新しいInMemoryHistoryオブジェクトを作成してstore変数に追加します。
コードの実行:
get_by_session_id("1")でセッションID "1" の履歴を取得します。
history.add_message(AIMessage(content="hello"))で"hello"という内容のAIからのメッセージを履歴に追加します。
print(store)でstore変数の内容、つまりメモリ上に保存された履歴を表示します。
出力結果:
{'1': {'messages': [AIMessage(content='hello', additional_kwargs={})], 'lc_serializable': True}}
これは、セッションID "1" の履歴に "hello" というメッセージが保存されたことを示しています。
公式コードを元にもう少し情報を入手しよう
OpenAIのチャットモデルと連携してユーザーの質問に応答する仕組みを確認します。また、セッションIDに基づいてチャットメッセージ履歴を管理するための機能も実装します。invokeに伴うメタデータも確認します。
確認用サンプルコード
前半部分は公式コードと同じですが、諸々の確認のためのコードを付加します。コード全体を示します。
basechat02.py
from operator import itemgetter
from typing import List
from langchain_openai.chat_models import ChatOpenAI
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import (
RunnableLambda,
ConfigurableFieldSpec,
RunnablePassthrough,
)
from langchain_core.runnables.history import RunnableWithMessageHistory
class InMemoryHistory(BaseChatMessageHistory, BaseModel):
"""In memory implementation of chat message history."""
messages: List[BaseMessage] = Field(default_factory=list)
def add_messages(self, messages: List[BaseMessage]) -> None:
"""Add a list of messages to the store"""
self.messages.extend(messages)
def clear(self) -> None:
self.messages = []
# Here we use a global variable to store the chat message history.
# This will make it easier to inspect it to see the underlying results.
store = {}
def get_by_session_id(session_id: str) -> BaseChatMessageHistory:
if session_id not in store:
store[session_id] = InMemoryHistory()
return store[session_id]
history = get_by_session_id("1")
history.add_messages([AIMessage(content="hello")])
print(store) # noqa: T201
#exit() # 公式サンプル終わり
# Example where the wrapped Runnable takes a dictionary input:
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
prompt = ChatPromptTemplate.from_messages([
("system", "You're an assistant who's good at {ability}"),
MessagesPlaceholder(variable_name="history"),
("human", "{question}"),
])
chain = prompt | ChatOpenAI(model="gpt-3.5-turbo")
chain_with_history = RunnableWithMessageHistory(
chain,
# Uses the get_by_session_id function defined in the example
# above.
get_by_session_id,
input_messages_key="question",
history_messages_key="history",
)
print(chain_with_history.invoke( # noqa: T201
{"ability": "math", "question": "What does cosine mean?"},
config={"configurable": {"session_id": "foo"}}
))
# Uses the store defined in the example above.
print(store) # noqa: T201
print(chain_with_history.invoke( # noqa: T201
{"ability": "math", "question": "What's its inverse"},
config={"configurable": {"session_id": "foo"}}
))
print(store) # noqa: T201
出力結果
実行結果が内容に比べて異常に長くなっていて申し訳ありません。
実際正常に動いたので、わざわざ中身の確認は必要ない感じなのですが、一度は出しておかないとどんな情報が含まれているかわからなくなるので、頑張りました。
読み飛ばして問題ないです。一応節目に区切り線を入れておきました。
出力結果の解説
最初の出力結果
{'1': InMemoryHistory(messages=[AIMessage(content='hello')])}session_idが"1"のチャットメッセージ履歴が表示されています。
InMemoryHistoryインスタンスのmessagesリストには、AIMessage(content='hello')が含まれています。
これは、history.add_messages([AIMessage(content="hello")])によって追加されたメッセージです。
最初のinvokeメソッドの出力結果
content='Cosine is a trigonometric function ...' 以下略"foo"セッションに対して、「What does cosine mean?」という質問が投げられ、AIモデルからの応答が返されています。
response_metadataには、トークン使用量やモデル名などの詳細情報が含まれています。
応答内容は「Cosine is a trigonometric function ...」という説明です。
2つ目の出力結果
InMemoryHistory(messages=[HumanMessage(content='What does cosine mean?'), AIMessage(content='Cosine is a trigonometric function ...')])}session_idが"foo"のチャットメッセージ履歴が表示されています。
HumanMessage(content='What does cosine mean?')と、それに対するAIの応答AIMessage(content='Cosine is a trigonometric function ...')が含まれています。
2度目のinvokeメソッドの出力結果
lled the arccosine ...' response_metadata={'token_usage': 以下略"foo"セッションに対して、「What's its inverse」という質問が投げられ、AIモデルからの応答が返されています。
response_metadataには、トークン使用量やモデル名などの詳細情報が含まれています。
応答内容は「The inverse of the cosine function is called the arccosine ...」という説明です。
3つ目の出力結果
{'1': InMemoryHistory(messages=[AIMessage(content='hello')]), 'foo': InMemoryHistory(messages=[HumanMessage(content='What does cosine mean?'), AIMessage(content='Cosine is a trigonometric function ...'), HumanMessage(content="What's its inverse"), AIMessage(content='The inverse of the cosine function is called the arccosine ...')])}session_idが"foo"のチャットメッセージ履歴が更新されています。
HumanMessage(content="What's its inverse")と、それに対するAIの応答
色々読みましたが、結果として
正常動作:出力結果から、コードは期待通りに動作していることが確認できました。
セッションIDに基づくメッセージ履歴の管理が正しく行われています。
ChatOpenAIモデルを使用して、適切な応答が生成されています。
トークン使用量やモデル名などのメタデータも正しく取得されています。
後半コードの概略
念の為この後半コードのエッセンスを説明します。
目的
チャットプロンプトの生成:ユーザーの質問に応答するためのプロンプトを生成。
履歴の管理:セッションごとにチャットメッセージの履歴を管理し、履歴に基づいた応答を生成。
AIモデルの連携:OpenAIのチャットモデルを使用して、ユーザーの質問に対する応答を生成。
コードの各部分の解説
1. チャットプロンプトの生成
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
prompt = ChatPromptTemplate.from_messages([
("system", "You're an assistant who's good at {ability}"),
MessagesPlaceholder(variable_name="history"),
("human", "{question}"),
])
ChatPromptTemplateの生成:from_messagesメソッドを使用して、チャットプロンプトのテンプレートを生成します。
メッセージのプレースホルダー:
("system", "You're an assistant who's good at {ability}"):システムメッセージで、アシスタントの能力を指定します。
MessagesPlaceholder(variable_name="history"):履歴のプレースホルダーです。過去のメッセージ履歴がここに挿入されます。
("human", "{question}"):ユーザーの質問をプレースホルダーとして指定します。
2. チェーンの作成
chain = prompt | ChatOpenAI(model="gpt-3.5-turbo")
プロンプトとAIモデルの連結:promptとChatOpenAIモデルをパイプラインのように連結します。これにより、プロンプトの生成とAIモデルの応答が一連の流れで処理されます。
3. 履歴の管理とRunnableの作成
from langchain_core.runnables.history import RunnableWithMessageHistory
chain_with_history = RunnableWithMessageHistory(
chain,
get_by_session_id,
input_messages_key="question",
history_messages_key="history",
)
RunnableWithMessageHistoryの作成:RunnableWithMessageHistoryクラスを使用して、履歴を管理するRunnableを作成します。
get_by_session_idの使用:セッションIDに基づいて履歴を取得するために、先ほど定義したget_by_session_id関数を使用します。
キーの設定:
input_messages_key="question":ユーザーの質問をインプットとして指定。
history_messages_key="history":メッセージ履歴を指定。
4. 実行と出力
print(chain_with_history.invoke( # noqa: T201
{"ability": "math", "question": "What does cosine mean?"},
config={"configurable": {"session_id": "foo"}}
))
# Uses the store defined in the example above.
print(store) # noqa: T201
print(chain_with_history.invoke( # noqa: T201
{"ability": "math", "question": "What's its inverse"},
config={"configurable": {"session_id": "foo"}}
))
print(store) # noqa: T201
invokeメソッドの使用:invokeメソッドを使用して、ユーザーの質問に対する応答を生成します。
{"ability": "math", "question": "What does cosine mean?"}:質問内容とアシスタントの能力を指定します。
config={"configurable": {"session_id": "foo"}}:セッションIDを指定して、履歴を管理します。
出力:
初回の出力:AIモデルからの応答が生成され、履歴に保存されます。
storeの出力:履歴の内容が表示されます。
再度のinvokeメソッドの使用:新しい質問を投げ、再度AIモデルからの応答を生成します。
storeの再度の出力:更新された履歴が表示されます。
元のチャットプログラムをRunnableWithMessageHsitoryで書き換える
さて、気持ちとしては下記のように単純な置き換えでも行けそうですが、すでに確認したようにチャットメモリ構造など大きく変更があるので、この機械に、公式サンプルベースで新しく基本チャット構造を作ったほうが将来的に安心できそうです。
単純な置き換え方針 ボツ
この単純な置き換え方針もメモとして残しておきます。
例:ConversationChainからRunnableWithMessageHistoryへの移行
# 旧コード
from langchain_core import ConversationChain
conversation = ConversationChain()
# 新コード
from langchain_core.runnables.history import RunnableWithMessageHistory
conversation = RunnableWithMessageHistory()
そのままでも動きそうですが、メモリ周りも含めて公式サンプルを参考に書いていきます。
基本的な読み落としがあるのかもしれませんが、試してみると、historyの受け渡し構造が想定と異なる類のメッセージがでて、もぐらたたきのように対応してとても面倒だったので、諦めることにしました。(一応動かすことはできたのですが、とても将来にわたって保守できるような構造にならなかったので、潔く辞めることにしました。なにかおまじない的な互換性コードが有るのかもしれませんが、これもきっと覚えていられないでしょう)
改定チャットプログラム
プログラムの概要
本プログラムの概要です。
コンソールプログラム
入力 input()文
出力 print()文実行方法
python basechat03.py -c chat_session -m llm_model です。
デフォルト値は、chat_session="chat" , llm_model="gpt-3.5-turbo"です。
OpenAI以外のモデルが使われることはありません。システムプロンプト
ロール設定は日本語のアドバイザーです。以下のようなシステムプロンプトを利用します。
"""
あなたは、各分野の専門家として、ユーザーの入力に対し、以下の条件を守って、わかりやすく解説します。
条件:
1.出力は、平易な日本語の平文、スライド、プログラムコードからなります。
2.スライドは、VS CodeのMarp extensionで確認するので、そのまま使えるmarkdown形式で出力してください。
3.プログラムコードは、特に指定がなければpythonで出力してください。
4.その他、特にプロンプトで指定された場合は、その指示に従ってください。"""チャットループ
チャット本体は、シンプルにexitが入力するまで、最新の応答のみをsh津力する無限ループです。
while True:
user=input()
if user=="exit":
exit
chain_with_history.invoke({,,,"quesiont":user,,,},{config={}) #user以外は適切なパラメータをいれる
output=store.xxxxx #最新の応答のみテキスト化
print(output)
end phase履歴の保存
ストアにある会話内容をjson形式で保存する
filename="log"+chat_session+".json"
基本的な構造
["human":input},
{"Ai":output},,,,,,
実装コード basechat03.py
import argparse
import json
from operator import itemgetter
from typing import List
from langchain_openai.chat_models import ChatOpenAI
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import BaseMessage, AIMessage
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.runnables import RunnableLambda, ConfigurableFieldSpec, RunnablePassthrough
from langchain_core.runnables.history import RunnableWithMessageHistory
class InMemoryHistory(BaseChatMessageHistory, BaseModel):
"""In memory implementation of chat message history."""
messages: List[BaseMessage] = Field(default_factory=list)
def add_messages(self, messages: List[BaseMessage]) -> None:
"""Add a list of messages to the store"""
self.messages.extend(messages)
def clear(self) -> None:
self.messages = []
# Here we use a global variable to store the chat message history.
store = {}
def get_by_session_id(session_id: str) -> BaseChatMessageHistory:
if session_id not in store:
store[session_id] = InMemoryHistory()
return store[session_id]
def main():
parser = argparse.ArgumentParser(description="Basic Chat Program")
parser.add_argument('-c', '--chat_session', type=str, default="chat", help="Chat session ID")
parser.add_argument('-m', '--llm_model', type=str, default="gpt-3.5-turbo", help="LLM model to use")
args = parser.parse_args()
chat_session = args.chat_session
llm_model = args.llm_model
history = get_by_session_id(chat_session)
prompt = ChatPromptTemplate.from_messages([
("system", "あなたは、各分野の専門家として、ユーザーの入力に対し、以下の条件を守って、わかりやすく解説します。条件:\n 1.出力は、平易な日本語の平文、スライド、プログラムコードからなります。\n 2.スライドは、VS CodeのMarp extensionで確認するので、そのまま使えるmarkdown形式で出力してください。\n 3.プログラムコードは、特に指定がなければpythonで出力してください。\n 4.その他、特にプロンプトで指定された場合は、その指示に従ってください。"),
MessagesPlaceholder(variable_name="history"),
("human", "{question}"),
])
chain = prompt | ChatOpenAI(model=llm_model)
chain_with_history = RunnableWithMessageHistory(
chain,
get_by_session_id,
input_messages_key="question",
history_messages_key="history",
)
while True:
user_input = input("あなた: ")
if user_input.lower() == "exit":
break
response = chain_with_history.invoke(
{"question": user_input},
config={"configurable": {"session_id": chat_session}}
)
# Get the latest AI response
latest_response = store[chat_session].messages[-1].content
print("AI: ", latest_response)
# Save chat history to a JSON file
filename = f"log_{chat_session}.json"
chat_log = [
{"human": msg.content} if isinstance(msg, BaseMessage) else {"AI": msg.content}
for msg in store[chat_session].messages
]
with open(filename, 'w', encoding='utf-8') as f:
json.dump(chat_log, f, ensure_ascii=False, indent=2)
if __name__ == "__main__":
main()
実行結果
本プログラムの実行結果(log_chat.jsonの内容)です。
基本的に動いていますね。履歴に基づく会話も問題ないようです。「今までの会話」をこのチャットセッション以前の会話として認識しているのが多少問題(この会話がセッションの内容らしい)ですが、これは、OpenAIのモデルを選択や、文脈によって変わるかもしれませんね。このチャットメモリーの責任ではないでしょう。
まとめ
以上、langchainの新しいチャット履歴管理であるlangchain_core.runnables.history.RunnableWithMessageHistory
を試してみました。
従来のConversationChainとConversationBufferMemoryの組み合わせは将来的いに使用できなくなる可能性があります。
RunnableWithMessageHistoryが利用できることが確認できました。
若干手続きが面倒な気がしますが、同じ枠組みでチャットセッションによって、複数のチャット履歴を管理できそうなので、移行する意味は十分あるでしょう。
今後もう少し調査しながら、移行を進めようと思います。例えば、
チャットクラスとしての描き下ろし
チャットメモリーのトークン数管理
複数チャットメモリの統合
メモリの保存、読み込み
ConversationChain
ConversationBufferMemory
ConversationChainクラス非推奨
この記事が気に入ったらサポートをしてみませんか?