Bedrock(Converse API)によるドキュメントチャットをChainlitとLangChainで構築する

:::note info
本記事で使用している langchain-aws パッケージの ChatBedrockConverse クラスは 2024/7/4 時点でベータ扱いです。今後仕様は変更される可能性があります。
:::

## はじめに
Amazon Bedrock の Converse API がドキュメントチャットをサポートしていると話題です。

https://qiita.com/moritalous/items/ff2763bcd9408a1b395a

Converse API は Amazon Bedrock がサポートする複数の基盤モデルを統一のインターフェースで呼び出すことができる API です。

https://aws.amazon.com/jp/about-aws/whats-new/2024/05/amazon-bedrock-new-converse-api/

少し乗り遅れましたが、Chainlit × LangChain × Bedrock (Converse API) でドキュメントアップロードに対応したチャットボットアプリを作ってみます。

## 動作イメージ
こんな感じのチャットボットアプリをサクッとデプロイできます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/1db309c5-b31c-7bb3-9493-911056b392e9.png)

Claude 3 シリーズおよび、Claude 3.5 Sonnet のみが画像入力をサポートします。ドキュメントはプロバイダーやモデルごとに対応状況が異なります。基盤モデルごとのモダリティのサポート状況は以下のドキュメントを参照してください。

https://docs.aws.amazon.com/bedrock/latest/userguide/conversation-inference.html#conversation-inference-supported-models-features

:::note info
Anthropic 社の最新モデルである Claude 3.5 Sonnet は 2024/7/4 時点でドキュメントの入力をサポートしていません。
:::

Chainlit の Chat settings 機能を通じて Bedrock が提供する複数の Text モデルに切り替えて利用することもできます。会話履歴はセッション内でのみ保持しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/c23660e1-ac7c-5654-9c58-91fe745b2035.png)

ソースコードは以下で公開しています。

https://github.com/hayao-k/Bedrock-AIChatbot-Sample

実際には以下の記事で紹介しているコードを最新のライブラリバージョンや Bedrock の Converse API などに対応させた形でアップデートしたものです。

https://qiita.com/hayao_k/items/a3f7a893e4f6b71dc0b7

## 動作環境
以下の環境で動作確認しています。

* 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 フレームワークです。

https://github.com/Chainlit/chainlit

Chainlit は OpenAI のライブラリや LangChain, LlamaIndex のような人気のフレームワークとの Integration を提供しており、簡単にチャットボットアプリケーションに組み込むことができます。

作成したアプリケーションコードは以下です。Chainlit や LangChain (LCEL) 周りのコード解説は [前回の記事](https://qiita.com/hayao_k/items/a3f7a893e4f6b71dc0b7) を参照いただければと思います。

```app.py
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 に変換するといった処理が不要になったので、少しだけコードもすっきりしました。

:::note info
Base64 へ変換が不要なのは厳密には AWS SDK を使用している場合のみです。API を直接叩く場合などは変換が必要です。
:::

https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_DocumentSource.html

> **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 をサポートする [チャットモデル](https://python.langchain.com/v0.1/docs/modules/model_io/chat/) です。

```py
from langchain_aws import ChatBedrockConverse

llm = ChatBedrockConverse(
    model="anthropic.claude-3-5-sonnet-20240620-v1:0",
    temperature=0.3,
    max_tokens=1024
)
```
:::note info
langchain-aws パッケージの ChatBedrockConverse クラスは 2024/7/4 時点でベータ扱いです。今後仕様は変更される可能性があります。
:::

既存のチャットモデルである ChatBedrock に対し、`beta_use_converse_api=True` を指定することで Converse API を使用することもできます (内部で ChatBedrockConverse クラスが使用されます。)

```py
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 で議論されていました。

https://github.com/langchain-ai/langchain-aws/pull/74

### ドキュメントの処理
デコレーターにより、Chainlit の UI からユーザーがメッセージを送信したときに on_message() 関数が呼び出されます。
```py
@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() 関数では以下のような処理が実行されます。

1. ユーザーセッションから memory と runnable オブジェクトを取得
1. メッセージにファイルが添付されていた場合、ファイルをオープン
1. MIME タイプによって画像 or ドキュメントを判定し、リクエストに必要な [ContentBlock](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ContentBlock.html) を追加
1. runnable.astream() を使って、作成したデータを言語モデルに渡し、生成された応答をチャンクごとにストリーミングしながら表示
1. 応答が完了したら、ユーザーのメッセージと生成された応答を memory に追加

Chainlit にはマルチモーダルの機能が組み込まれており、UI からファイルを添付できます。

https://docs.chainlit.io/advanced-features/multi-modal

:::note info
Amazon Bedrock は、コンテンツとして提供された画像、ドキュメントを保存しません。データは応答を生成するためにのみ使用されます。モデル呼び出しログを S3 バケットに対して配信している場合は、Converse API に渡された画像やドキュメントも S3 にバケットへ記録されます。
:::

https://docs.aws.amazon.com/bedrock/latest/userguide/conversation-inference.html#conversation-inference-supported-models-features

> Amazon Bedrock doesn't store any text, images, or documents that you provide as content. The data is only used to generate the response.

https://docs.aws.amazon.com/bedrock/latest/userguide/model-invocation-logging.html

> 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 リポジトリ](https://github.com/hayao-k/Bedrock-AIChatbot-Sample) にはチャットボットアプリケーションを AWS App Runner にデプロイするためのマニフェストファイルが含まれています。以下の手順でデプロイできます。Copilot CLI を使用せずに Dockerfile で手動でイメージをビルドし、任意のインフラにデプロイすることも可能です。

### 事前準備
AWS Copilot CLI のインストール

https://aws.github.io/copilot-cli/docs/getting-started/install/

```shell-session
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
```

### デプロイ
```shell-session
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 へアクセスします。以下のような画面が表示されたらデプロイ成功です！

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/9d8fbf1d-95e1-f9df-c9b4-5504f4348e2b.png)

:::note info
Chainlit の バージョン 1.1.300 以降、ユーザーに会話のきっかけを提案するための [Starters](https://docs.chainlit.io/concepts/starters) が実装されたため、以前のバージョンでチャットボットアクセス時に表示されていた Readme は表示されなくなりました (サイドバーのリンクから表示することは可能です)。
:::

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/125105/f36d4ea3-e49b-39d0-5d70-cf6e66175485.png)

### アプリケーションの削除

環境を削除する際は以下のコマンドを実行します。CloudWatch Logs group は残存してしまうので手動で削除してください。

```shell-session
copilot app delete
```

### ローカルでの起動方法
以下のコマンドを実行し、`http://localhost:8000` にアクセスします。
```shell-session
pip install -r requirements.txt
export AWS_REGION=us-east-1
chainlit run app.py
```



以上です。
参考になれば幸いです。