"""Photograph your ingredients; a small VLM lists what it sees. Uses MiniCPM-V 4.6 (openbmb) — an open-vocabulary vision-language model that can name produce, pantry items and even read spice-jar labels. We ask it for a plain ingredient list, then hand that straight to the seasoning planner. (LocateAnything is for bounding-box grounding of known targets; here we want open discovery, so a describe-and-list VLM is the right tool.) ~2B params, so it sits comfortably alongside Mellum 2's 12B under the 32B cap. Real inference runs only on a GPU (the Space). With MOCK_VISION=1 — or whenever torch/the model can't load — `detect_ingredients` returns a clearly-labelled sample list so the photo→pantry→plan flow is demonstrable offline. """ import os import re VISION_MODEL_ID = os.environ.get("VISION_MODEL_ID", "openbmb/MiniCPM-V-4.6") # Vision has its own backend because, unlike the reasoning model, MiniCPM-V has a # free hosted API — so you can run it real with no GPU at all. # openbmb : OpenBMB/ModelBest free hosted API (no GPU, recommended real path) # modal : your Modal endpoint | zerogpu : in-Space GPU VISION_BACKEND = os.environ.get("VISION_BACKEND", "zerogpu") # Mock only when asked, OR in MOCK_LLM dev runs that DIDN'T pick a real vision # backend — so `MOCK_LLM=1 VISION_BACKEND=openbmb` gives real vision + scripted reasoning. MOCK_VISION = os.environ.get("MOCK_VISION") == "1" or ( os.environ.get("MOCK_LLM") == "1" and "VISION_BACKEND" not in os.environ) MODAL_VISION_URL = os.environ.get("MODAL_VISION_URL", "") # OpenBMB hosted API (OpenAI-compatible). Public free key ships as the default but # is shared/rate-limited — override with your own from platform.modelbest.cn. OPENBMB_API_URL = os.environ.get("MINICPM_API_URL", "https://api.modelbest.cn/v1/chat/completions") OPENBMB_API_KEY = os.environ.get("MINICPM_API_KEY", "sk-pQ8L2zF3XmR5kY9wV4jB7hN1tC6vM0xG3aD5sH2bJ9lK4cZ8") OPENBMB_API_MODEL = os.environ.get("MINICPM_API_MODEL", "MiniCPM-V-4.6-Instruct") DETECT_PROMPT = ( "You are looking at a photo of someone's kitchen ingredients. List ONLY the " "specific food ingredients you can clearly identify. Use canonical SINGULAR " "names a recipe would use (e.g. 'cumin', 'garlic', 'lentil', 'tomato', " "'papaya'). Do NOT use vague category words like 'produce', 'vegetables', " "'spices', or 'herbs' — name the actual item. Reply with a single " "comma-separated list and nothing else." ) # Shown offline so the demo flow works without a GPU — clearly not a real read. _MOCK_DETECTION = [ "lentil", "cumin", "coriander seed", "turmeric", "garlic", "onion", "ginger", "tomato", "lemon", ] _model = None _tokenizer = None def _load(): """Lazy-load MiniCPM-V. Module-level load is fine on ZeroGPU, but lazy keeps startup cheap when vision isn't used.""" global _model, _tokenizer if _model is not None: return import torch from transformers import AutoModel, AutoTokenizer _tokenizer = AutoTokenizer.from_pretrained(VISION_MODEL_ID, trust_remote_code=True) _model = AutoModel.from_pretrained( VISION_MODEL_ID, trust_remote_code=True, dtype=torch.bfloat16 ).eval().to("cuda") def _to_image(image): """Accept a filepath or a PIL image; return RGB PIL.""" from PIL import Image if isinstance(image, str): return Image.open(image).convert("RGB") return image.convert("RGB") def _parse_list(text: str) -> list[str]: """Turn the model's reply into clean, de-duplicated ingredient names.""" text = re.sub(r"^[^:]*:", "", text.strip()) # drop any "I can see:" preamble items, seen = [], set() for part in re.split(r"[,\n;]+", text): name = re.sub(r"^[\s\-\*\d\.\)]+", "", part).strip().lower() # strip bullets/numbering name = re.sub(r"\s+", " ", name).strip(" .") if 1 < len(name) <= 30 and name not in seen: seen.add(name) items.append(name) return items[:20] def _detect_zerogpu(pil) -> str: _load() msgs = [{"role": "user", "content": [pil, DETECT_PROMPT]}] reply = _model.chat(image=None, msgs=msgs, tokenizer=_tokenizer, sampling=False, max_new_tokens=200) return reply if isinstance(reply, str) else str(reply) def _detect_modal(pil) -> str: import base64 import io import httpx buf = io.BytesIO() pil.save(buf, format="JPEG") payload = {"image_b64": base64.b64encode(buf.getvalue()).decode(), "prompt": DETECT_PROMPT} resp = httpx.post(MODAL_VISION_URL, json=payload, timeout=180) resp.raise_for_status() return resp.json()["text"] def _detect_openbmb(pil) -> str: """Call OpenBMB's free hosted MiniCPM-V API (OpenAI-compatible). No GPU.""" import base64 import io import httpx buf = io.BytesIO() pil.save(buf, format="JPEG") data_uri = "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode() payload = { "model": OPENBMB_API_MODEL, "messages": [{"role": "user", "content": [ {"type": "text", "text": DETECT_PROMPT}, {"type": "image_url", "image_url": {"url": data_uri}}, ]}], } resp = httpx.post(OPENBMB_API_URL, json=payload, timeout=120, headers={"Authorization": f"Bearer {OPENBMB_API_KEY}"}) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"] def detect_ingredients(image) -> tuple[list[str], str]: """Return (ingredient_names, source_note). Never raises — vision failures degrade to an empty list with an explanatory note so the UI stays usable.""" if image is None: return [], "No photo provided." if MOCK_VISION: return list(_MOCK_DETECTION), "🔬 mock vision (deploy on GPU for a real read)" try: pil = _to_image(image) if VISION_BACKEND == "openbmb": raw = _detect_openbmb(pil) elif VISION_BACKEND == "modal": raw = _detect_modal(pil) else: raw = _detect_zerogpu(pil) names = _parse_list(raw) return names, f"👁️ detected by MiniCPM-V ({VISION_BACKEND})" except Exception as exc: return [], f"Vision unavailable ({type(exc).__name__}: {exc}). Type your pantry instead."