""" Pantry API — Qwen2.5-VL-7B-Instruct su Hugging Face Spaces. Stateless: nessun gr.State, nessuna cronologia. Il contesto della dispensa arriva a ogni chiamata e non viene conservato dal codice applicativo. Endpoint: /genera proxy 1:1 con AiRequest dell'app Android /analizza_foto foto + contesto -> {"prodotti":[...]} /piano contesto -> {"giorni":[...]} /comando contesto + messaggio -> {"azione":...,"prodotti":[...]} /chat contesto + domanda -> testo """ from __future__ import annotations import functools import json import os import re import threading from typing import Optional os.environ.setdefault("GRADIO_ANALYTICS_ENABLED", "False") import gradio as gr import torch from PIL import Image from qwen_vl_utils import process_vision_info from transformers import AutoProcessor, Qwen2_5_VLForConditionalGeneration try: import spaces def gpu(fn=None, **kwargs): return spaces.GPU(**kwargs)(fn) if fn is not None else spaces.GPU(**kwargs) except ImportError: # Permette di aprire il file in locale senza il pacchetto spaces. def gpu(fn=None, **kwargs): def deco(f): @functools.wraps(f) def wrapper(*args, **kw): return f(*args, **kw) return wrapper return deco(fn) if fn is not None else deco MODEL_ID = os.getenv("PANTRY_MODEL_ID", "Qwen/Qwen2.5-VL-7B-Instruct") QUANTIZZAZIONE = os.getenv("PANTRY_QUANTIZZAZIONE", "bf16").lower() MAX_TOKENS_CAP = 8192 DEFAULT_MAX_TOKENS = 2048 _modello = None _processore = None _lock_modello = threading.Lock() def _carica_modello(): """Caricamento pigro: la GPU di ZeroGPU esiste solo dentro @gpu.""" global _modello, _processore if _modello is not None: return _modello, _processore with _lock_modello: if _modello is not None: return _modello, _processore if not torch.cuda.is_available(): raise gr.Error( "Questo Space richiede una GPU. Imposta Hardware su GPU " "(ZeroGPU o dedicata) nelle impostazioni dello Space." ) dtype = torch.bfloat16 if torch.cuda.is_bf16_supported() else torch.float16 _processore = AutoProcessor.from_pretrained(MODEL_ID, trust_remote_code=True) parametri = { "torch_dtype": dtype, "device_map": "auto", "trust_remote_code": True, } if QUANTIZZAZIONE == "4bit": from transformers import BitsAndBytesConfig parametri.pop("torch_dtype", None) parametri["quantization_config"] = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_compute_dtype=dtype, bnb_4bit_quant_type="nf4", ) _modello = Qwen2_5_VLForConditionalGeneration.from_pretrained( MODEL_ID, **parametri ).eval() return _modello, _processore # --------------------------------------------------------------------------- # Schemi: devono restare allineati a Schemi.kt dell'app. # --------------------------------------------------------------------------- SCHEMA_PRODOTTI = r""" { "type": "object", "properties": { "azione": { "type": "string", "enum": ["aggiungi", "finito", "usa", "nessuna"] }, "prodotti": { "type": "array", "items": { "type": "object", "properties": { "nome": {"type": "string"}, "quantita": {"type": "number"}, "unita": {"type": "string", "enum": ["g", "ml", "pz"]}, "posto": {"type": "string"}, "scadenza": {"type": "string"}, "sicurezza": {"type": "number"} }, "required": ["nome"] } } }, "required": ["azione", "prodotti"] } """ SCHEMA_PIANO = r""" { "type": "object", "properties": { "giorni": { "type": "array", "items": { "type": "object", "properties": { "numero": {"type": "integer"}, "pasti": { "type": "array", "items": { "type": "object", "properties": { "tipo": { "type": "string", "enum": ["colazione", "pranzo", "cena", "spuntino"] }, "nome": {"type": "string"}, "ingredienti": { "type": "array", "items": { "type": "object", "properties": { "voceId": {"type": "integer"}, "nome": {"type": "string"}, "quantita": {"type": "number"}, "unita": {"type": "string", "enum": ["g", "ml", "pz"]} }, "required": ["voceId", "nome", "quantita", "unita"] } } }, "required": ["tipo", "nome", "ingredienti"] } } }, "required": ["numero", "pasti"] } } }, "required": ["giorni"] } """ # --------------------------------------------------------------------------- # Prompt e generazione # --------------------------------------------------------------------------- def _componi_sistema(system_instruction: str, json_output: bool, schema_json: str) -> str: sistema = (system_instruction or "").strip() if json_output: sistema += ( "\n\nRispondi SOLO con un oggetto JSON valido, senza markdown, " "senza commenti e senza testo prima o dopo. Usa esattamente le " "chiavi e i tipi dello schema." ) if schema_json and schema_json.strip(): sistema += "\n\nSchema JSON da rispettare:\n" + schema_json.strip() return sistema def _prepara_ingressi( system_instruction: str, user_text: str, immagine: Optional[Image.Image], json_output: bool, schema_json: str, ): modello, processore = _carica_modello() sistema = _componi_sistema(system_instruction, json_output, schema_json) contenuto = [] if immagine is not None: contenuto.append({"type": "image", "image": immagine}) contenuto.append({"type": "text", "text": user_text}) messaggi = [ {"role": "system", "content": [{"type": "text", "text": sistema}]}, {"role": "user", "content": contenuto}, ] testo_prompt = processore.apply_chat_template( messaggi, tokenize=False, add_generation_prompt=True ) immagini, video = process_vision_info(messaggi) ingressi = processore( text=[testo_prompt], images=immagini, videos=video, padding=True, return_tensors="pt", ).to(modello.device) return modello, processore, ingressi def _genera_grezzo( system_instruction: str, user_text: str, immagine: Optional[Image.Image], json_output: bool, schema_json: str, max_tokens: int, temperatura: float, ) -> str: modello, processore, ingressi = _prepara_ingressi( system_instruction, user_text, immagine, json_output, schema_json ) token_max = max(16, min(int(max_tokens or DEFAULT_MAX_TOKENS), MAX_TOKENS_CAP)) parametri = dict(max_new_tokens=token_max) if temperatura > 0: parametri.update( do_sample=True, temperature=max(0.0, min(float(temperatura), 1.5)), top_p=0.95, ) else: parametri.update(do_sample=False) with torch.inference_mode(): generati = modello.generate(**ingressi, **parametri) ritagliati = [ uscita[len(ingresso):] for ingresso, uscita in zip(ingressi.input_ids, generati) ] testo = processore.batch_decode( ritagliati, skip_special_tokens=True, clean_up_tokenization_spaces=False, )[0] return testo.strip() def _estrai_json(testo: str) -> str: testo = (testo or "").strip() if testo.startswith("```"): testo = re.sub(r"^```(?:json)?\s*", "", testo, flags=re.IGNORECASE) testo = re.sub(r"\s*```$", "", testo) inizio = testo.find("{") fine = testo.rfind("}") if inizio >= 0 and fine > inizio: return testo[inizio:fine + 1] return testo def _valida_json(testo: str, schema_json: str) -> str: if not (schema_json or "").strip(): return testo import jsonschema schema = json.loads(schema_json) dati = json.loads(testo) jsonschema.validate(dati, schema) return json.dumps(dati, ensure_ascii=False) def _genera_con_validazione( system_instruction: str, user_text: str, immagine: Optional[Image.Image], json_output: bool, schema_json: str, max_tokens: int, temperatura: float, ) -> str: primo = _genera_grezzo( system_instruction, user_text, immagine, json_output, schema_json, max_tokens, temperatura, ) if not json_output: return primo candidato = _estrai_json(primo) try: return _valida_json(candidato, schema_json) except Exception as errore: # Un solo tentativo di riparazione: il modello vede l'errore e riscrive. riparazione = _genera_grezzo( system_instruction=system_instruction, user_text=( f"{user_text}\n\nLa risposta precedente era:\n{primo}\n\n" f"Non è valida: {errore}. Riscrivi l'oggetto completo, solo JSON." ), immagine=immagine, json_output=True, schema_json=schema_json, max_tokens=max_tokens, temperatura=0.0, ) try: return _valida_json(_estrai_json(riparazione), schema_json) except Exception as secondo: raise gr.Error(f"Il modello non ha prodotto un JSON valido: {secondo}") from secondo # --------------------------------------------------------------------------- # Endpoint # --------------------------------------------------------------------------- @gpu(duration=60) def genera( system_instruction: str, user_text: str, immagine: Optional[Image.Image], json_output: bool, schema_json: str, max_tokens: int = DEFAULT_MAX_TOKENS, temperatura: float = 0.2, ) -> str: return _genera_con_validazione( system_instruction, user_text, immagine, json_output, schema_json, max_tokens, temperatura, ) @gpu(duration=60) def analizza_foto(immagine: Optional[Image.Image], contesto: str) -> str: if immagine is None: raise gr.Error("Carica una foto del frigo, della dispensa o dello scontrino.") sistema = ( "Sei l'assistente della dispensa di casa.\n" "Leggi la foto della spesa e riconosci i prodotti acquistati.\n" "Per ogni prodotto scrivi nome, quantita' e unita' (solo g, ml o pz).\n" "Scegli il posto fra quelli elencati nel contesto.\n" "Aggiungi sicurezza da 0 a 1: quanto sei sicuro della lettura.\n" "Per una foto, azione vale sempre 'nessuna'.\n" "Non inventare prodotti che nella foto non si vedono.\n" "Rispondi solo con il JSON dello schema, senza spiegazioni.\n\n" + (contesto or "") ) return _genera_con_validazione( sistema, "Leggi questa foto della spesa e riconosci i prodotti.", immagine, True, SCHEMA_PRODOTTI, 1024, 0.0, ) @gpu(duration=60) def piano(contesto: str) -> str: sistema = ( "Sei il coach della dispensa di casa.\n" "Prepara un piano di 3 giorni: colazione, pranzo e cena per ogni giorno.\n" "Aggiungi gli spuntini solo se il profilo prevede piu' di 3 pasti al giorno.\n" "Rispetta calorie, macro, tempo ai fornelli e livello di cucina del profilo.\n" "Sale, olio e acqua si considerano sempre presenti.\n" "Per ogni altro ingrediente usa SOLO le voci dell'elenco, indicando il loro " "identificativo in voceId.\n" "Rispondi solo con il JSON dello schema, senza spiegazioni.\n\n" + (contesto or "") ) return _genera_con_validazione( sistema, "Prepara il piano dei 3 giorni.", None, True, SCHEMA_PIANO, MAX_TOKENS_CAP, 0.2, ) @gpu(duration=60) def comando(contesto: str, messaggio: str) -> str: sistema = ( "Sei l'assistente della dispensa di casa.\n" "Capisci se il messaggio dell'utente modifica la dispensa e lo traduci " "nei dati richiesti.\n" "- azione 'aggiungi': ha comprato qualcosa (serve quantita' e unita').\n" "- azione 'finito': una cosa e' finita (basta il nome).\n" "- azione 'usa': ne ha consumata una quantita' (serve quantita' e unita').\n" "- azione 'nessuna': non e' un comando sulla dispensa.\n" "Usa solo i posti e le unita' elencati; non convertire mai fra g, ml e pz.\n" "Rispondi solo con il JSON dello schema, senza spiegazioni.\n\n" + (contesto or "") ) return _genera_con_validazione( sistema, messaggio, None, True, SCHEMA_PRODOTTI, 1024, 0.0 ) @gpu(duration=60) def chat(contesto: str, domanda: str) -> str: sistema = ( "Sei il coach della dispensa di casa.\n" "Rispondi in italiano, in modo concreto e breve.\n" "Usa SOLO le voci elencate nel contesto, con le quantita' indicate.\n" "Sale, olio e acqua si considerano sempre presenti.\n" "Se un piatto richiede qualcosa che non c'e', non proporlo.\n\n" + (contesto or "") ) return _genera_con_validazione( sistema, domanda, None, False, "", 2048, 0.4 ) # --------------------------------------------------------------------------- # Interfaccia # --------------------------------------------------------------------------- with gr.Blocks(title="Pantry API — Qwen 7B") as demo: gr.Markdown( "# Pantry API\n" "Qwen2.5-VL-7B-Instruct. Lo Space è senza stato: a ogni chiamata " "arriva il contesto completo della dispensa." ) with gr.Tab("Generico"): g_sistema = gr.Textbox(label="System instruction", lines=6) g_utente = gr.Textbox(label="User text", lines=4) g_immagine = gr.Image(type="pil", label="Immagine (opzionale)") g_json = gr.Checkbox(label="JSON output", value=False) g_schema = gr.Textbox(label="Schema JSON", lines=4) g_max = gr.Slider(64, MAX_TOKENS_CAP, value=DEFAULT_MAX_TOKENS, step=64, label="Max new tokens") g_temp = gr.Slider(0.0, 1.0, value=0.2, step=0.1, label="Temperatura") g_out = gr.Textbox(label="Risposta", lines=14) gr.Button("Genera").click( genera, [g_sistema, g_utente, g_immagine, g_json, g_schema, g_max, g_temp], g_out, api_name="/genera", ) with gr.Tab("Foto"): f_immagine = gr.Image(type="pil", label="Foto frigo / dispensa / scontrino") f_contesto = gr.Textbox(label="Contesto dispensa", lines=8) f_out = gr.Code(label="Prodotti (JSON)", language="json") gr.Button("Analizza").click( analizza_foto, [f_immagine, f_contesto], f_out, api_name="/analizza_foto", ) with gr.Tab("Piano"): p_contesto = gr.Textbox(label="Contesto dispensa", lines=10) p_out = gr.Code(label="Piano (JSON)", language="json") gr.Button("Prepara piano").click( piano, [p_contesto], p_out, api_name="/piano" ) with gr.Tab("Comando"): c_contesto = gr.Textbox(label="Contesto dispensa", lines=8) c_messaggio = gr.Textbox(label="Messaggio") c_out = gr.Code(label="Comando (JSON)", language="json") gr.Button("Interpreta").click( comando, [c_contesto, c_messaggio], c_out, api_name="/comando" ) with gr.Tab("Chat"): h_contesto = gr.Textbox(label="Contesto dispensa", lines=8) h_domanda = gr.Textbox(label="Domanda") h_out = gr.Textbox(label="Risposta", lines=12) gr.Button("Chiedi").click( chat, [h_contesto, h_domanda], h_out, api_name="/chat" ) demo.queue(default_concurrency_limit=1, max_size=8) if __name__ == "__main__": demo.launch()