Gemma 4のリリースにより、オンデバイスAIの開発は新しい段階に入りましました。E2Bモデルがわずか2GB程度のメモリで動作し、E4Bが関数呼び出し対応で4B以下の有効パラメータを実現します。これは単なるデモ用途ではなく、プロダクション品質のAI機能をエッジデバイスに直接組み込める時代が来たことを意味しています。
オンデバイスAIのアーキテクチャ設計
プロダクション品質のオンデバイスAIを実現するには、モデル選定だけでなく、推論パイプライン全体の設計が重要です。Gemma 4の4つのモデルサイズを前提に、実践的なアーキテクチャパターンを見ていきましょう。
ハイブリッドアーキテクチャ: エッジ + クラウドの最適な分担
すべてのAI処理をデバイス上で行う必要はありません。最も効率的なアプローチは、タスクの性質に応じてエッジとクラウドを使い分けるハイブリッド構成です。
# ハイブリッド推論ルーターの実装例
class HybridInferenceRouter:
"""タスクの複雑さに応じてエッジとクラウドを自動振り分け"""
def __init__(self, edge_model, cloud_client):
self.edge_model = edge_model # Gemma 4 E4B (ローカル)
self.cloud_client = cloud_client # Gemma 4 31B (サーバー)
self.complexity_threshold = 0.7
async def route(self, task: str, context: dict) -> str:
complexity = self._estimate_complexity(task, context)
if complexity < self.complexity_threshold:
# 単純なタスク → エッジで即座に処理
return await self.edge_model.generate(task)
else:
# 複雑なタスク → クラウドに委譲
if self._is_network_available():
return await self.cloud_client.generate(task)
else:
# オフライン時はエッジでベストエフォート
return await self.edge_model.generate(
task,
max_tokens=2048,
temperature=0.3 # 精度重視
)
def _estimate_complexity(self, task: str, context: dict) -> float:
"""タスクの複雑さを0.0〜1.0で推定"""
score = 0.0
if len(task.split()) > 200:
score += 0.3
if context.get("code_lines", 0) > 500:
score += 0.3
if any(kw in task for kw in ["分析", "比較", "設計", "リファクタ"]):
score += 0.2
return min(score, 1.0)
def _is_network_available(self) -> bool:
"""ネットワーク接続状態を確認"""
import socket
try:
socket.create_connection(("8.8.8.8", 53), timeout=1)
return True
except OSError:
return False
# 使用例
# router = HybridInferenceRouter(edge_model, cloud_client)
# result = await router.route("このコードをレビューして", {"code_lines": 50})メモリ管理とモデルライフサイクル
エッジデバイスではメモリが制約となります。Gemma 4のモデルを効率的にロード・アンロードするライフサイクル管理が不可欠です。
// iOS向け: Core MLとGemma 4のメモリ管理
import CoreML
class GemmaModelManager {
private var loadedModel: MLModel?
private let modelURL: URL
private let memoryThresholdMB: Int = 150
init(modelPath: String) {
self.modelURL = Bundle.main.url(
forResource: modelPath,
withExtension: "mlmodelc"
)!
}
/// メモリ状態に応じてモデルをロード
func loadIfNeeded() throws -> MLModel {
if let model = loadedModel {
return model
}
let availableMB = getAvailableMemoryMB()
guard availableMB > memoryThresholdMB else {
throw GemmaError.insufficientMemory(
available: availableMB,
required: memoryThresholdMB
)
}
let config = MLModelConfiguration()
config.computeUnits = .cpuAndNeuralEngine
let model = try MLModel(contentsOf: modelURL, configuration: config)
self.loadedModel = model
NotificationCenter.default.addObserver(
self,
selector: #selector(handleMemoryWarning),
name: UIApplication.didReceiveMemoryWarningNotification,
object: nil
)
return model
}
@objc private func handleMemoryWarning() {
loadedModel = nil
print("[Gemma] Memory warning — model unloaded")
}
private func getAvailableMemoryMB() -> Int {
var info = mach_task_basic_info()
var count = mach_msg_type_number_t(
MemoryLayout<mach_task_basic_info>.size
) / 4
let result = withUnsafeMutablePointer(to: &info) {
$0.withMemoryRebound(
to: integer_t.self, capacity: 1
) {
task_info(mach_task_self_, task_flavor_t(MACH_TASK_BASIC_INFO), $0, &count)
}
}
guard result == KERN_SUCCESS else { return 0 }
let usedMB = Int(info.resident_size) / 1_048_576
let totalMB = Int(ProcessInfo.processInfo.physicalMemory) / 1_048_576
return totalMB - usedMB
}
}
enum GemmaError: Error {
case insufficientMemory(available: Int, required: Int)
}ファインチューニングによるカスタムモデル構築
Gemma 4はApache 2.0ライセンスのため、自由にファインチューニングが可能です。ドメイン固有の知識を持つカスタムモデルを構築する手順を解説します。
LoRAによる効率的なファインチューニング
フルパラメータのファインチューニングは計算コストが高いため、LoRA（Low-Rank Adaptation）を活用します。Gemma 4の26B MoEモデルであれば、16GBのGPUメモリでもファインチューニングが可能です。
# Gemma 4 LoRAファインチューニングの実装
from transformers import AutoModelForCausalLM, AutoTokenizer, BitsAndBytesConfig
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training
import torch
# 4bit量子化でメモリ使用量を削減
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_compute_dtype=torch.bfloat16,
bnb_4bit_use_double_quant=True
)
model_id = "google/gemma-4-27b-it"
tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoModelForCausalLM.from_pretrained(
model_id,
quantization_config=bnb_config,
device_map="auto",
torch_dtype=torch.bfloat16
)
lora_config = LoraConfig(
r=16,
lora_alpha=32,
target_modules=[
"q_proj", "k_proj", "v_proj",
"o_proj", "gate_proj",
"up_proj", "down_proj"
],
lora_dropout=0.05,
bias="none",
task_type="CAUSAL_LM"
)
model = prepare_model_for_kbit_training(model)
model = get_peft_model(model, lora_config)
trainable_params = sum(p.numel() for p in model.parameters() if p.requires_grad)
total_params = sum(p.numel() for p in model.parameters())
print(f"学習対象: {trainable_params:,} / 全体: {total_params:,}")
print(f"学習対象の割合: {100 * trainable_params / total_params:.2f}%")
# 出力例: 学習対象: 41,943,040 / 全体: 27,227,128,832
# 出力例: 学習対象の割合: 0.15%学習データの準備と品質管理
ファインチューニングの成否は学習データの品質で決まります。以下は、ドメイン固有の質問応答データセットを構築する例です。
# 学習データの構造化と品質チェック
from datasets import Dataset
import json
def prepare_training_data(raw_data_path: str) -> Dataset:
"""学習データを読み込み、Gemma 4の形式に変換"""
with open(raw_data_path, "r", encoding="utf-8") as f:
raw_data = json.load(f)
formatted = []
skipped = 0
for item in raw_data:
if len(item["question"]) < 10:
skipped += 1
continue
if len(item["answer"]) < 50:
skipped += 1
continue
text = (
f"<start_of_turn>user\n{item['question']}<end_of_turn>\n"
f"<start_of_turn>model\n{item['answer']}<end_of_turn>"
)
formatted.append({"text": text})
print(f"有効データ: {len(formatted)} / スキップ: {skipped}")
return Dataset.from_list(formatted)
# 使用例
# dataset = prepare_training_data("training_data.json")
# 出力例: 有効データ: 4,523 / スキップ: 47学習の実行とモデルの評価
# Transformers Trainerを使った学習実行
from transformers import TrainingArguments, Trainer
training_args = TrainingArguments(
output_dir="./gemma4-custom",
num_train_epochs=3,
per_device_train_batch_size=4,
gradient_accumulation_steps=4,
learning_rate=2e-4,
weight_decay=0.01,
warmup_ratio=0.03,
lr_scheduler_type="cosine",
logging_steps=10,
save_strategy="epoch",
bf16=True,
optim="paged_adamw_8bit",
gradient_checkpointing=True,
max_grad_norm=0.3,
)
trainer = Trainer(
model=model,
args=training_args,
train_dataset=dataset,
tokenizer=tokenizer,
)
trainer.train()
model.save_pretrained("./gemma4-custom-lora")
print("✅ カスタムモデルの保存完了")
# 出力: ✅ カスタムモデルの保存完了Antigravityエージェントとのカスタムモデル統合
ファインチューニングしたカスタムGemma 4モデルを、Antigravityのエージェントシステムに組み込む方法を解説します。
カスタムモデルサーバーの構築
まず、カスタムモデルをOllamaでホスティングできる形式に変換します。
# カスタムモデルのOllamaフォーマット変換
modelfile_content = """
FROM ./gemma4-custom-merged.gguf
PARAMETER temperature 0.7
PARAMETER top_p 0.9
PARAMETER num_ctx 8192
SYSTEM あなたはドメイン固有の専門知識を持つAIアシスタントです。
正確で実用的な回答を心がけてください。
"""
with open("Modelfile", "w") as f:
f.write(modelfile_content)
# シェルで実行:
# ollama create gemma4-custom -f Modelfile
# ollama run gemma4-custom "テスト質問"Antigravity agents.md でのカスタムモデル定義
<!-- .antigravity/agents.md -->
# Domain Expert Agent
## 概要
カスタムファインチューニングしたGemma 4モデルを使用し、
プロジェクト固有のドメイン知識に基づいた支援を行う。
## モデル設定
- ベース: Gemma 4 26B MoE
- アダプター: LoRA (ドメイン特化)
- エンドポイント: http://localhost:11434/api/generate
- モデル名: gemma4-custom
## タスク
1. プロジェクト固有の設計パターンに基づくコード提案
2. ドメイン用語を理解した上でのドキュメント生成
3. 過去の実装パターンに基づくバグ検出
## コンテキスト
- /docs/** のドキュメントを参照可能
- /src/** のソースコードを解析可能MCP（Model Context Protocol）サーバーとしての統合
Gemma 4カスタムモデルをMCPサーバーとして公開することで、Antigravityから直接利用できるようになります。
// MCP Server: Gemma 4 カスタムモデル
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
const server = new Server(
{ name: "gemma4-custom", version: "1.0.0" },
{ capabilities: { tools: {} } }
);
server.setRequestHandler("tools/list", async () => ({
tools: [{
name: "domain_query",
description: "ドメイン固有の質問に対してカスタムGemma 4で回答",
inputSchema: {
type: "object",
properties: {
question: { type: "string", description: "質問内容" },
context: { type: "string", description: "追加コンテキスト" }
},
required: ["question"]
}
}]
}));
server.setRequestHandler("tools/call", async (request) => {
if (request.params.name === "domain_query") {
const { question, context } = request.params.arguments as {
question: string;
context?: string;
};
const response = await fetch("http://localhost:11434/api/generate", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
model: "gemma4-custom",
prompt: context
? `コンテキスト:\n${context}\n\n質問: ${question}`
: question,
stream: false
})
});
const data = await response.json();
return { content: [{ type: "text", text: data.response }] };
}
throw new Error(`Unknown tool: ${request.params.name}`);
});
const transport = new StdioServerTransport();
await server.connect(transport);
// 起動コマンド: npx tsx mcp-gemma4-server.tsエッジデプロイメントの最適化
カスタムモデルをエッジデバイスにデプロイする際の最適化テクニックを解説します。
量子化とモデル圧縮
# GGUF形式への変換と量子化
# llama.cppを使用した量子化パイプライン
# 手順1: HuggingFace形式 → GGUF変換
# python convert_hf_to_gguf.py ./gemma4-custom-merged \
# --outfile gemma4-custom-f16.gguf --outtype f16
# 手順2: 量子化（Q4_K_M推奨 — 品質とサイズのバランスが最良）
# ./quantize gemma4-custom-f16.gguf gemma4-custom-q4km.gguf Q4_K_M
# 量子化方式の比較:
# Q8_0: 約27GB → 高品質だがエッジには大きすぎる
# Q4_K_M: 約15GB → 品質・サイズのバランスが最良（推奨）
# Q4_0: 約14GB → 最小サイズだが品質がやや劣化
def evaluate_quantized_model(
original_model_path: str,
quantized_model_path: str,
test_prompts: list[str]
) -> dict:
"""量子化前後のモデル出力品質を比較"""
from rouge_score import rouge_scorer
scorer = rouge_scorer.RougeScorer(
["rouge1", "rouge2", "rougeL"],
use_stemmer=True
)
results = {"rouge1": [], "rouge2": [], "rougeL": []}
for prompt in test_prompts:
original_output = generate(original_model_path, prompt)
quantized_output = generate(quantized_model_path, prompt)
scores = scorer.score(original_output, quantized_output)
for key in results:
results[key].append(scores[key].fmeasure)
avg_results = {
k: sum(v) / len(v) for k, v in results.items()
}
print(f"量子化品質スコア: {avg_results}")
return avg_results
# 出力例: 量子化品質スコア: {'rouge1': 0.92, 'rouge2': 0.87, 'rougeL': 0.90}Android向け LiteRT-LMデプロイパイプライン
// Android: Gemma 4カスタムモデルのプロダクション統合
class GemmaInferenceManager(
private val context: Context
) {
private var session: LlmInferenceSession? = null
companion object {
private const val MODEL_FILE = "gemma4-e4b-custom.task"
private const val MAX_TOKENS = 1024
private const val TEMPERATURE = 0.7f
}
suspend fun initialize(): Result<Unit> = withContext(Dispatchers.IO) {
runCatching {
val options = LlmInference.LlmInferenceOptions.builder()
.setModelPath(getModelPath())
.setMaxTokens(MAX_TOKENS)
.setTemperature(TEMPERATURE)
.setTopK(40)
.setRandomSeed(42)
.build()
val inference = LlmInference.createFromOptions(context, options)
session = inference.createSession()
}
}
fun generateStream(
prompt: String,
onToken: (String) -> Unit,
onComplete: () -> Unit,
onError: (Exception) -> Unit
) {
val activeSession = session ?: run {
onError(IllegalStateException("Model not initialized"))
return
}
try {
activeSession.addQueryChunk(prompt)
activeSession.generateResponseAsync(
object : LlmInferenceSession.ResponseCallback {
override fun onPartialResponse(text: String) {
onToken(text)
}
override fun onResponse(text: String) {
onComplete()
}
override fun onError(e: Exception) {
onError(e)
}
}
)
} catch (e: Exception) {
onError(e)
}
}
private fun getModelPath(): String {
return File(context.filesDir, MODEL_FILE).absolutePath
}
fun release() {
session?.close()
session = null
}
}
// Jetpack Compose での使用例
// val manager = remember { GemmaInferenceManager(context) }
// LaunchedEffect(Unit) { manager.initialize() }プロダクション運用ワークフロー
カスタムモデルを本番環境で安定運用するためのワークフローを構築します。
A/Bテストフレームワーク
# モデルバージョン間のA/Bテスト実装
import json
from datetime import datetime
from typing import Optional
class ModelABTest:
"""カスタムGemma 4モデルのA/Bテストフレームワーク"""
def __init__(
self,
model_a: str,
model_b: str,
traffic_ratio: float = 0.5,
log_path: str = "ab_test_logs.jsonl"
):
self.model_a = model_a
self.model_b = model_b
self.traffic_ratio = traffic_ratio
self.log_path = log_path
def select_model(self, user_id: str) -> str:
"""ユーザーIDに基づく一貫したモデル割り当て"""
hash_val = hash(user_id) % 100
selected = self.model_b if hash_val < (self.traffic_ratio * 100) else self.model_a
return selected
async def generate_and_log(
self,
user_id: str,
prompt: str,
feedback: Optional[str] = None
) -> dict:
"""推論実行 + ログ記録"""
model = self.select_model(user_id)
start = datetime.now()
response = await self._call_model(model, prompt)
latency_ms = (datetime.now() - start).total_seconds() * 1000
log_entry = {
"timestamp": datetime.now().isoformat(),
"user_id": user_id,
"model": model,
"prompt_length": len(prompt),
"response_length": len(response),
"latency_ms": round(latency_ms, 2),
"feedback": feedback
}
with open(self.log_path, "a") as f:
f.write(json.dumps(log_entry) + "\n")
return {"response": response, "model": model, "latency_ms": latency_ms}
async def _call_model(self, model: str, prompt: str) -> str:
"""Ollama経由でモデルを呼び出し"""
import httpx
async with httpx.AsyncClient(timeout=30.0) as client:
resp = await client.post(
"http://localhost:11434/api/generate",
json={"model": model, "prompt": prompt, "stream": False}
)
return resp.json()["response"]
# 使用例
# ab_test = ModelABTest("gemma4-custom-v1", "gemma4-custom-v2", traffic_ratio=0.2)
# result = await ab_test.generate_and_log("user_123", "質問内容")モニタリングとアラート
# カスタムモデルの本番モニタリング
from dataclasses import dataclass, field
from collections import deque
import statistics
@dataclass
class ModelMetrics:
"""推論メトリクスの収集と異常検知"""
window_size: int = 100
latencies: deque = field(default_factory=lambda: deque(maxlen=100))
error_count: int = 0
total_requests: int = 0
latency_p95_threshold_ms: float = 3000.0
error_rate_threshold: float = 0.05
def record_request(self, latency_ms: float, success: bool):
self.total_requests += 1
self.latencies.append(latency_ms)
if not success:
self.error_count += 1
def check_health(self) -> dict:
if len(self.latencies) < 10:
return {"status": "warming_up", "alerts": []}
alerts = []
sorted_latencies = sorted(self.latencies)
p95_idx = int(len(sorted_latencies) * 0.95)
p95 = sorted_latencies[p95_idx]
if p95 > self.latency_p95_threshold_ms:
alerts.append(
f"⚠️ P95レイテンシ超過: {p95:.0f}ms "
f"(閾値: {self.latency_p95_threshold_ms:.0f}ms)"
)
error_rate = self.error_count / self.total_requests
if error_rate > self.error_rate_threshold:
alerts.append(
f"🚨 エラー率超過: {error_rate:.2%} "
f"(閾値: {self.error_rate_threshold:.2%})"
)
return {
"status": "healthy" if not alerts else "degraded",
"p50_ms": statistics.median(self.latencies),
"p95_ms": p95,
"error_rate": error_rate,
"total_requests": self.total_requests,
"alerts": alerts
}
# 使用例
# metrics = ModelMetrics()
# metrics.record_request(latency_ms=250.0, success=True)
# health = metrics.check_health()
# 出力例: {"status": "healthy", "p50_ms": 245.0, "p95_ms": 890.0, ...}高度なワークフロー: CI/CDパイプラインへの統合
モデルの学習からデプロイまでを自動化するCI/CDパイプラインの構築例です。
# .github/workflows/gemma4-model-pipeline.yml
name: Gemma 4 Custom Model Pipeline
on:
push:
paths:
- 'training_data/**'
- 'model_config/**'
jobs:
train-and-evaluate:
runs-on: [self-hosted, gpu]
steps:
- uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Install dependencies
run: pip install -r requirements-training.txt
- name: Train LoRA adapter
run: python scripts/train_lora.py --config model_config/config.yaml
env:
HF_TOKEN: ${{ secrets.HF_TOKEN }}
- name: Evaluate model quality
run: |
python scripts/evaluate.py \
--model ./output/gemma4-custom-lora \
--benchmark ./eval_data/benchmark.json \
--threshold 0.85
- name: Convert to GGUF
if: success()
run: |
python scripts/merge_and_convert.py \
--base google/gemma-4-27b-it \
--lora ./output/gemma4-custom-lora \
--output ./output/gemma4-custom.gguf \
--quantize Q4_K_M
- name: Upload model artifact
if: success()
uses: actions/upload-artifact@v4
with:
name: gemma4-custom-model
path: ./output/gemma4-custom-q4km.gguf
deploy:
needs: train-and-evaluate
runs-on: ubuntu-latest
steps:
- name: Download model artifact
uses: actions/download-artifact@v4
with:
name: gemma4-custom-model
- name: Deploy to model server
run: |
scp gemma4-custom-q4km.gguf ${{ secrets.MODEL_SERVER }}:/models/
ssh ${{ secrets.MODEL_SERVER }} "ollama create gemma4-custom -f /models/Modelfile"
echo "✅ デプロイ完了"本番運用パターン — 推論キャッシング・バッチ処理・パラメータ最適化
ファインチューニングしたカスタムモデルを本番運用する際、コスト効率と応答速度を両立させるには推論レイヤーでの工夫が要ります。同一クエリのキャッシング、複数クエリのバッチ化、推論パラメータの調整について、実践的なパターンを整理します。
推論キャッシング — 同一クエリの高速化
本番環境では、同じプロンプトが何度も送信されます。キャッシングにより、2回目以降の推論を 0.1 秒以下に短縮できます。
import hashlib
import time
from typing import Optional
class GemmaInferenceCache:
def __init__(self, max_cache_size: int = 1000):
self.cache = {}
self.access_times = {}
self.max_size = max_cache_size
def _hash_prompt(self, prompt: str, model: str) -> str:
key_str = f"{model}:{prompt}"
return hashlib.sha256(key_str.encode()).hexdigest()
def get(self, prompt: str, model: str = "gemma4-instruct") -> Optional[str]:
key = self._hash_prompt(prompt, model)
if key in self.cache:
self.access_times[key] = time.time()
return self.cache[key]
return None
def set(self, prompt: str, result: str, model: str = "gemma4-instruct"):
key = self._hash_prompt(prompt, model)
if len(self.cache) >= self.max_size:
oldest_key = min(self.access_times, key=self.access_times.get)
del self.cache[oldest_key]
del self.access_times[oldest_key]
self.cache[key] = result
self.access_times[key] = time.time()バッチ処理 — 複数クエリの効率化
複数のプロンプトをまとめて処理する場合、逐次実行ではなくバッチ処理を使うことで、全体の処理時間を削減できます。
from concurrent.futures import ThreadPoolExecutor
from typing import List, Tuple
class GemmaBatchProcessor:
"""
マルチスレッド対応のバッチ処理エンジン
"""
def __init__(self, max_workers: int = 4):
self.executor = ThreadPoolExecutor(max_workers=max_workers)
self.model = "gemma4-instruct"
def _process_single(self, prompt: str) -> Tuple[str, str]:
"""単一プロンプトの処理"""
try:
import ollama
response = ollama.generate(
model=self.model,
prompt=prompt,
stream=False,
options={"num_predict": 256}
)
return (prompt, response["response"])
except Exception as e:
return (prompt, f"ERROR: {str(e)}")
def process_batch(self, prompts: List[str]) -> List[Tuple[str, str]]:
"""バッチ処理の実行"""
import time
start = time.time()
futures = [self.executor.submit(self._process_single, p) for p in prompts]
results = [f.result() for f in futures]
elapsed = time.time() - start
print(f"⏱️ バッチ処理完了: {len(prompts)} クエリを {elapsed:.2f} 秒で処理")
return results推論パラメータの最適化
Gemma 4 の出力品質は、サンプリングパラメータに大きく左右されます。
from dataclasses import dataclass
@dataclass
class InferenceConfig:
"""推論パラメータの推奨設定"""
CONFIGS = {
"precise": {
"top_p": 0.1,
"top_k": 5,
"temperature": 0.3,
"description": "事実ベースの回答（FAQ、データ抽出）"
},
"balanced": {
"top_p": 0.9,
"top_k": 40,
"temperature": 0.7,
"description": "通常の質問応答"
},
"creative": {
"top_p": 0.95,
"top_k": 50,
"temperature": 1.0,
"description": "創作、ブレインストーミング"
}
}
@classmethod
def get_config(cls, mode: str = "balanced") -> dict:
"""推奨パラメータを取得"""
if mode not in cls.CONFIGS:
raise ValueError(f"不正なモード: {mode}")
config = cls.CONFIGS[mode]
description = config.pop("description")
print(f"📌 モード: {mode} ({description})")
return config本番デプロイチェックリスト
プロダクション環境に Gemma 4 + Antigravity を導入する前に、以下の項目を必ずチェックすること：
- [ ] メモリ容量が 13GB 以上確保できているか
- [ ] キャッシング機能が実装され、キャッシュヒット率が 30% 以上
- [ ] バッチ処理が導入され、複数クエリの処理時間が 40% 削減
- [ ] API キーが環境変数から取得される（コード直書き禁止）
- [ ] レート制限が実装されている
- [ ] 推論パラメータが用途に応じて設定されている
- [ ] エラーハンドリングとフォールバック処理がある
- [ ] 推論速度が 2 秒以下
- [ ] メモリ使用量が 13GB を超えていない
- [ ] ログ出力が実装されている
Android AICore でのオンデバイス実装
Android プラットフォームで Gemma 4 をオンデバイス実行する場合、Google が提供する AICore ランタイムが選択肢になります。ファインチューニング済みモデルをエージェントから呼び出す具体的な実装と、クラウド API とのハイブリッド戦略を扱います。
Gemma 4 AICore の基本実装
ステップ1: モデルの利用可能性チェック
// AICoreManager.kt
import com.google.android.aicore.AICore
import com.google.android.aicore.GenerativeModel
import com.google.android.aicore.AvailabilityStatus
class AICoreManager(private val context: Context) {
/**
* Gemma 4がデバイスで利用可能かチェック
* @return true: 利用可能、false: 利用不可（クラウドAPIにフォールバック）
*/
fun checkAvailability(
onAvailable: (GenerativeModel) -> Unit,
onUnavailable: (String) -> Unit
) {
AICore.checkAvailability(
context = context,
modelName = "gemma-4-e2b-it" // Edge 2B Instruction Tuned
).addOnSuccessListener { status ->
when (status) {
AvailabilityStatus.AVAILABLE -> {
// モデルが既にダウンロード済み
initModel(onAvailable, onUnavailable)
}
AvailabilityStatus.DOWNLOADABLE -> {
// ダウンロードが必要
downloadAndInit(onAvailable, onUnavailable)
}
AvailabilityStatus.NOT_SUPPORTED -> {
onUnavailable("このデバイスはAICoreをサポートしていません")
}
}
}.addOnFailureListener { e ->
onUnavailable("利用可能性チェックに失敗: ${e.message}")
}
}
private fun initModel(
onAvailable: (GenerativeModel) -> Unit,
onUnavailable: (String) -> Unit
) {
AICore.getGenerativeModel(
modelName = "gemma-4-e2b-it",
context = context
).addOnSuccessListener { model ->
onAvailable(model)
}.addOnFailureListener { e ->
onUnavailable("モデル初期化に失敗: ${e.message}")
}
}
}ステップ2: オンデバイスAIエージェントの実装
// OnDeviceAIAgent.kt
import com.google.android.aicore.GenerativeModel
import com.google.android.aicore.Content
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.suspendCancellableCoroutine
import kotlin.coroutines.resume
import kotlin.coroutines.resumeWithException
class OnDeviceAIAgent(private val model: GenerativeModel) {
private val conversationHistory = mutableListOf<Content>()
// システムプロンプト（エージェントの役割定義）
private val systemPrompt = """
あなたは親切なアシスタントです。
ユーザーの質問に簡潔に日本語で答えてください。
センシティブな情報は扱わないでください。
""".trimIndent()
/**
* テキストを生成（suspend関数・コルーチン対応）
*/
suspend fun generate(userMessage: String): String =
suspendCancellableCoroutine { continuation ->
// 会話履歴にユーザーメッセージを追加
conversationHistory.add(
Content.Builder()
.setRole("user")
.addText(userMessage)
.build()
)
// AICore APIでオンデバイス推論実行
model.generateContent(conversationHistory)
.addOnSuccessListener { response ->
val text = response.text ?: ""
// AIの応答を履歴に追加
conversationHistory.add(
Content.Builder()
.setRole("model")
.addText(text)
.build()
)
continuation.resume(text)
}
.addOnFailureListener { e ->
continuation.resumeWithException(e)
}
}
/**
* ストリーミング生成（リアルタイムレスポンス）
*/
fun generateStream(userMessage: String): Flow<String> = flow {
model.generateContentStream(
listOf(
Content.Builder()
.setRole("user")
.addText(userMessage)
.build()
)
).forEach { chunk ->
chunk.text?.let { emit(it) }
}
}
fun clearHistory() {
conversationHistory.clear()
}
}ステップ3: UIとの統合（ViewModel）
// AIAgentViewModel.kt
import androidx.lifecycle.ViewModel
import androidx.lifecycle.viewModelScope
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.launch
class AIAgentViewModel : ViewModel() {
private var agent: OnDeviceAIAgent? = null
private var aicoreManager: AICoreManager? = null
private val _uiState = MutableStateFlow<AgentUiState>(AgentUiState.Initializing)
val uiState: StateFlow<AgentUiState> = _uiState
private val _messages = MutableStateFlow<List<ChatMessage>>(emptyList())
val messages: StateFlow<List<ChatMessage>> = _messages
fun initialize(context: Context) {
aicoreManager = AICoreManager(context)
aicoreManager?.checkAvailability(
onAvailable = { model ->
agent = OnDeviceAIAgent(model)
_uiState.value = AgentUiState.Ready
},
onUnavailable = { reason ->
// オンデバイス不可 → クラウドAPIにフォールバック
_uiState.value = AgentUiState.FallingBackToCloud(reason)
}
)
}
fun sendMessage(text: String) {
viewModelScope.launch {
// ユーザーメッセージを表示
_messages.value += ChatMessage(role = "user", text = text)
_uiState.value = AgentUiState.Generating
try {
// オンデバイス推論実行
val response = agent?.generate(text) ?: "エラー: エージェントが初期化されていません"
_messages.value += ChatMessage(role = "assistant", text = response)
_uiState.value = AgentUiState.Ready
} catch (e: Exception) {
_uiState.value = AgentUiState.Error(e.message ?: "不明なエラー")
}
}
}
}
sealed class AgentUiState {
object Initializing : AgentUiState()
object Ready : AgentUiState()
object Generating : AgentUiState()
data class FallingBackToCloud(val reason: String) : AgentUiState()
data class Error(val message: String) : AgentUiState()
}
data class ChatMessage(val role: String, val text: String)クラウドAPIとのハイブリッド戦略
すべてのデバイスがAICoreをサポートするわけではないため、フォールバック戦略が重要です。
// HybridAIStrategy.kt
class HybridAIStrategy(
private val onDeviceAgent: OnDeviceAIAgent?,
private val cloudApi: GeminiApiClient
) {
suspend fun generate(prompt: String): String {
return if (onDeviceAgent != null) {
// オンデバイス優先（高速・無料・プライバシー保護）
try {
onDeviceAgent.generate(prompt)
} catch (e: Exception) {
// 失敗時はクラウドにフォールバック
cloudApi.generate(prompt)
}
} else {
// デバイス非対応：クラウドAPI使用
cloudApi.generate(prompt)
}
}
}Android Studio + Antigravity の開発ガイドも参照すると、Android開発環境全体のセットアップ方法が確認できます。
全体を振り返って
Gemma 4のオンデバイスAI統合は、プロトタイプから本番運用まで一貫したワークフローで実現できる段階に到達しています。ハイブリッドアーキテクチャの設計、LoRAファインチューニングによるカスタムモデル構築、Antigravityエージェントとの統合、そしてCI/CDパイプラインによる自動化まで、本記事で解説した手法を組み合わせれば、エッジAIを軸とした競争力のあるプロダクトを開発できます。
まずは自分のプロジェクトで最も効果が高い部分から始めてみてください。エッジデバイスでのAI体験をさらに深めたい方はAntigravity × Core ML: iOSオンデバイスAI開発の完全ガイドも参照してください。AgentKit 2.0のマルチエージェント機能と組み合わせるならAntigravity AgentKit 2.0 マルチエージェント開発実践ガイドが役立つでしょう。
モデル最適化からデプロイまでの知識を総合的に身につけることができます。