【LangChain(LCEL) / AWS】会話履歴を踏まえた回答の生成

こんにちは。AWS CLIが好きな福島です。

はじめに
参考
Amazon Bedrock(Claude)に質問
会話履歴を踏まえた回答
- サンプルコード
会話履歴をメモリに保存
- サンプルコード
- ポイント
会話履歴をDynamoDBに保存(自己履歴管理)
会話履歴をDynamoDBに保存(自動履歴管理)
終わりに

はじめに

今回は、LangChain(LCEL)とAWS(Amazon Bedrock, Amazon DynamoDB)を利用して、会話履歴を踏まえた回答を生成するサンプルコードをご紹介します。

LCELについては、以下のブログに書いているため、ご興味がある方はご覧ください。

blog.serverworks.co.jp

参考

python.langchain.com

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

終わりに

今回は会話履歴を踏まえた回答を生成するサンプルコードをご紹介しました。どなたかのお役に立てれば幸いです。

はじめに

参考

Amazon Bedrock(Claude)に質問

サンプルコード

実行結果

補足

会話履歴を踏まえた回答

サンプルコード

会話履歴をメモリに保存

サンプルコード

ポイント

会話履歴をDynamoDBに保存(自己履歴管理)

DynamoDBの構築

サンプルコード

実行結果

メモリに保存する場合との変更点

DynamoDBに登録されたデータの確認

補足(質問や回答以外の情報を保存したい場合)

会話履歴をDynamoDBに保存(自動履歴管理)

実行結果

自己履歴管理との違い

RunnableWithMessageHistoryの動き

DynamoDBに登録されたデータの確認

終わりに

Amazon EventBridgeで実現するマルチアカウントイベントの集約管理

ACM for Nitro Enclaves を利用してNginxへのHTTPS通信を実装してみた

ISUCON過去問のボトルネックをNew Relic APMで確認する

AWSから抜け出す方法も考えておく【ガバメントクラウド】

AWS BuilderCardsを遊んでみた