Spaces:
Sleeping
Sleeping
| """ | |
| 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): | |
| 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 | |
| # --------------------------------------------------------------------------- | |
| 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, | |
| ) | |
| 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, | |
| ) | |
| 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, | |
| ) | |
| 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 | |
| ) | |
| 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() |