LoginSignup
5
6

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@moritalous

Claude 3.7 Sonnetの思考プロセスも含めてStreamlitでカタカタ言わす

Posted at

Claude 3.7 Sonnetが発表されたのでなにかしたいと思いまして、Streamlitを使ったチャットアプリを思考プロセスの可視化を含めてストリーミングで行うものを作成しました。

試行錯誤を繰り返し、なんとか完成しました。

その後、3.7 Sonnetにリファクタリングしてもらい、ついでに本投稿の下書きも作成してもらいました（ちゃんとレビューしてちょいちょい直しました）。

それでは、どうぞ御覧ください。

こんにちは！今回はAmazon Bedrock上のClaude 3.7 Sonnetを利用したチャットアプリケーションの実装について解説します。

特にこのアプリケーションの特徴は、LLMの「思考プロセス」を可視化できる点です。それでは、コードの詳細を見ていきましょう。

技術スタックの概要

このアプリケーションでは以下の技術を使用しています：

  • Streamlit - Pythonベースの対話型Webアプリケーションフレームワーク
  • Amazon Bedrock - AWSのフルマネージド型生成AIサービス
  • Claude 3.7 Sonnet - Anthropic社の最新LLMモデル

コード解説

1. ライブラリのインポートと定数定義

まずは必要なライブラリをインポートし、使用するモデルIDと思考プロセスのトークン予算を定数として定義しています。THINKING_BUDGETはClaude 3.7が思考プロセスに使用できるトークン数の上限です。

import boto3
import streamlit as st
from typing import Dict, Any, List

# 定数定義
MODEL_ID = "us.anthropic.claude-3-7-sonnet-20250219-v1:0"
THINKING_BUDGET = 4000

2. セッション管理

Streamlitのセッション状態を初期化する関数を作成します。チャット履歴を保存するためのmessages変数がない場合は初期化します。セッション状態を使うことで、複数ターンにわたるチャットが可能になります。（ブラウザをリロードすると初期化されます）

def initialize_session_state():
    """セッション状態を初期化する"""
    if "messages" not in st.session_state:
        st.session_state.messages = []
    return st.session_state.messages

3. チャット履歴の表示

会話履歴を表示する関数です。Streamlitのchat_messageコンポーネントを使って、ユーザーとアシスタントの会話をチャットUIで表示します。

def display_chat_history(messages: List[Dict[str, Any]]):
    """過去のチャット履歴を表示する"""
    for message in messages:
        with st.chat_message(message["role"]):
            st.write(message["content"][0]["text"])

4. Bedrockクライアントの取得

AWS SDKのboto3を使って、Amazon Bedrockのランタイムクライアントを取得します。このクライアントを通じてLLMと通信します。

def get_bedrock_client():
    """Bedrock APIクライアントを取得する"""
    return boto3.client("bedrock-runtime")

5. モデルレスポンスの処理

この関数はBedrockからのストリーミングレスポンスを処理する部分です。ここが最も重要と思います。ポイントは以下のとおりです。

  1. ストリーミングイベントを順次処理
  2. コンテンツブロックごとに「思考プロセス」と「回答テキスト」を分けて管理
  3. 思考プロセスは拡張可能なコンポーネント内に表示
  4. 回答テキストはチャットメッセージとして表示
  5. UIをリアルタイムで更新しながらコンテンツを蓄積
def process_model_response(response, messages: List[Dict[str, Any]]):
    """モデルからのレスポンスを処理する"""
    contents = {}  # コンテンツ保存用辞書
    outputs = {}  # UI出力要素の辞書

    for event in response["stream"]:
        if "contentBlockDelta" not in event:
            continue

        block = event["contentBlockDelta"]
        delta = block["delta"]
        index = str(block["contentBlockIndex"])

        # 新しいコンテンツブロックの初期化
        if index not in contents:
            contents[index] = {"text": "", "thinking": ""}

        # 思考プロセスの処理
        if "reasoningContent" in delta and "text" in delta["reasoningContent"]:
            contents[index]["thinking"] += delta["reasoningContent"]["text"]

            if index not in outputs:
                outputs[index] = st.expander("Thinking...", expanded=True).empty()

            outputs[index].write(contents[index]["thinking"])

        # 回答テキストの処理
        if "text" in delta:
            contents[index]["text"] += delta["text"]

            if index not in outputs:
                outputs[index] = st.chat_message("assistant").empty()

            outputs[index].write(contents[index]["text"])

    # メッセージ履歴に回答を追加
    update_message_history(messages, contents)

    return contents

6. メッセージ履歴の更新

レスポンスを履歴に追加します。思考プロセスは履歴に保存せず、最終的な回答テキストのみを保存する点に注意してください。

def update_message_history(
    messages: List[Dict[str, Any]], contents: Dict[str, Dict[str, str]]
):
    """応答内容をメッセージ履歴に追加する"""
    for content in contents.values():
        if content["text"]:
            messages.append(
                {"role": "assistant", "content": [{"text": content["text"]}]}
            )

7. メイン関数

メイン関数では、作成した関数を呼び出して以下の処理を行っています。

  1. Streamlitアプリのタイトル設定
  2. Bedrockクライアントの初期化
  3. セッション状態の初期化
  4. チャット履歴の表示
  5. ユーザー入力の受け取りと処理
  6. Bedrock APIへのリクエスト（重要な点としてthinking機能を有効化）
  7. ストリーミングレスポンスの処理
def main():
    """メイン関数"""
    st.title("Claude 3.7 Sonnet on Bedrock")
    st.subheader("with extended thinking")

    client = get_bedrock_client()
    messages = initialize_session_state()

    # チャット履歴の表示
    display_chat_history(messages)

    # ユーザー入力の処理
    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

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

        try:
            # APIリクエスト
            response = client.converse_stream(
                modelId=MODEL_ID,
                messages=messages,
                additionalModelRequestFields={
                    "thinking": {"type": "enabled", "budget_tokens": THINKING_BUDGET},
                },
            )

            # レスポンス処理
            process_model_response(response, messages)

        except Exception as e:
            st.error(f"エラーが発生しました: {str(e)}")

実装のポイント

思考プロセスの可視化

このアプリケーションの最大の特徴は、Claude 3.7 Sonnetの思考プロセスをリアルタイムで表示できる点です。additionalModelRequestFieldsパラメータでthinking機能を有効化し、budget_tokensで思考プロセスに割り当てるトークン数を指定しています。

additionalModelRequestFields={
    "thinking": {"type": "enabled", "budget_tokens": THINKING_BUDGET},
}

これにより、モデルのレスポンスにはreasoningContentとしてLLMの思考プロセスが含まれます。

st.emptyを活用したストリーミングUI実装

このアプリケーションのもう一つの重要な技術的ポイントは、st.empty()メソッドを使用したストリーミングUIの実装です。

st.empty()は単一の要素を保持できるコンテナを挿入するStreamlitの機能で、次のような特徴があります。

  1. 動的コンテンツの更新 - 同じ場所に新しいコンテンツを繰り返し書き込める
  2. 効率的な再レンダリング - 全画面を再描画せずに特定の部分だけを更新できる
  3. ストリーミングに最適 - 継続的に更新されるコンテンツを効率的に表示できる

このアプリケーションでは、以下のようにexpanderやチャットメッセージの中に空のコンテナを作成し：

outputs[index] = st.expander("Thinking...", expanded=True).empty()

outputs[index] = st.chat_message("assistant").empty()

モデルがトークンを生成するたびに、同じst.empty()コンテナに対して.write()メソッドを呼び出すことで、テキストを滑らかに追加しています。

outputs[index].write(contents[index]["thinking"])

outputs[index].write(contents[index]["text"])

これにより、ユーザーはモデルからの回答と思考プロセスをリアルタイムで見ることができ、応答を待つ際のユーザー体験が大幅に向上します。

まとめ

今回は、Amazon Bedrock上のClaude 3.7 Sonnetを活用したチャットアプリケーションを実装し、特にLLMの思考プロセスを可視化する機能について解説しました。st.empty()を活用したストリーミングUI実装により、ユーザーはLLMの推論過程と最終的な回答をリアルタイムで確認することができます。

この実装により、LLMがどのように推論を行い結論に至るかをユーザーに示すことができ、AIの説明可能性を高め、より信頼性の高いAIシステムを構築するための一歩となります。Streamlitの柔軟なUIコンポーネントとAmazon Bedrockのストリーミング機能を組み合わせることで、シンプルながらも実用的なアプリケーションとなっています。

最適なストリーミングUI実装のためには、st.empty()のような効率的なコンポーネントの活用が鍵であり、このアプローチはLLMを活用した他のアプリケーション開発にも応用できるでしょう。

ソースコード全体がこちらです。

import boto3
import streamlit as st
from typing import Dict, Any, List

# 定数定義
MODEL_ID = "us.anthropic.claude-3-7-sonnet-20250219-v1:0"
THINKING_BUDGET = 4000


def initialize_session_state():
    """セッション状態を初期化する"""
    if "messages" not in st.session_state:
        st.session_state.messages = []
    return st.session_state.messages


def display_chat_history(messages: List[Dict[str, Any]]):
    """過去のチャット履歴を表示する"""
    for message in messages:
        with st.chat_message(message["role"]):
            st.write(message["content"][0]["text"])


def get_bedrock_client():
    """Bedrock APIクライアントを取得する"""
    return boto3.client("bedrock-runtime")


def process_model_response(response, messages: List[Dict[str, Any]]):
    """モデルからのレスポンスを処理する"""
    contents = {}  # コンテンツ保存用辞書
    outputs = {}  # UI出力要素の辞書

    for event in response["stream"]:
        if "contentBlockDelta" not in event:
            continue

        block = event["contentBlockDelta"]
        delta = block["delta"]
        index = str(block["contentBlockIndex"])

        # 新しいコンテンツブロックの初期化
        if index not in contents:
            contents[index] = {"text": "", "thinking": ""}

        # 思考プロセスの処理
        if "reasoningContent" in delta and "text" in delta["reasoningContent"]:
            contents[index]["thinking"] += delta["reasoningContent"]["text"]

            if index not in outputs:
                outputs[index] = st.expander("Thinking...", expanded=True).empty()

            outputs[index].write(contents[index]["thinking"])

        # 回答テキストの処理
        if "text" in delta:
            contents[index]["text"] += delta["text"]

            if index not in outputs:
                outputs[index] = st.chat_message("assistant").empty()

            outputs[index].write(contents[index]["text"])

    # メッセージ履歴に回答を追加
    update_message_history(messages, contents)

    return contents


def update_message_history(
    messages: List[Dict[str, Any]], contents: Dict[str, Dict[str, str]]
):
    """応答内容をメッセージ履歴に追加する"""
    for content in contents.values():
        if content["text"]:
            messages.append(
                {"role": "assistant", "content": [{"text": content["text"]}]}
            )


def main():
    """メイン関数"""
    st.title("Claude 3.7 Sonnet on Bedrock")
    st.subheader("with extended thinking")

    client = get_bedrock_client()
    messages = initialize_session_state()

    # チャット履歴の表示
    display_chat_history(messages)

    # ユーザー入力の処理
    if prompt := st.chat_input():
        with st.chat_message("user"):
            st.write(prompt)

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

        try:
            # APIリクエスト
            response = client.converse_stream(
                modelId=MODEL_ID,
                messages=messages,
                additionalModelRequestFields={
                    "thinking": {"type": "enabled", "budget_tokens": THINKING_BUDGET},
                },
            )

            # レスポンス処理
            process_model_response(response, messages)

        except Exception as e:
            st.error(f"エラーが発生しました: {str(e)}")


if __name__ == "__main__":
    main()
5
6
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
5
6

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?