import os import logging import uvicorn from fastapi import FastAPI, HTTPException from pydantic import BaseModel import rdflib from rdflib.plugins.sparql.parser import parseQuery from huggingface_hub import InferenceClient import re import torch from transformers import DistilBertForSequenceClassification, DistilBertTokenizer # --------------------------------------------------------------------------- # CONFIGURAZIONE LOGGING # --------------------------------------------------------------------------- logging.basicConfig( level=logging.DEBUG, # Utilizziamo il livello DEBUG per un log più dettagliato format="%(asctime)s - %(levelname)s - %(message)s", handlers=[logging.FileHandler("app.log"), logging.StreamHandler()] ) logger = logging.getLogger(__name__) # Determina il device (GPU se disponibile, altrimenti CPU) device = torch.device("cuda" if torch.cuda.is_available() else "cpu") logger.info(f"Device per il classificatore: {device}") # Carica il modello e il tokenizer del classificatore fine-tuned try: logger.info("Caricamento del modello di classificazione fine-tuned da 'finetuned-bert-model'.") classifier_model = DistilBertForSequenceClassification.from_pretrained("finetuned-bert-model") classifier_tokenizer = DistilBertTokenizer.from_pretrained("finetuned-bert-model") classifier_model.to(device) logger.info("Modello di classificazione caricato correttamente.") except Exception as e: logger.error(f"Errore nel caricamento del modello di classificazione: {e}") classifier_model = None explanation_dict = {} # --------------------------------------------------------------------------- # COSTANTI / CHIAVI / MODELLI # --------------------------------------------------------------------------- # Nota: HF_API_KEY deve essere impostata a una chiave valida di Hugging Face. HF_API_KEY = os.getenv("HF_API_KEY") if not HF_API_KEY: # Se la chiave API non è impostata, solleva un errore logger.error("HF_API_KEY non impostata.") raise EnvironmentError("HF_API_KEY non impostata.") # Nome del modello Hugging Face per generare query SPARQL e risposte finali HF_MODEL = "meta-llama/Llama-3.3-70B-Instruct" # Nome del modello Hugging Face per rilevamento lingua LANG_DETECT_MODEL = "papluca/xlm-roberta-base-language-detection" # Prefisso per i modelli di traduzione su Hugging Face TRANSLATOR_MODEL_PREFIX = "Helsinki-NLP/opus-mt" # --------------------------------------------------------------------------- # INIZIALIZZAZIONE CLIENT HUGGING FACE (una volta sola) # --------------------------------------------------------------------------- """ Qui inizializziamo i client necessari. In questo modo, evitiamo di istanziare continuamente nuovi oggetti InferenceClient a ogni chiamata delle funzioni. - hf_generation_client: per generare query SPARQL e risposte stile "guida museale" - lang_detect_client: per rilevare la lingua della domanda e della risposta """ try: logger.info("[Startup] Inizializzazione client HF per generazione (modello di LLM).") hf_generation_client = InferenceClient( token=HF_API_KEY, model=HF_MODEL ) logger.info("[Startup] Inizializzazione client HF per rilevamento lingua.") lang_detect_client = InferenceClient( token=HF_API_KEY, model=LANG_DETECT_MODEL ) except Exception as ex: logger.error(f"Errore inizializzazione dei client Hugging Face: {ex}") raise HTTPException(status_code=500, detail="Impossibile inizializzare i modelli Hugging Face.") # --------------------------------------------------------------------------- # CARICAMENTO ONTOLOGIA # --------------------------------------------------------------------------- """ Carichiamo il file RDF/XML contenente l'ontologia del museo. Questo file è fondamentale per l'esecuzione di query SPARQL, in quanto definisce le classi, le proprietà e le istanze presenti nell'ontologia del museo. """ BASE_DIR = os.path.dirname(os.path.abspath(__file__)) RDF_FILE = os.path.join(BASE_DIR, "Ontologia_corretto-2.rdf") ontology_graph = rdflib.Graph() try: logger.info(f"Caricamento ontologia da file: {RDF_FILE}") # Indichiamo che l'ontologia è in formato RDF/XML ontology_graph.parse(RDF_FILE, format="xml") logger.info("Ontologia RDF caricata correttamente (formato XML).") except Exception as e: logger.error(f"Errore nel caricamento dell'ontologia: {e}") raise e # --------------------------------------------------------------------------- # DEFINIZIONE DELL'APP FASTAPI # --------------------------------------------------------------------------- app = FastAPI() # --------------------------------------------------------------------------- # Pydantic Model per la richiesta # --------------------------------------------------------------------------- class AssistantRequest(BaseModel): """ Questo modello Pydantic definisce lo schema della richiesta che riceverà l'endpoint /assistant. Contiene: - message: la domanda del visitatore - max_tokens: max di token per le risposte (di default 512) - temperature: temperatura di generazione (di default 0.5) """ message: str max_tokens: int = 512 temperature: float = 0.5 # --------------------------------------------------------------------------- # FUNZIONI DI SUPPORTO (Prompts, validazione SPARQL, correzioni, ecc.) # --------------------------------------------------------------------------- def create_system_prompt_for_classification(ontology_turtle:str) -> str: prompt = f""" SEI UN CLASSIFICATORE DI DOMANDE NEL CONTESTO DI UN MUSEO. DEVI ANALIZZARE LA DOMANDA DELL'UTENTE E, BASANDOTI SUL CONTESTO DELL'ONTOLOGIA (CHE TI VIENE FORNITA QUI SOTTO), RISPONDERE SOLAMENTE CON "SI" SE LA DOMANDA È PERTINENTE AL MUSEO, O CON "NO" SE LA DOMANDA NON È PERTINENTE. RICORDA: - "PERTINENTE" SIGNIFICA CHE LA DOMANDA RIGUARDA OPERE D'ARTE, ESPOSTE, BIGLIETTI, VISITATORI, CURATORI, RESTAURI, STANZE E TUTTI GLI ASPETTI RELATIVI ALL'AMBIENTE MUSEALE. - "NON PERTINENTE" SIGNIFICA CHE LA DOMANDA RIGUARDA ARGOMENTI ESTERNI AL CONTESTO MUSEALE (PER ESEMPIO, TECNOLOGIA, SPORT, POLITICA, CUCINA, ECC.). NON FORNIRE ULTERIORI SPIEGAZIONI O COMMENTI: LA TUA RISPOSTA DEVE ESSERE ESCLUSIVAMENTE "SI" O "NO". ONTOLOGIA (TURTLE/XML/ALTRO FORMATO): {ontology_turtle} FINE ONTOLOGIA. """ return prompt def create_system_prompt_for_sparql(ontology_turtle: str) -> str: """ Genera il testo di prompt che istruisce il modello su come costruire SOLO UNA query SPARQL, in un'unica riga, o in alternativa 'NO_SPARQL' se la domanda non è pertinente all'ontologia. Il prompt include regole di formattazione, esempi di domanda-risposta SPARQL e regole rigorose per la gestione della posizione dell'opera. Parametri: - ontology_turtle: una stringa con l'ontologia in formato Turtle (o simile). Ritorna: - Il testo da usare come "system prompt" per il modello generativo. """ prompt = f"""SEI UN GENERATORE DI QUERY SPARQL PER L'ONTOLOGIA DI UN MUSEO. DEVI GENERARE SOLO UNA QUERY SPARQL (IN UNA SOLA RIGA) SE LA DOMANDA RIGUARDA INFORMAZIONI NELL'ONTOLOGIA. SE LA DOMANDA NON È ATTINENTE, RISPONDI 'NO_SPARQL'. REGOLE SINTATTICHE RIGOROSE: 1) Usare: PREFIX progettoMuseo: 2) Query in UNA SOLA RIGA (niente a capo), forma: PREFIX progettoMuseo: <...> SELECT ?x WHERE {{ ... }} LIMIT N 3) Attento agli spazi: - Dopo SELECT: es. SELECT ?autore - Tra proprietà e variabile: es. progettoMuseo:autoreOpera ?autore . - Non incollare il '?' a 'progettoMuseo:'. - Ogni tripla termina con un punto. 4) Se non puoi generare una query valida, rispondi solo 'NO_SPARQL'. Esempi di Domande Specifiche e relative Query: 1) Utente: Chi ha creato l'opera 'Afrodite di Milo'? Risposta: PREFIX progettoMuseo: SELECT ?autore WHERE {{ progettoMuseo:AfroditeDiMilo progettoMuseo:autoreOpera ?autore . }} LIMIT 10 2) Utente: Quali sono le tecniche utilizzate nelle opere? Risposta: PREFIX progettoMuseo: SELECT ?opera ?tecnica WHERE {{ ?opera progettoMuseo:tecnicaOpera ?tecnica . }} LIMIT 100 3) Utente: Quali sono le dimensioni delle opere? Risposta: PREFIX progettoMuseo: SELECT ?opera ?dimensione WHERE {{ ?opera progettoMuseo:dimensioneOpera ?dimensione . }} LIMIT 100 4) Utente: Quali opere sono esposte nella stanza Greca? Risposta: PREFIX progettoMuseo: SELECT ?opera WHERE {{ progettoMuseo:StanzaGrecia progettoMuseo:Espone ?opera . }} LIMIT 100 5) Utente: Quali sono le proprietà e i tipi delle proprietà nell'ontologia? Risposta: PREFIX rdf: PREFIX owl: PREFIX progettoMuseo: SELECT DISTINCT ?property ?type WHERE {{ ?property rdf:type ?type . FILTER(?type IN (owl:ObjectProperty, owl:DatatypeProperty)) }} 6) Utente: Recupera tutti i biglietti e i tipi di biglietto. Risposta: PREFIX progettoMuseo: SELECT ?biglietto ?tipoBiglietto WHERE {{ ?biglietto rdf:type progettoMuseo:Biglietto . ?biglietto progettoMuseo:tipoBiglietto ?tipoBiglietto . }} LIMIT 100 7) Utente: Recupera tutti i visitatori e i tour a cui partecipano. Risposta: PREFIX progettoMuseo: SELECT ?visitatore ?tour WHERE {{ ?visitatore progettoMuseo:Partecipazione_a_Evento ?tour . }} LIMIT 100 8) Utente: Recupera tutte le stanze tematiche e le opere esposte. Risposta: PREFIX progettoMuseo: SELECT ?stanza ?opera WHERE {{ ?stanza rdf:type progettoMuseo:Stanza_Tematica . ?stanza progettoMuseo:Espone ?opera . }} LIMIT 100 9) Utente: Recupera tutte le opere con materiale 'Marmo'. Risposta: PREFIX progettoMuseo: SELECT ?opera WHERE {{ ?opera progettoMuseo:materialeOpera "Marmo"@it . }} LIMIT 100 10) Utente: Recupera tutti i visitatori con data di nascita dopo il 2000. Risposta: PREFIX progettoMuseo: SELECT ?visitatore WHERE {{ ?visitatore rdf:type progettoMuseo:Visitatore_Individuale . ?visitatore progettoMuseo:dataDiNascitaVisitatore ?data . FILTER(?data > "2000-01-01T00:00:00"^^xsd:dateTime) . }} LIMIT 100 NUOVE REGOLE RIGUARDANTI LA POSIZIONE DELL'OPERA: - SE la domanda include richieste come "fammi vedere l'opera X", "mi fai vedere l'opera X", "portami all'opera X", "voglio vedere l'opera X" o simili, la query SPARQL DEVE restituire la posizione dell'opera includendo la proprietà progettoMuseo:posizioneOpera. - SE la domanda include espressioni come "dove si trova l'opera X", "ubicazione dell'opera X" o simili, la query SPARQL NON deve restituire la proprietà progettoMuseo:posizioneOpera, limitandosi al massimo a riportare la relazione esistente. Esempi Aggiuntivi: 11) Utente: Fammi vedere l'opera 'Discobolo'. Risposta: PREFIX progettoMuseo: SELECT ?opera ?posizione WHERE {{ progettoMuseo:Discobolo progettoMuseo:posizioneOpera ?posizione . }} LIMIT 10 12) Utente: Mi fai vedere l'opera 'Gioconda'? Risposta: PREFIX progettoMuseo: SELECT ?opera ?posizione WHERE {{ progettoMuseo:Gioconda progettoMuseo:posizioneOpera ?posizione . }} LIMIT 10 13) Utente: Portami all'opera 'Autoritratto'. Risposta: PREFIX progettoMuseo: SELECT ?opera ?posizione WHERE {{ progettoMuseo:Autoritratto progettoMuseo:posizioneOpera ?posizione . }} LIMIT 10 14) Utente: Dove si trova l'opera 'La Notte Stellata'? Risposta: PREFIX progettoMuseo: SELECT ?opera WHERE {{ progettoMuseo:LaNotteStellata ?rel ?info . FILTER(?rel != progettoMuseo:posizioneOpera) }} LIMIT 10 15) Utente: Qual è l'ubicazione dell'opera 'Ramo Di Mandorlo Fiorito'? Risposta: PREFIX progettoMuseo: SELECT ?opera WHERE {{ progettoMuseo:RamoDiMandorloFiorito ?rel ?info . FILTER(?rel != progettoMuseo:posizioneOpera) }} LIMIT 10 ECCO L'ONTOLOGIA (TURTLE) PER CONTESTO: {ontology_turtle} FINE ONTOLOGIA. """ logger.debug("[create_system_prompt_for_sparql] Prompt generato con esempi originali e nuove regole rigorose sulla posizione.") return prompt def classify_and_translate(question_text: str, model_answer_text: str): """ Classifica la lingua della domanda e della risposta, quindi traduce la risposta se la lingua è diversa da quella della domanda. L'idea è di restituire una risposta nella stessa lingua dell'utente. Parametri: - question_text: Testo della domanda dell'utente. - model_answer_text: Risposta del modello (in qualsiasi lingua). Restituisce: - La risposta tradotta nella lingua della domanda o la risposta originale se entrambe le lingue coincidono. NB: Qui l'oggetto 'lang_detect_client' (per rilevamento lingua) è già stato inizializzato all'avvio dell'app. Mentre il 'translator_client' viene creato 'al volo' poiché la direzione di traduzione dipende dalle due lingue effettive. """ # Rileva la lingua della domanda try: question_lang_result = lang_detect_client.text_classification(text=question_text) question_lang = question_lang_result[0]['label'] explanation_dict['language_detection'] = f"Domanda in: {question_lang}." logger.info(f"[LangDetect] Lingua della domanda: {question_lang}") except Exception as e: logger.error(f"Errore nel rilevamento della lingua della domanda: {e}") question_lang = "en" # Fallback se non riusciamo a rilevare la lingua explanation_dict['language_detection'] = "Lingua domanda non rilevata, impostata a 'en'." # Rileva la lingua della risposta try: answer_lang_result = lang_detect_client.text_classification(text=model_answer_text) answer_lang = answer_lang_result[0]['label'] explanation_dict['language_detection'] += f" Risposta in: {answer_lang}." logger.info(f"[LangDetect] Lingua della risposta: {answer_lang}") except Exception as e: logger.error(f"Errore nel rilevamento della lingua della risposta: {e}") answer_lang = "it" # Fallback se non riusciamo a rilevare la lingua explanation_dict['language_detection'] += " Lingua risposta non rilevata, impostata a 'it'." # Se domanda e risposta sono nella stessa lingua, non traduciamo if question_lang == answer_lang: logger.info("[Translate] Nessuna traduzione necessaria: stessa lingua.") explanation_dict['translation'] = "Nessuna traduzione necessaria." return model_answer_text # Altrimenti, costruiamo "al volo" il modello di traduzione appropriato # (es: "Helsinki-NLP/opus-mt-en-it", "Helsinki-NLP/opus-mt-fr-en", ecc.) translator_model = f"{TRANSLATOR_MODEL_PREFIX}-{answer_lang}-{question_lang}" translator_client = InferenceClient( token=HF_API_KEY, model=translator_model ) explanation_dict['translation'] = f"Usato modello: {translator_model}." # Traduzione della risposta try: translation_result = translator_client.translation(text=model_answer_text) translated_answer = translation_result["translation_text"] explanation_dict['translation'] += " Traduzione riuscita." except Exception as e: logger.error(f"Errore nella traduzione {answer_lang} -> {question_lang}: {e}") explanation_dict['translation'] += " Traduzione fallita, risposta originale usata." # Se fallisce, restituiamo la risposta originale come fallback explanation_dict['translation'] = "Errore inizializzazione traduttore." translated_answer = model_answer_text return translated_answer def create_system_prompt_for_guide() -> str: """ Genera un testo di prompt che istruisce il modello a rispondere come "guida museale virtuale", in modo breve (~50 parole), riassumendo i risultati SPARQL (se presenti) o fornendo comunque una risposta in base alle conoscenze pregresse. """ prompt = ( "SEI UNA GUIDA MUSEALE VIRTUALE. " "RISPONDI IN MODO DISCORSIVO E NATURALE (circa 100 parole), SENZA SALUTI O INTRODUZIONI PROLISSE. " "SE HAI RISULTATI SPARQL, USALI; SE NON CI SONO, RISPONDI BASANDOTI SULLE TUE CONOSCENZE. " "QUALORA LA DOMANDA CONTENGA ESPRESSIONI COME 'fammi vedere', 'portami', 'mi fai vedere', O SIMILI, " "TRADUCI LA RICHIESTA IN UN INVITO ALL'AZIONE, AD ESEMPIO 'Adesso ti accompagno all'opera', " "SENZA RIPETERE DETTAGLI SPAZIALI O TECNICI (ES. COORDINATE, DISTANZE, POSITIONI FISICHE). " "PER ALTRE DOMANDE, RISPONDI IN MODO DESCRITTIVO MA SENZA INCLUDERE INFORMAZIONI TECNICHE SULLA POSIZIONE." ) logger.debug("[create_system_prompt_for_guide] Prompt per la risposta guida museale generato.") return prompt def correct_sparql_syntax_advanced(query: str) -> str: """ Applica correzioni sintattiche (euristiche) su una query SPARQL eventualmente mal formattata, generata dal modello. Passi: 1. Rimuove newline. 2. Verifica l'esistenza di 'PREFIX progettoMuseo:' e lo aggiunge se mancante. 3. Inserisce spazi dopo SELECT, WHERE (se mancanti). 4. Se c'è 'progettoMuseo:autoreOpera?autore' lo trasforma in 'progettoMuseo:autoreOpera ?autore'. 5. Rimuove spazi multipli. 6. Aggiunge '.' prima di '}' se manca. 7. Aggiunge la clausola WHERE se non presente. Parametri: - query: stringa con la query SPARQL potenzialmente mal formattata. Ritorna: - La query SPARQL corretta se possibile, in singola riga. """ original_query = query logger.debug(f"[correct_sparql_syntax_advanced] Query originaria:\n{original_query}") # 1) Rimuoviamo newline e normalizziamo a una singola riga query = query.replace('\n', ' ').replace('\r', ' ') # 2) Se manca il PREFIX museo, lo aggiungiamo in testa if 'PREFIX progettoMuseo:' not in query: logger.debug("[correct_sparql_syntax_advanced] Aggiungo PREFIX progettoMuseo.") query = ( "PREFIX progettoMuseo: " + query ) # 3) Spazio dopo SELECT se manca (SELECT?autore => SELECT ?autore) query = re.sub(r'(SELECT)(\?|\*)', r'\1 \2', query, flags=re.IGNORECASE) # 4) Spazio dopo WHERE se manca (WHERE{ => WHERE {) query = re.sub(r'(WHERE)\{', r'\1 {', query, flags=re.IGNORECASE) # 5) Correggiamo le incollature: progettoMuseo:autoreOpera?autore => progettoMuseo:autoreOpera ?autore query = re.sub(r'(progettoMuseo:\w+)\?(\w+)', r'\1 ?\2', query) # 6) Rimuoviamo spazi multipli query = re.sub(r'\s+', ' ', query).strip() # 7) Aggiungiamo '.' prima di '}' se manca query = re.sub(r'(\?\w+)\s*\}', r'\1 . }', query) # 8) Se manca la clausola WHERE, la aggiungiamo if 'WHERE' not in query.upper(): query = re.sub(r'(SELECT\s+[^\{]+)\{', r'\1 WHERE {', query, flags=re.IGNORECASE) # 9) Pulizia spazi superflui query = re.sub(r'\s+', ' ', query).strip() logger.debug(f"[correct_sparql_syntax_advanced] Query dopo correzioni:\n{query}") return query def is_sparql_query_valid(query: str) -> bool: """ Verifica la validità sintattica di una query SPARQL usando rdflib. Ritorna True se la query è sintatticamente corretta, False altrimenti. """ logger.debug(f"[is_sparql_query_valid] Validazione SPARQL: {query}") try: parseQuery(query) logger.debug("[is_sparql_query_valid] Query SPARQL sintatticamente corretta.") return True except Exception as ex: logger.warning(f"[is_sparql_query_valid] Query non valida: {ex}") return False # --------------------------------------------------------------------------- # ENDPOINT UNICO: /assistant # --------------------------------------------------------------------------- @app.post("/assistant") def assistant_endpoint(req: AssistantRequest): explanation_dict = {} """ Endpoint che gestisce l'intera pipeline: 1) Genera una query SPARQL dal messaggio dell'utente (prompt dedicato). 2) Verifica la validità della query e, se valida, la esegue sull'ontologia RDF. 3) Crea un "prompt da guida museale" e genera una risposta finale breve (max ~50 parole). 4) Eventualmente, traduce la risposta nella lingua dell'utente. Parametri: - req (AssistantRequest): un oggetto contenente: - message (str): la domanda dell'utente - max_tokens (int, opzionale): numero massimo di token per la generazione - temperature (float, opzionale): temperatura per la generazione Ritorna: - Un JSON con: { "query": , "response": } """ logger.info("Ricevuta chiamata POST su /assistant") # Estraggo i campi dal body della richiesta user_message = req.message max_tokens = req.max_tokens temperature = req.temperature logger.debug(f"Parametri utente: message='{user_message}', max_tokens={max_tokens}, temperature={temperature}") # ------------------------------- # CLASSIFICAZIONE DEL TESTO RICEVUTO # ------------------------------- if classifier_model is not None: try: # Prepara l'input per il modello di classificazione inputs = classifier_tokenizer(user_message, return_tensors="pt", truncation=True, padding=True) inputs = {k: v.to(device) for k, v in inputs.items()} # Disattiva il calcolo del gradiente per velocizzare l'inferenza with torch.no_grad(): outputs = classifier_model(**inputs) logits = outputs.logits pred = torch.argmax(logits, dim=1).item() # Mappa l'etichetta numerica a una stringa (modifica secondo la tua logica) label_mapping = {0: "NON PERTINENTE", 1: "PERTINENTE"} classification_result = label_mapping.get(pred, f"Etichetta {pred}") logger.info(f"[Classificazione] La domanda classificata come: {classification_result}") explanation_dict['classification'] = f"Risultato classificazione: {classification_result}" except Exception as e: logger.error(f"Errore durante la classificazione della domanda: {e}") explanation_dict['classification'] = f"Errore classificazione: {e}" else: logger.warning("Modello di classificazione non disponibile.") explanation_dict['classification'] = "Modello di classificazione non disponibile." # ----------------------------------------------------------------------- # STEP 1: Generazione della query SPARQL # ----------------------------------------------------------------------- try: # Serializziamo l'ontologia in XML per fornirla al prompt (anche se si chiama 'turtle' va bene così). ontology_turtle = ontology_graph.serialize(format="xml") logger.debug("Ontologia serializzata con successo (XML).") explanation_dict['ontology'] = "Ontologia serializzata." except Exception as e: logger.warning(f"Impossibile serializzare l'ontologia in formato XML: {e}") explanation_dict['ontology'] = f"Errore serializzazione: {e}." ontology_turtle = "" # Creiamo il prompt di sistema per la generazione SPARQL system_prompt_sparql = create_system_prompt_for_sparql(ontology_turtle) explanation_dict['sparql_prompt'] = "Prompt SPARQL creato." # Chiamata al modello per generare la query SPARQL try: logger.debug("[assistant_endpoint] Chiamata HF per generare la query SPARQL...") explanation_dict['sparql_generation'] = "Generazione query SPARQL iniziata." gen_sparql_output = hf_generation_client.chat.completions.create( messages=[ {"role": "system", "content": system_prompt_sparql}, {"role": "user", "content": user_message} ], max_tokens=512, # max_tokens per la generazione della query temperature=0.2 # temperatura bassa per avere risposte più "deterministiche" ) possible_query = gen_sparql_output["choices"][0]["message"]["content"].strip() logger.info(f"[assistant_endpoint] Query generata dal modello: {possible_query}") explanation_dict['sparql_generation'] += f" Query: {possible_query}." except Exception as ex: logger.error(f"Errore nella generazione della query SPARQL: {ex}") explanation_dict['sparql_generation'] = f"Errore generazione SPARQL: {ex}." # Se fallisce la generazione, consideriamo la query come "NO_SPARQL" possible_query = "NO_SPARQL" # Verifichiamo se la query è "NO_SPARQL" if possible_query.upper().startswith("NO_SPARQL"): generated_query = None logger.debug("[assistant_endpoint] Modello indica 'NO_SPARQL', quindi nessuna query generata.") else: # Applichiamo la correzione avanzata advanced_corrected = correct_sparql_syntax_advanced(possible_query) explanation_dict['sparql_correction'] = "Sintassi SPARQL corretta." # Verifichiamo la validità della query if is_sparql_query_valid(advanced_corrected): generated_query = advanced_corrected logger.debug(f"[assistant_endpoint] Query SPARQL valida dopo correzione avanzata: {generated_query}") explanation_dict['sparql_validation'] = "Query SPARQL valida." else: logger.debug("[assistant_endpoint] Query SPARQL non valida. Verrà ignorata.") explanation_dict['sparql_validation'] = "Query SPARQL non valida." generated_query = None # ----------------------------------------------------------------------- # STEP 2: Esecuzione della query, se disponibile # ----------------------------------------------------------------------- results = [] if generated_query: logger.debug(f"[assistant_endpoint] Esecuzione della query SPARQL:\n{generated_query}") explanation_dict['sparql_execution'] = "Esecuzione query SPARQL." try: query_result = ontology_graph.query(generated_query) results = list(query_result) logger.info(f"[assistant_endpoint] Query eseguita con successo. Numero risultati = {len(results)}") explanation_dict['sparql_execution'] += f" Risultati: {len(results)}." except Exception as ex: logger.error(f"[assistant_endpoint] Errore nell'esecuzione della query: {ex}") explanation_dict['sparql_execution'] = f"Errore esecuzione: {ex}." results = [] else: explanation_dict['sparql_execution'] = "Nessuna query eseguita." # ----------------------------------------------------------------------- # STEP 3: Generazione della risposta finale stile "guida museale" # ----------------------------------------------------------------------- system_prompt_guide = create_system_prompt_for_guide() explanation_dict['guide_prompt'] = "Prompt guida museale creato." if generated_query and results: # Caso: query generata + risultati SPARQL # Convertiamo i risultati in una stringa più leggibile results_str = "\n".join( f"{idx+1}) " + ", ".join(f"{var}={row[var]}" for var in row.labels) for idx, row in enumerate(results) ) second_prompt = ( f"{system_prompt_guide}\n\n" f"Domanda utente: {user_message}\n" f"Query generata: {generated_query}\n" f"Risultati:\n{results_str}\n" "Rispondi in modo breve (max ~50 parole)." ) logger.debug("[assistant_endpoint] Prompt di risposta con risultati SPARQL.") explanation_dict['guide_prompt'] += " Inclusi risultati SPARQL." elif generated_query and not results: # Caso: query valida ma 0 risultati second_prompt = ( f"{system_prompt_guide}\n\n" f"Domanda utente: {user_message}\n" f"Query generata: {generated_query}\n" "Nessun risultato dalla query. Prova comunque a rispondere con le tue conoscenze." ) logger.debug("[assistant_endpoint] Prompt di risposta: query valida ma senza risultati.") explanation_dict['guide_prompt'] += " Query valida senza risultati." else: # Caso: nessuna query generata second_prompt = ( f"{system_prompt_guide}\n\n" f"Domanda utente: {user_message}\n" "Nessuna query SPARQL generata. Rispondi come puoi, riarrangiando le tue conoscenze." ) logger.debug("[assistant_endpoint] Prompt di risposta: nessuna query generata.") explanation_dict['guide_prompt'] += " Nessuna query generata." # Chiamata finale al modello per la risposta "guida museale" try: logger.debug("[assistant_endpoint] Chiamata HF per generare la risposta finale...") explanation_dict['response_generation'] = "Generazione risposta finale iniziata." final_output = hf_generation_client.chat.completions.create( messages=[ {"role": "system", "content": second_prompt}, {"role": "user", "content": "Fornisci la risposta finale."} ], max_tokens=1024, temperature=0.8 ) final_answer = final_output["choices"][0]["message"]["content"].strip() logger.info(f"[assistant_endpoint] Risposta finale generata: {final_answer}") explanation_dict['response_generation'] += " Risposta generata." except Exception as ex: logger.error(f"Errore nella generazione della risposta finale: {ex}") explanation_dict['response_generation'] = f"Errore generazione risposta finale: {ex}." raise HTTPException(status_code=500, detail="Errore nella generazione della risposta in linguaggio naturale.") # ----------------------------------------------------------------------- # STEP 4: Traduzione (se necessario) # ----------------------------------------------------------------------- final_ans = classify_and_translate(user_message, final_answer) final_ans = final_ans.replace('\\"', "").replace('\"', "") explanation_dict['translation_completion'] = "Traduzione completata." # ----------------------------------------------------------------------- # Restituzione in formato JSON # ----------------------------------------------------------------------- logger.debug("[assistant_endpoint] Fine elaborazione, restituzione risposta JSON.") return { "query": generated_query, "response": final_ans, "explanation": explanation_dict } # --------------------------------------------------------------------------- # ENDPOINT DI TEST / HOME # --------------------------------------------------------------------------- @app.get("/") def home(): """ Endpoint di test per verificare se l'applicazione è in esecuzione. """ logger.debug("Chiamata GET su '/' - home.") return { "message": "Endpoint attivo. Esempio di backend per generare query SPARQL e risposte guida museale." } # --------------------------------------------------------------------------- # ENDPOINT QUERY STANZE /QUERY_STANZE # --------------------------------------------------------------------------- @app.get("/query_stanze") def query_stanze_endpoint(): """ Endpoint per restituire le stanze con le opere esposte e relativi punti. La query restituisce, per ogni triple, la stanza, l'opera e il valore della proprietà progettoMuseo:posizioneOpera (facoltativo). Successivamente, in Python, si raggruppa per stanza in un dizionario. Per ogni stanza, viene creata una lista di dizionari, dove ogni dizionario rappresenta un'opera e contiene: - "nome": il localName dell'opera (la parte dopo il simbolo "#") - "punto": il valore della proprietà progettoMuseo:posizioneOpera (se presente, altrimenti una stringa vuota) Query usata (in una riga): PREFIX progettoMuseo: SELECT ?stanza ?opera ?p WHERE { ?opera progettoMuseo:èEsposto ?stanza. OPTIONAL { ?opera progettoMuseo:posizioneOpera ?p. } } """ # Definiamo la query in una singola riga query_str = ( "PREFIX progettoMuseo: " "SELECT ?stanza ?opera ?p WHERE { ?opera progettoMuseo:èEsposto ?stanza. " "OPTIONAL { ?opera progettoMuseo:posizioneOpera ?p. } }" ) # Applichiamo eventuali correzioni sintattiche corrected_query = correct_sparql_syntax_advanced(query_str) logger.debug(f"[query_stanze_endpoint] Query corretta:\n{corrected_query}") # Verifichiamo la validità della query utilizzando il parser di rdflib if not is_sparql_query_valid(corrected_query): logger.error("[query_stanze_endpoint] Query SPARQL non valida.") raise HTTPException(status_code=400, detail="Query SPARQL non valida.") try: query_result = ontology_graph.query(corrected_query) # Costruiamo un dizionario: per ogni stanza (localName), una lista di dizionari # con chiavi "nome" (localName dell'opera) e "punto" (valore di progettoMuseo:posizioneOpera) dict_stanze = {} for row in query_result: # Estraiamo il localName della stanza stanza_uri = str(row["stanza"]) stanza_local = stanza_uri.split("#")[-1].strip() if "#" in stanza_uri else stanza_uri # Estraiamo il localName dell'opera opera_uri = str(row["opera"]) opera_local = opera_uri.split("#")[-1].strip() if "#" in opera_uri else opera_uri # Estraiamo il punto (se presente) punto = str(row["p"]) if row["p"] is not None else "" # Costruiamo il dizionario per l'opera opera_dict = {"nome": opera_local, "punto": punto} # Raggruppiamo per stanza if stanza_local not in dict_stanze: dict_stanze[stanza_local] = [] dict_stanze[stanza_local].append(opera_dict) logger.info(f"[query_stanze_endpoint] Trovate {len(dict_stanze)} stanze.") except Exception as ex: logger.error(f"[query_stanze_endpoint] Errore nell'esecuzione della query: {ex}") raise HTTPException(status_code=500, detail="Errore nell'esecuzione della query SPARQL.") return { "stanze": dict_stanze } # --------------------------------------------------------------------------- # MAIN # --------------------------------------------------------------------------- if __name__ == "__main__": """ Avvio dell'applicazione FastAPI sulla porta 8000, utile se eseguito come script principale. """ logger.info("Avvio dell'applicazione FastAPI.") uvicorn.run(app, host="0.0.0.0", port=8000)