Run Gemma 4 Locally: Building Zero-Cost AI Apps

# Install Homebrew (if needed)
/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"
 
# Install tools
brew install python@3.11 git cmake

mkdir gemma4_app && cd gemma4_app
python3.11 -m venv venv
source venv/bin/activate
pip install --upgrade pip setuptools wheel

# llama.cpp is Metal-optimized
pip install llama-cpp-python transformers torch numpy
pip install huggingface-hub fastapi uvicorn

# download_model.py
from huggingface_hub import snapshot_download
import os
 
MODEL_ID = "google/gemma-1.1-31b-it"
DOWNLOAD_PATH = "./models"
 
os.makedirs(DOWNLOAD_PATH, exist_ok=True)
 
snapshot_download(
    MODEL_ID,
    local_dir=DOWNLOAD_PATH,
    token="YOUR_HUGGING_FACE_TOKEN"
)
print(f"Downloaded to {DOWNLOAD_PATH}")

# inference.py
from llama_cpp import Llama
 
llm = Llama(
    model_path="./models/gemma-1.1-31b-it.Q4_K_M.gguf",
    n_gpu_layers=-1,
    n_ctx=16000,
    temperature=0.7,
    max_tokens=512
)
 
def generate(prompt: str) -> str:
    response = llm(prompt, echo=False)
    return response["choices"][0]["text"]
 
if __name__ == "__main__":
    result = generate("Explain quantum computing")
    print(result)

# Flash Ubuntu Server 24.04 LTS via Raspberry Pi Imager
# Then SSH and continue:
 
sudo apt update && sudo apt upgrade -y
sudo apt install -y python3.11 python3.11-venv python3.11-dev
sudo apt install -y git cmake build-essential
 
# Swap config (optional but recommended)
sudo nano /etc/dphys-swapfile
# Set CONF_SWAPSIZE=2048
 
# Same setup as MacBook
mkdir gemma4_app && cd gemma4_app
python3.11 -m venv venv
source venv/bin/activate
pip install llama-cpp-python transformers huggingface-hub fastapi uvicorn

# CUDA 12.1 environment
pip install torch --index-url https://download.pytorch.org/whl/cu121
 
# llama.cpp with CUDA
CMAKE_ARGS="-DGGML_CUDA=on" pip install llama-cpp-python
 
# High-performance inference
pip install vllm

# app.py
from fastapi import FastAPI, WebSocket
from llama_cpp import Llama
import sqlite3
from datetime import datetime
import json
 
app = FastAPI()
 
llm = Llama(
    model_path="./models/gemma-1.1-31b-it.Q4_K_M.gguf",
    n_gpu_layers=-1,
    n_ctx=16000,
    temperature=0.7
)
 
def init_db():
    conn = sqlite3.connect("sessions.db")
    c = conn.cursor()
    c.execute("""
        CREATE TABLE IF NOT EXISTS chat_history (
            id INTEGER PRIMARY KEY,
            session_id TEXT,
            role TEXT,
            content TEXT,
            timestamp DATETIME
        )
    """)
    conn.commit()
    return conn
 
@app.websocket("/ws/chat/{session_id}")
async def websocket_endpoint(websocket: WebSocket, session_id: str):
    await websocket.accept()
    
    try:
        while True:
            data = await websocket.receive_text()
            user_msg = json.loads(data)["content"]
            
            response = llm(
                f"User: {user_msg}\nAssistant:",
                echo=False,
                max_tokens=512
            )
            
            bot_msg = response["choices"][0]["text"].strip()
            
            conn = sqlite3.connect("sessions.db")
            c = conn.cursor()
            c.execute("""
                INSERT INTO chat_history (session_id, role, content, timestamp)
                VALUES (?, ?, ?, ?)
            """, (session_id, "user", user_msg, datetime.now()))
            c.execute("""
                INSERT INTO chat_history (session_id, role, content, timestamp)
                VALUES (?, ?, ?, ?)
            """, (session_id, "assistant", bot_msg, datetime.now()))
            conn.commit()
            conn.close()
            
            await websocket.send_json({
                "role": "assistant",
                "content": bot_msg
            })
    
    except Exception as e:
        print(f"Error: {e}")
    finally:
        await websocket.close()
 
@app.on_event("startup")
async def startup():
    init_db()
 
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# long_doc_analysis.py
def analyze_document(file_path: str, question: str) -> str:
    with open(file_path, 'r', encoding='utf-8') as f:
        doc = f.read()
    
    prompt = f"""
    Analyze the following document:
    
    {doc[:100000]}
    
    Question: {question}
    
    Answer:
    """
    
    response = llm(prompt, max_tokens=2048)
    return response["choices"][0]["text"]
 
result = analyze_document("annual_report.txt", "What are the key metrics?")
print(result)

# image_analysis.py
import base64
import requests
 
def analyze_image(image_path: str, question: str) -> str:
    """Using LLaVA for vision"""
    with open(image_path, "rb") as f:
        encoded = base64.b64encode(f.read()).decode('utf-8')
    
    payload = {
        "model": "llava:34b",
        "prompt": question,
        "images": [encoded],
        "stream": False
    }
    
    response = requests.post(
        "http://localhost:11434/api/generate",
        json=payload
    )
    
    return response.json()["response"]
 
result = analyze_image("product.jpg", "Describe this product")
print(result)

# audio_processing.py
import subprocess
import json
 
def transcribe(audio_file: str) -> str:
    result = subprocess.run(
        ["whisper", audio_file, "--language", "en", "--output_format", "json"],
        capture_output=True,
        text=True
    )
    return json.loads(result.stdout)["text"]
 
def meeting_insights(audio_file: str) -> str:
    transcript = transcribe(audio_file)
    
    prompt = f"""
    Analyze this meeting transcript:
    
    {transcript}
    
    Provide:
    1. Main topics
    2. Key decisions
    3. Action items
    4. Next steps
    """
    
    response = llm(prompt, max_tokens=1024)
    return response["choices"][0]["text"]
 
result = meeting_insights("call.mp3")
print(result)

def batch_process(prompts: list) -> list:
    results = []
    for prompt in prompts:
        response = llm(prompt, max_tokens=256)
        results.append(response["choices"][0]["text"])
    return results
 
queries = [
    "What is AI?",
    "What is ML?",
    "What is DL?"
]
answers = batch_process(queries)

import psutil
import os
 
def check_memory():
    p = psutil.Process(os.getpid())
    info = p.memory_info()
    print(f"RAM: {info.rss / 1024**2:.2f} MB")
    print(f"VMS: {info.vms / 1024**2:.2f} MB")
 
check_memory()

# hybrid_routing.py
from llama_cpp import Llama
import anthropic
 
llm_local = Llama(
    model_path="./models/gemma-1.1-31b-it.Q4_K_M.gguf",
    n_gpu_layers=-1
)
 
client = anthropic.Anthropic(api_key="YOUR_CLAUDE_KEY")
 
def smart_route(query: str) -> dict:
    token_count = len(query.split())
    
    if token_count < 50:
        # Local (free)
        response = llm_local(query, max_tokens=512)
        return {
            "source": "local",
            "cost": 0,
            "response": response["choices"][0]["text"]
        }
    else:
        # Cloud (precise)
        msg = client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=1024,
            messages=[{"role": "user", "content": query}]
        )
        return {
            "source": "claude-api",
            "cost": "$0.003",
            "response": msg.content[0].text
        }
 
result = smart_route("Explain machine learning fundamentals")
print(f"Source: {result['source']}")
print(result['response'])

import sqlite3
from datetime import datetime, timedelta
 
def log_call(source: str, tokens: int, cost: float):
    conn = sqlite3.connect("api_usage.db")
    c = conn.cursor()
    c.execute("""
        CREATE TABLE IF NOT EXISTS usage (
            source TEXT, tokens INT, cost FLOAT, timestamp DATETIME
        )
    """)
    c.execute(
        "INSERT INTO usage VALUES (?, ?, ?, ?)",
        (source, tokens, cost, datetime.now())
    )
    conn.commit()
    conn.close()
 
def cost_report(days=30):
    conn = sqlite3.connect("api_usage.db")
    c = conn.cursor()
    since = datetime.now() - timedelta(days=days)
    
    c.execute("""
        SELECT source, COUNT(*), SUM(tokens), SUM(cost)
        FROM usage WHERE timestamp > ? GROUP BY source
    """, (since,))
    
    for src, calls, tokens, cost in c.fetchall():
        print(f"{src}: {calls} calls, {tokens} tokens, ${cost:.4f}")
    
    conn.close()
 
cost_report(30)