本記事で使用している langchain-aws パッケージの ChatBedrockConverse クラスは 2024/7/4 時点でベータ扱いです。今後仕様は変更される可能性があります。
はじめに
Amazon Bedrock の Converse API がドキュメントチャットをサポートしていると話題です。
Converse API は Amazon Bedrock がサポートする複数の基盤モデルを統一のインターフェースで呼び出すことができる API です。
少し乗り遅れましたが、Chainlit × LangChain × Bedrock (Converse API) でドキュメントアップロードに対応したチャットボットアプリを作ってみます。
動作イメージ
こんな感じのチャットボットアプリをサクッとデプロイできます。
Claude 3 シリーズおよび、Claude 3.5 Sonnet のみが画像入力をサポートします。ドキュメントはプロバイダーやモデルごとに対応状況が異なります。基盤モデルごとのモダリティのサポート状況は以下のドキュメントを参照してください。
Anthropic 社の最新モデルである Claude 3.5 Sonnet は 2024/7/4 時点でドキュメントの入力をサポートしていません。
Chainlit の Chat settings 機能を通じて Bedrock が提供する複数の Text モデルに切り替えて利用することもできます。会話履歴はセッション内でのみ保持しています。
ソースコードは以下で公開しています。
実際には以下の記事で紹介しているコードを最新のライブラリバージョンや Bedrock の Converse API などに対応させた形でアップデートしたものです。
動作環境
以下の環境で動作確認しています。
- Python: v3.12.4
- boto3: v1.34.138
- chainlit: 1.1.306
- langchain: 0.2.6
- langchain-aws: 0.1.9
ソースコード
UI には Chainlit を使用しています。Chainlit は会話型 AI を構築するためのオープンソースの Python フレームワークです。
Chainlit は OpenAI のライブラリや LangChain, LlamaIndex のような人気のフレームワークとの Integration を提供しており、簡単にチャットボットアプリケーションに組み込むことができます。
作成したアプリケーションコードは以下です。Chainlit や LangChain (LCEL) 周りのコード解説は 前回の記事 を参照いただければと思います。
import mimetypes
import os
import re
from operator import itemgetter
from pathlib import Path
import boto3
import chainlit as cl
from chainlit.input_widget import Select, Slider
from langchain.memory import ConversationBufferMemory
from langchain_aws import ChatBedrockConverse
from langchain_core.messages import HumanMessage
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables import RunnableConfig, RunnableLambda, RunnablePassthrough
AWS_REGION = os.environ["AWS_REGION"]
EXCLUDE_MODELS = ['meta.llama2-13b-v1', 'meta.llama2-70b-v1']
PATTERN = re.compile(r'v\d+(?!.*\d[kK]$)')
PROVIDER = ""
@cl.on_chat_start
async def main():
bedrock = boto3.client("bedrock", region_name=AWS_REGION)
response = bedrock.list_foundation_models(
byOutputModality="TEXT"
)
# オンデマンドスループットのモデル ID のみをリスト化
model_ids = [
item['modelId']
for item in response["modelSummaries"]
if PATTERN.search(item['modelId']) and item['modelId'] not in EXCLUDE_MODELS
]
settings = await cl.ChatSettings(
[
Select(
id="Model",
label="Amazon Bedrock - Model",
values=model_ids,
initial_index=model_ids.index(
"anthropic.claude-3-5-sonnet-20240620-v1:0"
),
),
Slider(
id="Temperature",
label="Temperature",
initial=0.3,
min=0,
max=1,
step=0.1,
),
Slider(
id="MAX_TOKEN_SIZE",
label="Max Token Size",
initial=2048,
min=256,
max=8192,
step=256,
),
]
).send()
await setup_runnable(settings)
@cl.on_settings_update
async def setup_runnable(settings):
cl.user_session.set(
"memory", ConversationBufferMemory(return_messages=True)
)
memory = cl.user_session.get("memory")
bedrock_model_id = settings["Model"]
llm = ChatBedrockConverse(
model=bedrock_model_id,
temperature=settings["Temperature"],
max_tokens=int(settings["MAX_TOKEN_SIZE"])
)
global PROVIDER
PROVIDER = bedrock_model_id.split(".")[0]
prompt = ChatPromptTemplate.from_messages(
[
("system", "You are a helpful chatbot"),
MessagesPlaceholder(variable_name="history"),
MessagesPlaceholder(variable_name="human_message")
]
)
runnable = (
RunnablePassthrough.assign(
history=RunnableLambda(memory.load_memory_variables) | itemgetter("history")
)
| prompt
| llm
| StrOutputParser()
)
cl.user_session.set("runnable", runnable)
@cl.on_message
async def on_message(message: cl.Message):
memory = cl.user_session.get("memory")
runnable = cl.user_session.get("runnable")
content = []
for file in (message.elements or []):
mime_type, _ = mimetypes.guess_type(file.name)
file_name_path = Path(file.name)
file_name = file_name_path.stem.replace('.', '-')
file_format = file_name_path.suffix.lstrip('.')
with open(file.path, "rb") as f:
file_data = f.read()
if mime_type:
if mime_type.startswith("image"):
content.append({
"image": {
"format": file_format,
"source": {"bytes": file_data}
}
})
elif mime_type.startswith("application") or mime_type.startswith("text"):
content.append({
"document": {
"name": file_name,
"format": file_format,
"source": {"bytes": file_data}
}
})
content_text = {"type": "text", "text": message.content}
content.append(content_text)
runnable_message_data = {"human_message": [HumanMessage(content=content)]}
res = cl.Message(content="", author=f'Assistant: {PROVIDER.capitalize()}')
async for chunk in runnable.astream(
runnable_message_data,
config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()]),
):
await res.stream_token(chunk)
await res.send()
memory.chat_memory.add_user_message(message.content)
memory.chat_memory.add_ai_message(res.content)
Converse API によって、モデルのプロバイダー毎に異なるパラメーターを設定したり、画像を Base64 に変換するといった処理が不要になったので、少しだけコードもすっきりしました。
Base64 へ変換が不要なのは厳密には AWS SDK を使用している場合のみです。API を直接叩く場合などは変換が必要です。
bytes
The raw bytes for the document. If you use an AWS SDK, you don't need to encode the bytes in base64.
Type: Base64-encoded binary data object
コードの補足
ChatBedrockConverse クラスの利用
langchain-aws パッケージの v0.1.8 以降で ChatBedrockConverse クラスを使用できます。その名のとおり Converse API をサポートする チャットモデル です。
from langchain_aws import ChatBedrockConverse
llm = ChatBedrockConverse(
model="anthropic.claude-3-5-sonnet-20240620-v1:0",
temperature=0.3,
max_tokens=1024
)
langchain-aws パッケージの ChatBedrockConverse クラスは 2024/7/4 時点でベータ扱いです。今後仕様は変更される可能性があります。
既存のチャットモデルである ChatBedrock に対し、beta_use_converse_api=True を指定することで Converse API を使用することもできます (内部で ChatBedrockConverse クラスが使用されます。)
from langchain_aws import ChatBedrock
llm = ChatBedrock(
beta_use_converse_api=True,
model_id="anthropic.claude-3-5-sonnet-20240620-v1:0",
model_kwargs={
"temperature": 0.3,
"max_tokens": 1024
}
)
現状の仕様や今後については以下の Pull Request で議論されていました。
ドキュメントの処理
デコレーターにより、Chainlit の UI からユーザーがメッセージを送信したときに on_message() 関数が呼び出されます。
@cl.on_message
async def on_message(message: cl.Message):
memory = cl.user_session.get("memory")
runnable = cl.user_session.get("runnable")
content = []
for file in (message.elements or []):
mime_type, _ = mimetypes.guess_type(file.name)
file_name_path = Path(file.name)
file_name = file_name_path.stem.replace('.', '-')
file_format = file_name_path.suffix.lstrip('.')
with open(file.path, "rb") as f:
file_data = f.read()
if mime_type:
if mime_type.startswith("image"):
content.append({
"image": {
"format": file_format,
"source": {"bytes": file_data}
}
})
elif mime_type.startswith("application") or mime_type.startswith("text"):
content.append({
"document": {
"name": file_name,
"format": file_format,
"source": {"bytes": file_data}
}
})
content_text = {"type": "text", "text": message.content}
content.append(content_text)
runnable_message_data = {"human_message": [HumanMessage(content=content)]}
res = cl.Message(content="", author=f'Assistant: {PROVIDER.capitalize()}')
async for chunk in runnable.astream(
runnable_message_data,
config=RunnableConfig(callbacks=[cl.LangchainCallbackHandler()]),
):
await res.stream_token(chunk)
await res.send()
memory.chat_memory.add_user_message(message.content)
memory.chat_memory.add_ai_message(res.content)
on_message() 関数では以下のような処理が実行されます。
- ユーザーセッションから memory と runnable オブジェクトを取得
- メッセージにファイルが添付されていた場合、ファイルをオープン
- MIME タイプによって画像 or ドキュメントを判定し、リクエストに必要な ContentBlock を追加
- runnable.astream() を使って、作成したデータを言語モデルに渡し、生成された応答をチャンクごとにストリーミングしながら表示
- 応答が完了したら、ユーザーのメッセージと生成された応答を memory に追加
Chainlit にはマルチモーダルの機能が組み込まれており、UI からファイルを添付できます。
Amazon Bedrock は、コンテンツとして提供された画像、ドキュメントを保存しません。データは応答を生成するためにのみ使用されます。モデル呼び出しログを S3 バケットに対して配信している場合は、Converse API に渡された画像やドキュメントも S3 にバケットへ記録されます。
Amazon Bedrock doesn't store any text, images, or documents that you provide as content. The data is only used to generate the response.
When using the Converse API, any image or document data that you pass is logged in Amazon S3 (if you have enabled delivery and image logging in Amazon S3).
参考: デプロイ方法
冒頭で紹介した GitHub リポジトリ にはチャットボットアプリケーションを AWS App Runner にデプロイするためのマニフェストファイルが含まれています。以下の手順でデプロイできます。Copilot CLI を使用せずに Dockerfile で手動でイメージをビルドし、任意のインフラにデプロイすることも可能です。
事前準備
AWS Copilot CLI のインストール
sudo curl -Lo /usr/local/bin/copilot https://github.com/aws/copilot-cli/releases/latest/download/copilot-linux && sudo chmod +x /usr/local/bin/copilot
デプロイ
git clone https://github.com/hayao-k/Bedrock-AIChatbot-Sample
cd Bedrock-AIChatbot-Sample
export AWS_REGION=us-east-1
copilot app init bedrockchat-app
copilot deploy --name bedrockchat --env dev
デプロイが成功したらメッセージに出力された URL へアクセスします。以下のような画面が表示されたらデプロイ成功です！
Chainlit の バージョン 1.1.300 以降、ユーザーに会話のきっかけを提案するための Starters が実装されたため、以前のバージョンでチャットボットアクセス時に表示されていた Readme は表示されなくなりました (サイドバーのリンクから表示することは可能です)。
アプリケーションの削除
環境を削除する際は以下のコマンドを実行します。CloudWatch Logs group は残存してしまうので手動で削除してください。
copilot app delete
ローカルでの起動方法
以下のコマンドを実行し、http://localhost:8000 にアクセスします。
pip install -r requirements.txt
export AWS_REGION=us-east-1
chainlit run app.py
以上です。
参考になれば幸いです。