Gemma 4 をAntigravityで動かすエージェント構築：ツール呼び出しとメモリ設計の実践

pip install google-cloud-aiplatform antigravity-sdk

import vertexai
from antigravity import AgentBuilder, Tool
from vertexai.preview.generative_models import GenerativeModel
 
# Vertex AI の初期化
vertexai.init(project="your-project-id", location="us-central1")
 
# Gemma 4 モデルの接続
model = GenerativeModel("gemma-4-pro")  # または gemma-4-it
 
# エージェントビルダーの初期化
agent_builder = AgentBuilder(model=model)

from antigravity import Tool, ToolParameter
 
# 検索ツールの定義
search_tool = Tool(
    name="web_search",
    description="最新情報や事実確認が必要な場合にウェブを検索する。一般的な知識には使わない。",
    parameters=[
        ToolParameter(
            name="query",
            type="string",
            description="検索クエリ。具体的な検索語を使い、自然文は避ける。",
            required=True
        ),
        ToolParameter(
            name="max_results",
            type="integer",
            description="取得する結果の最大数（1〜5）",
            required=False,
            default=3
        )
    ],
    function=execute_web_search  # 実際の検索を行う関数
)
 
# 計算ツールの定義
calculator_tool = Tool(
    name="calculate",
    description="数値計算を行う。掛け算・割り算・複雑な数式の評価など。",
    parameters=[
        ToolParameter(
            name="expression",
            type="string",
            description="評価する数式。例: '1234 * 5678' または '(100 * 0.08) + 1000'",
            required=True
        )
    ],
    function=safe_calculate
)
 
# エージェントにツールを登録
agent = agent_builder.build(
    tools=[search_tool, calculator_tool],
    system_instruction="""
    あなたは業務支援アシスタントです。
    ウェブ検索は最新情報が必要なときだけ使ってください。
    すでに知っている情報にはツールを使わないでください。
    計算は必ず calculator ツールを使ってください。頭の中で計算しないこと。
    """
)

from antigravity.memory import SummaryMemory, ExternalMemory
 
# サマリーメモリの設定
memory = SummaryMemory(
    max_recent_messages=10,    # 最新10件はそのまま保持
    summary_interval=20,        # 20件ごとに要約を生成
    summary_model=model         # 要約にも同じモデルを使用
)
 
# または外部メモリ（大規模ナレッジが必要な場合）
external_memory = ExternalMemory(
    vector_store="vertex-ai-vector-search",
    index_id="your-index-id",
    top_k=5  # 関連する過去情報を上位5件取得
)
 
agent = agent_builder.build(
    tools=[search_tool, calculator_tool],
    memory=memory,  # または external_memory
    system_instruction="..."
)

import asyncio
 
async def run_agent_task(user_request: str) -> str:
    """エージェントにタスクを依頼し、結果を受け取る"""
    
    response = await agent.run_async(
        user_message=user_request,
        session_id="session-001"  # セッションを継続する場合に指定
    )
    
    # ツール呼び出しのログを確認（デバッグ時に有用）
    for step in response.steps:
        if step.type == "tool_call":
            print(f"ツール使用: {step.tool_name}({step.arguments})")
            print(f"結果: {step.result[:100]}...")
    
    return response.final_answer
 
# 実行例
async def main():
    result = await run_agent_task(
        "2026年4月時点のPythonの最新バージョンを調べて、"
        "それを使った仮想環境の作成コマンドを教えてください。"
    )
    print(result)
 
asyncio.run(main())

import time
from functools import wraps
 
def rate_limited(calls_per_second: float):
    """シンプルなレート制限デコレータ"""
    min_interval = 1.0 / calls_per_second
    last_called = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            if elapsed < min_interval:
                time.sleep(min_interval - elapsed)
            result = func(*args, **kwargs)
            last_called[0] = time.time()
            return result
        return wrapper
    return decorator
 
@rate_limited(calls_per_second=1)
def execute_web_search(query: str, max_results: int = 3) -> str:
    # 実際の検索処理
    ...