AmazonBedrock(Claude3 Vision)をLangChain+Streamlitで使う

AWSサンプルのAmazon Bedrock QuickStartのサンプルがClaude3になっていなかったので作ってみました。

「claude_3_examples/image_api_st.py」が追加されていましたが、Stable DiffusionかAmazon Titanで制定した画像のタイトルをClaude3に問い合わせるものであり、思ったものと違いました。

以下がAmazon Bedrock(Claude3 sonnet)をLangChain+Streamlitで動かすPythonコードです。
テキストだけでなく、画像の入力にも対応しています。

chat_bedrock_st.py

import time

import boto3
import base64
import streamlit as st

from langchain_community.chat_models import BedrockChat
from langchain_core.messages import HumanMessage
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

st.title("ChatBedrock")

# Setup bedrock
bedrock_runtime = boto3.client(
    service_name="bedrock-runtime",
    region_name="us-east-1",
)

@st.cache_resource
def load_chat():
    chat = BedrockChat(
        client=bedrock_runtime,
        model_id="anthropic.claude-3-sonnet-20240229-v1:0",
        streaming=True,
        callbacks=[StreamingStdOutCallbackHandler()],
        model_kwargs = {"temperature": 0.0}
    )

    return chat

chat = load_chat()

uploaded_file = st.file_uploader("Image", type="jpg")

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

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

if prompt := st.chat_input("What is up?"):
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.markdown(prompt)

    with st.chat_message("assistant"):
        message_placeholder = st.empty()
        full_response = ""
        
        content = []
        
        if uploaded_file:
            image_data = base64.b64encode(uploaded_file.read()).decode('utf-8')
            content.append(
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/jpeg",
                        "data": image_data
                    }
                }
            )
    
        content.append(
            {
                "type": "text",
                "text": prompt
            }
        )

        messages = [
            HumanMessage(
                content=content
            )
        ]

        # prompt = prompt_fixer(prompt)
        result = chat.invoke(messages)

        # Simulate stream of response with milliseconds delay
        for chunk in result.content.split(' '): # fix for https://github.com/streamlit/streamlit/issues/868
            full_response += chunk + ' '
            if chunk.endswith('\n'):
                full_response += ' '
            time.sleep(0.05)
            # Add a blinking cursor to simulate typing
            message_placeholder.markdown(full_response + "▌")

        message_placeholder.markdown(full_response)

    st.session_state.messages.append({"role": "assistant", "content": full_response})

以下のサイトを参考にしました。

以下のような感じで動作します。よく見ると、ClaudeがOpenAIになっています。

入力した画像は以下の通りです。