Claude Computer Use Demo - エージェントループの詳細解説

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

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

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

## 主要なコンポーネント

### APIプロバイダーの定義

```python
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）をサポート
- 各プロバイダーに対応するデフォルトのモデル名を定義

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

```python
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>
"""
```

システムプロンプトでは：
- 利用可能な環境の説明
- 使用可能なツールの説明
- 制約事項や推奨事項の提示
- 重要な注意事項の明記

#### 日本語版

```python
# このシステムプロンプトは、このリポジトリの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で直接読み取ってください。
</重要>"""
```

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

```python
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"})
```

### 補助機能の実装

#### 画像フィルタリング
```python
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
```

#### ツール実行結果の変換
```python
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,
    }
```

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

### 全体の処理フロー

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

```mermaid
flowchart TD
    subgraph Init ["初期化フェーズ"]
        A[開始] --> B[ツールコレクションの初期化]
        B --> C[システムプロンプトの設定]
        C --> D[APIクライアントの準備]
    end

    subgraph MainLoop ["メインループ"]
        E[古い画像の削除] --> F[APIリクエスト送信]
        F --> G[レスポンスの受信と解析]
        G --> H{ツール使用要求?}
        H -- Yes --> I[ツール実行]
        I --> J[結果を履歴に追加]
        J --> E
        H -- No --> K[会話終了]
    end
    
    D --> E
    K --> L[終了]

    subgraph ToolExecution ["ツール実行フロー"]
        I1[ツール名の特定] --> I2[入力パラメータの検証]
        I2 --> I3[ツールの実行]
        I3 --> I4[結果の変換]
        I4 --> I5[コールバック処理]
    end

```

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

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

```mermaid
sequenceDiagram
    participant U as User
    participant L as Loop
    participant A as Claude API
    participant T as Tools

    U->>L: ユーザーメッセージ
    L->>A: メッセージ履歴送信
    A-->>L: レスポンス
    
    rect rgb(240, 240, 255)
        Note over L,T: ツール実行フェーズ
        loop ツール使用要求がある場合
            L->>T: ツール実行要求
            T-->>L: 実行結果
            L->>A: ツール実行結果送信
            A-->>L: 次のアクション
        end
    end
    
    L-->>U: 最終結果表示
```

### 処理の詳細

1. **初期化フェーズ**
   - ツールコレクションの作成
   - システムプロンプトの設定
   - APIクライアントの準備

2. **メインループ**
   - 古い画像の削除（設定されている場合）
   - APIリクエストの送信
   - レスポンスの処理
   - ツール実行の管理
   - 結果の履歴への追加

3. **終了条件**
   - ツール実行がない場合にループを終了
   - 最終的なメッセージ履歴を返却

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

1. **画像管理の最適化**
   - 古い画像の効率的な削除
   - キャッシュ効率を考慮したチャンク単位の処理

2. **エラー処理**
   - ツール実行時のエラーハンドリング
   - API通信のエラー処理
   - システムメッセージの適切な付与

## 拡張性と保守性

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

1. **APIプロバイダーの追加**
   - 新しいプロバイダーの追加が容易
   - プロバイダー固有の設定の管理が可能

2. **ツールの追加**
   - ToolCollectionを通じた新規ツールの追加
   - 既存ツールの機能拡張

3. **コールバックの活用**
   - 出力処理のカスタマイズ
   - 監視やログ機能の追加

## まとめ

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

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

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