Streamlit公式のAgent Skillsを使ってAmazon BedrockのチャットUIをClaude Codeに作らせてみた

# Bedrock チャットデモ用カスタムテーマ
#
# ライト = 白基調 + 緑アクセント / ダーク = 黒基調 + 青アクセント
# [theme.light] と [theme.dark] を両方定義しているため、設定メニューから
# ライト/ダークを切り替えても、それぞれのカスタム配色が維持される。

[theme]
# 両モード共通の設定（色は light/dark 側で上書き）
font = "Inter:https://fonts.googleapis.com/css2?family=Inter:wght@400;500;600;700&display=swap"
baseRadius = "8px"
linkUnderline = false
showSidebarBorder = true

# ライトテーマ: 白基調 + 緑アクセント
[theme.light]
primaryColor = "#15803d"
backgroundColor = "#ffffff"
secondaryBackgroundColor = "#eef5f0"
textColor = "#1f2a24"
linkColor = "#15803d"
borderColor = "#cde2d4"
greenColor = "#15803d"
codeBackgroundColor = "#eef5f0"

[theme.light.sidebar]
backgroundColor = "#f1f7f3"
secondaryBackgroundColor = "#e2efe7"
textColor = "#1f2a24"
borderColor = "#cde2d4"
primaryColor = "#15803d"

# ダークテーマ: 黒基調 + 青アクセント
[theme.dark]
primaryColor = "#3b82f6"
backgroundColor = "#0a0e14"
secondaryBackgroundColor = "#141b24"
textColor = "#e6edf3"
linkColor = "#60a5fa"
borderColor = "#222c38"
blueColor = "#3b82f6"
codeBackgroundColor = "#141b24"

[theme.dark.sidebar]
backgroundColor = "#06090d"
secondaryBackgroundColor = "#141b24"
textColor = "#e6edf3"
borderColor = "#222c38"
primaryColor = "#3b82f6"

_MESSAGE_ACTIONS = st.components.v2.component(
    "message_actions",
    html=_ACTIONS_HTML,   # コピー / 👍 / 👎 ボタン
    css=_ACTIONS_CSS,     # 配色は var(--st-text-color) / var(--st-primary-color) など
    js=_ACTIONS_JS,       # copy: setTriggerValue("copied") / 評価: setStateValue("feedback")
)


def message_actions(idx, text, feedback, *, on_copied=None, on_feedback=None):
    return _MESSAGE_ACTIONS(
        key=f"msg-actions-{idx}",
        data={"text": text, "feedback": feedback},
        on_feedback_change=on_feedback or (lambda: None),
        on_copied_change=on_copied or (lambda: None),
    )

def stream_bedrock(
    region: str, model_id: str, history: list[dict], system_prompt: str = ""
) -> Iterator[str]:
    """bedrock-runtime の converse_stream を呼び出しテキスト差分を逐次 yield する。"""
    client = bedrock_runtime_client(region)
    params: dict = {
        "modelId": model_id,
        "messages": to_converse_messages(history),
        "inferenceConfig": {"maxTokens": MAX_TOKENS},
    }
    if system_prompt.strip():
        params["system"] = [{"text": system_prompt}]
    response = client.converse_stream(**params)
    for event in response["stream"]:
        if "contentBlockDelta" in event:
            yield event["contentBlockDelta"]["delta"].get("text", "")

"""AWS Bedrock チャットUI デモ

ローカル認証情報（boto3 既定のクレデンシャルチェーン）を使い、bedrock-runtime の
Converse Stream API で Anthropic モデルを呼び出してストリーミング応答する。
サイドバーのモデル / リージョンのセレクタは、アカウントで実際に利用可能なものだけを
ランタイムに検出して表示する。
"""

from collections.abc import Callable, Iterator

import boto3
import streamlit as st
from botocore.config import Config
from botocore.exceptions import (
    BotoCoreError,
    ClientError,
    NoCredentialsError,
    SSOTokenLoadError,
    TokenRetrievalError,
)

# 検出対象とする Bedrock 商用リージョンの候補。利用できないものは検出時に除外する。
CANDIDATE_REGIONS: list[str] = [
    "us-east-1",
    "us-east-2",
    "us-west-2",
    "ap-northeast-1",
    "ap-northeast-2",
    "ap-northeast-3",
    "ap-south-1",
    "ap-southeast-1",
    "ap-southeast-2",
    "ca-central-1",
    "eu-central-1",
    "eu-west-1",
    "eu-west-2",
    "eu-west-3",
    "sa-east-1",
]

# リージョンコードと表示名（場所）の対応
REGION_LABELS: dict[str, str] = {
    "us-east-1": "米国東部（バージニア北部）",
    "us-east-2": "米国東部（オハイオ）",
    "us-west-2": "米国西部（オレゴン）",
    "ap-northeast-1": "アジアパシフィック（東京）",
    "ap-northeast-2": "アジアパシフィック（ソウル）",
    "ap-northeast-3": "アジアパシフィック（大阪）",
    "ap-south-1": "アジアパシフィック（ムンバイ）",
    "ap-southeast-1": "アジアパシフィック（シンガポール）",
    "ap-southeast-2": "アジアパシフィック（シドニー）",
    "ca-central-1": "カナダ（中部）",
    "eu-central-1": "欧州（フランクフルト）",
    "eu-west-1": "欧州（アイルランド）",
    "eu-west-2": "欧州（ロンドン）",
    "eu-west-3": "欧州（パリ）",
    "sa-east-1": "南米（サンパウロ）",
}

# 認証情報そのものが見つからない / 失効している場合に送出される例外
CREDENTIAL_ERRORS = (NoCredentialsError, TokenRetrievalError, SSOTokenLoadError)

# 制御プレーン（モデル一覧の検出）は短いタイムアウトで素早く失敗させる
_DISCOVERY_CONFIG = Config(
    retries={"max_attempts": 2, "mode": "adaptive"},
    connect_timeout=3,
    read_timeout=8,
)
# 推論呼び出しは本番想定のアダプティブリトライ
_RUNTIME_CONFIG = Config(
    retries={"max_attempts": 5, "mode": "adaptive"},
    connect_timeout=5,
    read_timeout=60,
)

# 出力トークン上限は必ず明示的に設定する（未設定はモデル最大値を予約し
# 想定外の ThrottlingException の原因になるため）。
# temperature は新しめの Claude では非対応のため送信しない。
MAX_TOKENS = 1024


# ---------------------------------------------------------------------------
# boto3 クライアント（リージョン単位で 1 つだけ生成して使い回す）
# ---------------------------------------------------------------------------
@st.cache_resource(show_spinner=False)
def bedrock_control_client(region: str):
    """制御プレーン（bedrock）クライアント。モデル一覧の検出に使う。"""
    return boto3.client("bedrock", region_name=region, config=_DISCOVERY_CONFIG)


@st.cache_resource(show_spinner=False)
def bedrock_runtime_client(region: str):
    """データプレーン（bedrock-runtime）クライアント。推論呼び出しに使う。"""
    return boto3.client("bedrock-runtime", region_name=region, config=_RUNTIME_CONFIG)


# ---------------------------------------------------------------------------
# アカウントで実際に利用可能なリージョン / モデルの検出
# ---------------------------------------------------------------------------
@st.cache_data(ttl=3600, show_spinner="利用可能なリージョンを確認中...")
def discover_regions() -> list[str]:
    """候補リージョンを走査し、Anthropic モデルを列挙できたものだけ返す。

    認証情報が無い / 失効している場合は CREDENTIAL_ERRORS を送出する。
    """
    available: list[str] = []
    for region in CANDIDATE_REGIONS:
        try:
            resp = bedrock_control_client(region).list_foundation_models(
                byProvider="anthropic", byOutputModality="TEXT"
            )
        except CREDENTIAL_ERRORS:
            raise
        except (ClientError, BotoCoreError):
            # そのリージョンで Bedrock 非対応 / 権限不足などはスキップ
            continue
        if resp.get("modelSummaries"):
            available.append(region)
    return available


@st.cache_data(ttl=3600, show_spinner="モデルを確認中...")
def discover_models(region: str) -> list[tuple[str, str]]:
    """リージョンで呼び出し可能な Anthropic テキストモデルを返す。

    返り値は (表示名, 呼び出しID) のリスト。クロスリージョン推論プロファイルが
    存在する場合はそのプロファイルIDを優先する（新しめのモデルはプロファイル
    必須のことが多いため）。
    """
    control = bedrock_control_client(region)

    summaries = control.list_foundation_models(
        byProvider="anthropic", byOutputModality="TEXT"
    ).get("modelSummaries", [])

    # ベースモデルID -> 推論プロファイルID の対応表を構築
    profile_for_model: dict[str, str] = {}
    try:
        profiles = control.list_inference_profiles(
            typeEquals="SYSTEM_DEFINED"
        ).get("inferenceProfileSummaries", [])
    except (ClientError, BotoCoreError):
        profiles = []
    for profile in profiles:
        if profile.get("status") != "ACTIVE":
            continue
        for model in profile.get("models", []):
            arn = model.get("modelArn", "")
            base_id = arn.split("foundation-model/")[-1]
            if base_id:
                profile_for_model.setdefault(base_id, profile["inferenceProfileId"])

    models: list[tuple[str, str]] = []
    seen: set[str] = set()
    for fm in summaries:
        if fm.get("modelLifecycle", {}).get("status") != "ACTIVE":
            continue
        if not fm.get("responseStreamingSupported"):
            continue
        base_id = fm["modelId"]
        types = fm.get("inferenceTypesSupported", [])
        if base_id in profile_for_model:
            invoke_id = profile_for_model[base_id]
        elif "ON_DEMAND" in types:
            invoke_id = base_id
        else:
            # オンデマンド非対応かつ対応プロファイルなし -> 呼び出せない
            continue
        if invoke_id in seen:
            continue
        seen.add(invoke_id)
        label = fm.get("modelName") or base_id
        models.append((label, invoke_id))

    models.sort(key=lambda m: m[0])
    return models


# ---------------------------------------------------------------------------
# Converse Stream による応答（st.write_stream に渡すジェネレーター）
# ---------------------------------------------------------------------------
def to_converse_messages(history: list[dict]) -> list[dict]:
    """セッションの会話履歴を Converse API のメッセージ形式へ変換する。"""
    return [
        {"role": m["role"], "content": [{"text": m["content"]}]}
        for m in history
        if m["role"] in ("user", "assistant")
    ]


def stream_bedrock(
    region: str, model_id: str, history: list[dict], system_prompt: str = ""
) -> Iterator[str]:
    """bedrock-runtime の converse_stream を呼び出しテキスト差分を逐次 yield する。"""
    client = bedrock_runtime_client(region)
    params: dict = {
        "modelId": model_id,
        "messages": to_converse_messages(history),
        "inferenceConfig": {"maxTokens": MAX_TOKENS},
    }
    if system_prompt.strip():
        params["system"] = [{"text": system_prompt}]
    response = client.converse_stream(**params)
    for event in response["stream"]:
        if "contentBlockDelta" in event:
            yield event["contentBlockDelta"]["delta"].get("text", "")


# ---------------------------------------------------------------------------
# カスタムコンポーネント (CCv2): 回答メッセージのコピー / 評価ボタン
# ---------------------------------------------------------------------------
# 標準ウィジェットには無いクリップボードコピーと評価操作を双方向コンポーネント
# として実装する。配色は --st-* テーマ変数を参照し、ライト/ダーク・カスタム
# テーマに自動追従する（既定の isolate_styles=True により Shadow DOM 内に隔離）。

_ACTIONS_HTML = """
<div id="actions">
  <button id="copy" type="button" title="回答をコピー">
    <span class="ico">&#x2398;</span><span>コピー</span>
  </button>
  <button id="up" type="button" title="参考になった">&#128077;</button>
  <button id="down" type="button" title="参考にならなかった">&#128078;</button>
</div>
"""

_ACTIONS_CSS = """
#actions { display: flex; gap: 6px; margin-top: 4px; }
#actions button {
  display: inline-flex; align-items: center; gap: 4px;
  font-size: 12px; line-height: 1;
  padding: 5px 9px;
  background: transparent;
  color: var(--st-text-color);
  border: 1px solid var(--st-border-color, rgba(128,128,128,0.35));
  border-radius: var(--st-base-radius, 8px);
  cursor: pointer;
  font-family: var(--st-font, inherit);
}
#actions button:hover { background: var(--st-secondary-background-color); }
#actions button.selected {
  border-color: var(--st-primary-color);
  color: var(--st-primary-color);
  font-weight: 600;
}
#actions .ico { font-size: 13px; }
"""

_ACTIONS_JS = """
export default function (component) {
  const { data, parentElement, setStateValue, setTriggerValue } = component
  const root = parentElement.querySelector("#actions")
  if (!root) return

  const copyBtn = root.querySelector("#copy")
  const upBtn = root.querySelector("#up")
  const downBtn = root.querySelector("#down")

  const fb = (data && data.feedback) || null
  upBtn.classList.toggle("selected", fb === "up")
  downBtn.classList.toggle("selected", fb === "down")

  copyBtn.onclick = async () => {
    try {
      await navigator.clipboard.writeText((data && data.text) || "")
    } catch (e) {
      /* クリップボード権限が無い環境では黙ってスキップ */
    }
    setTriggerValue("copied", true)
  }
  upBtn.onclick = () => setStateValue("feedback", fb === "up" ? null : "up")
  downBtn.onclick = () => setStateValue("feedback", fb === "down" ? null : "down")
}
"""

_MESSAGE_ACTIONS = st.components.v2.component(
    "message_actions",
    html=_ACTIONS_HTML,
    css=_ACTIONS_CSS,
    js=_ACTIONS_JS,
)


def message_actions(
    idx: int,
    text: str,
    feedback: str | None,
    *,
    on_copied: Callable[[], None] | None = None,
    on_feedback: Callable[[], None] | None = None,
):
    """1 件のアシスタント回答に対する操作ボタンを描画する。

    戻り値の result から feedback（state: "up"/"down"/None）と
    copied（trigger: コピー実行時に True）を読み取れる。
    """
    return _MESSAGE_ACTIONS(
        key=f"msg-actions-{idx}",
        data={"text": text, "feedback": feedback},
        on_feedback_change=on_feedback or (lambda: None),
        on_copied_change=on_copied or (lambda: None),
    )


# ---------------------------------------------------------------------------
# ページ設定
# ---------------------------------------------------------------------------
st.set_page_config(page_title="Bedrock チャット", page_icon=":material/smart_toy:")

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

# ---------------------------------------------------------------------------
# サイドバー: モデル / リージョン選択（実利用可能なもののみ）、履歴クリア
# ---------------------------------------------------------------------------
with st.sidebar:
    st.header(":material/settings: 設定")

    credentials_ok = True
    regions: list[str] = []
    try:
        regions = discover_regions()
    except CREDENTIAL_ERRORS:
        credentials_ok = False

    if not credentials_ok:
        st.error(
            "ローカル AWS 認証情報が見つからない、または失効しています。"
            "`aws configure` / SSO ログイン / 環境変数などで認証情報を設定してください。"
        )
        model_name = model_id = region = None
    elif not regions:
        st.error(
            "Bedrock で Anthropic モデルを利用できるリージョンが見つかりませんでした。"
            "アカウントのモデルアクセス設定をご確認ください。"
        )
        model_name = model_id = region = None
    else:
        default_region = boto3.Session().region_name
        region_index = regions.index(default_region) if default_region in regions else 0
        region = st.selectbox(
            "リージョン（場所）",
            options=regions,
            index=region_index,
            format_func=lambda r: f"{REGION_LABELS.get(r, r)} ({r})",
            help="アカウントで Anthropic モデルを検出できたリージョンのみ表示",
        )

        models = discover_models(region)
        if not models:
            st.warning("このリージョンで呼び出し可能な Anthropic モデルがありません。")
            model_name = model_id = None
        else:
            labels = [m[0] for m in models]
            model_name = st.selectbox(
                "推論モデル",
                options=labels,
                help="選択リージョンでストリーミング推論可能な Anthropic モデルのみ表示",
            )
            model_id = dict(models)[model_name]
            st.caption(f"呼び出しID: `{model_id}`")

    st.text_area(
        "システムプロンプト",
        key="system_prompt",
        placeholder="例: あなたは簡潔に回答する日本語アシスタントです。",
        help="モデルに与える役割や制約。空欄の場合は送信しません。",
    )

    st.divider()

    if st.button(
        "会話履歴をクリア",
        icon=":material/delete_sweep:",
        width="stretch",
        disabled=not st.session_state.messages,
    ):
        st.session_state.messages = []
        st.rerun()

# ---------------------------------------------------------------------------
# メイン: チャット画面
# ---------------------------------------------------------------------------
st.title(":material/smart_toy: Bedrock チャット")
st.caption("AWS Bedrock 経由で Anthropic モデルと対話します")

# 既存の会話履歴を表示。アシスタント回答にはコピー/評価ボタンを付与する。
for i, msg in enumerate(st.session_state.messages):
    with st.chat_message(msg["role"]):
        if msg.get("error"):
            st.error(msg["content"])
        else:
            st.write(msg["content"])
        if msg["role"] == "assistant" and not msg.get("error"):
            result = message_actions(i, msg["content"], msg.get("feedback"))
            if result.feedback != msg.get("feedback"):
                msg["feedback"] = result.feedback
            if result.copied:
                st.toast("回答をコピーしました", icon=":material/content_copy:")

# モデル選択ができている場合のみ入力を受け付ける
ready = model_id is not None
if prompt := st.chat_input("メッセージを入力してください", disabled=not ready):
    st.session_state.messages.append({"role": "user", "content": prompt})
    with st.chat_message("user"):
        st.write(prompt)

    runtime = bedrock_runtime_client(region)
    with st.chat_message("assistant"):
        try:
            response = st.write_stream(
                stream_bedrock(
                    region,
                    model_id,
                    st.session_state.messages,
                    st.session_state.get("system_prompt", ""),
                )
            )
            st.session_state.messages.append(
                {"role": "assistant", "content": response, "feedback": None}
            )
        except runtime.exceptions.AccessDeniedException:
            st.session_state.messages.append({
                "role": "assistant",
                "content": (
                    "このモデルへのアクセスが有効化されていません。"
                    "Bedrock のモデルアクセス設定でこのモデルを有効化してください。"
                ),
                "error": True,
            })
        except runtime.exceptions.ValidationException as e:
            st.session_state.messages.append({
                "role": "assistant",
                "content": f"リクエストが不正です: {e.response['Error']['Message']}",
                "error": True,
            })
        except ClientError as e:
            err = e.response["Error"]
            st.session_state.messages.append({
                "role": "assistant",
                "content": f"Bedrock 呼び出しエラー: {err['Code']} - {err['Message']}",
                "error": True,
            })
    st.rerun()

# サイドバーにフィードバック集計を表示（履歴処理後の最新値で算出）
up = sum(1 for m in st.session_state.messages if m.get("feedback") == "up")
down = sum(1 for m in st.session_state.messages if m.get("feedback") == "down")
if up or down:
    with st.sidebar:
        st.divider()
        st.caption("フィードバック")
        c1, c2 = st.columns(2)
        c1.metric("良い", up)
        c2.metric("いまいち", down)