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

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

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

Converse APIとは

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

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

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

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

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

作ったもの

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

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

リージョン一覧の取得

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

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

モデル一覧の取得

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

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もあります。

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

コード全体

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

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がいますね？