💎 Build an offline field-inspection assistant with Gemma 4

import os

# Pick by field hardware — e4b for laptops and mini PCs, e2b for tablet-class devices
MODEL = os.environ.get("GEMMA_MODEL", "gemma4:e4b")

# Inspection reports need consistent judgments, so use a low temperature (the model card default is 1.0)
REPORT_OPTIONS = {"temperature": 0.2, "num_ctx": 8192}

import sys
from typing import Literal
from ollama import chat
from pydantic import BaseModel, Field
from config import MODEL, REPORT_OPTIONS


class Finding(BaseModel):
    component: str = Field(description="Component name (e.g. pipe joint, pressure gauge)")
    issue: str = Field(description="The observed problem in one sentence")
    severity: Literal["ok", "watch", "action", "stop"]


class InspectionReport(BaseModel):
    asset_id: str | None = Field(default=None, description="The asset tag in the photo if visible, otherwise null")
    readings: list[str] = Field(description="Values read from gauges and displays")
    findings: list[Finding]
    overall: Literal["ok", "watch", "action", "stop"]


SYSTEM = """You are an equipment inspection assistant. Base the report only on what is visible in the photo.
severity: ok (no issue) / watch (recheck next inspection) / action (maintenance request needed) / stop (consider shutting down now)
Leaks, smoke, sparks or heavy deformation are action or higher; when unsure, go one level higher.
overall is the highest severity among the findings."""


def inspect_photo(path: str, note: str = "") -> InspectionReport:
    resp = chat(
        model=MODEL,
        messages=[
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": f"Inspection photo. {note}".strip(), "images": [path]},
        ],
        format=InspectionReport.model_json_schema(),
        options=REPORT_OPTIONS,
    )
    return InspectionReport.model_validate_json(resp.message.content)


if __name__ == "__main__":
    report = inspect_photo(sys.argv[1] if len(sys.argv) > 1 else "photos/pump-07.jpg")
    print(report.model_dump_json(indent=2))

{
  "asset_id": "PUMP-07",
  "readings": ["pressure 7.8 bar"],
  "findings": [{"component": "pipe joint", "issue": "water is leaking from the joint", "severity": "action"}],
  "overall": "action"
}

from ollama import chat
from config import MODEL
from inspection import InspectionReport

SERIOUS = {"action", "stop"}


def review(report: InspectionReport, photo: str) -> str | None:
    """Re-check only maintenance-level findings in thinking mode — spend time only where a wrong call is costly"""
    if report.overall not in SERIOUS:
        return None
    resp = chat(
        model=MODEL,
        messages=[{
            "role": "user",
            "content": (
                "Check whether this inspection finding matches the photo, and list three steps the field worker should take right now.\n"
                f"Finding: {report.model_dump_json()}"
            ),
            "images": [photo],
        }],
        think=True,
    )
    print("[thinking]", (resp.message.thinking or "")[:200], "...")
    return resp.message.content

import sqlite3
from datetime import date

DB = sqlite3.connect("plant.db", check_same_thread=False)
DB.executescript("""
CREATE TABLE IF NOT EXISTS equipment (asset_id TEXT PRIMARY KEY, type TEXT, location TEXT, last_inspection TEXT);
CREATE TABLE IF NOT EXISTS work_orders (id INTEGER PRIMARY KEY AUTOINCREMENT, asset_id TEXT, severity TEXT, summary TEXT, created TEXT);
INSERT OR IGNORE INTO equipment VALUES
  ('PUMP-07', 'centrifugal pump', 'Building B, level B2', '2026-08-14'),
  ('VALVE-12', 'gate valve', 'Building A, rooftop', '2026-09-02');
""")


def lookup_equipment(asset_id: str) -> dict:
    """Look up equipment type, location and last inspection date by asset ID.

    Args:
      asset_id: Asset ID (e.g. PUMP-07)
    """
    row = DB.execute("SELECT type, location, last_inspection FROM equipment WHERE asset_id = ?", (asset_id.upper(),)).fetchone()
    if not row:
        return {"error": f"Equipment {asset_id} not found"}
    return {"asset_id": asset_id.upper(), "type": row[0], "location": row[1], "last_inspection": row[2]}


def create_work_order(asset_id: str, severity: str, summary: str) -> dict:
    """Create a maintenance work order. Call it only when severity is action or stop.

    Args:
      asset_id: Asset ID
      severity: action or stop
      summary: One sentence on why maintenance is needed
    """
    if severity not in {"action", "stop"}:
        return {"error": "Work orders can only be created for action or stop findings"}
    cur = DB.execute(
        "INSERT INTO work_orders (asset_id, severity, summary, created) VALUES (?, ?, ?, ?)",
        (asset_id.upper(), severity, summary, date.today().isoformat()),
    )
    DB.commit()
    return {"work_order_id": cur.lastrowid, "asset_id": asset_id.upper(), "severity": severity}


TOOLS = {"lookup_equipment": lookup_equipment, "create_work_order": create_work_order}

from ollama import chat
from assets import TOOLS, create_work_order, lookup_equipment
from config import MODEL
from inspection import InspectionReport

SYSTEM = """You are an equipment inspection assistant.
1) If the report has an asset ID, check its location and last inspection date with lookup_equipment.
2) If overall is action or stop, create a work order with create_work_order. Do not create one for ok or watch.
3) Finish with a three-line summary for the field worker."""


def process(report: InspectionReport, max_steps: int = 4) -> str:
    messages = [
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": f"Inspection report: {report.model_dump_json()}"},
    ]
    for _ in range(max_steps):
        resp = chat(model=MODEL, messages=messages, tools=[lookup_equipment, create_work_order])
        messages.append(resp.message)
        if not resp.message.tool_calls:
            return resp.message.content
        for call in resp.message.tool_calls:
            fn = TOOLS.get(call.function.name)
            result = fn(**call.function.arguments) if fn else {"error": f"unknown tool: {call.function.name}"}
            print(f"[tool] {call.function.name}({call.function.arguments}) -> {result}")
            messages.append({"role": "tool", "tool_name": call.function.name, "content": str(result)})
    return "Automatic processing did not finish. Please review the report."

import sys
from transformers import AutoModelForMultimodalLM, AutoProcessor

# Audio input is supported only by E2B, E4B and 12B. Up to 30 seconds per input.
MODEL_ID = "google/gemma-4-E2B-it"

processor = AutoProcessor.from_pretrained(MODEL_ID)
model = AutoModelForMultimodalLM.from_pretrained(MODEL_ID, dtype="auto", device_map="auto")


def transcribe(wav_path: str) -> str:
    messages = [{
        "role": "user",
        "content": [
            {"type": "text", "text": "Transcribe this field voice memo in its original language. Output only the transcription, and write numbers as digits."},
            {"type": "audio", "audio": wav_path},
        ],
    }]
    inputs = processor.apply_chat_template(
        messages,
        tokenize=True,
        return_dict=True,
        return_tensors="pt",
        add_generation_prompt=True,
        enable_thinking=False,
    ).to(model.device)
    input_len = inputs["input_ids"].shape[-1]
    outputs = model.generate(**inputs, max_new_tokens=256)
    response = processor.decode(outputs[0][input_len:], skip_special_tokens=False)
    parsed = processor.parse_response(response, prefix=inputs["input_ids"])
    return parsed["content"] if isinstance(parsed, dict) else str(parsed)


if __name__ == "__main__":
    print(transcribe(sys.argv[1] if len(sys.argv) > 1 else "memos/pump-07.wav"))

import json
import sys
import time
from pathlib import Path
from inspection import inspect_photo

photo_dir = Path(sys.argv[1] if len(sys.argv) > 1 else "photos")
out_path = Path("data/reports.jsonl")
out_path.parent.mkdir(exist_ok=True)

# Skip photos already processed — if the battery dies or the run stops, rerunning picks up where it left off
done = set()
if out_path.exists():
    done = {json.loads(line)["photo"] for line in out_path.read_text(encoding="utf-8").splitlines() if line}

photos = sorted(p for p in photo_dir.iterdir() if p.suffix.lower() in {".jpg", ".jpeg", ".png"})
with out_path.open("a", encoding="utf-8") as out:
    for photo in photos:
        if photo.name in done:
            continue
        started = time.perf_counter()
        try:
            report = inspect_photo(str(photo))
            row = {"photo": photo.name, "ok": True, "report": report.model_dump()}
        except Exception as e:   # one failed photo does not stop the rest
            row = {"photo": photo.name, "ok": False, "error": str(e)}
        row["seconds"] = round(time.perf_counter() - started, 2)
        out.write(json.dumps(row) + "\n")
        out.flush()
        print(f"{photo.name}: {row.get('report', {}).get('overall', 'ERROR')} ({row['seconds']}s)")

import csv
import json
from collections import Counter

LEVEL = {"ok": 0, "watch": 1, "action": 2, "stop": 3}

labels = {row["photo"]: row["severity"] for row in csv.DictReader(open("data/labels.csv", encoding="utf-8"))}
reports = [json.loads(line) for line in open("data/reports.jsonl", encoding="utf-8") if line.strip()]

result = Counter()
for r in reports:
    expected = labels.get(r["photo"])
    if expected is None or not r["ok"]:
        result["skipped"] += 1
        continue
    got = r["report"]["overall"]
    if got == expected:
        result["match"] += 1
    elif LEVEL[got] < LEVEL[expected]:
        result["under"] += 1   # risk rated too low — the most dangerous error
        print(f"  ⚠ under-rated {r['photo']}: inspector {expected} / model {got}")
    else:
        result["over"] += 1    # rated too high — the cost of an unnecessary dispatch

judged = result["match"] + result["under"] + result["over"]
print(f"match {result['match']}/{judged}, under-rated {result['under']}, over-rated {result['over']}, skipped {result['skipped']}")

import shutil
import tempfile
from pathlib import Path
from fastapi import FastAPI, File, Form, HTTPException, UploadFile
from agent import process
from inspection import inspect_photo
from review import review

app = FastAPI(title="Field Inspection Assistant")
MAX_BYTES = 10 * 1024 * 1024


@app.post("/inspections")
def create_inspection(photo: UploadFile = File(...), note: str = Form("")) -> dict:
    if photo.content_type not in {"image/jpeg", "image/png"}:
        raise HTTPException(status_code=415, detail="Only JPEG or PNG photos are accepted")
    with tempfile.TemporaryDirectory() as tmp:
        path = Path(tmp) / (photo.filename or "photo.jpg")
        with path.open("wb") as f:
            shutil.copyfileobj(photo.file, f)
        if path.stat().st_size > MAX_BYTES:
            raise HTTPException(status_code=413, detail="Photos must be 10 MB or smaller")
        report = inspect_photo(str(path), note)
        return {
            "report": report.model_dump(),
            "review": review(report, str(path)),
            "summary": process(report),
        }