Gemma 4 × Antigravity — iOS・Android 両対応オンデバイスAI 実装ガイド

# Python 環境（モデル変換に使用）
python3 -m venv gemma4-env
source gemma4-env/bin/activate
pip install torch==2.3.0 coremltools==7.2 tensorflow==2.16.0 \
  ai-edge-torch==0.2.0 keras-nlp==0.14.0

{
  "context": {
    "pythonEnv": "./gemma4-env",
    "platforms": ["ios", "android"],
    "modelFamily": "gemma4",
    "targetSizes": ["1b", "4b"]
  },
  "rules": [
    "Use coremltools 7.2 for iOS conversion",
    "Use ai-edge-torch for Android TFLite conversion",
    "Apply INT8 quantization unless precision test fails"
  ]
}

# convert_gemma4_coreml.py
# 何をするか: Gemma 4 1B PyTorch モデルを Core ML 形式に変換する
# なぜこう書くか: coremltools 7.2 は LLM 変換に新しい StatefulModel API を使うため
 
import coremltools as ct
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM
import numpy as np
 
MODEL_ID = "google/gemma-4-1b-pt"
OUTPUT_PATH = "Gemma4_1B.mlpackage"
 
def load_model():
    tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
    model = AutoModelForCausalLM.from_pretrained(
        MODEL_ID,
        torch_dtype=torch.float32,  # float16はCore ML変換で不安定
        device_map="cpu"
    )
    return tokenizer, model
 
def trace_and_convert(model, tokenizer):
    # 代表的な入力でトレース
    sample_input = tokenizer(
        "こんにちは、今日の天気は",
        return_tensors="pt",
        max_length=128,
        padding="max_length"
    )
 
    # Core ML 変換: stateful cache を使うと生成速度が向上する
    traced = torch.jit.trace(
        model,
        (sample_input["input_ids"], sample_input["attention_mask"]),
        strict=False
    )
 
    mlmodel = ct.convert(
        traced,
        inputs=[
            ct.TensorType(name="input_ids", shape=(1, 128), dtype=np.int32),
            ct.TensorType(name="attention_mask", shape=(1, 128), dtype=np.int32),
        ],
        outputs=[ct.TensorType(name="logits")],
        compute_units=ct.ComputeUnit.CPU_AND_NE,  # Neural Engine を優先
        minimum_deployment_target=ct.target.iOS17,
        # INT8量子化: 精度とサイズのバランスが最も良い
        pass_pipeline=ct.PassPipeline.DEFAULT_PALETTIZATION,
    )
 
    mlmodel.save(OUTPUT_PATH)
    print(f"✅ Core ML モデルを保存しました: {OUTPUT_PATH}")
    print(f"   サイズ: {os.path.getsize(OUTPUT_PATH) / 1e6:.1f} MB")
 
if __name__ == "__main__":
    tokenizer, model = load_model()
    trace_and_convert(model, tokenizer)

// Gemma4InferenceEngine.swift
// 何をするか: Core ML モデルをラップし、トークン逐次生成を AsyncStream で提供する
// なぜ AsyncStream: SwiftUI の .task モディファイアと相性が良く、キャンセル対応も簡単
 
import CoreML
import Combine
 
@MainActor
final class Gemma4InferenceEngine: ObservableObject {
    @Published var generatedText: String = ""
    @Published var isGenerating: Bool = false
    @Published var errorMessage: String?
 
    private var mlModel: MLModel?
    private var tokenizer: BPETokenizer  // 自前実装 or SentencePiece Swift バインディング
 
    init() {
        loadModel()
    }
 
    private func loadModel() {
        do {
            let config = MLModelConfiguration()
            config.computeUnits = .cpuAndNeuralEngine
            // On-Demand Resource でダウンロードしたモデルのパスを使う
            let modelURL = Bundle.main.url(
                forResource: "Gemma4_1B",
                withExtension: "mlpackage"
            )!
            mlModel = try MLModel(contentsOf: modelURL, configuration: config)
        } catch {
            errorMessage = "モデルの読み込みに失敗しました: \(error.localizedDescription)"
        }
    }
 
    func generate(prompt: String, maxNewTokens: Int = 200) -> AsyncStream<String> {
        AsyncStream { continuation in
            Task.detached(priority: .userInitiated) {
                await MainActor.run {
                    self.isGenerating = true
                    self.generatedText = ""
                }
 
                guard let model = self.mlModel else {
                    continuation.finish()
                    return
                }
 
                // トークナイズ
                let inputIds = self.tokenizer.encode(prompt, maxLength: 512)
                var outputTokens: [Int] = []
 
                // 逐次生成ループ
                for _ in 0..<maxNewTokens {
                    do {
                        let input = try MLDictionaryFeatureProvider(dictionary: [
                            "input_ids": MLMultiArray(inputIds + outputTokens),
                            "attention_mask": MLMultiArray(
                                Array(repeating: 1, count: (inputIds + outputTokens).count)
                            )
                        ])
                        let output = try model.prediction(from: input)
                        guard let logits = output.featureValue(for: "logits")?.multiArrayValue else {
                            break
                        }
 
                        // グリーディデコード（本番では temperature sampling を推奨）
                        let nextToken = argmax(logits: logits)
                        if nextToken == tokenizer.eosTokenId { break }
 
                        outputTokens.append(nextToken)
                        let decodedToken = self.tokenizer.decode([nextToken])
 
                        await MainActor.run {
                            self.generatedText += decodedToken
                        }
                        continuation.yield(decodedToken)
 
                    } catch {
                        // エラー時は中断（ループ継続しない）
                        await MainActor.run {
                            self.errorMessage = error.localizedDescription
                        }
                        break
                    }
                }
 
                await MainActor.run { self.isGenerating = false }
                continuation.finish()
            }
        }
    }
 
    private func argmax(logits: MLMultiArray) -> Int {
        var maxVal: Double = -Double.infinity
        var maxIdx: Int = 0
        for i in 0..<logits.count {
            let val = logits[i].doubleValue
            if val > maxVal { maxVal = val; maxIdx = i }
        }
        return maxIdx
    }
}

struct ChatView: View {
    @StateObject private var engine = Gemma4InferenceEngine()
    @State private var inputText = ""
 
    var body: some View {
        VStack {
            ScrollView {
                Text(engine.generatedText)
                    .padding()
                    .frame(maxWidth: .infinity, alignment: .leading)
            }
            if engine.isGenerating {
                ProgressView("生成中...")
            }
            HStack {
                TextField("メッセージを入力", text: $inputText)
                    .textFieldStyle(.roundedBorder)
                Button("送信") {
                    let prompt = inputText
                    inputText = ""
                    Task {
                        for await _ in engine.generate(prompt: prompt) {}
                    }
                }
                .disabled(engine.isGenerating)
            }
            .padding()
        }
    }
}

# convert_gemma4_tflite.py
# 何をするか: Gemma 4 1B を TFLite 形式に変換し、INT8 量子化を適用する
# なぜ ai-edge-torch: Google 公式の変換ライブラリで Gemma 系モデルへの最適化が含まれる
 
import ai_edge_torch
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
 
MODEL_ID = "google/gemma-4-1b-pt"
OUTPUT_PATH = "gemma4_1b.tflite"
 
def convert_to_tflite():
    tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
    model = AutoModelForCausalLM.from_pretrained(
        MODEL_ID,
        torch_dtype=torch.float32,
    )
    model.eval()
 
    # ai-edge-torch による変換（TFLite + INT8 量子化）
    sample_inputs = (
        torch.randint(0, 256000, (1, 512)),  # input_ids
        torch.ones(1, 512, dtype=torch.long),  # attention_mask
    )
 
    edge_model = ai_edge_torch.convert(
        model,
        sample_inputs,
        quant_config=ai_edge_torch.quantize.pt2e_quantizer.PT2EQuantConfig(
            global_config=ai_edge_torch.quantize.quant_config.QuantConfig(
                weight_kwargs={"num_bits": 8, "symmetric": True},
                activation_kwargs={"num_bits": 8, "symmetric": False},
            )
        ),
    )
 
    edge_model.export(OUTPUT_PATH)
 
    import os
    size_mb = os.path.getsize(OUTPUT_PATH) / 1e6
    print(f"✅ TFLite モデル保存完了: {OUTPUT_PATH} ({size_mb:.1f} MB)")
    return size_mb
 
if __name__ == "__main__":
    size_mb = convert_to_tflite()
    if size_mb > 1000:
        print("⚠️  サイズが 1GB を超えています。INT4量子化も検討してください")

// Gemma4AiCoreEngine.kt
// 何をするか: Android AI Core API 経由で Gemma 4 を呼び出す
// なぜ AI Core: システムが管理する共有モデルを使えるため、アプリのバイナリに含める必要がない
 
import android.ai.core.AIModel
import android.ai.core.GenerativeModel
import android.ai.core.GenerativeModelParams
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
 
class Gemma4AiCoreEngine(private val context: Context) {
 
    private var generativeModel: GenerativeModel? = null
 
    suspend fun initialize(): Result<Unit> = runCatching {
        val params = GenerativeModelParams.Builder()
            .setModelName("gemma-4-1b")  // AI Core が管理するモデル名
            .setMaxOutputTokens(512)
            .setTemperature(0.7f)
            .build()
 
        generativeModel = AIModel.getGenerativeModel(context, params)
    }
 
    fun generate(prompt: String): Flow<String> = flow {
        val model = generativeModel
            ?: throw IllegalStateException("モデルが初期化されていません。initialize() を先に呼んでください")
 
        // ストリーミング生成
        model.generateContentStream(prompt).collect { chunk ->
            chunk.text?.let { emit(it) }
        }
    }
 
    // TFLite フォールバック（AI Core 非対応端末向け）
    fun isAiCoreAvailable(): Boolean {
        return try {
            AIModel.isAvailable(context, "gemma-4-1b")
        } catch (e: Exception) {
            false
        }
    }
}
 
// 使用側: AI Core と TFLite を自動切り替え
class InferenceRepository(private val context: Context) {
    private val aiCoreEngine = Gemma4AiCoreEngine(context)
    private val tfliteEngine = Gemma4TfliteEngine(context)  // TFLite実装
 
    suspend fun generate(prompt: String): Flow<String> {
        return if (aiCoreEngine.isAiCoreAvailable()) {
            aiCoreEngine.initialize()
            aiCoreEngine.generate(prompt)
        } else {
            tfliteEngine.generate(prompt)  // フォールバック
        }
    }
}

<!-- PrivacyInfo.xcprivacy -->
<?xml version="1.0" encoding="UTF-8"?>
<!DOCTYPE plist PUBLIC "-//Apple//DTD PLIST 1.0//EN"
  "http://www.apple.com/DTDs/PropertyList-1.0.dtd">
<plist version="1.0">
<dict>
  <key>NSPrivacyAccessedAPITypes</key>
  <array>
    <dict>
      <key>NSPrivacyAccessedAPIType</key>
      <string>NSPrivacyAccessedAPICategoryFileTimestamp</string>
      <key>NSPrivacyAccessedAPITypeReasons</key>
      <array>
        <string>C617.1</string>
        <!-- キャッシュファイルの更新確認に使用 -->
      </array>
    </dict>
  </array>
  <key>NSPrivacyCollectedDataTypes</key>
  <array/>
  <!-- オンデバイス AI のため外部送信データなし -->
  <key>NSPrivacyTracking</key>
  <false/>
</dict>
</plist>

// 正しい実装: AssetPackManager 経由でモデルファイルを取得する
class ModelDownloader(private val activity: ComponentActivity) {
    private val assetPackManager = AssetPackManagerFactory.getInstance(activity)
 
    fun downloadModelIfNeeded(onReady: (File) -> Unit, onProgress: (Float) -> Unit) {
        val packNames = listOf("gemma4_model_pack")
        val packStates = assetPackManager.getPackStates(packNames).result
 
        val packState = packStates.packStates()["gemma4_model_pack"]
        when (packState?.status()) {
            AssetPackStatus.DOWNLOADING -> {
                val total = packState.totalBytesToDownload()
                val downloaded = packState.bytesDownloaded()
                onProgress(downloaded.toFloat() / total)
            }
            AssetPackStatus.COMPLETED -> {
                val assetDir = assetPackManager.getPackLocation("gemma4_model_pack")
                    ?.assetsPath()
                val modelFile = File("$assetDir/models/gemma4_1b.tflite")
                onReady(modelFile)
            }
            else -> {
                assetPackManager.fetch(packNames)
            }
        }
    }
}

# 日本語を含むキャリブレーションデータセットの用意
CALIBRATION_DATA = [
    "今日の天気は晴れです。",
    "Pythonでリストを逆順にするには reversed() を使います。",
    "機械学習モデルの評価指標には精度・再現率・F値があります。",
    "SwiftUIでViewを構築するときは body プロパティを実装します。",
    # ...実際の用途に近い100〜500サンプルを用意する
]