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 # --------------------------------------------------------------------------- # CONFIGURAZIONE LOGGING # --------------------------------------------------------------------------- logging.basicConfig( level=logging.DEBUG, # DEBUG per un log più dettagliato format="%(asctime)s - %(levelname)s - %(message)s", handlers=[logging.FileHandler("app.log"), logging.StreamHandler()] ) logger = logging.getLogger(__name__) # Categorie di zero-shot classification CANDIDATE_LABELS = ["domanda_museo", "small_talk", "fuori_contesto"] HF_API_KEY = os.getenv("HF_API_KEY") HF_MODEL = "meta-llama/Llama-3.3-70B-Instruct" # modello per query SPARQL e risposte ZERO_SHOT_MODEL = "facebook/bart-large-mnli" # modello per zero-shot classification if not HF_API_KEY: logger.error("HF_API_KEY non impostata.") raise EnvironmentError("HF_API_KEY non impostata.") # --------------------------------------------------------------------------- # INIZIALIZZIAMO IL CLIENT PER ZERO-SHOT # --------------------------------------------------------------------------- try: logger.info("Inizializzazione del client per Zero-Shot Classification.") client_cls = InferenceClient( token=HF_API_KEY, model=ZERO_SHOT_MODEL ) logger.info("Client zero-shot creato con successo.") except Exception as ex: logger.error(f"Errore nell'inizializzazione del client zero-shot: {ex}") raise ex # --------------------------------------------------------------------------- # FUNZIONE DI CLASSIFICAZIONE # --------------------------------------------------------------------------- def classify_message_inference_api(text: str) -> str: """ Usa client_cls.zero_shot_classification(...) per classificare 'domanda_museo', 'small_talk' o 'fuori_contesto'. Restituisce la label top. """ try: hypothesis_template = "Questa domanda è inerente all'arte o all'ontologia di un museo ({}), oppure no?" # multi_label=False => elegge UNA sola label top results = client_cls.zero_shot_classification( text=text, candidate_labels=CANDIDATE_LABELS, multi_label=False, hypothesis_template=hypothesis_template ) # results è una lista di ZeroShotClassificationOutputElement # es: [ZeroShotClassificationOutputElement(label='domanda_museo', score=0.85), ...] top_label = results[0].label top_score = results[0].score logger.info(f"[ZeroShot] top_label={top_label}, score={top_score}") return top_label except Exception as e: logger.error(f"Errore nella zero-shot classification: {e}") return "fuori_contesto" # fallback in caso di errore # Inizializziamo la nostra ontologia BASE_DIR = os.path.dirname(os.path.abspath(__file__)) RDF_FILE = os.path.join(BASE_DIR, "Ontologia_corretto.rdf") client_cls = InferenceClient(token=HF_API_KEY) ontology_graph = rdflib.Graph() try: # L'ontologia è in formato RDF/XML logger.info(f"Caricamento ontologia da file: {RDF_FILE}") 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() # Modello di request class AssistantRequest(BaseModel): message: str max_tokens: int = 512 temperature: float = 0.5 # --------------------------------------------------------------------------- # FUNZIONI DI SUPPORTO (Prompts, validazione SPARQL, correzione) # --------------------------------------------------------------------------- def create_system_prompt_for_sparql(ontology_turtle: str) -> str: """ PRIMO PROMPT DI SISTEMA molto prolisso e stringente sulle regole SPARQL, con i vari esempi (1-10) inclusi. """ 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 ECCO L'ONTOLOGIA (TURTLE) PER CONTESTO: {ontology_turtle} FINE ONTOLOGIA. """ logger.debug("[create_system_prompt_for_sparql] Prompt generato con ESEMPI e regole SPARQL.") return prompt def create_system_prompt_for_guide() -> str: """ SECONDO PROMPT DI SISTEMA: - Risponde in stile "guida museale" in modo breve (max ~50 parole). - Se c'è una query e risultati, descrive brevemente. - Se non c'è query o non ci sono risultati, prova comunque a dare una risposta. """ prompt = ( "SEI UNA GUIDA MUSEALE VIRTUALE. " "RISPONDI IN MODO BREVE (~50 PAROLE), SENZA SALUTI O INTRODUZIONI PROLISSE. " "SE HAI RISULTATI SPARQL, USALI. " "SE NON HAI RISULTATI O NON HAI UNA QUERY, RISPONDI COMUNQUE CERCANDO DI RIARRANGIARE LE TUE CONOSCENZE." ) logger.debug("[create_system_prompt_for_guide] Prompt per la risposta guida museale generato.") return prompt def correct_sparql_syntax_advanced(query: str) -> str: """ Corregge in maniera più complessa gli errori sintattici comuni generati dal modello nelle query SPARQL, tramite euristiche: - Spazi dopo SELECT, WHERE - Rimozione di '?autore' attaccato a 'progettoMuseo:autoreOpera?autore' - Aggiunta di PREFIX se assente - Rimozione newline (una riga) - Aggiunta di '.' se manca a fine tripla - Pulizia di spazi doppi """ original_query = query logger.debug(f"[correct_sparql_syntax_advanced] Query originaria:\n{original_query}") # 1) Rimuoviamo newline e forziamo un'unica riga query = query.replace('\n', ' ').replace('\r', ' ') # 2) Se manca il PREFIX, lo aggiungiamo in testa (solo se notiamo che non c'è "PREFIX progettoMuseo:") 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 query = re.sub(r'(SELECT)(\?|\*)', r'\1 \2', query, flags=re.IGNORECASE) # 4) Spazio dopo WHERE se manca query = re.sub(r'(WHERE)\{', r'\1 {', query, flags=re.IGNORECASE) # 5) Correggiamo i punti interrogativi attaccati alle proprietà: # "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 '.' a fine tripla prima del '}' se manca query = re.sub(r'(\?\w+)\s*\}', r'\1 . }', query) # 8) Se manca la clausola WHERE, proviamo ad aggiungerla if 'WHERE' not in query.upper(): query = re.sub(r'(SELECT\s+[^\{]+)\{', r'\1 WHERE {', query, flags=re.IGNORECASE) # 9) Pulizia finale di spazi 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 sintassi SPARQL tramite rdflib.""" 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 # --------------------------------------------------------------------------- @app.post("/assistant") def assistant_endpoint(req: AssistantRequest): """ Endpoint UNICO con due step interni: 1) Genera la query SPARQL (prompt prolisso). 2) Esegue la query (se valida) e fornisce una risposta breve stile "guida museale", anche se i risultati sono vuoti o la query non esiste. """ logger.info("Ricevuta chiamata POST su /assistant") user_message = req.message max_tokens = req.max_tokens temperature = req.temperature label = classify_message_inference_api(user_message) logger.info(label) logger.debug(f"Parametri utente: message='{user_message}', max_tokens={max_tokens}, temperature={temperature}") # STEP 1: Generazione SPARQL try: logger.debug("Serializzazione dell'ontologia in formato Turtle per contesto nel prompt.") ontology_turtle = ontology_graph.serialize(format="xml") logger.debug("Ontologia serializzata con successo (XML).") except Exception as e: logger.warning(f"Impossibile serializzare l'ontologia in Turtle: {e}") ontology_turtle = "" system_prompt_sparql = create_system_prompt_for_sparql(ontology_turtle) # Inizializziamo client Hugging Face try: logger.debug(f"Inizializzazione InferenceClient con modello='{HF_MODEL}'.") hf_client = InferenceClient(model=HF_MODEL, token=HF_API_KEY) except Exception as ex: logger.error(f"Errore inizializzazione HF client: {ex}") raise HTTPException(status_code=500, detail="Impossibile inizializzare il modello Hugging Face.") # Chiediamo al modello la query SPARQL (fase interna 1) try: logger.debug("[assistant_endpoint] Chiamata HF per generare la query SPARQL...") gen_sparql_output = hf_client.chat.completions.create( messages=[ {"role": "system", "content": system_prompt_sparql}, {"role": "user", "content": user_message} ], max_tokens=512, temperature=0.3 ) possible_query = gen_sparql_output["choices"][0]["message"]["content"].strip() logger.info(f"[assistant_endpoint] Query generata dal modello: {possible_query}") except Exception as ex: logger.error(f"Errore nella generazione della query SPARQL: {ex}") # Se fallisce la generazione, consideriamo la query come "NO_SPARQL" possible_query = "NO_SPARQL" # Verifica se la query è NO_SPARQL if possible_query.upper().startswith("NO_SPARQL"): generated_query = None logger.debug("[assistant_endpoint] Modello indica 'NO_SPARQL', nessuna query generata.") else: # Correggiamo in modo avanzato advanced_corrected = correct_sparql_syntax_advanced(possible_query) # Dopo la correzione, verifichiamo se è valida if is_sparql_query_valid(advanced_corrected): generated_query = advanced_corrected logger.debug(f"[assistant_endpoint] Query SPARQL valida dopo correzione avanzata: {generated_query}") else: logger.debug("[assistant_endpoint] Query SPARQL non valida dopo correzione avanzata. La ignoriamo.") generated_query = None # STEP 2: Esecuzione query (se presente) e risposta guida results = [] if generated_query: logger.debug(f"[assistant_endpoint] Esecuzione della query SPARQL:\n{generated_query}") 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)}") except Exception as ex: logger.error(f"[assistant_endpoint] Errore nell'esecuzione della query: {ex}") results = [] # Creiamo il prompt di sistema "guida museale" system_prompt_guide = create_system_prompt_for_guide() if generated_query and results: # Abbiamo query + risultati # 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.") elif generated_query and not results: # 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 nessun risultato.") else: # 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.") # Ultima chiamata al modello per la risposta finale try: logger.debug("[assistant_endpoint] Chiamata HF per la risposta guida museale...") final_output = hf_client.chat.completions.create( messages=[ {"role": "system", "content": second_prompt}, {"role": "user", "content": "Fornisci la risposta finale."} ], max_tokens=512, temperature=0.7 ) final_answer = final_output["choices"][0]["message"]["content"].strip() logger.info(f"[assistant_endpoint] Risposta finale generata: {final_answer}") except Exception as ex: logger.error(f"Errore nella generazione della risposta finale: {ex}") raise HTTPException(status_code=500, detail="Errore nella generazione della risposta in linguaggio naturale.") # Risposta JSON logger.debug("[assistant_endpoint] Fine elaborazione. Restituzione risposta.") return { "query": generated_query, "response": final_answer } # --------------------------------------------------------------------------- # ENDPOINT DI TEST # --------------------------------------------------------------------------- @app.get("/") def home(): logger.debug("Chiamata GET su '/' - home.") return { "message": "Endpoint con ESEMPI di query SPARQL + correzione avanzata + risposta guida museale." } # --------------------------------------------------------------------------- # MAIN # --------------------------------------------------------------------------- if __name__ == "__main__": logger.info("Avvio dell'applicazione FastAPI sulla porta 8000.")