こんにちは。AWS CLIが好きな福島です。
- はじめに
- 参考
- Amazon Bedrock(Claude)に質問
- 会話履歴を踏まえた回答
- 会話履歴をメモリに保存
- 会話履歴をDynamoDBに保存(自己履歴管理)
- 会話履歴をDynamoDBに保存(自動履歴管理)
- 終わりに
はじめに
今回は、LangChain(LCEL)とAWS(Amazon Bedrock, Amazon DynamoDB)を利用して、 会話履歴を踏まえた回答を生成するサンプルコードをご紹介します。
LCELについては、以下のブログに書いているため、ご興味がある方はご覧ください。
参考
Amazon Bedrock(Claude)に質問
まずは単純にAmazon Bedrockに質問するサンプルコードです。
サンプルコード
from langchain_community.chat_models.bedrock import BedrockChat from langchain_core.messages import HumanMessage chat = BedrockChat( model_id="anthropic.claude-v2:1", model_kwargs={"temperature": 0.2} ) result = chat.invoke( [HumanMessage(content="この文を日本語から英語に翻訳してください: 私はプログラミングが大好きです.")] ) print(result)
実行結果
$ python test.py content=' Here is the translation to English:\n\nI love programming.' $
補足
補足ですが、ChatModelは入力として、文字列型, ChatMessageのリスト, PromptValueを受け取ることができるため、 以下のように実行することも可能です。どの実行方法でも出力形式は変わらず、ChatMessageになります。
- 文字列型
chat.invoke("この文を日本語から英語に翻訳してください: 私はプログラミングが大好きです.")
- ChatMessageのリスト
chat.invoke([{
"role": "human",
"content": "この文を日本語から英語に翻訳してください: 私はプログラミングが大好きです."
}])
会話履歴を踏まえた回答
本来、LLMは過去のやり取りを覚えていませんが、 会話履歴も含めてLLMに質問することで会話履歴を踏まえた回答を生成することができます。
以下は、会話履歴を踏まえて「何と言いましたか？」という質問に答えるサンプルコードになります。
サンプルコード
from langchain_community.chat_models.bedrock import BedrockChat from langchain_core.messages import HumanMessage, AIMessage chat = BedrockChat( model_id="anthropic.claude-v2:1", model_kwargs={"temperature": 0.2} ) chat.invoke( [ HumanMessage( content="この文を日本語から英語に翻訳してください: 私はプログラミングが大好きです." ), AIMessage(content="Here is the translation to English:\n\nI love programming."), HumanMessage(content="何と言いましたか？"), ] )
$ python test.py content=' 申し訳ありません、最初のメッセージは英語でした。日本語訳は以下の通りです:\n\n私はプログラミングが大好きです。' $
これは文字列型では表現できないですが、 ChatMessageのリスト形式を使い、以下のように実行することも可能です。
chat.invoke(
[
{
"role":"human",
"content":"この文を日本語から英語に翻訳してください: 私はプログラミングが大好きです."
},
{
"role":"assistant",
"content":"Here is the translation to English:\n\nI love programming."
},
{
"role":"human",
"content":"何と言いましたか？"
},
]
)
会話履歴をメモリに保存
ChatMessageHistoryクラスを利用することでメモリ内に会話履歴を保存できます。 また必須ではないのですが、以下ではプロンプトテンプレートを活用し、promptとchatのChainを組んでいます。
サンプルコード
from langchain_community.chat_models.bedrock import BedrockChat from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.memory import ChatMessageHistory ## クラスのインスタンス化 chat_history = ChatMessageHistory() ## ユーザーとして、メッセージの追加 chat_history.add_user_message("この文を日本語から英語に翻訳してください: 私はプログラミングが大好きです.") ## AIとして、メッセージの追加 chat_history.add_ai_message("Here is the translation to English:\n\nI love programming.") ## ユーザーとして、メッセージの追加 chat_history.add_user_message("何と言いましたか？") ## プロンプトテンプレートを作成 prompt = ChatPromptTemplate.from_messages( [ ( "system", "あなたは役に立つアシスタントです。 すべての質問にできる限り答えてください。", ), ## chain.invokeの引数に指定したmessages(Key)の値(Value)が含まれます。 MessagesPlaceholder(variable_name="messages"), ] ) chat = BedrockChat( model_id="anthropic.claude-v2:1", model_kwargs={"temperature": 0.2} ) ## promptとchatのChainを組む chain = prompt | chat ## Chainの実行 result = chain.invoke( { "messages": chat_history.messages } ) print(result)
- 実行結果
$ python test.py content=' ごめんなさい、最初のメッセージは英語でした。日本語への翻訳は以下の通りです:\n\n私はプログラミングが大好きです。' $
ポイント
chat_history.messagesを実行結果は、以下の通りでPromptValue形式になっています。
[HumanMessage(content='この文を日本語から英語に翻訳してください: 私はプログラミングが大好きです.'), AIMessage(content='Here is the translation to English:\n\nI love programming.'), HumanMessage(content='何と言いましたか？')]
また、chatに渡される値を確認したい場合は、prompt.invoke({"messages": chat_history.messages})を実行することで確認することができます。
result = prompt.invoke(
{
"messages": chat_history.messages
}
)
prompt.invokeの実行結果は以下の通りです。
$ python test.py messages=[SystemMessage(content='あなたは役に立つアシスタントです。 すべての質問にできる限り答えてください。'), HumanMessage(content='この文を日本語から英語に翻訳してください: 私はプログラミングが大好きです.'), AIMessage(content='Here is the translation to English:\n\nI love programming.'), HumanMessage(content='何と言いましたか？')] $
会話履歴をDynamoDBに保存(自己履歴管理)
メモリに会話履歴を保存してもプログラムの実行が終了した際に揮発されるため、 実際のシステムのことを考慮すると、永続化する必要があります。
今回は、DynamoDBに保存するやり方をご紹介いたします。 DynamoDBに保存する場合は、DynamoDBChatMessageHistoryクラスを利用します。
また、本ブログでは、後述の自動履歴管理と対比して、自己履歴管理と表現しています。
DynamoDBの構築
まずは、DynamoDBを構築します。 テーブル名は、chat-history-dynamodbとし、パーティションキーは、SessionId(String)としています。
aws dynamodb create-table \
--table-name chat-history-dynamodb \
--attribute-definitions \
AttributeName=SessionId,AttributeType=S \
--key-schema \
AttributeName=SessionId,KeyType=HASH \
--provisioned-throughput \
ReadCapacityUnits=5,WriteCapacityUnits=5
サンプルコード
import sys from langchain_community.chat_models.bedrock import BedrockChat from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_community.chat_message_histories import DynamoDBChatMessageHistory ## クラスのインスタンス化 chat_history = DynamoDBChatMessageHistory( table_name="chat-history-dynamodb", session_id=sys.argv[1], ) ## ユーザーとして、メッセージの追加 chat_history.add_user_message(sys.argv[2]) ## プロンプトテンプレートを作成 prompt = ChatPromptTemplate.from_messages( [ ( "system", "あなたは役に立つアシスタントです。 すべての質問にできる限り答えてください。", ), ## chain.invokeの引数に指定したmessages(Key)の値(Value)が含まれます。 MessagesPlaceholder(variable_name="messages"), ] ) chat = BedrockChat( model_id="anthropic.claude-v2:1", model_kwargs={"temperature": 0.2} ) ## promptとchatのChainを組む chain = prompt | chat ## Chainの実行 result = chain.invoke( { "messages": chat_history.messages } ) ## AIとして、メッセージの追加 chat_history.add_ai_message(result.content)
実行結果
pythonを2回実行していますが、2回目の「何と言いましたか？」という質問に会話履歴を踏まえて回答してくれていることが分かります。 第1引数にセッションIDとして、session-id-1を指定しています。(ここは文字列型であればなんでもOKです。) そして、第2引数に質問文を指定しています。
$ python test.py "session-id-1" "この文を日本語から英語に翻訳してください: 私はプログラミングが大好きです." content=' I love programming.' $ python test.py "session-id-1" "何と言いましたか？" content=' ご質問の日本語の文を英語に翻訳しました。\n\n「私はプログラミングが大好きです。」\n\n英語訳:\n\n"I love programming."' $
メモリに保存する場合との変更点
1つ目が以下の点になります。 今回、session_idはpython実行時の引数から受け取ります。 同じ値を設定することでpythonを複数回実行しても同じ会話のやりとりとして判断してくれます。
## 変更前 from langchain.memory import ChatMessageHistory chat_history = ChatMessageHistory() ## 変更後 from langchain_community.chat_message_histories import DynamoDBChatMessageHistory chat_history = DynamoDBChatMessageHistory( table_name="chat-history-dynamodb", session_id=sys.argv[1], )
2つ目がpython実行時の引数に質問を含めるようにしています。 これは動作確認しやすいように変更しています。
## ユーザーとして、メッセージの追加 chat_history.add_user_message(sys.argv[2])
3つ目が生成された回答をDynamoDBに保存するため、以下のコードを最終行に追加してます。
## AIとして、メッセージの追加
chat_history.add_ai_message(result.content)
DynamoDBに登録されたデータの確認
DynamoDBにどのようにデータが保存されているかは以下のコマンドで確認できます。
aws dynamodb query \
--table-name chat-history-dynamodb \
--key-condition-expression "SessionId = :sessionid" \
--expression-attribute-values '{":sessionid":{"S":"session-id-1"}}' \
--return-consumed-capacity TOTAL
大枠として、SessionIdとHistoryというキーが存在します。 SessionIdには会話履歴を特定する識別子を入れて、Historyには会話履歴がリスト形式で保存されています。
{ "Items": [ { "SessionId": { "S": "session-id-1" }, "History": { "L": [ { "M": { "type": { "S": "human" }, "data": { "M": { "type": { "S": "human" }, "content": { "S": "この文を日本語から英語に翻訳してください: 私はプログラミングが大好きです." }, "additional_kwargs": { "M": {} }, "example": { "BOOL": false } } } } }, { "M": { "type": { "S": "ai" }, "data": { "M": { "type": { "S": "ai" }, "content": { "S": "Here is the translation to English:\n\nI love programming." }, "additional_kwargs": { "M": {} }, "example": { "BOOL": false } } } } }, { "M": { "type": { "S": "human" }, "data": { "M": { "type": { "S": "human" }, "content": { "S": "何と言いましたか？" }, "additional_kwargs": { "M": {} }, "example": { "BOOL": false } } } } } ] } } ], "Count": 1, "ScannedCount": 1, "ConsumedCapacity": { "TableName": "chat-history-dynamodb", "CapacityUnits": 0.5 } }
補足(質問や回答以外の情報を保存したい場合)
質問や回答以外の情報を保存したいケース(RAGを実装した際に参考となるドキュメントの情報など)もあるかと思います。 その場合は、add_messageメソッドを使い、以下のようにadditional_kwargsキーを指定することで対応できます。
以下は、時刻を保存する例です。
from datetime import datetime from langchain_core.messages.human import HumanMessage from langchain_core.messages.ai import AIMessage ## ユーザーメッセージに時刻を追加 chat_history.add_message( HumanMessage( content=sys.argv[2], additional_kwargs={'time': str(datetime.now())} ) ) ## AIメッセージに時刻を追加 chat_history.add_message( AIMessage( content=result.content, additional_kwargs={'time': str(datetime.now())} ) )
一部抜粋していますが、DynamoDBにはadditional_kwargsに以下のように保存されます。
{ "M": { "type": { "S": "human" }, "data": { "M": { "type": { "S": "human" }, "content": { "S": "この文を日本語から英語からに翻訳してください: 私はプログラミングが大好きです." }, "additional_kwargs": { "M": { "time": { "S": "2024-02-04 13:21:16.161654" } } }, "example": { "BOOL": false } } } } }, { "M": { "type": { "S": "ai" }, "data": { "M": { "type": { "S": "ai" }, "content": { "S": " Here is the English translation of that Japanese sentence:\n\nI love programming." }, "additional_kwargs": { "M": { "time": { "S": "2024-02-04 13:21:16.188391" } } }, "example": { "BOOL": false } } } } }
会話履歴をDynamoDBに保存(自動履歴管理)
DynamoDBへの会話履歴の保存を自動的にやる方法もあります。 それが、RunnableWithMessageHistoryクラスを利用する方法です。
import sys from langchain_community.chat_models.bedrock import BedrockChat from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_community.chat_message_histories import DynamoDBChatMessageHistory from langchain_core.runnables.history import RunnableWithMessageHistory ## プロンプトテンプレートを作成 prompt = ChatPromptTemplate.from_messages( [ ( "system", "あなたは役に立つアシスタントです。 すべての質問にできる限り答えてください。", ), ## chain.invokeの引数に指定したmessages(Key)の値(Value)が含まれます。 MessagesPlaceholder(variable_name="chat_history"), ("human", "{messages}"), ] ) chat = BedrockChat( model_id="anthropic.claude-v2:1", model_kwargs={"temperature": 0.2} ) ## promptとchatのChainを組む chain = prompt | chat ## RunnableWithMessageHistoryでchainをラップ chain_with_message_history = RunnableWithMessageHistory( chain, lambda session_id : DynamoDBChatMessageHistory( table_name="chat-history-dynamodb", session_id=session_id ), input_messages_key="messages", history_messages_key="chat_history", ) ## Chainの実行 result = chain_with_message_history.invoke( { "messages": sys.argv[2] }, { 'configurable': { 'session_id': sys.argv[1] } } ) print(result)
実行結果
実行方法および結果は先ほどと同様です。 (pythonを2回実行していますが、2回目の「何と言いましたか？」という質問に会話履歴を踏まえて回答してくれていることが分かります。)
$ python test.py "session-id-2" "この文を日本語から英語に翻訳してください: 私はプログラミングが大好きです." content=' I love programming.' $ python test.py "session-id-2" "何と言いましたか？" content=' ご質問の日本語の文を英語に翻訳しました。\n\n「私はプログラミングが大好きです。」\n\nI love programming.' $
自己履歴管理との違い
自己履歴管理との違いは以下の通りです。 add_user_messageやadd_ai_messageをプログラミングする必要がないため、 会話履歴の保存を忘れずに済みそうです。
## 自己履歴管理 chat_history = DynamoDBChatMessageHistory( table_name="chat-history-dynamodb", session_id=sys.argv[1], ) chat_history.add_user_message(sys.argv[2]) --- 中略 --- chain = prompt | chat result = chain.invoke( { "messages": chat_history.messages } ) chat_history.add_ai_message(result.content) ## 自動履歴管理 chain = prompt | chat chain_with_message_history = RunnableWithMessageHistory( chain, lambda session_id : DynamoDBChatMessageHistory( table_name="chat-history-dynamodb", session_id=session_id ), input_messages_key="messages", history_messages_key="chat_history", ) result = chain_with_message_history.invoke( { "messages": sys.argv[2] }, { 'configurable': { 'session_id': sys.argv[1] } } )
RunnableWithMessageHistoryの動き
RunnableWithMessageHistoryについての私の理解での解説です。
chain_with_message_history = RunnableWithMessageHistory(
chain,
lambda session_id : DynamoDBChatMessageHistory(
table_name="chat-history-dynamodb",
session_id=session_id
),
input_messages_key="messages",
history_messages_key="chat_history",
)
第1引数には、ラップするchainを指定します。
第2引数には、DynamoDBChatMessageHistoryが実行される無名関数を定義します。
これは、chain_with_message_history.invoke時の {'configurable': {'session_id': sys.argv[1]}} の引数を受け取ります。
つまり、上記実行結果の例の場合はsession-id-2が無名関数の引数(session_id)に渡されます。
input_messages_keyには、chain_with_message_history.invoke時の {"messages": sys.argv[2]} の値が入ります。
history_messages_keyには、第2引数で指定したクラスのインスタンスから会話履歴が出力された結果(値)を保持するキーを指定します。
つまり、chain_with_message_history.invokeの実行時の動きとしては以下のイメージになると思います。
① input_messages_keyに指定したキーと値を受け取る ② DynamoDBChatMessageHistoryクラスのインスタンスから会話履歴を取得 ③ history_messages_keyに指定したKeyのValueに②の結果が含まれる ④ ③のデータを引数にchain.invokeが実行される ⑤ 質問と回答がDynamoDBに保存される
③の時点で以下のようなデータが生成されます。
{ "messages" : '何と言いましたか？', "chat_history" : "[HumanMessage(content='この文を日本語から英語に翻訳してください: 私はプログラミングが大好きです.'), AIMessage(content='Here is the translation to English:\n\nI love programming.')" }
DynamoDBに登録されたデータの確認
先ほどと同様のコマンドでDynamoDBに会話履歴が保存されていることを確認することができます。
aws dynamodb query \
--table-name chat-history-dynamodb \
--key-condition-expression "SessionId = :sessionid" \
--expression-attribute-values '{":sessionid":{"S":"session-id-2"}}' \
--return-consumed-capacity TOTAL
終わりに
今回は会話履歴を踏まえた回答を生成するサンプルコードをご紹介しました。 どなたかのお役に立てれば幸いです。
