Deploying Gemma on Edge Devices: The Complete Guide to On-Device Edge AI in 2025

The AI revolution is moving from the cloud to your pocket. While GPT-4 and Claude dominate headlines, a quieter revolution is happening at the edge: Small Language Models (SLMs) powered by neural processing units (NPUs), running locally with zero latency and complete privacy.

Google's Gemma represents the cutting edge of efficient AI. Today, you'll learn how to deploy Gemma 2B on edge devices and build production-ready applications using quantization, model optimization, and RAG (Retrieval-Augmented Generation).

Why Gemma for Edge AI?

Gemma 2B is Google's answer to democratizing AI. At just 2 billion parameters, it delivers performance comparable to much larger models while being small enough for smartphones, IoT devices, and edge servers.

The value proposition is clear:

• Privacy-first: Data never leaves the device
• Zero-latency inference: No network round-trips
• Cost-effective: No API fees or cloud compute
• Offline-capable: Works without internet connectivity
• Carbon-efficient: Lower energy consumption than cloud inference

Understanding the Edge AI Landscape in 2025

The convergence of TinyML, neuromorphic computing, and federated learning is reshaping AI deployment. Key trends include:

• Multi-modal SLMs: Vision + language in under 4GB
• Quantization-aware training: Models designed for INT4/INT8 from scratch
• Neural architecture search (NAS): Auto-optimized models for specific hardware
• On-device fine-tuning: Personalized AI without cloud sync

Environment Setup

# Create isolated environment
python -m venv gemma_edge_env
source gemma_edge_env/bin/activate  # Windows: gemma_edge_env\Scripts\activate

# Install core dependencies
pip install torch transformers accelerate sentencepiece protobuf
pip install bitsandbytes  # For quantization
pip install optimum onnx onnxruntime  # For model optimization
pip install huggingface-hub  # For model downloads

Step 1: Loading Gemma 2B with Quantization

Gemma requires Hugging Face authentication. Get your token at https://huggingface.co/settings/tokens

import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
import warnings
warnings.filterwarnings('ignore')

# Authenticate with Hugging Face
from huggingface_hub import login
login(token="your_hf_token_here")  # Replace with your token

# 4-bit quantization for edge deployment
quantization_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_compute_dtype=torch.bfloat16,  # Use bfloat16 for better stability
    bnb_4bit_use_double_quant=True,  # Nested quantization for extra compression
    bnb_4bit_quant_type="nf4"  # NormalFloat4 quantization
)

# Load Gemma 2B
model_name = "google/gemma-2b-it"  # Instruction-tuned variant

print("Loading Gemma 2B with 4-bit quantization...")
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    quantization_config=quantization_config,
    device_map="auto",
    torch_dtype=torch.bfloat16
)

print(f"✓ Model loaded successfully!")
print(f"Memory footprint: {model.get_memory_footprint() / 1e9:.2f} GB")
# Expected: ~1.2-1.5GB (vs 5GB unquantized)

Step 2: Building a High-Performance Inference Pipeline

def generate_response(
    prompt,
    model,
    tokenizer,
    max_new_tokens=256,
    temperature=0.7,
    top_p=0.9,
    repetition_penalty=1.1
):
    """
    Optimized inference for edge devices with Gemma
    """
    # Format for instruction-tuned Gemma
    formatted_prompt = f"<start_of_turn>user\n{prompt}<end_of_turn>\n<start_of_turn>model\n"

    # Tokenize
    inputs = tokenizer(formatted_prompt, return_tensors="pt").to(model.device)

    # Generate with optimized settings
    with torch.inference_mode():  # Faster than torch.no_grad()
        outputs = model.generate(
            **inputs,
            max_new_tokens=max_new_tokens,
            temperature=temperature,
            top_p=top_p,
            repetition_penalty=repetition_penalty,
            do_sample=True,
            pad_token_id=tokenizer.pad_token_id,
            eos_token_id=tokenizer.eos_token_id
        )

    # Decode and clean response
    full_response = tokenizer.decode(outputs[0], skip_special_tokens=True)
    # Extract only the model's response
    response = full_response.split("<start_of_turn>model\n")[-1].strip()

    return response

# Test inference
prompt = "Explain neural networks in 3 sentences."
response = generate_response(prompt, model, tokenizer)
print(f"\nPrompt: {prompt}")
print(f"Response: {response}")

Use Case #1: On-Device Personal Assistant

Build a privacy-preserving personal assistant that handles queries without cloud connectivity.

class GemmaPersonalAssistant:
    def __init__(self, model_name="google/gemma-2b-it"):
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)

        quantization_config = BitsAndBytesConfig(
            load_in_4bit=True,
            bnb_4bit_compute_dtype=torch.bfloat16,
            bnb_4bit_use_double_quant=True
        )

        self.model = AutoModelForCausalLM.from_pretrained(
            model_name,
            quantization_config=quantization_config,
            device_map="auto"
        )

        # Conversation history
        self.conversation_history = []

    def chat(self, user_message, max_history=5):
        """
        Interactive chat with conversation memory
        """
        # Add user message to history
        self.conversation_history.append(f"User: {user_message}")

        # Build context from recent history
        context = "\n".join(self.conversation_history[-max_history:])

        prompt = f"""You are a helpful personal assistant. Based on the conversation:

{context}

Provide a concise and helpful response."""

        response = generate_response(
            prompt,
            self.model,
            self.tokenizer,
            max_new_tokens=200,
            temperature=0.7
        )

        # Add assistant response to history
        self.conversation_history.append(f"Assistant: {response}")

        return response

    def clear_history(self):
        self.conversation_history = []

# Usage
assistant = GemmaPersonalAssistant()

print("\n=== Personal Assistant Demo ===")
print(assistant.chat("What's a good breakfast for energy?"))
print("\n" + assistant.chat("Make it vegetarian"))
print("\n" + assistant.chat("How many calories would that be?"))

Use Case #2: Edge-Based Content Moderation

Real-time content filtering for social platforms, chat apps, and gaming communities.

class EdgeContentModerator:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.categories = [
            "hate_speech",
            "violence",
            "adult_content",
            "harassment",
            "spam"
        ]

    def moderate(self, text):
        """
        Classify content safety in real-time
        """
        prompt = f"""Analyze this text for policy violations. Respond with ONLY a JSON object.

Text: "{text}"

Categories: {', '.join(self.categories)}

Format:
{{"is_safe": true/false, "violations": ["category1", "category2"], "confidence": 0.0-1.0}}

Response:"""

        response = generate_response(
            prompt,
            self.model,
            self.tokenizer,
            max_new_tokens=100,
            temperature=0.3  # Lower temperature for consistent formatting
        )

        # Parse response (in production, use proper JSON parsing with error handling)
        import json
        try:
            result = json.loads(response.strip())
            return result
        except:
            return {"is_safe": True, "violations": [], "confidence": 0.5}

    def batch_moderate(self, texts):
        """Process multiple texts efficiently"""
        return [self.moderate(text) for text in texts]

# Demo
moderator = EdgeContentModerator(model, tokenizer)

test_cases = [
    "Great product, highly recommend!",
    "This is spam buy now!!!",
    "Check out my blog for cooking tips"
]

print("\n=== Content Moderation Demo ===")
for text in test_cases:
    result = moderator.moderate(text)
    print(f"\nText: {text}")
    print(f"Result: {result}")

Use Case #3: Privacy-Preserving Healthcare Assistant

Medical information assistant for clinical settings where HIPAA compliance is critical.

class HealthcareAssistant:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.disclaimer = "\n\n⚕️ This is for educational purposes only. Consult healthcare professionals for medical advice."

    def medical_query(self, question, patient_context=None):
        """
        Answer medical questions with context
        """
        base_prompt = f"""You are a medical information assistant. Provide accurate, evidence-based information.

Question: {question}"""

        if patient_context:
            base_prompt += f"\n\nPatient context: {patient_context}"

        base_prompt += "\n\nProvide a clear, concise answer:"

        response = generate_response(
            base_prompt,
            self.model,
            self.tokenizer,
            max_new_tokens=300,
            temperature=0.4
        )

        return response + self.disclaimer

    def symptom_checker(self, symptoms):
        """
        Preliminary symptom analysis
        """
        prompt = f"""Based on these symptoms, suggest possible conditions and when to seek care:

Symptoms: {symptoms}

Provide:
1. Possible conditions (most to least likely)
2. Severity assessment (mild/moderate/urgent)
3. Recommended actions"""

        response = generate_response(
            prompt,
            self.model,
            self.tokenizer,
            max_new_tokens=400,
            temperature=0.3
        )

        return response + self.disclaimer

# Demo
health_assistant = HealthcareAssistant(model, tokenizer)

print("\n=== Healthcare Assistant Demo ===")
query = health_assistant.medical_query(
    "What are the common side effects of statins?",
    patient_context="65-year-old with high cholesterol"
)
print(query)

Use Case #4: On-Device Code Assistant

Code generation and debugging without sending proprietary code to cloud APIs.

class EdgeCodeAssistant:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer

    def generate_code(self, description, language="python"):
        """
        Generate code from natural language
        """
        prompt = f"""Generate clean, efficient {language} code for:

Task: {description}

Requirements:
- Include comments
- Follow best practices
- Handle edge cases

Code:"""

        response = generate_response(
            prompt,
            self.model,
            self.tokenizer,
            max_new_tokens=400,
            temperature=0.5
        )

        return response

    def debug_code(self, code, error_message):
        """
        Debug code with error context
        """
        prompt = f"""Debug this code:

Code:


\{code}\



Error: {error_message}

Provide:
1. Root cause
2. Fixed code
3. Explanation"""

        response = generate_response(
            prompt,
            self.model,
            self.tokenizer,
            max_new_tokens=500,
            temperature=0.3
        )

        return response

    def explain_code(self, code):
        """
        Explain code functionality
        """
        prompt = f"""Explain this code in simple terms:



\{code\}



Explanation:"""

        response = generate_response(
            prompt,
            self.model,
            self.tokenizer,
            max_new_tokens=300
        )

        return response

# Demo
code_assistant = EdgeCodeAssistant(model, tokenizer)

print("\n=== Code Assistant Demo ===")
code = code_assistant.generate_code(
    "Create a function to calculate Fibonacci numbers using memoization"
)
print("Generated Code:")
print(code)

Use Case #5: RAG-Powered Knowledge Base

Combine Gemma with vector search for intelligent document retrieval.

import numpy as np
from typing import List, Tuple

class SimpleVectorStore:
    """Minimal vector store for demonstration"""
    def __init__(self):
        self.documents = []
        self.embeddings = []

    def add_document(self, text: str, embedding: np.ndarray):
        self.documents.append(text)
        self.embeddings.append(embedding)

    def search(self, query_embedding: np.ndarray, top_k: int = 3) -> List[str]:
        """Simple cosine similarity search"""
        if not self.embeddings:
            return []

        # Calculate similarities
        similarities = []
        for emb in self.embeddings:
            similarity = np.dot(query_embedding, emb) / (
                np.linalg.norm(query_embedding) * np.linalg.norm(emb)
            )
            similarities.append(similarity)

        # Get top-k
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.documents[i] for i in top_indices]

class RAGAssistant:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.vector_store = SimpleVectorStore()

    def simple_embed(self, text: str) -> np.ndarray:
        """Simple embedding using token frequencies (production: use proper embeddings)"""
        tokens = self.tokenizer.encode(text)
        # Create basic embedding from token statistics
        embedding = np.zeros(300)
        for i, token in enumerate(tokens[:300]):
            embedding[i % 300] += token
        return embedding / (np.linalg.norm(embedding) + 1e-8)

    def add_knowledge(self, documents: List[str]):
        """Add documents to knowledge base"""
        for doc in documents:
            embedding = self.simple_embed(doc)
            self.vector_store.add_document(doc, embedding)

    def query(self, question: str) -> str:
        """Query with retrieval augmentation"""
        # Retrieve relevant documents
        query_embedding = self.simple_embed(question)
        relevant_docs = self.vector_store.search(query_embedding, top_k=2)

        # Build context
        context = "\n\n".join([f"Document {i+1}: {doc}"
                               for i, doc in enumerate(relevant_docs)])

        prompt = f"""Use the following documents to answer the question accurately.

{context}

Question: {question}

Answer based on the documents:"""

        response = generate_response(
            prompt,
            self.model,
            self.tokenizer,
            max_new_tokens=300
        )

        return response

# Demo
rag_assistant = RAGAssistant(model, tokenizer)

# Add company knowledge base
knowledge_base = [
    "Our company offers 24/7 customer support through phone, email, and live chat. Support hours are Monday-Friday 9AM-9PM EST.",
    "We have a 30-day return policy for all products. Items must be unused and in original packaging. Refunds are processed within 5-7 business days.",
    "Shipping is free for orders over $50. Standard shipping takes 3-5 business days. Express shipping is available for $9.99 and takes 1-2 business days.",
    "We accept Visa, Mastercard, American Express, PayPal, and Apple Pay. All transactions are encrypted and PCI compliant."
]

rag_assistant.add_knowledge(knowledge_base)

print("\n=== RAG Assistant Demo ===")
print(rag_assistant.query("What is your return policy?"))
print("\n" + rag_assistant.query("How long does shipping take?"))

Use Case #6: Real-Time Translation for IoT

Low-latency translation for smart devices and wearables.

class EdgeTranslator:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer

    def translate(self, text, source_lang, target_lang):
        """
        Translate text between languages
        """
        prompt = f"""Translate this text from {source_lang} to {target_lang}. Provide ONLY the translation.

Text: {text}

Translation:"""

        response = generate_response(
            prompt,
            self.model,
            self.tokenizer,
            max_new_tokens=200,
            temperature=0.3
        )

        return response.strip()

    def detect_language(self, text):
        """
        Detect input language
        """
        prompt = f"""Detect the language of this text. Respond with ONLY the language name.

Text: {text}

Language:"""

        response = generate_response(
            prompt,
            self.model,
            self.tokenizer,
            max_new_tokens=10,
            temperature=0.1
        )

        return response.strip()

# Demo
translator = EdgeTranslator(model, tokenizer)

print("\n=== Translation Demo ===")
print(translator.translate("Hello, how are you?", "English", "Spanish"))
print(translator.translate("Bonjour le monde", "French", "English"))

Step 3: Advanced Optimization with ONNX Runtime

Convert Gemma to ONNX for 2-3x faster inference on edge devices.

from optimum.onnxruntime import ORTModelForCausalLM
from optimum.onnxruntime.configuration import AutoQuantizationConfig

# Export to ONNX with dynamic quantization
print("\n=== Converting to ONNX ===")
onnx_path = "./gemma_onnx"

# Load and export
ort_model = ORTModelForCausalLM.from_pretrained(
    model_name,
    export=True,
    provider="CPUExecutionProvider"  # Or CUDAExecutionProvider for GPU
)

# Apply quantization
quantization_config = AutoQuantizationConfig.avx512_vnni(
    is_static=False,
    per_channel=True
)

# Save optimized model
ort_model.save_pretrained(onnx_path)
print(f"✓ ONNX model saved to {onnx_path}")

# Load for inference
tokenizer_onnx = AutoTokenizer.from_pretrained(model_name)
optimized_model = ORTModelForCausalLM.from_pretrained(
    onnx_path,
    provider="CPUExecutionProvider"
)

print("✓ Optimized ONNX model loaded and ready for inference")

Performance Benchmarking Suite

import time
import psutil
import os
from statistics import mean, stdev

class EdgeBenchmark:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.process = psutil.Process(os.getpid())

    def benchmark_latency(self, prompts, num_runs=10):
        """
        Measure inference latency
        """
        # Warm-up
        generate_response(prompts[0], self.model, self.tokenizer, max_new_tokens=50)

        latencies = []
        for prompt in prompts:
            for _ in range(num_runs):
                start = time.perf_counter()
                _ = generate_response(prompt, self.model, self.tokenizer, max_new_tokens=50)
                latencies.append((time.perf_counter() - start) * 1000)

        return {
            "mean_ms": mean(latencies),
            "std_ms": stdev(latencies) if len(latencies) > 1 else 0,
            "min_ms": min(latencies),
            "max_ms": max(latencies),
            "p95_ms": sorted(latencies)[int(len(latencies) * 0.95)]
        }

    def benchmark_throughput(self, prompts, duration_seconds=60):
        """
        Measure queries per second
        """
        start_time = time.time()
        queries_processed = 0

        while time.time() - start_time < duration_seconds:
            for prompt in prompts:
                _ = generate_response(prompt, self.model, self.tokenizer, max_new_tokens=50)
                queries_processed += 1

                if time.time() - start_time >= duration_seconds:
                    break

        elapsed = time.time() - start_time
        return {
            "qps": queries_processed / elapsed,
            "total_queries": queries_processed,
            "duration_seconds": elapsed
        }

    def benchmark_memory(self):
        """
        Measure memory footprint
        """
        return {
            "rss_mb": self.process.memory_info().rss / 1024 / 1024,
            "vms_mb": self.process.memory_info().vms / 1024 / 1024,
            "model_gb": self.model.get_memory_footprint() / 1e9
        }

    def run_full_benchmark(self):
        """
        Complete benchmark suite
        """
        test_prompts = [
            "What is machine learning?",
            "Explain quantum computing briefly.",
            "Write a haiku about AI."
        ]

        print("\n=== Edge AI Benchmark Results ===")

        # Latency
        latency_results = self.benchmark_latency(test_prompts)
        print(f"\nLatency Metrics:")
        print(f"  Mean: {latency_results['mean_ms']:.2f}ms")
        print(f"  Std Dev: {latency_results['std_ms']:.2f}ms")
        print(f"  P95: {latency_results['p95_ms']:.2f}ms")
        print(f"  Min: {latency_results['min_ms']:.2f}ms")
        print(f"  Max: {latency_results['max_ms']:.2f}ms")

        # Memory
        memory_results = self.benchmark_memory()
        print(f"\nMemory Footprint:")
        print(f"  Model Size: {memory_results['model_gb']:.2f} GB")
        print(f"  RSS: {memory_results['rss_mb']:.2f} MB")

        return {
            "latency": latency_results,
            "memory": memory_results
        }

# Run benchmark
benchmark = EdgeBenchmark(model, tokenizer)
results = benchmark.run_full_benchmark()

Production Deployment Checklist

1. Model Optimization Pipeline

def optimize_for_production(model_name, output_path):
    """
    Complete optimization pipeline for edge deployment
    """
    from optimum.onnxruntime import ORTOptimizer, ORTModelForCausalLM
    from optimum.onnxruntime.configuration import OptimizationConfig

    # Load model
    model = ORTModelForCausalLM.from_pretrained(model_name, export=True)

    # Configure optimization
    optimization_config = OptimizationConfig(
        optimization_level=2,  # O2: extended graph optimizations
        optimize_for_gpu=False,  # CPU optimization
        fp16=False,  # Keep FP32 for CPU
        enable_transformers_specific_optimizations=True
    )

    # Optimize
    optimizer = ORTOptimizer.from_pretrained(model)
    optimizer.optimize(save_dir=output_path, optimization_config=optimization_config)

    print(f"✓ Production-optimized model saved to {output_path}")

# Example usage
# optimize_for_production("google/gemma-2b-it", "./gemma_production")

2. Error Handling and Fallbacks

class RobustEdgeAssistant:
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.max_retries = 3
        self.timeout_seconds = 30

    def safe_generate(self, prompt, fallback_response="I'm having trouble processing that request."):
        """
        Generation with error handling and timeouts
        """
        for attempt in range(self.max_retries):
            try:
                response = generate_response(
                    prompt,
                    self.model,
                    self.tokenizer,
                    max_new_tokens=200
                )

                # Validate response
                if response and len(response) > 10:
                    return response

            except torch.cuda.OutOfMemoryError:
                print("⚠️ GPU OOM, clearing cache...")
                torch.cuda.empty_cache()

            except Exception as e:
                print(f"⚠️ Attempt {attempt + 1} failed: {str(e)}")

        return fallback_response

3. Model Serving with FastAPI

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn

app = FastAPI(title="Gemma Edge API")

# Global model instance
edge_model = None
edge_tokenizer = None

class GenerationRequest(BaseModel):
    prompt: str
    max_tokens: int = 200
    temperature: float = 0.7

@app.on_event("startup")
async def load_model():
    global edge_model, edge_tokenizer
    print("Loading Gemma model...")
    edge_tokenizer = AutoTokenizer.from_pretrained("google/gemma-2b-it")

    quantization_config = BitsAndBytesConfig(
        load_in_4bit=True,
        bnb_4bit_compute_dtype=torch.bfloat16
    )

    edge_model = AutoModelForCausalLM.from_pretrained(
        "google/gemma-2b-it",
        quantization_config=quantization_config,
        device_map="auto"
    )
    print("✓ Model loaded successfully")

@app.post("/generate")
async def generate_text(request: GenerationRequest):
    if edge_model is None:
        raise HTTPException(status_code=503, detail="Model not loaded")

    try:
        response = generate_response(
            request.prompt,
            edge_model,
            edge_tokenizer,
            max_new_tokens=request.max_tokens,
            temperature=request.temperature
        )
        return {"response": response, "model": "gemma-2b-it"}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "model_loaded": edge_model is not None,
        "memory_gb": edge_model.get_memory_footprint() / 1e9 if edge_model else 0
    }

# Run with: uvicorn script_name:app --host 0.0.0.0 --port 8000

Best Practices for Edge AI in 2025

1. Hardware-Aware Optimization

def detect_and_optimize():
    """
    Auto-detect hardware and optimize accordingly
    """
    import platform

    # Check for Apple Silicon
    if platform.machine() == "arm64" and platform.system() == "Darwin":
        print("✓ Apple Silicon detected, using MPS backend")
        device = "mps"
    # Check for CUDA
    elif torch.cuda.is_available():
        print(f"✓ CUDA detected: {torch.cuda.get_device_name(0)}")
        device = "cuda"
    else:
        print("✓ Using CPU")
        device = "cpu"

    return device

2. Adaptive Batch Processing

def adaptive_batch_generate(prompts, model, tokenizer, max_memory_gb=2.0):
    """
    Dynamically adjust batch size based on available memory
    """
    # Start with batch size 1 and increase
    batch_size = 1
    results = []

    for i in range(0, len(prompts), batch_size):
        batch = prompts[i:i+batch_size]

        try:
            # Process batch
            batch_results = [
                generate_response(p, model, tokenizer, max_new_tokens=100)
                for p in batch
            ]
            results.extend(batch_results)

            # Increase batch size if memory allows
            current_memory = model.get_memory_footprint() / 1e9
            if current_memory < max_memory_gb * 0.8:
                batch_size = min(batch_size + 1, 8)

        except RuntimeError as e:
            if "out of memory" in str(e).lower():
                batch_size = max(1, batch_size // 2)
                torch.cuda.empty_cache()

    return results

3. Model Compression Techniques

# Pruning (conceptual example)
def prune_model(model, pruning_ratio=0.3):
    """
    Remove less important weights to reduce model size
    """
    import torch.nn.utils.prune as prune

    for name, module in model.named_modules():
        if isinstance(module, torch.nn.Linear):
            prune.l1_unstructured(module, name='weight', amount=pruning_ratio)
            prune.remove(module, 'weight')

    return model

# Knowledge distillation (conceptual)
def distill_knowledge(teacher_model, student_model, dataset):
    """
    Transfer knowledge from larger to smaller model
    """
    # This is a simplified outline
    # Production implementation requires full training loop
    pass

Monitoring and Observability

import logging
from datetime import datetime

class EdgeMonitor:
    def __init__(self, log_file="edge_ai_metrics.log"):
        logging.basicConfig(
            filename=log_file,
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger(__name__)
        self.metrics = {
            "total_requests": 0,
            "successful_requests": 0,
            "failed_requests": 0,
            "total_latency_ms": 0
        }

    def log_request(self, prompt, response, latency_ms, success=True):
        """Log inference request with metrics"""
        self.metrics["total_requests"] += 1

        if success:
            self.metrics["successful_requests"] += 1
            self.metrics["total_latency_ms"] += latency_ms

            self.logger.info(
                f"Request successful | Latency: {latency_ms:.2f}ms | "
                f"Prompt length: {len(prompt)} | Response length: {len(response)}"
            )
        else:
            self.metrics["failed_requests"] += 1
            self.logger.error(f"Request failed | Prompt: {prompt[:100]}")

    def get_metrics(self):
        """Get aggregated metrics"""
        if self.metrics["successful_requests"] > 0:
            avg_latency = (
                self.metrics["total_latency_ms"] /
                self.metrics["successful_requests"]
            )
        else:
            avg_latency = 0

        return {
            **self.metrics,
            "average_latency_ms": avg_latency,
            "success_rate": (
                self.metrics["successful_requests"] /
                max(self.metrics["total_requests"], 1)
            ) * 100
        }

    def alert_if_degraded(self, latency_threshold_ms=1000, error_rate_threshold=0.1):
        """Alert if performance degrades"""
        metrics = self.get_metrics()

        if metrics["average_latency_ms"] > latency_threshold_ms:
            self.logger.warning(
                f"⚠️ High latency detected: {metrics['average_latency_ms']:.2f}ms"
            )

        error_rate = 1 - (metrics["success_rate"] / 100)
        if error_rate > error_rate_threshold:
            self.logger.warning(
                f"⚠️ High error rate: {error_rate*100:.2f}%"
            )

# Usage
monitor = EdgeMonitor()

# Wrap inference with monitoring
def monitored_generate(prompt, model, tokenizer):
    start = time.perf_counter()
    try:
        response = generate_response(prompt, model, tokenizer)
        latency = (time.perf_counter() - start) * 1000
        monitor.log_request(prompt, response, latency, success=True)
        return response
    except Exception as e:
        latency = (time.perf_counter() - start) * 1000
        monitor.log_request(prompt, "", latency, success=False)
        raise e

print("\n=== Monitoring Demo ===")
monitored_generate("What is edge AI?", model, tokenizer)
print(f"Metrics: {monitor.get_metrics()}")

Mobile Deployment: Android & iOS

TensorFlow Lite Conversion

# Convert Gemma to TFLite for mobile deployment
def convert_to_tflite(model_path, output_path="gemma_mobile.tflite"):
    """
    Convert to TensorFlow Lite for mobile devices
    Note: This is a conceptual outline - full conversion requires additional steps
    """
    print("Converting to TensorFlow Lite...")

    # Step 1: Export to ONNX
    from optimum.onnxruntime import ORTModelForCausalLM
    ort_model = ORTModelForCausalLM.from_pretrained(model_path, export=True)
    ort_model.save_pretrained("./temp_onnx")

    # Step 2: Convert ONNX to TensorFlow (requires onnx-tf)
    # Step 3: Convert TensorFlow to TFLite
    # This requires tensorflow and additional conversion tools

    print("""
    Complete TFLite conversion requires:
    1. ONNX export (✓ done above)
    2. onnx-tf conversion: pip install onnx-tf
    3. TFLite conversion with quantization

    Android Integration:
    - Add TFLite dependency to build.gradle
    - Load model in Kotlin/Java
    - Use GPU delegate for acceleration

    iOS Integration:
    - Use Core ML converter
    - Integrate with Swift via Core ML framework
    - Leverage Neural Engine on A-series chips
    """)

# convert_to_tflite("google/gemma-2b-it")

Edge TPU Deployment

# Compile for Google Coral Edge TPU
def compile_for_edge_tpu():
    """
    Prepare model for Edge TPU deployment
    """
    print("""
    Edge TPU Compilation Steps:

    1. Convert to TFLite (INT8 quantized)
    2. Compile with Edge TPU compiler:
       $ edgetpu_compiler model.tflite

    3. Deploy on Coral devices:
       - Dev Board
       - USB Accelerator
       - PCIe Accelerator
       - M.2 Accelerator

    Performance: ~4 TOPS at 2W power consumption
    """)

Advanced Use Cases

Use Case #7: Federated Learning on Edge

class FederatedEdgeModel:
    """
    Conceptual federated learning for privacy-preserving model updates
    """
    def __init__(self, model, tokenizer):
        self.model = model
        self.tokenizer = tokenizer
        self.local_updates = []

    def local_train(self, training_data, epochs=1):
        """
        Simulate local fine-tuning on device-specific data
        """
        print("Training on local data (privacy preserved)...")
        # In production: actual fine-tuning with LoRA/QLoRA
        # This maintains privacy by never sending raw data

        # Collect gradient updates
        self.local_updates.append({
            "timestamp": datetime.now(),
            "samples": len(training_data),
            "device_id": "edge_device_001"
        })

    def aggregate_updates(self, global_model_updates):
        """
        Receive aggregated updates from central server
        """
        print("Receiving aggregated model updates...")
        # Apply federated averaging updates
        # This enables collaborative learning without data sharing

    def get_differential_privacy_noise(self, epsilon=1.0):
        """
        Add differential privacy noise to protect individual contributions
        """
        # Implement DP-SGD noise mechanism
        pass

# Demo concept
federated_model = FederatedEdgeModel(model, tokenizer)
print("\n=== Federated Learning (Conceptual) ===")
print("✓ Privacy-preserving collaborative learning")
print("✓ No raw data leaves the device")
print("✓ Model improves through decentralized training")