Converse APIの登場によりBedrockに新モデルが登場したらいち早く試せる画面が完成しました

Amazon Bedrockに突如新APIが登場しました。驚きです。その名も **Converse API**

![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/41574/c6f4fdeb-1b0e-3705-0e1d-9ddfbb6f03d5.png)

注：Titan Image Generator G1で生成しました（プロンプト：Converse, Bedrock, API）

## Converse APIとは

Converse APIとは、会話型アプリケーションを**単一のAPIで**実行できるものです。

Bedrockにこれまであった、Invoke Model APIは、APIは単一でしたが、パラメータの形式がモデルごとに異なりますので、モデルにあったパラメータを把握して利用する必要がありました。

Converse APIは、パラメータの構造も含めて共通化が進んでおり、モデルIDを変える**だけ**で別のモデルでも動作するようになりました。ClaudeからCommandに切り替えたりが簡単にできるようになりました。

:::note
モデルごとに対応非対応の機能もありますので、詳細はドキュメントを参照してください。

https://docs.aws.amazon.com/bedrock/latest/userguide/conversation-inference.html
:::

## 作ったもの

今後新しいモデルが出た場合も、モデルIDだけ新しくすれば動作する可能性が高いので、モデルIDを選んでチャットする画面を作っておけば、新しいモデルが出てもいち早く試せるはずだと思い、Streamlitで作ってみました。

左のメニューでリージョンやモデルを選ぶと、そのモデルを使用してBedrockにリクエストが投げられます。


![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/41574/368bfac9-6d10-eb37-c1a6-105cbda65d05.png)

### リージョン一覧の取得

普段あまり見慣れない`account`に、`list_regions`というAPIがあります。

```python
client = boto3.client("account")
client.list_regions()
```

### モデル一覧の取得

`list_foundation_models`を使用します。
テキストを扱うモデルで、プロビジョンではなくオンデマンドのものだけを抽出します。

```python
client = boto3.client("bedrock")
client.list_foundation_models(
    byOutputModality="TEXT", byInferenceType="ON_DEMAND"
)
```

## Converse APIの呼び出し

ちょっと階層が深いJSONを作ります。
Invoke Model APIでは、`json.dumps()`を使って文字列にする必要がありましたが、Converse APIではオブジェクトのまま渡せます。

ストリーミングレスポンスに対応した`converse_stream`もあります。

```python
client = boto3.client("bedrock-runtime")
client.converse_stream(
    modelId=modelId, messages=[{
        "role": "user",
        "content": [{"text": prompt}],
    }]
)
```


## コード全体

そんなこんなでできたのがこちら

```python:app.py
import streamlit as st
import boto3

st.title("Bedrock")

def create_client():
    st.session_state["client"] = {
        "bedrock": boto3.Session(region_name=st.session_state["select_region"]).client(
            "bedrock"
        ),
        "bedrock-runtime": boto3.Session(
            region_name=st.session_state["select_region"]
        ).client("bedrock-runtime"),
    }


def change_region():
    create_client()
    del st.session_state["models"]


with st.spinner():

    if "regions" not in st.session_state:
        account_client = boto3.client("account")
        regions = account_client.list_regions(MaxResults=50)["Regions"]
        regions = list(filter(lambda x: x["RegionOptStatus"] != "DISABLED", regions))
        st.session_state["regions"] = list(map(lambda x: x["RegionName"], regions))
    regions = st.session_state["regions"]

    if "select_region" not in st.session_state:
        st.session_state["select_region"] = "us-east-1"

    if "client" not in st.session_state:
        create_client()
    bedrock_client = st.session_state["client"]["bedrock"]

    bedrock_runtime_client = st.session_state["client"]["bedrock-runtime"]

    if "models" not in st.session_state:
        st.session_state["models"] = {}

        models = {}

        foundation_models = bedrock_client.list_foundation_models(
            byOutputModality="TEXT", byInferenceType="ON_DEMAND"
        )

        modelSummaries = foundation_models["modelSummaries"]

        for model in modelSummaries:
            if model["providerName"] not in models:
                models[model["providerName"]] = []
            models[model["providerName"]].append(
                {"modelName": model["modelName"], "modelId": model["modelId"]}
            )

        st.session_state["models"] = models

    models = st.session_state["models"]

    with st.sidebar:

        st.selectbox(
            "リージョン", regions, on_change=change_region, key="select_region"
        )

        providerName = st.selectbox("providerName", models.keys())

        modelName = st.selectbox(
            "モデル", options=list(map(lambda x: x["modelName"], models[providerName]))
        )

        modelId = list(
            filter(lambda x: x["modelName"] == modelName, models[providerName])
        )[0]["modelId"]

        st.text_input("モデルID", value=modelId, disabled=True)


if "messages" not in st.session_state:
    st.session_state["messages"] = []

for message in st.session_state["messages"]:
    with st.chat_message(name=message["role"]):
        st.markdown(message["content"][0]["text"])

if prompt := st.chat_input("何でも聞いて下さい"):

    with st.chat_message(name="user"):
        st.markdown(prompt)

    st.session_state["messages"].append(
        {
            "role": "user",
            "content": [{"text": prompt}],
        }
    )

    res = bedrock_runtime_client.converse_stream(
        modelId=modelId, messages=st.session_state["messages"]
    )

    def stream():
        stream = res["stream"]
        if stream:
            for event in stream:
                if "contentBlockDelta" in event:
                    yield event["contentBlockDelta"]["delta"]["text"]

    with st.chat_message(name="assistant"):
        answer = st.write_stream(stream)

    st.session_state["messages"].append(
        {
            "role": "assistant",
            "content": [{"text": answer}],
        }
    )
```

おや？
us-west-2にも`Olympus 1 Premier`がいますね？

![](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/41574/7d1de7cf-a269-9bbf-be6c-6f95db8fcd3e.png)