Update app.py

app.py

@@ -9,15 +9,21 @@ class MariannaBot:
     def __init__(self):
         self.data_path_main = "dati_per_database_riassunti.pkl"
         self.data_path_legends = "legends.pkl"
+        self.data_path_exp = "secondDB.pkl"
 
         print("Inizializzazione di MariannaBot (senza DB)...")  # Debug
 
         self.database = self.load_data_from_pickle(self.data_path_main)
         self.database_legends = self.load_data_from_pickle(self.data_path_legends)
-        self.database = self.database + self.database_legends
+        self.database_expansion = self.load_data_from_pickle(self.data_path_exp)
+
+        self.database = self.database + self.database_legends + self.database_expansion
+        self.further_dataset = self.load_further_info_as_dataset(self.database)
+        self.further_dataset = self.further_dataset + self.database
 
         self.db_keys = [el[0] for el in self.database] if isinstance(self.database, list) else []
         self.db_keys_legends = [el[0] for el in self.database] if isinstance(self.database, list) else []
+        self.db_keys_further = [el[0] for el in self.further_dataset] if isinstance(self.further_dataset, list) else []
         # print("Chiavi principali caricate:", len(self.db_keys))  # Debug
         # print("Chiavi leggende caricate:", len(self.db_keys_legends))  # Debug
 
@@ -27,9 +33,20 @@ class MariannaBot:
 
         self.reset_state()
 
+    def load_further_info_as_dataset(self,dataset):
+        nuova_lista = []
+        for chiave_principale, info in dataset:
+          nuovo_dizionario = {'intro': info['intro']}
+          if 'further_info' in info:
+            for chiave_secondaria in info['further_info']:
+              nuova_lista.append((f"{chiave_secondaria} ({chiave_principale})", {'intro': info['further_info'][chiave_secondaria]}))
+          else:
+            nuova_lista.append((chiave_principale, nuovo_dizionario))
+        return nuova_lista
+
     def load_queries_dataset(self):
         """Loads queries dataset"""
-        return {"si, certo, certamente, ok, assolutamente si, sicuro, sisi":"si","no, non ho domande, non mi interessa, niente, nulla":"no","non so, scegli tu, fai tu, casuale, lascio a te, decidi tu, pensaci tu, sorprendimi":"non so","stronzo, vaffanculo, ti odio, pezzo di merda, cazzo":"parolacce"}
+        return {"si, certo, certamente, ok, assolutamente si, sicuro, sisi, continua, prosegui":"si","no, non ho domande, non mi interessa, niente, nulla":"no","non so, scegli tu, fai tu, casuale, lascio a te, decidi tu, pensaci tu, sorprendimi":"non so","stronzo, vaffanculo, ti odio, pezzo di merda, cazzo":"parolacce"}
 
     def load_data_from_pickle(self, file_path):
         """Loads data from a pickle file."""
@@ -60,6 +77,7 @@ class MariannaBot:
             self.db_keys_embeddings = self.encoder.encode(self.db_keys, convert_to_tensor=True)
             self.db_keys_legends_embeddings = self.encoder.encode(self.db_keys_legends, convert_to_tensor=True)
             self.first_query_emb = self.encoder.encode(self.query_dic_keys, convert_to_tensor=True)
+            self.further_embeddings = self.encoder.encode(self.db_keys_further, convert_to_tensor=True)
 
             print(f"Encoder initialized with {len(self.db_keys)} keys.")
             return True
@@ -121,14 +139,46 @@ class MariannaBot:
         # per recuperare il contenuto effettivo.
         return self.database_legends.get(key) if isinstance(self.database_legends, dict) else None
 
-    def get_value(self, key):
+    def get_value(self, key,state):
         """Retrieve a value from the loaded main data by key."""
-        for k, v in self.database:
+        if state=="A":
+          for k, v in self.database:
+            if k == key:
+                return v
+        else:
+          for k, v in self.further_dataset:
             if k == key:
                 return v
         return None
 
 
+    def deeper_handle_query(self,message,query_embedding,CS_old):
+        print('Ricerca in profondità')
+        try:
+          semantic_hits = util.semantic_search(query_embedding, self.further_embeddings, top_k=3)
+          semantic_hits = semantic_hits[0]
+
+          cross_inp = [(message, self.db_keys_further[hit['corpus_id']]) for hit in semantic_hits]
+          cross_scores = self.cross_encoder.predict(cross_inp)
+          cross_scores = cross_scores + CS_old
+
+          reranked_hits = sorted(
+              [{'corpus_id': hit['corpus_id'], 'cross-score': score}
+                for hit, score in zip(semantic_hits, cross_scores)],
+              key=lambda x: x['cross-score'], reverse=True
+          )
+          for h in reranked_hits: 
+            print(self.db_keys_further[h['corpus_id']],h['cross-score'])
+          best_hit = reranked_hits[0]
+          best_title = self.db_keys_further[best_hit['corpus_id']]
+          best_score = best_hit['cross-score']
+          return reranked_hits
+        except Exception as e:
+            print(e)
+
+
+
+
 
     def handle_query(self, message):
         """Handle user queries by searching the database"""
@@ -148,65 +198,74 @@ class MariannaBot:
                  for hit, score in zip(semantic_hits, cross_scores)],
                 key=lambda x: x['cross-score'], reverse=True
             )
-
+            
+            for h in reranked_hits: 
+              print(self.db_keys[h['corpus_id']],h['cross-score'])
+            chiavi = self.db_keys
             best_hit = reranked_hits[0]
             best_title = self.db_keys[best_hit['corpus_id']]
             best_score = best_hit['cross-score']
-            # print(best_title, best_score)
-
+            state="A"
+
+            if best_score < 0.75:
+              reranked_hits = self.deeper_handle_query(message,query_embedding,cross_scores)
+              best_hit = reranked_hits[0]
+              best_title = self.db_keys_further[best_hit['corpus_id']]
+              best_score = best_hit['cross-score']
+              state="B"
+            
             # Main treshold = 0.75
             similarity_threshold = 0.75
 
             # treshold granularity
             if best_score < similarity_threshold:
                 # low confidence (< 0.35)
-                if best_score < 0.55:
-                    return random.choice(["Mi dispiace, non ho informazioni su questo argomento. Puoi chiedermi di altro sulla città di Napoli.",
+              if best_score < 0.55:
+                return random.choice(["Mi dispiace, non ho informazioni su questo argomento. Puoi chiedermi di altro sulla città di Napoli.",
                                           "Purtroppo non riesco a rammentare questo argomento, la mia memoria non è più quella di un tempo. Chiedimi qualcos'altro su Napoli e le sue bellezze!",
-                                          "Mi dispiace tantissimo, ma non riesco a ricordare altro. Vuoi chiedermi altro sulla città di Napoli?"])
+                                          "Mi dispiace tantissimo, ma non riesco a ricordare. Vuoi chiedermi altro sulla città di Napoli?"])
 
                 # medium confidence(0.55 - 0.75)
+              else:
+                alternative_hits = [self.db_keys[hit['corpus_id']] for hit in reranked_hits[:2]]
+                suggestions = " o ".join(alternative_hits)
+                value = self.get_value(best_title,state)
+                if value:
+                    partial_info = value.get('short_intro', value.get('intro', '').split('.')[0] + '.')
+                    self.state = "query"
+                    self.is_telling_stories = False
+                    return random.choice([f"Potrei avere alcune informazioni su {best_title}, ma non sono completamente sicura sia ciò che stai cercando. I miei suggerimenti sono {suggestions}. \n\nCosa ti interessa?",
+                                          f"Credo che tu stia parlando de {best_title}, ma per essere sicura di ciò che vuoi sapere, potresti specificare se parli di {suggestions}?",
+                                          f"Per assicurarmi di aver capito bene, vuoi che ti parli di {suggestions}?"])
                 else:
-                    alternative_hits = [self.db_keys[hit['corpus_id']] for hit in reranked_hits[:2]]
-                    suggestions = ", ".join(alternative_hits)
-                    value = self.get_value(best_title)
-                    if value:
-                        partial_info = value.get('short_intro', value.get('intro', '').split('.')[0] + '.')
-                        self.state = "query"
-                        self.is_telling_stories = False
-                        return random.choice([f"Potrei avere alcune informazioni su {best_title}, ma non sono completamente sicura sia ciò che stai cercando. I miei suggerimenti sono {suggestions}. \n\nCosa ti interessa?",
-                                              f"Credo che tu stia parlando de {best_title}, ma per essere sicura di ciò che vuoi sapere, potresti specificare se parli di {suggestions}?",
-                                              f"Per assicurarmi di aver capito bene, vuoi che ti parli di {suggestions}?"])
-                    else:
-                        return f"Ho trovato qualcosa su {best_title}, ma non sono completamente sicura. Vuoi saperne di più?"
+                    return f"Ho trovato qualcosa su {best_title}, ma non sono completamente sicura. Vuoi saperne di più?"
 
-            # high confidence (above the threshold)
+          # high confidence (above the threshold)
             if best_title is not None:
-                value = self.get_value(best_title)
+              value = self.get_value(best_title,state)
 
-                if value:
-                    key = best_title
-                    self.main_k.append(key)
-                    self.state = "follow_up"
-                    self.is_telling_stories = False
-                    response = value.get('intro', '')
-                    if isinstance(value, dict):
-                        self.current_further_info_values = list(value.get('further_info', {}).values())
-                    else:
-                        self.current_further_info_values = []  # Se il valore non è un dizionario
-                    self.current_index = 0
-                    return f"{response}\n\nVuoi sapere altro su {self.main_k[-1]}?"
+              if value:
+                key = best_title
+                self.main_k.append(key)
+                self.state = "follow_up"
+                self.is_telling_stories = False
+                response = value.get('intro', '')
+                if isinstance(value, dict):
+                    self.current_further_info_values = list(value.get('further_info', {}).values())
                 else:
-                    return random.choice(["Mi dispiace, non ho informazioni su questo argomento. Puoi chiedermi di altro sulla città di Napoli.",
-                                          "Purtroppo non riesco a rammentare altro su questo argomento, la mia memoria non è più quella di un tempo. Chiedimi qualcos'altro su Napoli e le sue bellezze!",
-                                          "Mi dispiace tantissimo, ma non riesco a ricordare altro. Vuoi chiedermi altro sulla città di Napoli?"])
-
+                    self.current_further_info_values = []  # Se il valore non è un dizionario
+                self.current_index = 0
+                return f"{response}\n\nVuoi sapere altro su {self.main_k[-1]}?"
+            else:
+                return random.choice(["Mi dispiace, non ho informazioni su questo argomento. Puoi chiedermi di altro sulla città di Napoli.",
+                                      "Purtroppo non riesco a rammentare altro su questo argomento, la mia memoria non è più quella di un tempo. Chiedimi qualcos'altro su Napoli e le sue bellezze!",
+                                      "Mi dispiace tantissimo, ma non riesco a ricordare altro. Vuoi chiedermi altro sulla città di Napoli?"])
         except Exception as e:
             print(e)
             self.state = "initial"
             return random.choice(["Mi dispiace, c'è stato un errore. Puoi riprovare con un'altra domanda? ",
                                   "Scusami, sto facendo confusione. Puoi farmi un'altra domanda?",
-                                  "Mi dispiace, non ho capito. Puoi essere più preciso?"])	
+                                  "Mi dispiace, non ho capito. Puoi essere più preciso?"])
 
 
     def first_query(self, message):
@@ -217,7 +276,6 @@ class MariannaBot:
             # Perform semantic search on the keys
             semantic_hits = util.semantic_search(query_embedding, self.first_query_emb, top_k=4)
             semantic_hits = semantic_hits[0]
-            print(semantic_hits)
             cross_inp = [(message, self.query_dic_keys[hit['corpus_id']]) for hit in semantic_hits]
             cross_scores = self.cross_encoder.predict(cross_inp)
             reranked_hits = sorted(
@@ -228,7 +286,7 @@ class MariannaBot:
             best_hit = reranked_hits[0]
             best_title = self.query_dic[self.query_dic_keys[best_hit['corpus_id']]]
             best_score = best_hit['cross-score']
-            print(best_title, best_score)
+            print(message,best_title, best_score)
             # Main treshold = 0.75
             similarity_threshold = 0.35
 
@@ -252,14 +310,14 @@ class MariannaBot:
             self.state = "initial"
             return random.choice(["Mi dispiace, c'è stato un errore. Puoi riprovare con un'altra domanda? ",
                                   "Scusami, sto facendo confusione. Puoi farmi un'altra domanda?",
-                                  "Mi dispiace, non ho capito. Puoi essere più preciso?"])	
+                                  "Mi dispiace, non ho capito. Puoi essere più preciso?"])
 
 
     def respond(self, message, history):
         if not message:
             return random.choice(["Mi dispiace, c'è stato un errore. Puoi riprovare con un'altra domanda? ",
                                   "Scusami, sto facendo confusione. Puoi farmi un'altra domanda?",
-                                  "Mi dispiace, non ho capito. Puoi essere più preciso?"])	
+                                  "Mi dispiace, non ho capito. Puoi essere più preciso?"])
 
         message = message.lower().strip()