pixagram-search v4 embeddings (deploy.sh)

README.md

@@ -1,13 +1,186 @@
 ---
-title: Pixagram Search V4
-emoji: 💻
-colorFrom: indigo
-colorTo: yellow
+title: pixagram-search embeddings
+emoji: 🟪
+colorFrom: purple
+colorTo: indigo
 sdk: gradio
-sdk_version: 6.30.0
-python_version: '3.13'
+sdk_version: 6.28.0
 app_file: app.py
 pinned: false
+license: apache-2.0
+short_description: SigLIP image/text embeddings for the Pixagram search engine
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# SigLIP embedding Space for pixagram-search
+
+One model, two towers. The Cloudflare Worker sends artwork PNGs here while indexing and the
+query text here at search time; both land in the same vector space, which is what makes
+text → image search work. `app.py` is the Space entry point; it serves:
+
+| route | |
+|---|---|
+| `POST /embed` | `{"inputs": {"images": ["<base64>", …], "texts": ["…", …]}}` → `{"model", "dim", "embeddings": [[…], …], "calibration", "max_num_patches"}` (images first, then texts; L2-normalised; `max_num_patches` for NaFlex models) |
+| `GET /health` | `{"ok", "model", "dim", "ready", "stub", "calibration", "max_num_patches"}` — `ready` flips to true once the weights are loaded |
+| `GET /` | Gradio page: embed a text or an image, score an image against captions |
+
+`calibration` is the model's learned `{"logit_scale", "logit_bias"}`:
+`sigmoid(logit_scale · cos + logit_bias)` is SigLIP's own text↔image match probability. The v3
+Worker stores it (KV `calib:<model>`) and uses it until it has a background sample of the
+corpus; older Spaces without the field still work (the Worker knows the SigLIP 2 base values).
+
+## Deploy as a Space
+
+1. Create a Space (SDK: **Gradio**, hardware: **CPU basic** is enough — a SigLIP-base pass on
+   one artwork takes ~0.25 s on 2 vCPU, ~0.5 s for NaFlex at 576 patches). The Space computes
+   one embedding per CPU at once and queues the rest; the Worker reads that number from
+   `/health` (`concurrency`) every ten minutes and sends as many per consumer invocation, so
+   bigger hardware is used with nothing else to change: on CPU upgrade (8 vCPU) it computes 8
+   at once instead of 2.
+2. Upload this folder — `README.md` (the frontmatter above is the Space config), `app.py`,
+   `siglip.py`, `requirements.txt`; `handler.py` can stay, it is only used by Inference
+   Endpoints. From the repository root:
+
+   ```bash
+   pip install -U huggingface_hub
+   hf auth login                                                   # older CLI: huggingface-cli login
+   hf upload <owner>/<space> hf . --repo-type space
+   ```
+
+   `scripts/deploy.sh` does all of this for the v4 Space (`primerz/pixagram-search-v4`); the v3
+   repository did it for `primerz/pixagram-search-v3`.
+3. Visibility:
+   * **Private** Space (recommended): every request must carry an HF token that can read the
+     Space — set that token as the Worker's `HF_TOKEN` secret. Leave `API_TOKEN` unset.
+   * **Public** Space: set a Space secret `API_TOKEN` to a long random string and use the same
+     string as the Worker's `HF_TOKEN`; `/embed` then rejects anything else with 401.
+4. Worker: `npx wrangler secret put HF_TOKEN`, and set
+   `HF_EMBED_URL = "https://<owner>-<space-name>.hf.space/embed"` in `wrangler.jsonc`
+   (the host is `<owner>-<space-name>` with dots in the owner name replaced by dashes; the
+   Space settings page shows the exact "Direct URL").
+5. Deploy the Worker, then `scripts/admin.sh reindex-all embed,text` to fill the vectors in.
+
+Cold start: the SigLIP base checkpoints are ~1.5 GB; the Space answers `/health`
+in seconds and `/embed` blocks until the weights are loaded (about 20 s once cached, a few
+minutes on the first build). Free CPU Spaces **sleep after 48 h without traffic**; the Worker
+treats the wake-up page as a transient error and retries with backoff, and search simply runs
+without the semantic leg until the Space is back. If that gap matters, use paid CPU hardware
+and set the sleep time to "never".
+
+## Concurrency
+
+The Space computes `MAX_CONCURRENCY` embeddings at once (default: one per CPU), each with
+`CPUs / MAX_CONCURRENCY` PyTorch threads (`TORCH_THREADS` overrides), on threads that live as long
+as the server. Search queries (texts only) have threads of their own and never wait behind
+images being indexed. `/health` answers at once even under load.
+
+Measured on 2 vCPU (Oct 2026): 24 NaFlex images at 576 patches from 4 clients, alone, then with
+a search text every second:
+
+| | images alone | images with searches | a search text meanwhile | `/health` meanwhile |
+|---|---|---|---|---|
+| before: one request at a time, 2 threads | 1.81 / s | 1.76 / s | median 1.25 s, worst 1.9 s | blocked (seconds) |
+| **one per CPU, 1 thread each** | 1.89 / s | 1.77 / s | median 0.22 s, worst 0.29 s | 3 ms |
+| one at a time, 2 threads | 1.55 / s | | | |
+
+Two images at once with one thread each beat one image with two threads (1.89 against 1.55
+images/s here), which is why the default is one per CPU. Fewer at once with more threads each
+(`MAX_CONCURRENCY=1`) answers a single query sooner (~70 ms instead of ~125 ms for a text on 2
+vCPU) at the cost of throughput.
+
+`python3 hf/loadtest.py https://<owner>-<space>.hf.space <token>` measures a running Space the
+same way (the token is the Worker's `HF_TOKEN`, `SPACE_API_TOKEN` in `~/.pixagram-search-v4.json`):
+run it before and after changing the hardware.
+
+## Troubleshooting
+
+* Text embeddings taking 8-10 s on a CPU Space: `os.cpu_count()` inside the container reports
+  the host's cores (64-96) while the cgroup allows 2-8, and PyTorch spin-waits on the
+  oversubscribed thread pool. `siglip.py` sizes everything from the cgroup quota
+  (`effective_cpus()`); `/health` reports `cpus`, `concurrency` (embeddings at once) and
+  `threads` (PyTorch threads per embedding) so you can see what it picked.
+
+* `[Errno 98] error while attempting to bind on address ('0.0.0.0', 7860)` in the build/run
+  log: Spaces set `GRADIO_SSR_MODE=True`, which makes Gradio start a Node SSR server on 7860
+  before uvicorn. `app.py` mounts Gradio with `ssr_mode=False` for that reason — keep it.
+* `WARNING: Running pip as the 'root' user…` during the build is harmless; the Space image
+  installs requirements as root by design.
+* `Your space is in error` on the direct URL: open the Space page → *Logs* (Build, then
+  Container); `/health` only answers once the container is running.
+* Intermittent `502` HTML pages (fast, ~0.15 s, no `x-proxied-host` / `x-proxied-replica`
+  response header) while the Space shows *Running*: Hugging Face's edge fails before the
+  request reaches the container, so the Container log shows nothing. On 2026-09-28 this came in
+  windows of 10-50 s (up to 34 failures in a row). A quick retry does not bridge that. Search
+  falls back to full text, and the queue retries with backoff. Restart the Space. If it
+  persists, report it to HF with the `x-request-id` of a failed call, or move to paid hardware
+  or an Inference Endpoint (`handler.py`).
+
+## Test
+
+```bash
+curl -s https://<owner>-<space>.hf.space/health
+curl -s https://<owner>-<space>.hf.space/embed -H "Authorization: Bearer $HF_TOKEN" \
+  -H "Content-Type: application/json" -d '{"inputs":{"texts":["a swan on a lake at sunset"]}}' | jq '.dim'
+```
+
+Local run (any machine with Python 3.10+): `pip install -r requirements.txt && python app.py`,
+then point the Worker's `.dev.vars` at `HF_EMBED_URL=http://127.0.0.1:7860/embed`.
+`EMBED_STUB=1 python app.py` starts without weights and returns deterministic pseudo-vectors —
+handy for wiring tests, never for production (`/health` reports `"stub": true`).
+
+## Changing the model
+
+The code default is `google/siglip-base-patch16-256-multilingual`, which is what the
+production Space runs. Choose a model **per Space** with the `MODEL_ID` variable, so that
+uploading `hf/` never changes production's model:
+
+- the v2 Space (`primerz/pixagram-siglip2`) sets `MODEL_ID=google/siglip2-base-patch16-256`;
+- the v3 Space (`primerz/pixagram-search-v3`) and the v4 Space (`primerz/pixagram-search-v4`,
+  created by this repository's `scripts/deploy.sh`) set `MODEL_ID=google/siglip2-base-patch16-naflex`
+  and `MAX_NUM_PATCHES=576`: v4 embeds exactly as v3 does.
+
+### NaFlex
+
+NaFlex checkpoints keep each image's aspect ratio. The processor resizes the image to the
+largest size that fits `MAX_NUM_PATCHES` patches of 16×16 pixels (default 256), and batches mix
+shapes through an attention mask. Every reply reports `max_num_patches`. The Worker refuses
+image vectors whose budget differs from its `EMBED_PATCHES` (text vectors do not depend on it),
+and re-embeds when `EMBED_PATCHES` changes. On the Pixagram corpus, 256 patches did worse than
+the fixed 256 px model and 576 slightly better (README-V3.md, "SigLIP 2 NaFlex").
+
+Any SigLIP / SigLIP 2 checkpoint works unchanged: fixed-resolution ones load as `SiglipModel`,
+NaFlex ones as `Siglip2Model`. Stick to **multilingual** checkpoints, because queries arrive
+in any language: every SigLIP 2 model, `google/siglip-base-patch16-256-multilingual` or
+`google/siglip-so400m-patch16-256-i18n`. The other SigLIP 1 checkpoints have an English text
+tower.
+
+Measured on the 106 Pixagram artworks (+500 pixel-art distractors, 157 queries in EN/FR/DE/JA,
+2 vCPU, Sept 2026):
+
+| model | dim | ms / image | ms / query | notes |
+|---|---|---|---|---|
+| siglip-base-patch16-256-multilingual | 768 | 310 | 90 | production |
+| **siglip2-base-patch16-256** | 768 | 315 | 87 | v2 Space; same cost; EN/FR on par, DE/JA better |
+| siglip2-base-patch16-384 / -naflex (256 patches) | 768 | 650 / 305 | 85 | no gain over base-256 |
+| **siglip2-base-patch16-naflex, 576 patches** | 768 | ~530 | 85 | v3 Space: small gain, measured in Oct 2026 on 133 artworks (README-V3.md) |
+| siglip2-large-patch16-256 | 1024 | 980 | 290 | no gain |
+| siglip2-so400m-patch16-256 | 1152 | 1270 | 390 | small gain (JA) at 4× the CPU |
+| siglip-so400m-patch16-256-i18n | 1152 | 1280 | 395 | best measured, 4× the CPU |
+
+Switching a stack in place, same dimension (768 → 768): keep the Vectorize index. Set the
+Space's `MODEL_ID`, then set `EMBED_MODEL` to the same id in that stack's wrangler config
+and deploy. The Worker refuses vectors whose `model` differs from `EMBED_MODEL`, and
+re-embeds artworks whose `embed_model` differs (the 10-minute sweeper picks them up; to do it
+at once: `scripts/admin.sh reindex-all embed,text`). Then refresh the background sample
+(`scripts/admin.sh background`), since z-scores are per model.
+
+A different dimension also needs **new Vectorize indexes** (with the metadata indexes from
+`scripts/lib.sh`) that `VEC` and `VEC_TEXT` point at, and a new `EMBED_DIM`. Vectors from
+different models must never be compared.
+
+## Inference Endpoint instead of a Space
+
+`handler.py` implements the `EndpointHandler` contract for a dedicated Inference Endpoint
+(task *custom*, same `siglip.py`, same request/response shape). Use it when you want
+autoscaling, private networking, or GPU without the Space UI. Then `HF_EMBED_URL` is the
+endpoint URL itself and the Worker's `X-Scale-Up-Timeout: 600` header makes a scaled-to-zero
+replica wake up within the request instead of returning 503.

app.py

@@ -0,0 +1,153 @@
+"""
+Hugging Face Space entry point (sdk: gradio): SigLIP image + text embeddings for pixagram-search.
+
+Two faces on one server (port 7860):
+  POST /embed      JSON API used by the Cloudflare Worker (see siglip.Embedder.handle for the shape)
+  GET  /health     {"ok", "model", "dim", "ready", "calibration"}
+  GET  /           Gradio demo page: embed a text or an image, compare an image with a caption
+
+Space secrets / variables:
+  API_TOKEN   optional shared secret; when set, /embed requires "Authorization: Bearer <API_TOKEN>"
+              (for a *private* Space leave it unset — HF already gates requests with your HF token)
+  MODEL_ID    override the model (default google/siglip-base-patch16-256-multilingual, 768-d;
+              the v2 Space sets google/siglip2-base-patch16-256, the v3 Space
+              google/siglip2-base-patch16-naflex)
+  MAX_NUM_PATCHES  NaFlex models: patches per image (default 256)
+  MAX_CONCURRENCY  embeddings computed at once (default one per CPU; see siglip.py). /health
+                   reports it as "concurrency", and the Worker's queue consumer sends that many
+                   at once: a Space with more CPUs indexes faster without changing anything else.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from typing import Optional
+
+import gradio as gr
+import uvicorn
+from fastapi import FastAPI, Header, HTTPException, Request
+from fastapi.responses import JSONResponse
+from PIL import Image
+
+from siglip import MODEL_ID, Embedder, concurrency, effective_cpus, shared, threads_per_job, to_rgb
+
+API_TOKEN = os.environ.get("API_TOKEN") or None
+# Spaces expose port 7860 and set GRADIO_SERVER_PORT; PORT is honoured for other hosts.
+PORT = int(os.environ.get("GRADIO_SERVER_PORT") or os.environ.get("PORT") or "7860")
+HOST = os.environ.get("GRADIO_SERVER_NAME", "0.0.0.0")
+
+embedder: Embedder = shared()
+_started = time.time()
+
+# CONCURRENCY embeddings at once, each on a thread of its own that lives as long as the server (a
+# thread that runs PyTorch keeps its own OpenMP/MKL thread team: reusing the same threads keeps
+# that to one team per slot). The event loop stays free for /health and for queueing the rest.
+# Search queries (texts only) have threads of their own, so they never wait behind images being
+# indexed; a text takes a fraction of an image's time.
+CONCURRENCY = concurrency()
+_images = ThreadPoolExecutor(max_workers=CONCURRENCY, thread_name_prefix="embed-image")
+_texts = ThreadPoolExecutor(max_workers=CONCURRENCY, thread_name_prefix="embed-text")
+
+
+def _pool(payload) -> ThreadPoolExecutor:
+    inputs = payload.get("inputs", payload) if isinstance(payload, dict) else payload
+    return _images if isinstance(inputs, dict) and inputs.get("images") else _texts
+
+# Warm the model in the background so the first request does not pay the full load time.
+threading.Thread(target=embedder.load, name="siglip-warmup", daemon=True).start()
+
+api = FastAPI(title="pixagram-search embeddings", version="1.0")
+
+
+@api.get("/health")
+def health():
+    return {"ok": True, "model": embedder.model_id, "dim": embedder.dim, "ready": embedder.ready, "stub": embedder.stub,
+            "calibration": embedder.calibration, "max_num_patches": embedder.max_num_patches if embedder.naflex else None,
+            "concurrency": CONCURRENCY, "threads": threads_per_job(), "cpus": effective_cpus(),
+            "uptime_s": int(time.time() - _started)}
+
+
+@api.post("/embed")
+async def embed(request: Request, authorization: Optional[str] = Header(default=None)):
+    if API_TOKEN and authorization != f"Bearer {API_TOKEN}":
+        raise HTTPException(status_code=401, detail="unauthorized")
+    try:
+        payload = await request.json()
+    except Exception:
+        raise HTTPException(status_code=400, detail="body must be JSON")
+    try:
+        # embedder.handle blocks while the model finishes loading on a cold start; the Worker
+        # waits (its consumer has a 15-minute budget) rather than getting a 503.
+        return JSONResponse(await asyncio.get_running_loop().run_in_executor(_pool(payload), embedder.handle, payload))
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except Exception as e:  # decoding errors etc.
+        raise HTTPException(status_code=422, detail=f"{type(e).__name__}: {e}")
+
+
+# ---- Gradio demo -------------------------------------------------------------------------
+
+
+def _vector_view(vec):
+    return {"dim": len(vec), "first_8": [round(v, 5) for v in vec[:8]], "norm": round(sum(v * v for v in vec) ** 0.5, 6)}
+
+
+def ui_embed_text(text: str):
+    if not text or not text.strip():
+        return {"error": "type something"}
+    vec = embedder.embed_texts([text])[0]
+    return _vector_view(vec)
+
+
+def ui_embed_image(img: Optional[Image.Image]):
+    if img is None:
+        return {"error": "upload an image"}
+    vec = embedder.embed_images([to_rgb(img)])[0]
+    return _vector_view(vec)
+
+
+def ui_compare(img: Optional[Image.Image], texts: str):
+    if img is None or not texts.strip():
+        return {"error": "need an image and one caption per line"}
+    captions = [t.strip() for t in texts.splitlines() if t.strip()]
+    iv = embedder.embed_images([to_rgb(img)])[0]
+    tvs = embedder.embed_texts(captions)
+    scores = [(c, round(sum(a * b for a, b in zip(iv, tv)), 4)) for c, tv in zip(captions, tvs)]
+    scores.sort(key=lambda x: -x[1])
+    return {"cosine": scores}
+
+
+with gr.Blocks(title="pixagram-search embeddings") as demo:
+    gr.Markdown(
+        f"## pixagram-search embeddings\n"
+        f"Model `{MODEL_ID}` — the same vectors the search engine stores in Vectorize. "
+        f"The Worker calls `POST /embed`; this page is for eyeballing the model on pixel art."
+    )
+    with gr.Tab("Text → vector"):
+        t_in = gr.Textbox(label="query text", placeholder="a swan on a lake at sunset")
+        t_btn = gr.Button("Embed")
+        t_out = gr.JSON(label="vector")
+        t_btn.click(ui_embed_text, inputs=t_in, outputs=t_out, api_name="embed_text")
+    with gr.Tab("Image → vector"):
+        i_in = gr.Image(type="pil", label="artwork (webp/png)")
+        i_btn = gr.Button("Embed")
+        i_out = gr.JSON(label="vector")
+        i_btn.click(ui_embed_image, inputs=i_in, outputs=i_out, api_name="embed_image")
+    with gr.Tab("Compare"):
+        c_img = gr.Image(type="pil", label="artwork")
+        c_txt = gr.Textbox(label="captions, one per line", lines=4, value="a swan on a lake at sunset\na space invader\na portrait of a woman\nabstract shapes")
+        c_btn = gr.Button("Score")
+        c_out = gr.JSON(label="cosine similarity per caption")
+        c_btn.click(ui_compare, inputs=[c_img, c_txt], outputs=c_out, api_name="compare")
+
+# ssr_mode=False is essential on Spaces: the platform sets GRADIO_SSR_MODE=True, which would make
+# mount_gradio_app start Gradio's Node SSR server on 7860 before uvicorn binds it
+# ("[Errno 98] address already in use"). Client-side rendering needs no extra process.
+app = gr.mount_gradio_app(api, demo, path="/", ssr_mode=False)
+
+if __name__ == "__main__":
+    uvicorn.run(app, host=HOST, port=PORT)

handler.py

@@ -0,0 +1,22 @@
+"""
+Alternative deployment: a Hugging Face *Inference Endpoint* (dedicated) instead of a Space.
+Endpoints look for this file and the EndpointHandler class; the Space uses app.py.
+Same request/response contract, so the Worker does not care which one it talks to.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict
+
+from siglip import Embedder
+
+
+class EndpointHandler:
+    def __init__(self, path: str = "") -> None:
+        self.embedder = Embedder().load()
+
+    def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            return self.embedder.handle(data)
+        except ValueError as e:
+            return {"error": str(e), "model": self.embedder.model_id, "dim": self.embedder.dim, "embeddings": []}

loadtest.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python3
+"""Throughput of a running Space: 24 images from 4 clients at once (NaFlex-sized, 480x480), with a
+search text every second, then the same without the texts. Run it before and after changing the
+Space's hardware or MAX_CONCURRENCY.
+
+    python3 hf/loadtest.py https://primerz-pixagram-search-v4.hf.space [token]
+
+The token is the Worker's HF_TOKEN (SPACE_API_TOKEN in ~/.pixagram-search-v4.json) when the
+Space has an API_TOKEN; it is read from $SPACE_API_TOKEN when not given.
+"""
+import base64
+import io
+import json
+import os
+import sys
+import threading
+import time
+import urllib.request
+
+from PIL import Image
+
+BASE = sys.argv[1].rstrip("/")
+TOKEN = sys.argv[2] if len(sys.argv) > 2 else os.environ.get("SPACE_API_TOKEN", "")
+buf = io.BytesIO()
+Image.effect_noise((480, 480), 64).convert("RGB").save(buf, "PNG")
+IMG = base64.b64encode(buf.getvalue()).decode()
+
+
+def post(body):
+    headers = {"content-type": "application/json", **({"authorization": f"Bearer {TOKEN}"} if TOKEN else {})}
+    req = urllib.request.Request(BASE + "/embed", data=json.dumps(body).encode(), headers=headers)
+    with urllib.request.urlopen(req, timeout=600) as r:
+        return json.load(r)
+
+
+def run(with_texts: bool) -> str:
+    done, text_ms = False, []
+
+    def client():
+        for _ in range(6):
+            post({"inputs": {"images": [IMG]}})
+
+    def texts():
+        while not done:
+            time.sleep(1)
+            a = time.time()
+            post({"inputs": {"texts": ["a red dragon over a castle"]}})
+            text_ms.append((time.time() - a) * 1000)
+
+    t0 = time.time()
+    clients = [threading.Thread(target=client) for _ in range(4)]
+    tx = threading.Thread(target=texts) if with_texts else None
+    for c in clients:
+        c.start()
+    if tx:
+        tx.start()
+    for c in clients:
+        c.join()
+    wall = time.time() - t0
+    done = True
+    if tx:
+        tx.join()
+    out = f"{24 / wall:.2f} images/s"
+    if text_ms:
+        text_ms.sort()
+        out += f"; a search text meanwhile: median {text_ms[len(text_ms) // 2]:.0f} ms, worst {text_ms[-1]:.0f} ms"
+    return out
+
+
+if __name__ == "__main__":
+    with urllib.request.urlopen(BASE + "/health", timeout=60) as r:
+        h = json.load(r)
+    print("health:", {k: h.get(k) for k in ("model", "cpus", "concurrency", "threads")})
+    post({"inputs": {"texts": ["warm up"]}})
+    post({"inputs": {"images": [IMG]}})
+    print("images alone:       ", run(False))
+    print("images and searches:", run(True))

requirements.txt

@@ -0,0 +1,11 @@
+torch>=2.2
+# SigLIP 2 needs >= 4.49 (Gemma tokenizer config); tested with 5.17
+transformers>=4.49
+pillow>=10
+# sentencepiece: tokenizer of the previous model (siglip-base-patch16-256-multilingual) — keep for rollback
+sentencepiece>=0.2
+protobuf>=4
+# app.py (Space) — the gradio SDK image already ships gradio/fastapi/uvicorn; pins keep local runs identical
+gradio>=5
+fastapi>=0.110
+uvicorn>=0.29

siglip.py

@@ -0,0 +1,275 @@
+"""
+SigLIP embedder shared by app.py (Space) and handler.py (Inference Endpoint).
+
+One model, two towers: images for the indexing pipeline, texts for search queries — both land
+in the same vector space, which is what makes text → image search work. Vectors are
+L2-normalised so cosine similarity is a dot product (Vectorize metric: cosine).
+
+Text is lowercased here, before tokenisation. SigLIP 2 was trained on lowercased text and its
+Gemma tokenizer is case-sensitive: on Pixagram artworks, Title Case queries drop from 97 % to
+3 % recall@1. The original multilingual SigLIP tokenizer lowercases on its own, so this is a
+no-op for it. Doing it here covers every caller (search queries, blog résumés, the demo page).
+
+NaFlex models (google/siglip2-*-naflex) keep each image's aspect ratio: the processor resizes it to
+the largest size that fits MAX_NUM_PATCHES patches of 16x16 px, instead of squashing it to a fixed
+square. A 274x183 artwork is seen as about 19x13 patches rather than distorted to 16x16.
+
+Environment:
+  MODEL_ID         Hub id of a SigLIP/CLIP-family model (default: multilingual SigLIP base, 768-d;
+                   the v2 Space primerz/pixagram-siglip2 sets google/siglip2-base-patch16-256, the
+                   v3 Space google/siglip2-base-patch16-naflex)
+  MAX_NUM_PATCHES  NaFlex only: patch budget per image (default 256, the model's training default).
+                   Reported in every reply so the Worker can check it matches EMBED_PATCHES.
+  MAX_BATCH        images/texts per forward pass (default 32)
+  MAX_CONCURRENCY  embeddings computed at once (default: one per CPU, effective_cpus()); each
+                   gets CPUs / MAX_CONCURRENCY PyTorch threads. One per CPU gives the most
+                   throughput (a small forward pass does not spread well over many threads);
+                   fewer, with more threads each, answer a single query sooner.
+  TORCH_THREADS    PyTorch threads per embedding, overriding CPUs / MAX_CONCURRENCY
+  EMBED_STUB       "1" returns deterministic pseudo-vectors without loading any model — wiring tests only
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import io
+import math
+import os
+import threading
+from typing import Any, Dict, List, Optional
+
+from PIL import Image
+
+# Default = what the production Space runs. Change models per Space with the MODEL_ID variable,
+# so pushing hf/ never switches production's model by accident.
+MODEL_ID = os.environ.get("MODEL_ID", "google/siglip-base-patch16-256-multilingual")
+MAX_BATCH = int(os.environ.get("MAX_BATCH", "32"))
+MAX_NUM_PATCHES = int(os.environ.get("MAX_NUM_PATCHES", "256"))
+MAX_TEXT_TOKENS = 64  # SigLIP text-tower context
+STUB = os.environ.get("EMBED_STUB", "") == "1"
+STUB_DIM = int(os.environ.get("EMBED_STUB_DIM", "768"))
+
+
+def decode_image(b64: str) -> Image.Image:
+    """base64 (optionally a data URI) → RGB PIL image, transparency composited over white."""
+    if b64.startswith("data:"):
+        b64 = b64.split(",", 1)[1]
+    raw = base64.b64decode(b64)
+    img = Image.open(io.BytesIO(raw))
+    img.load()
+    return to_rgb(img)
+
+
+def to_rgb(img: Image.Image) -> Image.Image:
+    if img.mode in ("RGBA", "LA", "P"):
+        rgba = img.convert("RGBA")
+        bg = Image.new("RGBA", rgba.size, (255, 255, 255, 255))
+        img = Image.alpha_composite(bg, rgba)
+    return img.convert("RGB")
+
+
+class Embedder:
+    """Lazy-loading, thread-safe wrapper around a SigLIP model."""
+
+    def __init__(self, model_id: str = MODEL_ID, max_num_patches: int = MAX_NUM_PATCHES) -> None:
+        self.model_id = model_id
+        self._lock = threading.Lock()
+        self._model = None
+        self._processor = None
+        self.device = "cpu"
+        self.dim = STUB_DIM if STUB else 0
+        self.stub = STUB
+        # NaFlex: set on load when the image processor takes a patch budget
+        self.naflex = False
+        self.max_num_patches = max_num_patches
+        # sigmoid(logit_scale * cos + logit_bias) is the model's own text/image match probability
+        self.calibration: Optional[Dict[str, float]] = None
+
+    # ---- loading --------------------------------------------------------------------------
+
+    def load(self) -> "Embedder":
+        if self.stub or self._model is not None:
+            return self
+        with self._lock:
+            if self._model is not None:
+                return self
+            import torch
+            from transformers import AutoModel, AutoProcessor
+
+            self.device = "cuda" if torch.cuda.is_available() else "cpu"
+            # os.cpu_count() reports the HOST's cores inside a container (often 64-96 on Spaces)
+            # while the cgroup allows 2-8; oversubscribing PyTorch's thread pool that badly turns
+            # an 80 ms text embedding into 8-12 s of spin-waiting. Size the pool to what the
+            # container can actually run, shared by the embeddings computed at once.
+            torch.set_num_threads(threads_per_job())
+            model = AutoModel.from_pretrained(self.model_id).to(self.device).eval()
+            self._processor = AutoProcessor.from_pretrained(self.model_id)
+            self.naflex = hasattr(getattr(self._processor, "image_processor", None), "max_num_patches")
+            cfg = model.config
+            self.dim = int(getattr(cfg, "projection_dim", 0) or getattr(cfg.text_config, "projection_size", 0) or cfg.text_config.hidden_size)
+            # SigLIP's learned temperature and bias (exp() of the stored log-scale). CLIP-family models
+            # without a bias report none.
+            try:
+                scale = float(model.logit_scale.exp().item())
+                bias = float(model.logit_bias.item()) if getattr(model, "logit_bias", None) is not None else None
+                self.calibration = {"logit_scale": round(scale, 4), "logit_bias": round(bias, 4)} if bias is not None else None
+            except Exception:
+                self.calibration = None
+            self._model = model
+        # First inference pays one-off kernel/allocator warm-up (~2 s); do it here, not on a user query.
+        try:
+            self.embed_texts(["warm up"])
+            self.embed_images([Image.new("RGB", (32, 32), (128, 128, 128))])
+        except Exception:
+            pass
+        return self
+
+    @property
+    def ready(self) -> bool:
+        return self.stub or self._model is not None
+
+    # ---- embedding ------------------------------------------------------------------------
+
+    def embed_images(self, images: List[Image.Image]) -> List[List[float]]:
+        if self.stub:
+            return [_stub_vector(hashlib.sha256(im.tobytes()).hexdigest()) for im in images]
+        self.load()
+        import torch
+
+        out: List[List[float]] = []
+        # NaFlex: images of any aspect ratio share a batch; the processor pads them to the patch
+        # budget and passes the attention mask and each image's patch grid to the model.
+        kw = {"max_num_patches": self.max_num_patches} if self.naflex else {}
+        with torch.inference_mode():
+            for i in range(0, len(images), MAX_BATCH):
+                batch = [to_rgb(im) for im in images[i : i + MAX_BATCH]]
+                inputs = self._processor(images=batch, return_tensors="pt", **kw).to(self.device)
+                feats = self._model.get_image_features(**inputs)
+                feats = _pooled(feats)
+                feats = torch.nn.functional.normalize(feats, dim=-1)
+                out.extend(feats.cpu().float().tolist())
+        return out
+
+    def embed_texts(self, texts: List[str]) -> List[List[float]]:
+        if self.stub:
+            return [_stub_vector(hashlib.sha256(t.strip().lower().encode()).hexdigest()) for t in texts]
+        self.load()
+        import torch
+
+        out: List[List[float]] = []
+        with torch.inference_mode():
+            for i in range(0, len(texts), MAX_BATCH):
+                # Lowercase: SigLIP 2 was trained on lowercased text (see module docstring).
+                batch = [t.lower() if t.strip() else " " for t in texts[i : i + MAX_BATCH]]
+                # SigLIP was trained with padding="max_length"; keep it so query vectors match.
+                inputs = self._processor(
+                    text=batch, padding="max_length", truncation=True, max_length=MAX_TEXT_TOKENS, return_tensors="pt"
+                ).to(self.device)
+                feats = self._model.get_text_features(**inputs)
+                feats = _pooled(feats)
+                feats = torch.nn.functional.normalize(feats, dim=-1)
+                out.extend(feats.cpu().float().tolist())
+        return out
+
+    # ---- request handling shared by the Space route and the Endpoint handler -----------------
+
+    def handle(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        {"inputs": {"images": [b64, ...], "texts": [str, ...]}}  (either key optional)
+        → {"model", "dim", "embeddings": [image vectors..., text vectors...], "calibration": {"logit_scale", "logit_bias"}}
+        """
+        inputs = payload.get("inputs", payload) if isinstance(payload, dict) else payload
+        if isinstance(inputs, str):
+            inputs = {"texts": [inputs]}
+        if not isinstance(inputs, dict):
+            raise ValueError("body must be {\"inputs\": {\"images\": [...], \"texts\": [...]}}")
+        images_b64 = inputs.get("images") or []
+        texts = inputs.get("texts") or []
+        if isinstance(images_b64, str):
+            images_b64 = [images_b64]
+        if isinstance(texts, str):
+            texts = [texts]
+        if not images_b64 and not texts:
+            raise ValueError("provide inputs.images (base64) and/or inputs.texts")
+        if len(images_b64) + len(texts) > 256:
+            raise ValueError("at most 256 items per request")
+
+        embeddings: List[List[float]] = []
+        if images_b64:
+            embeddings.extend(self.embed_images([decode_image(b) for b in images_b64]))
+        if texts:
+            embeddings.extend(self.embed_texts([str(t) for t in texts]))
+        out: Dict[str, Any] = {"model": self.model_id if not self.stub else "stub", "dim": len(embeddings[0]) if embeddings else self.dim, "embeddings": embeddings}
+        if self.calibration:
+            out["calibration"] = self.calibration
+        if self.naflex:
+            out["max_num_patches"] = self.max_num_patches
+        return out
+
+
+def effective_cpus(cap: int = 16) -> int:
+    """CPUs this process may really use: affinity mask ∩ cgroup quota, capped."""
+    n = os.cpu_count() or 1
+    try:
+        n = min(n, len(os.sched_getaffinity(0)))
+    except Exception:
+        pass
+    try:  # cgroup v2
+        quota, period = open("/sys/fs/cgroup/cpu.max").read().split()[:2]
+        if quota != "max":
+            n = min(n, max(1, int(int(quota) / int(period))))
+    except Exception:
+        try:  # cgroup v1
+            quota = int(open("/sys/fs/cgroup/cpu/cpu.cfs_quota_us").read())
+            period = int(open("/sys/fs/cgroup/cpu/cpu.cfs_period_us").read())
+            if quota > 0:
+                n = min(n, max(1, quota // period))
+        except Exception:
+            pass
+    return max(1, min(n, cap))
+
+
+def concurrency() -> int:
+    """Embeddings computed at once: MAX_CONCURRENCY, else one per CPU."""
+    env = os.environ.get("MAX_CONCURRENCY", "")
+    return max(1, int(env)) if env.isdigit() else effective_cpus()
+
+
+def threads_per_job() -> int:
+    """PyTorch threads per embedding: TORCH_THREADS, else the CPUs shared by the embeddings at once."""
+    env = os.environ.get("TORCH_THREADS", "")
+    if env.isdigit() and int(env) > 0:
+        return int(env)
+    return max(1, effective_cpus() // concurrency())
+
+
+def _pooled(feats):
+    """transformers ≥5 may return a ModelOutput; older versions return the tensor directly."""
+    if hasattr(feats, "pooler_output") and feats.pooler_output is not None:
+        return feats.pooler_output
+    if hasattr(feats, "last_hidden_state") and not hasattr(feats, "shape"):
+        return feats.last_hidden_state
+    return feats
+
+
+def _stub_vector(seed_hex: str) -> List[float]:
+    """Deterministic unit vector from a hash — lets the Worker ↔ Space wiring be tested without weights."""
+    vals: List[float] = []
+    h = seed_hex
+    while len(vals) < STUB_DIM:
+        h = hashlib.sha256(h.encode()).hexdigest()
+        vals.extend(int(h[i : i + 2], 16) / 127.5 - 1.0 for i in range(0, 64, 2))
+    vals = vals[:STUB_DIM]
+    norm = math.sqrt(sum(v * v for v in vals)) or 1.0
+    return [v / norm for v in vals]
+
+
+_shared: Optional[Embedder] = None
+
+
+def shared() -> Embedder:
+    global _shared
+    if _shared is None:
+        _shared = Embedder()
+    return _shared