エージェントループの概要
エージェントループ（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,
}
エージェントループの動作フロー
全体の処理フロー
以下の図は、エージェントループの主要な処理フローを示しています：
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、およびツール間の相互作用を示しています：
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: 最終結果表示
処理の詳細
-
初期化フェーズ
-
ツールコレクションの作成
-
システムプロンプトの設定
-
APIクライアントの準備
-
メインループ
-
古い画像の削除（設定されている場合）
-
APIリクエストの送信
-
レスポンスの処理
-
ツール実行の管理
-
結果の履歴への追加
-
終了条件
-
ツール実行がない場合にループを終了
-
最終的なメッセージ履歴を返却
エラーハンドリングと最適化
-
画像管理の最適化
-
古い画像の効率的な削除
-
キャッシュ効率を考慮したチャンク単位の処理
-
エラー処理
-
ツール実行時のエラーハンドリング
-
API通信のエラー処理
-
システムメッセージの適切な付与
拡張性と保守性
エージェントループは以下の点で拡張性が高い設計となっています：
-
APIプロバイダーの追加
-
新しいプロバイダーの追加が容易
-
プロバイダー固有の設定の管理が可能
-
ツールの追加
-
ToolCollectionを通じた新規ツールの追加
-
既存ツールの機能拡張
-
コールバックの活用
-
出力処理のカスタマイズ
-
監視やログ機能の追加
まとめ
エージェントループ（loop.py）は、Computer Use Demoの中核として以下の特徴を持ちます：
- 柔軟なAPI対応
- 効率的なメッセージ管理
- 堅牢なツール実行制御
- 適切なエラーハンドリング
- 高い拡張性
これらの特徴により、Claude 3.5 Sonnetが安定的かつ効率的にコンピュータを操作することが可能となっています。
コメント