前回の記事では、Claude 3.7 Sonnetの思考プロセスを可視化するStreamlitアプリケーションを作成しました。今回は、そのアプリケーションをさらに発展させ、Web検索機能を追加した実装について解説します。

技術スタックの概要

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

Streamlit - Pythonベースの対話型Webアプリケーションフレームワーク
Amazon Bedrock - AWSのフルマネージド型生成AIサービス
Claude 3.7 Sonnet - Anthropic社の最新LLMモデル
Tavily API - Web検索機能を提供するAPIサービス

前回からの主な拡張ポイント

Web検索機能の追加: Tavily APIを利用したWeb検索機能をClaude 3.7に統合
ツール使用の可視化: モデルがWeb検索ツールを使用する過程を可視化
複数ターンのツール使用: モデルが複数回検索を行うことができるツールループ実装

これらの機能追加により、モデルはリアルタイムで最新の情報にアクセスし、より正確な回答を提供できるようになりました。

コード解説

1. ツール関連の実装

今回の実装で特に重要なのがツール（Web検索）機能の実装部分です。

ここでは次のことを行っています。

web_search関数の定義：Tavily APIを使用してWeb検索を実行する関数
ツールのマッピング：ツール名と関数の関連付け
ツール設定の定義：Amazon Bedrock APIに渡すツール仕様の設定

def web_search(query: str) -> dict:
    """
    Tavilyクライアントを使用してWeb検索を実行する関数

    Args:
        query: 検索クエリ文字列

    Returns:
        検索結果を含む辞書
    """
    tavily_client = TavilyClient()
    return tavily_client.search(query)


# 利用可能なツールのマッピング
tools = {"web_search": web_search}

# ツール設定
tool_config = {
    "tools": [
        {
            "toolSpec": {
                "name": "web_search",
                "description": "Web Search",
                "inputSchema": {
                    "json": {
                        "type": "object",
                        "properties": {
                            "query": {
                                "type": "string",
                                "description": "Search query",
                            }
                        },
                        "required": ["query"],
                    }
                },
            }
        }
    ]
}

2. ストリームレスポンスの処理

前回のアプリケーションと比較して、ストリームレスポンスの処理がより複雑になっています。ここでは、テキスト、思考プロセス、そしてツール使用情報も処理する必要があります。

このメソッドでは、レスポンスストリームから以下の情報を抽出・処理しています。

テキスト応答
思考プロセス（reasoning）
ツール使用情報（toolUse）

それぞれの情報タイプに応じて、Streamlitの異なるUIコンポーネントに表示しています。

def process_stream(response):
    """
    Bedrockからのストリームレスポンスを処理する関数

    Args:
        response: Bedrockからのレスポンスストリーム

    Returns:
        処理されたメッセージオブジェクト
    """
    content = []
    message = {"content": content}
    text = ""
    tool_use = {}
    reasoning = {}

    # UI出力要素を管理する辞書
    st_out = {}

    for chunk in response["stream"]:
        # メッセージ開始情報の処理
        if "messageStart" in chunk:
            message["role"] = chunk["messageStart"]["role"]

        # コンテンツブロック開始情報の処理
        elif "contentBlockStart" in chunk:
            tool = chunk["contentBlockStart"]["start"]["toolUse"]
            tool_use["toolUseId"] = tool["toolUseId"]
            tool_use["name"] = tool["name"]

        # コンテンツブロックのデルタ情報処理
        elif "contentBlockDelta" in chunk:
            # ...（省略）...

3. ツールループの実装

チャットボットがWeb検索を行い、その結果を利用して回答するための「ツールループ」を実装しています。

このループの重要なポイントは以下の通りです。

モデルからのレスポンスにToolUseが含まれているかチェック
ToolUseがある場合、そのツールを実行し結果を取得
ツールの実行結果をメッセージ履歴に追加
モデルに再度クエリを実行
ToolUseがなくなるまで繰り返す

これにより、モデルは必要に応じて複数回のWeb検索を行い、より正確で情報に基づいた回答を生成することができます。

# ユーザー入力の処理
if prompt := st.chat_input():
    # ユーザーメッセージの表示
    with st.chat_message("user"):
        st.write(prompt)

    # ユーザーメッセージの追加
    messages.append({"role": "user", "content": [{"text": prompt}]})

    # ツール使用のループ処理
    while True:
        # Bedrockモデルへのリクエスト
        response = client.converse_stream(
            modelId=MODEL_ID,
            messages=messages,
            toolConfig=tool_config,
            additionalModelRequestFields={
                "thinking": {
                    "type": "enabled",
                    "budget_tokens": 1024,
                },
            },
        )

        # レスポンスの処理とメッセージへの追加
        message = process_stream(response)
        messages.append(message)

        # ツール使用コンテンツのフィルタリング
        tool_use_content = list(
            filter(lambda x: "toolUse" in x.keys(), message["content"])
        )

        # ツール使用がなければループを終了
        if len(tool_use_content) == 0:
            break

        # 各ツール使用の処理
        for content in tool_use_content:
            tool_use_id = content["toolUse"]["toolUseId"]
            name = content["toolUse"]["name"]
            input = content["toolUse"]["input"]

            # ツールの実行と結果の取得
            result = tools[name](**input)

            # ツール結果メッセージの作成
            tool_result = {
                "toolUseId": tool_use_id,
                "content": [{"text": json.dumps(result, ensure_ascii=False)}],
            }

            # ツール結果メッセージの追加
            tool_result_message = {
                "role": "user",
                "content": [{"toolResult": tool_result}],
            }
            messages.append(tool_result_message)

実装の技術的なポイント

1. ToolUseの統合

Amazon BedrockのToolUseを利用することで、モデルの判断でWeb検索を実行できるようにしています。toolConfigパラメータを使用してツールの仕様を定義し、モデルがToolの使用有無を判断できるようにしています。

2. 複雑なストリーミングレスポンスの処理

このアプリケーションでは、Amazon Bedrockから返される複雑なストリーミングレスポンスを適切に処理する必要があります。レスポンスには、テキスト、思考プロセス、ToolUseが含まれており、それぞれを適切に抽出して表示しています。

3. 対話的なユーザーエクスペリエンス

Streamlitのst.empty()やst.expander()などの機能を活用して、ユーザーに対してモデルの思考プロセスやツール使用を視覚的に提示しています。これにより、モデルがどのように情報を収集し、それをどう処理しているかを透明に表示することができます。

4. 効率的なツールループ

WhileループとBedrockのconverse_stream APIを組み合わせることで、モデルが必要に応じて複数回のWeb検索を行うことができる効率的なツールループを実装しています。これはモデルが自律的に情報収集を行い、より充実した回答を提供するために重要な機能です。

まとめと今後の発展可能性

今回実装したWeb検索機能付きのClaude 3.7 Sonnetチャットアプリケーションは、単なるチャットボットを超えた、情報収集能力を持つインテリジェントなアシスタントへの一歩となっています。思考プロセスの可視化とWeb検索機能の組み合わせにより、ユーザーはモデルの推論過程を確認しながら、最新の情報に基づいた回答を得ることができます。

Amazon BedrockとStreamlitを組み合わせることで、比較的少ないコード量で高機能なAIアプリケーションを実現できることを示せたと思います。Claude 3.7 Sonnetの高度な思考能力と外部情報へのアクセス能力を組み合わせることで、実用的で透明性の高いAIアシスタントを構築することができました。

完全なソースコード

import json

import boto3
import streamlit as st
from tavily import TavilyClient


# Bedrock モデルの設定
MODEL_ID = "us.anthropic.claude-3-7-sonnet-20250219-v1:0"
client = boto3.client("bedrock-runtime")

# ==============================
# ツール関連の実装
# ==============================


def web_search(query: str) -> dict:
    """
    Tavilyクライアントを使用してWeb検索を実行する関数

    Args:
        query: 検索クエリ文字列

    Returns:
        検索結果を含む辞書
    """
    tavily_client = TavilyClient()
    return tavily_client.search(query)


# 利用可能なツールのマッピング
tools = {"web_search": web_search}

# ツール設定
tool_config = {
    "tools": [
        {
            "toolSpec": {
                "name": "web_search",
                "description": "Web Search",
                "inputSchema": {
                    "json": {
                        "type": "object",
                        "properties": {
                            "query": {
                                "type": "string",
                                "description": "Search query",
                            }
                        },
                        "required": ["query"],
                    }
                },
            }
        }
    ]
}


def process_stream(response):
    """
    Bedrockからのストリームレスポンスを処理する関数

    Args:
        response: Bedrockからのレスポンスストリーム

    Returns:
        処理されたメッセージオブジェクト
    """
    content = []
    message = {"content": content}
    text = ""
    tool_use = {}
    reasoning = {}

    # UI出力要素を管理する辞書
    st_out = {}

    for chunk in response["stream"]:
        # メッセージ開始情報の処理
        if "messageStart" in chunk:
            message["role"] = chunk["messageStart"]["role"]

        # コンテンツブロック開始情報の処理
        elif "contentBlockStart" in chunk:
            tool = chunk["contentBlockStart"]["start"]["toolUse"]
            tool_use["toolUseId"] = tool["toolUseId"]
            tool_use["name"] = tool["name"]

        # コンテンツブロックのデルタ情報処理
        elif "contentBlockDelta" in chunk:
            delta = chunk["contentBlockDelta"]["delta"]
            index = str(chunk["contentBlockDelta"]["contentBlockIndex"])

            # ツール使用情報の処理
            if "toolUse" in delta:
                if "input" not in tool_use:
                    tool_use["input"] = ""
                tool_use["input"] += delta["toolUse"]["input"]

                # UIへのツール使用情報の表示
                if index not in st_out:
                    st_out[index] = st.expander("Tool use...", expanded=False).empty()
                st_out[index].write(tool_use["input"])

            # テキスト情報の処理
            elif "text" in delta:
                text += delta["text"]

                # UIへのテキスト情報の表示
                if index not in st_out:
                    st_out[index] = st.chat_message("assistant").empty()
                st_out[index].write(text)

            # 推論内容の処理
            if "reasoningContent" in delta:
                if "reasoningText" not in reasoning:
                    reasoning["reasoningText"] = {"text": "", "signature": ""}

                if "text" in delta["reasoningContent"]:
                    reasoning["reasoningText"]["text"] += delta["reasoningContent"][
                        "text"
                    ]
                if "signature" in delta["reasoningContent"]:
                    reasoning["reasoningText"]["signature"] = delta["reasoningContent"][
                        "signature"
                    ]

                # UIへの推論内容の表示
                if index not in st_out:
                    st_out[index] = st.expander("Thinking...", expanded=True).empty()
                st_out[index].write(reasoning["reasoningText"]["text"])

        # コンテンツブロック終了情報の処理
        elif "contentBlockStop" in chunk:
            # ツール使用情報のコンテンツへの追加
            if "input" in tool_use:
                tool_use["input"] = json.loads(tool_use["input"])
                content.append({"toolUse": tool_use})
                tool_use = {}
            # 推論内容のコンテンツへの追加
            elif "reasoningText" in reasoning:
                content.append({"reasoningContent": reasoning})
                reasoning = {}
            # テキスト情報のコンテンツへの追加
            else:
                content.append({"text": text})
                text = ""

        # メッセージ終了情報の処理
        elif "messageStop" in chunk:
            stop_reason = chunk["messageStop"]["stopReason"]

    return message


# ==============================
# Streamlit UIの実装
# ==============================

st.title("Claude 3.7 Sonnet on Bedrock")
st.subheader("extended thinking with web search")

# セッション状態の初期化
if "messages" not in st.session_state:
    st.session_state.messages = []
messages = st.session_state.messages

# 過去のメッセージの表示
for message in messages:
    # テキストコンテンツのみをフィルタリング
    text_content = list(filter(lambda x: "text" in x.keys(), message["content"]))

    for content in text_content:
        with st.chat_message(message["role"]):
            st.write(content["text"])

# ユーザー入力の処理
if prompt := st.chat_input():
    # ユーザーメッセージの表示
    with st.chat_message("user"):
        st.write(prompt)

    # ユーザーメッセージの追加
    messages.append({"role": "user", "content": [{"text": prompt}]})

    # ツール使用のループ処理
    while True:
        # Bedrockモデルへのリクエスト
        response = client.converse_stream(
            modelId=MODEL_ID,
            messages=messages,
            toolConfig=tool_config,
            additionalModelRequestFields={
                "thinking": {
                    "type": "enabled",
                    "budget_tokens": 1024,
                },
            },
        )

        # レスポンスの処理とメッセージへの追加
        message = process_stream(response)
        messages.append(message)

        # ツール使用コンテンツのフィルタリング
        tool_use_content = list(
            filter(lambda x: "toolUse" in x.keys(), message["content"])
        )

        # ツール使用がなければループを終了
        if len(tool_use_content) == 0:
            break

        # 各ツール使用の処理
        for content in tool_use_content:
            tool_use_id = content["toolUse"]["toolUseId"]
            name = content["toolUse"]["name"]
            input = content["toolUse"]["input"]

            # ツールの実行と結果の取得
            result = tools[name](**input)

            # ツール結果メッセージの作成
            tool_result = {
                "toolUseId": tool_use_id,
                "content": [{"text": json.dumps(result, ensure_ascii=False)}],
            }

            # ツール結果メッセージの追加
            tool_result_message = {
                "role": "user",
                "content": [{"toolResult": tool_result}],
            }
            messages.append(tool_result_message)