Moving app AI work from runtime calls to a pre-ship batch pass

What moves before shipping, and what genuinely cannot

Four questions settled every case for me.

The first column carries most of the weight. When the answer there is yes, the other three usually follow.

Write the pre-ship pass assuming it will die halfway

Moving work before the ship date means running hundreds of items in one go. Connections drop. You stop the script yourself to go do something else. Restarting from zero every time is not a workable habit.

I keyed the pass on a content hash so finished items are skipped. The same asset is billed once no matter how often you re-run.

"""Classify wallpaper assets before shipping.
Interrupt it at any point; the next run picks up where it stopped."""
 
import hashlib
import json
import pathlib
from google import genai
from google.genai import types
 
CLIENT = genai.Client(api_key="YOUR_API_KEY")
MODEL = "gemini-2.5-flash"
LEDGER = pathlib.Path("classified.jsonl")   # append-only; this file is the deliverable
CATEGORIES = ["landscape", "portrait", "flower", "wave", "night", "uncategorized"]
 
SCHEMA = types.Schema(
    type=types.Type.OBJECT,
    required=["category", "confidence"],
    properties={
        "category": types.Schema(type=types.Type.STRING, enum=CATEGORIES),
        "confidence": types.Schema(type=types.Type.NUMBER),
    },
)
 
def content_key(path: pathlib.Path) -> str:
    """Identity comes from bytes, not filenames, so a rename never bills twice."""
    return hashlib.sha256(path.read_bytes()).hexdigest()[:16]
 
def load_done() -> dict:
    if not LEDGER.exists():
        return {}
    done = {}
    for line in LEDGER.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        rec = json.loads(line)
        done[rec["key"]] = rec       # later entries win if a key repeats
    return done
 
def classify(path: pathlib.Path) -> dict:
    res = CLIENT.models.generate_content(
        model=MODEL,
        contents=[
            types.Part.from_bytes(data=path.read_bytes(), mime_type="image/jpeg"),
            "Assign this image to exactly one of the given categories. "
            "Choose uncategorized when you cannot decide.",
        ],
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=SCHEMA,
            temperature=0,          # a drifting result makes the ledger much less useful
        ),
    )
    return json.loads(res.text)
 
def main(src="_preview"):
    done = load_done()
    files = sorted(pathlib.Path(src).glob("*.jpg"))
    skipped = called = failed = 0
 
    with LEDGER.open("a", encoding="utf-8") as fp:
        for path in files:
            key = content_key(path)
            if key in done:
                skipped += 1
                continue
            try:
                result = classify(path)
            except Exception as exc:            # one bad item must not stop the run
                failed += 1
                print(f"  fail {path.name}: {exc}")
                continue
            record = {"key": key, "file": path.name, **result}
            fp.write(json.dumps(record, ensure_ascii=False) + "\n")
            fp.flush()                          # survive a kill signal with work intact
            called += 1
 
    print(f"total={len(files)} called={called} skipped={skipped} failed={failed}")
 
if __name__ == "__main__":
    main()

The flush() on every line is deliberate. An earlier version buffered writes and lost several dozen classifications to a forced quit. Flushing per line makes the worst case predictable instead of surprising.

Failed items never enter the ledger, so the next run retries them automatically. I covered the row-level version of that idea in a per-row retry ledger for the Gemini Batch API.

The client just reads a static file

Once the work moves earlier, the API call disappears from the app entirely. What remains is reading a generated table.

// Ship the generated table with the app; at runtime we only read it.
type Classified = { file: string; category: string; confidence: number };
 
const KNOWN = new Set([
  "landscape", "portrait", "flower", "wave", "night", "uncategorized",
]);
 
export function groupByCategory(rows: Classified[]): Map<string, string[]> {
  const grouped = new Map<string, string[]>();
 
  for (const row of rows) {
    // Builds already in the field must survive categories added later,
    // because the pre-ship pass ships faster than app review does.
    const key = KNOWN.has(row.category) ? row.category : "uncategorized";
    const bucket = grouped.get(key) ?? [];
    bucket.push(row.file);
    grouped.set(key, bucket);
  }
  return grouped;
}

Those few lines that fold unknown categories into uncategorized are not optional. The batch pass can add a category whenever you like, but the shipped binary keeps whatever list passed review. Skip the guard and the day you add one category is the day older builds render an empty screen.

What surprised me after the move

I started this as a cost change. The parts that mattered turned out to be elsewhere.

The quality loop got much faster. While the call lives at runtime, every answer you already returned is gone. With the work sitting in a batch pass, editing the prompt and re-running replaces every classification at once. At three-digit call counts, "just redo all of it" is a real option.

Model choice stopped being constrained. Nobody is watching a spinner, so a few extra seconds cost nothing. Picking a cheaper, slower model became a free decision rather than a compromise.

Consistency turned out to be a feature. I had filed "the same image lands in different categories for different people" under acceptable variance. In practice, that is precisely the kind of inconsistency people write in about.

There is a real cost on the other side. New assets are not reflected the moment they land — classification is now tied to your release cadence. If you add assets daily, this design is not available to you. Mine arrive every few weeks, which is why it fits.

As a rule of thumb: if your asset set is finite and turns over more slowly than weekly, move the work before the ship date. If assets land daily, leave the call at runtime and put the ceilings from the next section in place first.

The one feature I kept at runtime

Search query interpretation stayed. The input is whatever someone types, so it cannot be enumerated.

I gave it three ceilings instead:

1. Try a dictionary first. If the query matches an existing category or tag, no API call happens. Many queries stop right there
2. Cap calls per device per day. Past the cap, the day falls back to substring search
3. Always degrade to the old search on failure. Not reaching the model must never mean not being able to search

With those in place, the runtime count tracks "queries that missed the dictionary" rather than "users × sessions." When you do decide something belongs at runtime, it is worth asking whether you can re-anchor it to a smaller quantity like that.

What to check next

Pick one AI feature in your own app and write down, in a single line, what its call count is proportional to. If you find yourself writing "users × sessions," that is the signal to go back and ask whether the input really differs per person.

For anything image-shaped, shrinking before upload moves the needle about as much as call placement does — I wrote that part up in why you should resize with Pillow before sending images to the Gemini API.

Until I made this switch, "adding AI" and "calling at runtime" were the same thought in my head. If you are carrying the same assumption, I hope this saves you an estimate you would otherwise run too late.