エージェントループの概要

エージェントループ（loop.py）は、Computer Use Demoの中核となるコンポーネントです。このモジュールは、以下の重要な役割を担っています：

Claude APIとの通信制御
ツールの実行管理
メッセージの履歴管理
結果のコールバック処理

主要なコンポーネント

APIプロバイダーの定義

class APIProvider(StrEnum):
    ANTHROPIC = "anthropic"
    BEDROCK = "bedrock"
    VERTEX = "vertex"

# 各プロバイダーのデフォルトモデル名の定義
PROVIDER_TO_DEFAULT_MODEL_NAME: dict[APIProvider, str] = {
    APIProvider.ANTHROPIC: "claude-3-5-sonnet-20241022",
    APIProvider.BEDROCK: "anthropic.claude-3-5-sonnet-20241022-v2:0",
    APIProvider.VERTEX: "claude-3-5-sonnet-v2@20241022",
}

このコードでは：

複数のAPIプロバイダー（Anthropic直接、AWS Bedrock、Google Vertex AI）をサポート
各プロバイダーに対応するデフォルトのモデル名を定義

システムプロンプトの設定

SYSTEM_PROMPT = f"""<SYSTEM_CAPABILITY>
* You are utilising an Ubuntu virtual machine using {platform.machine()} architecture with internet access.
* You can feel free to install Ubuntu applications with your bash tool. Use curl instead of wget.
* To open firefox, please just click on the firefox icon.  Note, firefox-esr is what is installed on your system.
* Using bash tool you can start GUI applications, but you need to set export DISPLAY=:1 and use a subshell...
[...]
</SYSTEM_CAPABILITY>
"""

システムプロンプトでは：

利用可能な環境の説明
使用可能なツールの説明
制約事項や推奨事項の提示
重要な注意事項の明記

日本語版

# このシステムプロンプトは、このリポジトリのDocker環境と
# 有効化されている特定のツールの組み合わせに最適化されています。
# モデルが実行環境のコンテキストを確実に理解し、
# タスクに役立つ可能性のある追加情報を提供するために、
# このシステムプロンプトの修正をお勧めします。

SYSTEM_PROMPT = f"""<システム機能>
* あなたはインターネットアクセス可能な{platform.machine()}アーキテクチャのUbuntu仮想マシンを使用しています。
* bashツールを使用してUbuntuアプリケーションを自由にインストールできます。wgetの代わりにcurlを使用してください。
* Firefoxを開くには、Firefoxアイコンをクリックするだけです。なお、システムにはfirefox-esrがインストールされています。
* bashツールを使用してGUIアプリケーションを起動できますが、export DISPLAY=:1を設定し、サブシェルを使用する必要があります。例：「(DISPLAY=:1 xterm &)」。bashツールで実行されるGUIアプリはデスクトップ環境内に表示されますが、表示されるまでに時間がかかる場合があります。スクリーンショットを撮って確認してください。
* 大量のテキスト出力が予想されるコマンドをbashツールで使用する場合は、一時ファイルにリダイレクトし、str_replace_editorまたは`grep -n -B <前の行数> -A <後の行数> <検索語> <ファイル名>`を使用して出力を確認してください。
* ページを表示する際は、ページ全体を見渡せるようにズームアウトすると便利です。または、何かが利用できないと判断する前に、必ずスクロールダウンしてすべてを確認してください。
* コンピュータ関数呼び出しを使用する際、実行と結果の送信に時間がかかります。可能な限り、複数の呼び出しを1つの関数呼び出しリクエストにまとめるようにしてください。
* 現在の日付は{datetime.today().strftime('%A, %B %-d, %Y')}です。
</システム機能>

<重要>
* Firefoxを使用する際、起動ウィザードが表示されても無視してください。「このステップをスキップ」もクリックしないでください。代わりに、「検索またはアドレスを入力」と表示されているアドレスバーをクリックし、適切な検索語やURLを入力してください。
* PDFを閲覧している場合、PDFの単一のスクリーンショットを撮った後、スクリーンショットとナビゲーションでPDFを読み続けるのではなく文書全体を読みたい場合は、URLを特定し、curlを使用してPDFをダウンロードし、pdftotextをインストールして使用してテキストファイルに変換し、そのテキストファイルをStrReplaceEditToolで直接読み取ってください。
</重要>"""

メインのサンプリングループ

async def sampling_loop(
    *,
    model: str,                     # 使用するモデル名
    provider: APIProvider,          # APIプロバイダー
    system_prompt_suffix: str,      # システムプロンプトの追加部分
    messages: list[BetaMessageParam], # メッセージ履歴
    output_callback: Callable,      # 出力用コールバック
    tool_output_callback: Callable, # ツール出力用コールバック
    api_response_callback: Callable, # API応答用コールバック
    api_key: str,                  # APIキー
    only_n_most_recent_images: int | None = None,  # 保持する画像の数
    max_tokens: int = 4096,        # 最大トークン数
):
    # ツールコレクションの初期化
    tool_collection = ToolCollection(
        ComputerTool(),  # コンピュータ操作用ツール
        BashTool(),      # Bashコマンド実行用ツール
        EditTool(),      # ファイル編集用ツール
    )
    
    # システムプロンプトの設定
    system = f"{SYSTEM_PROMPT}{' ' + system_prompt_suffix if system_prompt_suffix else ''}"

    while True:
        # 古い画像の削除処理
        if only_n_most_recent_images:
            _maybe_filter_to_n_most_recent_images(messages, only_n_most_recent_images)

        # APIクライアントの初期化
        if provider == APIProvider.ANTHROPIC:
            client = Anthropic(api_key=api_key)
        elif provider == APIProvider.VERTEX:
            client = AnthropicVertex()
        elif provider == APIProvider.BEDROCK:
            client = AnthropicBedrock()

        # APIリクエストの送信
        raw_response = client.beta.messages.with_raw_response.create(
            max_tokens=max_tokens,
            messages=messages,
            model=model,
            system=system,
            tools=tool_collection.to_params(),
            betas=["computer-use-2024-10-22"],
        )

        # コールバック処理
        api_response_callback(cast(APIResponse[BetaMessage], raw_response))
        response = raw_response.parse()

        # アシスタントのメッセージを履歴に追加
        messages.append({
            "role": "assistant",
            "content": cast(list[BetaContentBlockParam], response.content),
        })

        # ツール実行結果の処理
        tool_result_content: list[BetaToolResultBlockParam] = []
        for content_block in cast(list[BetaContentBlock], response.content):
            output_callback(content_block)
            if content_block.type == "tool_use":
                # ツールの実行と結果の取得
                result = await tool_collection.run(
                    name=content_block.name,
                    tool_input=cast(dict[str, Any], content_block.input),
                )
                tool_result_content.append(
                    _make_api_tool_result(result, content_block.id)
                )
                tool_output_callback(result, content_block.id)

        # ツール実行がない場合はループ終了
        if not tool_result_content:
            return messages

        # ツール実行結果をユーザーメッセージとして追加
        messages.append({"content": tool_result_content, "role": "user"})

補助機能の実装

画像フィルタリング

def _maybe_filter_to_n_most_recent_images(
    messages: list[BetaMessageParam],
    images_to_keep: int,
    min_removal_threshold: int = 10,
):
    """
    会話の進行に伴い価値が低下するスクリーンショットについて、
    指定された数だけ最新のものを残して削除する
    """
    if images_to_keep is None:
        return messages

    # ツール実行結果ブロックの取得
    tool_result_blocks = cast(
        list[ToolResultBlockParam],
        [
            item
            for message in messages
            for item in (
                message["content"] if isinstance(message["content"], list) else []
            )
            if isinstance(item, dict) and item.get("type") == "tool_result"
        ],
    )

    # 画像の総数をカウント
    total_images = sum(
        1
        for tool_result in tool_result_blocks
        for content in tool_result.get("content", [])
        if isinstance(content, dict) and content.get("type") == "image"
    )

    # 削除する画像数の計算
    images_to_remove = total_images - images_to_keep
    # キャッシュ効率のため、削除はチャンク単位で行う
    images_to_remove -= images_to_remove % min_removal_threshold

    # 古い画像の削除処理
    for tool_result in tool_result_blocks:
        if isinstance(tool_result.get("content"), list):
            new_content = []
            for content in tool_result.get("content", []):
                if isinstance(content, dict) and content.get("type") == "image":
                    if images_to_remove > 0:
                        images_to_remove -= 1
                        continue
                new_content.append(content)
            tool_result["content"] = new_content

ツール実行結果の変換

def _make_api_tool_result(
    result: ToolResult, 
    tool_use_id: str
) -> BetaToolResultBlockParam:
    """エージェントのToolResultをAPI用のToolResultBlockParamに変換"""
    tool_result_content: list[BetaTextBlockParam | BetaImageBlockParam] | str = []
    is_error = False

    # エラー処理
    if result.error:
        is_error = True
        tool_result_content = _maybe_prepend_system_tool_result(result, result.error)
    else:
        # 通常の出力処理
        if result.output:
            tool_result_content.append({
                "type": "text",
                "text": _maybe_prepend_system_tool_result(result, result.output),
            })
        if result.base64_image:
            tool_result_content.append({
                "type": "image",
                "source": {
                    "type": "base64",
                    "media_type": "image/png",
                    "data": result.base64_image,
                },
            })

    return {
        "type": "tool_result",
        "content": tool_result_content,
        "tool_use_id": tool_use_id,
        "is_error": is_error,
    }

エージェントループの動作フロー

全体の処理フロー

以下の図は、エージェントループの主要な処理フローを示しています：

メッセージとツールの連携

以下のシーケンス図は、ユーザー、エージェントループ、Claude API、およびツール間の相互作用を示しています：

処理の詳細

初期化フェーズ
- ツールコレクションの作成
- システムプロンプトの設定
- APIクライアントの準備
メインループ
- 古い画像の削除（設定されている場合）
- APIリクエストの送信
- レスポンスの処理
- ツール実行の管理
- 結果の履歴への追加
終了条件
- ツール実行がない場合にループを終了
- 最終的なメッセージ履歴を返却

エラーハンドリングと最適化

画像管理の最適化
- 古い画像の効率的な削除
- キャッシュ効率を考慮したチャンク単位の処理
エラー処理
- ツール実行時のエラーハンドリング
- API通信のエラー処理
- システムメッセージの適切な付与

拡張性と保守性

エージェントループは以下の点で拡張性が高い設計となっています：

APIプロバイダーの追加
- 新しいプロバイダーの追加が容易
- プロバイダー固有の設定の管理が可能
ツールの追加
- ToolCollectionを通じた新規ツールの追加
- 既存ツールの機能拡張
コールバックの活用
- 出力処理のカスタマイズ
- 監視やログ機能の追加

まとめ

エージェントループ（loop.py）は、Computer Use Demoの中核として以下の特徴を持ちます：

柔軟なAPI対応
効率的なメッセージ管理
堅牢なツール実行制御
適切なエラーハンドリング
高い拡張性

これらの特徴により、Claude 3.5 Sonnetが安定的かつ効率的にコンピュータを操作することが可能となっています。