【調査レポ10】Google Antigravity SDKでAIチームを結成！サブエージェント（subagents）による役割分担と協調ワークフロー

import os
import sys
import asyncio
import datetime

# 1. 安全対策：python-dotenv ライブラリの検出と環境変数の読み込み
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    print("注意: python-dotenv がインストールされていません。環境変数から直接APIキーを読み込みます。")

# 2. Google Antigravity SDK のインポート
try:
    from google.antigravity import Agent, LocalAgentConfig, types
except ImportError:
    print("エラー: google.antigravity パッケージが見つかりません。pip install google-antigravity で導入してください。", file=sys.stderr)
    sys.exit(1)

# 3. 専門ツールの定義

# 【リサーチ担当ツール】サーバーの生メトリクス・ログを収集するツール
def fetch_server_metrics(target_server: str) -> str:
    """指定されたサーバーの稼働メトリクスおよび最新ログを取得するツール。

    Args:
        target_server (str): 調査対象のサーバー名 (例: 'web-prod-01')

    Returns:
        str: サーバーの稼働データ
    """
    return (
        f"【サーバー '{target_server}' 調査データ】\n"
        f"- CPU平均使用率: 88% (ピーク時 96%)\n"
        f"- メモリ使用率: 82% (16GB中 13.1GB使用)\n"
        f"- DB接続プール待機数: 45件 (高負荷傾向)\n"
        f"- レスポンス遅延ログ: 'DB_CONNECTION_TIMEOUT' が過去1時間に12件発生"
    )

# 【データ分析担当ツール】メトリクスを解析してボトルネックを算出するツール
def calculate_optimization_impact(metric_type: str, current_value: str) -> str:
    """メトリクスデータを分析し、改善施策と期待されるパフォーマンス向上率を算出するツール。

    Args:
        metric_type (str): 分析対象のメトリクス種別 (例: 'db_pool', 'memory')
        current_value (str): 現在の測定値や課題

    Returns:
        str: 分析結果と改善インパクト
    """
    return (
        f"【最適化シミュレーション結果 - {metric_type}】\n"
        f"・主因: DBコネクションプールの枯渇によるリクエスト滞留\n"
        f"・推奨施策: コネクションプール上限を 50 ➔ 120 に拡張、キャッシュ層(Redis)の導入\n"
        f"・期待効果: 応答遅延の 75% 削減、CPU負荷の 20% 改善が見込めます。"
    )

# 4. メイン処理：subagents パラメータを活用したAIチーム編成
async def main():
    api_key = os.environ.get("GEMINI_API_KEY")
    if not api_key:
        print("エラー: GEMINI_API_KEY が設定されていません。.env ファイルを確認してください。", file=sys.stderr)
        return

    print("=== Google Antigravity SDK subagents パラメータによるAIチーム協調デモを開始します ===\n")
    user_request = "本番Webサーバー 'web-prod-01' で発生している高負荷の原因を調査し、システム改善提案書を作成してください。"
    print(f"【ユーザーからの依頼】: {user_request}\n")

    # ─── 1. サブエージェントの定義（types.SubagentConfig） ───
    # リサーチスペシャリスト（情報収集に特化）
    researcher_subagent = types.SubagentConfig(
        name="server_researcher",
        description="対象サーバーの稼働メトリクスやエラーログの生データを調査・収集する専門サブエージェント",
        system_instructions=(
            "あなたはシステム調査専門のリサーチスペシャリストです。"
            "fetch_server_metrics ツールを活用し、サーバーの稼働データやエラーログを正確に収集して報告してください。"
        ),
        tools=[fetch_server_metrics]
    )

    # データアナリスト（分析・シミュレーションに特化）
    analyst_subagent = types.SubagentConfig(
        name="performance_analyst",
        description="収集されたサーバーメトリクスを分析し、最適化シミュレーションと改善施策を算出する専門サブエージェント",
        system_instructions=(
            "あなたはシステム性能分析専門のデータアナリストです。"
            "calculate_optimization_impact ツールを活用してボトルネックの特定と改善効果を論理的に算出してください。"
        ),
        tools=[calculate_optimization_impact]
    )

    # ─── 2. 親エージェントの定義（LocalAgentConfig の subagents パラメータに一括登録） ───
    # ※サブエージェントが使用する全カスタムツールは親の tools リストにも登録します
    manager_config = LocalAgentConfig(
        system_instructions=(
            "あなたはAI開発チームを率いるプロジェクトマネージャーです。"
            "配下の専門サブエージェント（server_researcher, performance_analyst）を必要に応じて呼び出してタスクを委譲し、"
            "得られた結果を統合して、エンジニアや意思決定者にわかりやすい『システム改善提案書』としてまとめて出力してください。"
        ),
        tools=[fetch_server_metrics, calculate_optimization_impact],
        subagents=[researcher_subagent, analyst_subagent],
        model="gemini-3.1-flash-lite"
    )

    # ─── 3. 親エージェントの実行（親が自動でサブエージェントを統率・呼び出し） ───
    print("▶ プロジェクトマネージャー（親エージェント）がチームを起動し、タスクを開始します...\n")
    async with Agent(manager_config) as manager:
        response = await manager.chat(user_request)
        final_report = await response.text()

        print("==================================================")
        print(" 🏆 [AI専門チームによる最終納品物：システム改善提案書]")
        print("==================================================")
        print(f"{final_report}\n")

    print("=== subagents パラメータによるチーム協調タスクが正常に完了しました ===")

if __name__ == "__main__":
    asyncio.run(main())

# 仮想環境の有効化
source .venv/bin/activate

# パッケージのインストール
pip install google-antigravity python-dotenv

# APIキーの設定 (.env ファイルを作成)
echo 'GEMINI_API_KEY="YOUR_GEMINI_API_KEY_HERE"' > .env

python3 team_agent_demo.py

=== Google Antigravity SDK subagents パラメータによるAIチーム協調デモを開始します ===

【ユーザーからの依頼】: 本番Webサーバー 'web-prod-01' で発生している高負荷の原因を調査し、システム改善提案書を作成してください。

▶ プロジェクトマネージャー（親エージェント）がチームを起動し、タスクを開始します...

==================================================
 🏆 [AI専門チームによる最終納品物：システム改善提案書]
==================================================
サーバー 'web-prod-01' の負荷状況を調査するため、専門サブエージェント `server_researcher` を起動しました。現在、稼働メトリクスと最新のログの分析を進めています。

結果が得られ次第、`performance_analyst` と連携して詳細な分析と改善提案書の作成を行います。続報をお待ちください。サーバー 'web-prod-01' の調査結果を報告いたします。CPUおよびメモリ負荷の上昇は、DB接続プールの枯渇によって発生しているリクエストの滞留と、それに起因するタイムアウトエラーが主因であると特定しました。

詳細および改善施策を以下の報告書にまとめました。

# システム改善提案書: web-prod-01 の高負荷対策

## 1. 調査結果サマリー
- **CPU 使用率:** 88% (ピーク時 96%)
- **メモリ使用率:** 82% (13.1GB/16GB)
- **主要なボトルネック:** DB接続プールの待機数過多 (45件)
- **最新ログ:** `DB_CONNECTION_TIMEOUT` が過去1時間で12件発生しており、接続待ちによるパフォーマンス低下が顕著です。

## 2. 原因分析
アプリケーションからDBへの接続数がプールの制限値に達しており、後続のリクエストが接続待ち（ブロッキング）状態にあります。これがプロセス全体の負荷を高め、CPUおよびメモリリソースを圧迫する悪循環を生んでいます。

## 3. 改善施策案
最適化シミュレーションの結果、以下の対策を推奨します。

| 施策項目 | 内容 | 期待される効果 |
| :--- | :--- | :--- |
| **接続プールの拡充** | DB接続プールの最大値を 50 から 120 へ引き上げ | 接続待ちの劇的な解消 |
| **キャッシュ層の導入** | 高頻度アクセスデータの Redis キャッシュ化 | DB負荷の軽減とCPU効率化 |

これらの施策により、**応答遅延の約 75% 削減**、および **CPU 負荷の約 20% 改善**が見込まれます。

---
本件について承認いただければ、設定変更の準備を開始いたします。ご検討をお願いいたします。

=== subagents パラメータによるチーム協調タスクが正常に完了しました ===