はじめて画像入力つきのエージェントを本番に出した日のことを、今でも覚えています。テキストだけのときは想定内で動いていたものが、写真を一枚渡した途端に「見えているはずのもの」を取り違える。その戸惑いから、私のマルチモーダルエージェント設計は始まりました。
Antigravity AI（Google の Gemma 4 ベースのエージェント基盤）を使ったマルチモーダルエージェントは、「テキストだけ」のエージェントとは別次元の可能性があります。画像を理解して行動を起こす、スクリーンショットを見てUIを操作する、文書の図表を解析してレポートを生成する——これらがすべてひとつのエージェントシステムで実現できます。
ただし、マルチモーダルとエージェント機能を組み合わせると、実装の複雑さが一段上がります。ここではAntigravity API を使ったマルチモーダルエージェントの設計と本番実装を、動くコードと共に解説します。
Antigravity APIのマルチモーダル入力の基本
まず、Antigravity API でテキストと画像を組み合わせた入力を処理する基本的な方法を確認します。
import google.generativeai as genai
from PIL import Image
import requests
from io import BytesIO
import base64
genai.configure(api_key="YOUR_ANTIGRAVITY_API_KEY")
model = genai.GenerativeModel("gemini-2.5-pro-preview-05-06")
def load_image_from_url(url: str) -> Image.Image:
"""URLから画像を読み込む"""
response = requests.get(url, timeout=10)
return Image.open(BytesIO(response.content))
def load_image_from_file(path: str) -> Image.Image:
"""ファイルから画像を読み込む"""
return Image.open(path)
def analyze_image(image: Image.Image, prompt: str) -> str:
"""画像とテキストを組み合わせた分析"""
response = model.generate_content([prompt, image])
return response.text
# 使用例：商品画像の説明生成
product_image = load_image_from_file("product.jpg")
description = analyze_image(
product_image,
"この商品の特徴を3つの箇条書きで説明してください。EC サイトの商品説明として使えるよう、ターゲット顧客を意識した表現にしてください。"
)
print(description)
Function Calling（ツール使用）の実装
Antigravity API のエージェントが「考えて行動する」ために不可欠なのが Function Calling です。モデルが外部ツール（API・データベース・計算処理）を自律的に呼び出せるようにする機能です。
import json
from typing import Any, Callable
# ツール定義（モデルが利用できる外部機能を定義する）
tools = [
{
"function_declarations": [
{
"name": "search_product_database",
"description": "商品データベースを検索する。商品名・カテゴリ・価格帯で絞り込める。",
"parameters": {
"type": "OBJECT",
"properties": {
"query": {
"type": "STRING",
"description": "検索キーワード"
},
"category": {
"type": "STRING",
"description": "商品カテゴリ（electronics/clothing/food）",
"enum": ["electronics", "clothing", "food", "all"]
},
"max_price": {
"type": "NUMBER",
"description": "最大価格（円）"
}
},
"required": ["query"]
}
},
{
"name": "get_product_image",
"description": "商品IDから商品画像URLを取得する",
"parameters": {
"type": "OBJECT",
"properties": {
"product_id": {
"type": "STRING",
"description": "商品ID"
}
},
"required": ["product_id"]
}
},
{
"name": "add_to_cart",
"description": "カートに商品を追加する",
"parameters": {
"type": "OBJECT",
"properties": {
"product_id": {"type": "STRING"},
"quantity": {"type": "INTEGER", "description": "数量（デフォルト:1）"}
},
"required": ["product_id"]
}
}
]
}
]
# ツールの実際の実装（モックデータ）
def search_product_database(query: str, category: str = "all", max_price: float = None) -> dict:
"""商品検索の実装"""
products = [
{"id": "P001", "name": "ワイヤレスイヤホン Pro", "category": "electronics", "price": 12800},
{"id": "P002", "name": "ノイズキャンセリングヘッドホン", "category": "electronics", "price": 24000},
{"id": "P003", "name": "Bluetoothスピーカー", "category": "electronics", "price": 6500},
]
results = [
p for p in products
if query.lower() in p["name"].lower()
and (category == "all" or p["category"] == category)
and (max_price is None or p["price"] <= max_price)
]
return {"products": results, "total": len(results)}
def get_product_image(product_id: str) -> dict:
"""商品画像URLの取得"""
image_map = {
"P001": "https://example.com/images/P001.jpg",
"P002": "https://example.com/images/P002.jpg",
}
return {"image_url": image_map.get(product_id, ""), "product_id": product_id}
def add_to_cart(product_id: str, quantity: int = 1) -> dict:
"""カートへの追加"""
return {"success": True, "product_id": product_id, "quantity": quantity, "cart_total": quantity}
# ツール名とその実装をマッピング
tool_implementations: dict[str, Callable] = {
"search_product_database": search_product_database,
"get_product_image": get_product_image,
"add_to_cart": add_to_cart,
}
def execute_tool(tool_name: str, tool_args: dict) -> Any:
"""ツールを実行して結果を返す"""
impl = tool_implementations.get(tool_name)
if not impl:
return {"error": f"Unknown tool: {tool_name}"}
try:
return impl(**tool_args)
except Exception as e:
return {"error": str(e)}
マルチモーダルエージェントのメインループ
テキストと画像の両方を扱いながら、Function Calling でツールを実行するエージェントのコアロジックです。
from dataclasses import dataclass, field
from typing import Optional
import time
@dataclass
class AgentState:
"""エージェントの実行状態を管理するクラス"""
session_id: str
conversation_history: list = field(default_factory=list)
tool_calls_count: int = 0
total_input_tokens: int = 0
total_output_tokens: int = 0
start_time: float = field(default_factory=time.time)
max_tool_calls: int = 30 # 無限ループ防止
max_duration_seconds: float = 300 # 5分のタイムアウト
def is_budget_exceeded(self) -> bool:
return (self.tool_calls_count >= self.max_tool_calls or
time.time() - self.start_time > self.max_duration_seconds)
def record_api_call(self, input_tokens: int, output_tokens: int):
self.total_input_tokens += input_tokens
self.total_output_tokens += output_tokens
def summary(self) -> dict:
elapsed = time.time() - self.start_time
return {
"session_id": self.session_id,
"tool_calls": self.tool_calls_count,
"total_tokens": self.total_input_tokens + self.total_output_tokens,
"elapsed_seconds": round(elapsed, 1)
}
class MultimodalAgent:
"""マルチモーダル対応のAIエージェント"""
def __init__(self, model_name: str = "gemini-2.5-pro-preview-05-06"):
self.model = genai.GenerativeModel(
model_name=model_name,
tools=tools,
system_instruction="""あなたは親切なECサイトのアシスタントです。
ユーザーが商品を探したり、カートに追加するのをサポートします。
画像が提供された場合は、その内容を正確に分析してください。
ツールを使う際は、必要最小限の呼び出し数で目的を達成してください。"""
)
def run(
self,
user_message: str,
images: list[Image.Image] = None,
state: Optional[AgentState] = None
) -> dict:
"""エージェントを実行してレスポンスと状態を返す"""
if state is None:
import uuid
state = AgentState(session_id=str(uuid.uuid4())[:8])
# メッセージの構築（テキスト + オプションの画像）
content_parts = [user_message]
if images:
content_parts.extend(images)
# 会話履歴に追加
state.conversation_history.append({
"role": "user",
"parts": content_parts
})
# エージェントループ
chat = self.model.start_chat(history=[])
while not state.is_budget_exceeded():
# API呼び出し
response = chat.send_message(content_parts if not chat.history else "")
if response.usage_metadata:
state.record_api_call(
response.usage_metadata.prompt_token_count,
response.usage_metadata.candidates_token_count
)
# Function Callがない場合は最終回答
if not response.candidates[0].content.parts[0].function_call.name if \
hasattr(response.candidates[0].content.parts[0], 'function_call') else True:
final_text = response.text
state.conversation_history.append({
"role": "model",
"parts": [final_text]
})
return {
"response": final_text,
"state": state.summary()
}
# Function Callを処理
function_responses = []
for part in response.candidates[0].content.parts:
if hasattr(part, 'function_call') and part.function_call.name:
fc = part.function_call
state.tool_calls_count += 1
tool_result = execute_tool(fc.name, dict(fc.args))
function_responses.append(
genai.protos.Part(
function_response=genai.protos.FunctionResponse(
name=fc.name,
response={"result": json.dumps(tool_result, ensure_ascii=False)}
)
)
)
if function_responses:
content_parts = function_responses
return {
"response": "処理時間またはツール呼び出し上限に達しました。",
"state": state.summary()
}
画像とFunction Callingを組み合わせた実用例
agent = MultimodalAgent()
# テキストのみの検索
result = agent.run("ノイズキャンセリング機能があるイヤホンを探して、一番安いものをカートに入れてください")
print("テキスト検索結果:")
print(result["response"])
print(f"処理統計: {result['state']}")
# 画像を使った商品検索
import urllib.request
# テスト用の商品画像を取得（実際のユースケースでは実際の商品画像を使う）
product_img = Image.new("RGB", (200, 200), color=(100, 150, 200))
image_result = agent.run(
"この画像の商品に似たものを検索して、おすすめを教えてください",
images=[product_img]
)
print("\n画像検索結果:")
print(image_result["response"])
複数画像の一括処理：非同期パターン
大量の画像を処理する場合、非同期処理で大幅なパフォーマンス改善が可能です。
import asyncio
import google.generativeai as genai
async_model = genai.GenerativeModel("gemini-2.5-pro-preview-05-06")
async def analyze_image_async(image: Image.Image, prompt: str, image_id: str) -> dict:
"""単一画像の非同期分析"""
loop = asyncio.get_event_loop()
try:
response = await loop.run_in_executor(
None,
lambda: async_model.generate_content([prompt, image])
)
return {
"id": image_id,
"result": response.text,
"status": "success"
}
except Exception as e:
return {
"id": image_id,
"error": str(e),
"status": "failed"
}
async def batch_analyze_images(
images: list[dict], # [{"id": str, "image": Image, "prompt": str}]
concurrency_limit: int = 5
) -> list[dict]:
"""複数画像を並列で分析（同時実行数を制限）"""
semaphore = asyncio.Semaphore(concurrency_limit)
async def analyze_with_limit(item: dict) -> dict:
async with semaphore:
return await analyze_image_async(
item["image"], item["prompt"], item["id"]
)
tasks = [analyze_with_limit(item) for item in images]
return await asyncio.gather(*tasks, return_exceptions=True)
# 使用例
async def main():
image_tasks = [
{
"id": f"product_{i}",
"image": Image.new("RGB", (200, 200), color=(i*30, 100, 200)),
"prompt": "この商品カテゴリを判定してください"
}
for i in range(10)
]
results = await batch_analyze_images(image_tasks, concurrency_limit=3)
success = [r for r in results if isinstance(r, dict) and r.get("status") == "success"]
print(f"成功: {len(success)}/{len(image_tasks)}件")
本番運用のための状態管理とロギング
エージェントを本番環境で安定運用するために不可欠なのが、適切な状態管理とロギングです。
import logging
import json
from datetime import datetime
# 構造化ロギングの設定
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("antigravity_agent")
class ProductionAgent(MultimodalAgent):
"""本番運用向けの拡張エージェント"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.request_log = []
def run(self, user_message: str, images: list = None, state=None) -> dict:
request_id = f"req_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}"
logger.info(json.dumps({
"event": "agent_request_start",
"request_id": request_id,
"message_length": len(user_message),
"has_images": bool(images)
}))
try:
result = super().run(user_message, images, state)
logger.info(json.dumps({
"event": "agent_request_complete",
"request_id": request_id,
**result.get("state", {})
}))
result["request_id"] = request_id
return result
except Exception as e:
logger.error(json.dumps({
"event": "agent_request_error",
"request_id": request_id,
"error": str(e)
}))
raise
# 使用例
production_agent = ProductionAgent()
result = production_agent.run("ワイヤレスイヤホンの在庫を確認してください")
print(f"Request ID: {result.get('request_id')}")
print(f"応答: {result['response']}")
コスト推定と最適化
Antigravity API の利用コストを管理するためのユーティリティです。
@dataclass
class CostEstimator:
"""API利用コストの推定"""
# 2026年4月現在のGemini 2.5 Pro概算料金（$/1M tokens）
input_price_per_million: float = 1.25
output_price_per_million: float = 5.00
image_price_per_image: float = 0.001315 # 1枚あたり
def estimate(
self,
input_tokens: int,
output_tokens: int,
image_count: int = 0
) -> dict:
input_cost = input_tokens / 1_000_000 * self.input_price_per_million
output_cost = output_tokens / 1_000_000 * self.output_price_per_million
image_cost = image_count * self.image_price_per_image
total = input_cost + output_cost + image_cost
return {
"input_cost_usd": round(input_cost, 6),
"output_cost_usd": round(output_cost, 6),
"image_cost_usd": round(image_cost, 6),
"total_cost_usd": round(total, 6),
"total_cost_jpy": round(total * 150, 2) # 概算円換算
}
estimator = CostEstimator()
cost = estimator.estimate(input_tokens=50000, output_tokens=10000, image_count=5)
print(f"推定コスト: ${cost['total_cost_usd']:.4f} (約¥{cost['total_cost_jpy']})")
権限設計：マルチモーダルエージェントに「何を委ね、何を握るか」
ここまでのコードは、エージェントが自律的にツールを呼び出す前提で書いてきました。ですが本番運用では、add_to_cart のように状態を変える操作を、モデルの判断だけで無条件に実行させてよいのか——この問いが必ず立ち上がります。私自身、EC 系の試作でここを甘く見て、痛い思いをしました。
Antigravity 2.2.1（2026-06-25）で導入された統一パーミッション体系は、まさにこの迷いに輪郭を与えるものでした。エージェントの操作範囲を明示的に宣言し、範囲外の行為には人間の確認を挟む。この考え方は、API から自作するエージェントにもそのまま持ち込めます。
私は、ツールを「副作用の重さ」で三段階に分けています。読み取りだけの操作（search_product_database・get_product_image）は自動許可。状態を変えるが可逆な操作（add_to_cart）は自動だが監査ログ必須。不可逆・金銭が動く操作（Stripe での決済確定・在庫引き当て）は人間の承認を必須にする。この線引きを、実行層に一枚のゲートとして差し込みます。
from enum import Enum
from typing import Callable
class Risk(Enum):
"""ツールの副作用の重さ"""
READ = "read" # 読み取りのみ・自動許可
REVERSIBLE = "reversible" # 可逆な変更・自動だが監査
IRREVERSIBLE = "irreversible" # 不可逆/金銭・人間の承認必須
# ツール名 → リスク区分（統一パーミッション体系の考え方を自作エージェントへ）
TOOL_RISK: dict[str, Risk] = {
"search_product_database": Risk.READ,
"get_product_image": Risk.READ,
"add_to_cart": Risk.REVERSIBLE,
"checkout": Risk.IRREVERSIBLE, # Stripe 決済確定
}
class PermissionDenied(Exception):
pass
def guarded_execute(
tool_name: str,
tool_args: dict,
approver: Callable[[str, dict], bool] | None = None,
) -> Any:
"""権限ゲートを通してからツールを実行する"""
risk = TOOL_RISK.get(tool_name, Risk.IRREVERSIBLE) # 未知のツールは最も厳しく
if risk is Risk.IRREVERSIBLE:
# 人間（または上位ポリシー）の承認を必須にする
if approver is None or not approver(tool_name, tool_args):
logger.warning(json.dumps({
"event": "tool_blocked", "tool": tool_name, "risk": risk.value
}))
raise PermissionDenied(f"{tool_name} は承認が必要です")
result = execute_tool(tool_name, tool_args)
if risk is not Risk.READ:
# 可逆・不可逆な変更は必ず監査ログに残す
logger.info(json.dumps({
"event": "tool_mutation", "tool": tool_name,
"risk": risk.value, "args": tool_args
}, ensure_ascii=False))
return result
guarded_execute は、エージェントループの execute_tool をそのまま置き換えられます。落とし穴は「未知のツールを登録し忘れること」です。ここで未知のツールを最も厳しい区分にフォールバックさせておけば、区分の登録を忘れても危険側に倒れます。デフォルトで拒否する設計は、うっかりを事故に変えないための保険です。
なぜここまで慎重にするのか。マルチモーダルだからこそ、モデルの判断根拠が「画像の解釈」というブラックボックスを含むからです。テキストなら入力を読み返して検証できますが、画像の誤読は事後に追いにくい。だからこそ、副作用の重い操作ほど人間の側に手綱を残しておく。この非対称性が、個人開発でマルチモーダルを扱って学んだいちばん大きな注意点でした。
委ねる範囲を決める3つの判断軸
権限区分をコードに落とす前に、そもそも何を自動化してよいのかを判断する軸を持っておくと迷いが減ります。私は次の3点で見ています。
| 判断軸 | 自動に寄せてよい | 人間に残す |
|---|
| 失敗時の影響 | 表示や下書きの生成にとどまる | 金銭・在庫・ユーザー体験に波及する |
| 復旧の明文化 | ロールバック手順が自動化済み | 復旧が手作業・属人的 |
| 観測の粒度 | ログから人間が再現できる | 判断根拠が画像解釈に埋もれる |
この表は絶対の基準ではありません。事業のフェーズや、そのエージェントが触れる資産の性質で線は動きます。ただ「影響・復旧・観測」の三つを毎回自問する習慣そのものが、暴走を未然に防いでくれます。
本番投入までの検証ステップ
設計が固まったら、いきなり全開放にはしません。私はいつも次の三段階で本番運用に近づけていきます。
シャドー運用で却下ログを集める
まず、権限ゲートをすべて IRREVERSIBLE 相当に絞った状態で、自分だけのシャドー運用を回します。エージェントが「本当はこうしたかった」という却下ログを集めると、区分の甘い部分が見えてきます。この段階を飛ばさないことを強く推奨します。
観測メトリクスとアラートを設置する
次に、ツール呼び出し数・却下率・レイテンシ・推定コストを可視化し、異常な呼び出しパターンを検知できる状態を作ります。私の試作では、実トラフィックで会話が伸びて、推定コストが机上の見積もりの約1.4倍に膨らんだことがありました。観測がなければ、この乖離には気づけません。
限定ロールアウトで確かめる
そのうえで、一部ユーザー・一部時間帯だけに開放し、実トラフィックで挙動を確かめます。ここまで踏むと、コスト推定のコードで見積もった数字が現実とどれだけずれるかも同時に検証できます。机上のコスト計算は、実際の会話の長さや画像枚数の分布に驚くほど左右されるからです。
マルチモーダルエージェントは、視覚的な理解と行動を一つに束ねられる強力な仕組みです。だからこそ、その力に見合った権限設計と観測をセットで持たせたい。まずはお手元のユースケースでツールを副作用の重さで分類し、guarded_execute のような一枚のゲートを差し込むところから始めていただければ、事故の芽を早いうちに摘めるはずです。
Antigravity API の基本的な使い方についてはAntigravity 入門記事を、エージェント設計のさらなる深掘りにはAgentsカテゴリの記事一覧もあわせてご覧ください。実装の一助になれば幸いです。お読みいただきありがとうございました。