Update app.py

app.py

@@ -108,13 +108,10 @@ class MistralRAGChatbot:
     def __init__(self, vector_db_path: str, annoy_index_path: str):
         self.embeddings, self.texts = self.load_vector_db(vector_db_path)
         self.annoy_index = self.load_annoy_index(annoy_index_path, self.embeddings.shape[1])
-        self.tfidf_matrix, self.tfidf_vectorizer = self.calculate_tfidf(self.texts)
+        # self.tfidf_matrix, self.tfidf_vectorizer = self.calculate_tfidf(self.texts)
         self.bm25 = BM25Okapi([text.split() for text in self.texts])
-        self.word2vec_model = self.train_word2vec(self.texts)
+        # self.word2vec_model = self.train_word2vec(self.texts)
         self.reranking_methods = {
-            # 'reciprocal_rank_fusion': self.reciprocal_rank_fusion,
-            # 'weighted_score_fusion': self.weighted_score_fusion,
-            # 'semantic_similarity': self.semantic_similarity_reranking,
             'advanced_fusion': self.advanced_fusion_retrieval
             }
         logging.info("MistralRAGChatbot initialized successfully.")
@@ -134,17 +131,17 @@ class MistralRAGChatbot:
         logging.info(f"Loaded Annoy index from {annoy_index_path}.")
         return annoy_index
 
-    def calculate_tfidf(self, texts: List[str]) -> Tuple[np.ndarray, TfidfVectorizer]:
-        vectorizer = TfidfVectorizer(stop_words='english')
-        tfidf_matrix = vectorizer.fit_transform(texts)
-        logging.info("TF-IDF matrix calculated.")
-        return tfidf_matrix, vectorizer
+    # def calculate_tfidf(self, texts: List[str]) -> Tuple[np.ndarray, TfidfVectorizer]:
+    #     vectorizer = TfidfVectorizer(stop_words='english')
+    #     tfidf_matrix = vectorizer.fit_transform(texts)
+    #     logging.info("TF-IDF matrix calculated.")
+    #     return tfidf_matrix, vectorizer
 
-    def train_word2vec(self, texts: List[str]) -> Word2Vec:
-        tokenized_texts = [text.split() for text in texts]
-        model = Word2Vec(sentences=tokenized_texts, vector_size=100, window=5, min_count=1, workers=4)
-        logging.info("Word2Vec model trained.")
-        return model
+    # def train_word2vec(self, texts: List[str]) -> Word2Vec:
+    #     tokenized_texts = [text.split() for text in texts]
+    #     model = Word2Vec(sentences=tokenized_texts, vector_size=100, window=5, min_count=1, workers=4)
+    #     logging.info("Word2Vec model trained.")
+    #     return model
 
     async def get_text_embedding(self, text: str, model: str = "mistral-embed") -> np.ndarray:
         try:
@@ -250,12 +247,12 @@ class MistralRAGChatbot:
         logging.debug(f"Annoy retrieval returned {len(indices)} documents.")
         return indices, scores
 
-    def retrieve_with_tfidf(self, user_query: str, query_embedding: np.ndarray, top_k: int) -> Tuple[List[int], List[float]]:
-        query_vec = self.tfidf_vectorizer.transform([user_query])
-        similarities = cosine_similarity(query_vec, self.tfidf_matrix).flatten()
-        indices = np.argsort(-similarities)[:top_k]
-        logging.debug(f"TF-IDF retrieval returned {len(indices)} documents.")
-        return indices, similarities[indices].tolist()
+    # def retrieve_with_tfidf(self, user_query: str, query_embedding: np.ndarray, top_k: int) -> Tuple[List[int], List[float]]:
+    #     query_vec = self.tfidf_vectorizer.transform([user_query])
+    #     similarities = cosine_similarity(query_vec, self.tfidf_matrix).flatten()
+    #     indices = np.argsort(-similarities)[:top_k]
+    #     logging.debug(f"TF-IDF retrieval returned {len(indices)} documents.")
+    #     return indices, similarities[indices].tolist()
 
     def retrieve_with_bm25(self, user_query: str, query_embedding: np.ndarray, top_k: int) -> Tuple[List[int], List[float]]:
         tokenized_query = user_query.split()
@@ -264,38 +261,38 @@ class MistralRAGChatbot:
         logging.debug(f"BM25 retrieval returned {len(indices)} documents.")
         return indices, scores[indices].tolist()
 
-    def retrieve_with_word2vec(self, user_query: str, query_embedding: np.ndarray, top_k: int) -> Tuple[List[int], List[float]]:
-        query_tokens = user_query.split()
-        query_vec = np.mean([self.word2vec_model.wv[token] for token in query_tokens if token in self.word2vec_model.wv], axis=0)
-        expected_dim = query_vec.shape[0]
-        doc_vectors = []
-        for doc in self.texts:
-            word_vectors = [self.word2vec_model.wv[token] for token in doc.split() if token in self.word2vec_model.wv]
-            avg_vector = np.mean(word_vectors, axis=0) if word_vectors else np.zeros(expected_dim)
-            doc_vectors.append(avg_vector)
-        doc_vectors = np.array(doc_vectors)
-        similarities = cosine_similarity([query_vec], doc_vectors).flatten()
-        indices = np.argsort(-similarities)[:top_k]
-        return indices, similarities[indices].tolist()
-
-    def retrieve_with_euclidean(self, user_query: str, query_embedding: np.ndarray, top_k: int) -> Tuple[List[int], List[float]]:
-        distances = euclidean_distances([query_embedding], self.embeddings).flatten()
-        indices = np.argsort(distances)[:top_k]
-        logging.debug(f"Euclidean retrieval returned {len(indices)} documents.")
-        return indices, distances[indices].tolist()
-
-    def retrieve_with_jaccard(self, user_query: str, query_embedding: np.ndarray, top_k: int) -> Tuple[List[int], List[float]]:
-        query_set = set(user_query.lower().split())
-        scores = []
-        for doc in self.texts:
-            doc_set = set(doc.lower().split())
-            intersection = query_set.intersection(doc_set)
-            union = query_set.union(doc_set)
-            score = float(len(intersection)) / len(union) if union else 0
-            scores.append(score)
-        indices = np.argsort(-np.array(scores))[:top_k]
-        logging.debug(f"Jaccard retrieval returned {len(indices)} documents.")
-        return indices.tolist(), [scores[i] for i in indices]
+    # def retrieve_with_word2vec(self, user_query: str, query_embedding: np.ndarray, top_k: int) -> Tuple[List[int], List[float]]:
+    #     query_tokens = user_query.split()
+    #     query_vec = np.mean([self.word2vec_model.wv[token] for token in query_tokens if token in self.word2vec_model.wv], axis=0)
+    #     expected_dim = query_vec.shape[0]
+    #     doc_vectors = []
+    #     for doc in self.texts:
+    #         word_vectors = [self.word2vec_model.wv[token] for token in doc.split() if token in self.word2vec_model.wv]
+    #         avg_vector = np.mean(word_vectors, axis=0) if word_vectors else np.zeros(expected_dim)
+    #         doc_vectors.append(avg_vector)
+    #     doc_vectors = np.array(doc_vectors)
+    #     similarities = cosine_similarity([query_vec], doc_vectors).flatten()
+    #     indices = np.argsort(-similarities)[:top_k]
+    #     return indices, similarities[indices].tolist()
+
+    # def retrieve_with_euclidean(self, user_query: str, query_embedding: np.ndarray, top_k: int) -> Tuple[List[int], List[float]]:
+    #     distances = euclidean_distances([query_embedding], self.embeddings).flatten()
+    #     indices = np.argsort(distances)[:top_k]
+    #     logging.debug(f"Euclidean retrieval returned {len(indices)} documents.")
+    #     return indices, distances[indices].tolist()
+
+    # def retrieve_with_jaccard(self, user_query: str, query_embedding: np.ndarray, top_k: int) -> Tuple[List[int], List[float]]:
+    #     query_set = set(user_query.lower().split())
+    #     scores = []
+    #     for doc in self.texts:
+    #         doc_set = set(doc.lower().split())
+    #         intersection = query_set.intersection(doc_set)
+    #         union = query_set.union(doc_set)
+    #         score = float(len(intersection)) / len(union) if union else 0
+    #         scores.append(score)
+    #     indices = np.argsort(-np.array(scores))[:top_k]
+    #     logging.debug(f"Jaccard retrieval returned {len(indices)} documents.")
+    #     return indices.tolist(), [scores[i] for i in indices]
 
     def rerank_documents(
         self,
@@ -312,54 +309,6 @@ class MistralRAGChatbot:
 
         return reranked_docs
 
-    # def reciprocal_rank_fusion(self, user_query: str, docs: List[dict]) -> List[dict]:
-    #     k = 60  
-    #     method_ranks = {}
-    #     fused_scores = {} 
-    #     for doc in docs:
-    #         method = doc['method']
-    #         if method not in method_ranks:
-    #             method_ranks[method] = {doc['index']: 1}
-    #         else:
-    #             method_ranks[method][doc['index']] = len(method_ranks[method]) + 1
-    #     for doc in docs:
-    #         idx = doc['index']
-    #         if idx not in fused_scores:
-    #             fused_scores[idx] = sum(1 / (k + rank) for method_rank in method_ranks.values() for i, rank in method_rank.items() if i == idx)
-    #     reranked_docs = sorted(docs, key=lambda x: fused_scores.get(x['index'], 0), reverse=True)
-    #     for doc in reranked_docs:
-    #         doc['rrf_score'] = fused_scores.get(doc['index'], 0) 
-    #     return reranked_docs
-
-    # def weighted_score_fusion(self, user_query: str, docs: List[dict]) -> List[dict]:
-    #     method_weights = {
-    #         'annoy': 0.3,
-    #         'tfidf': 0.2,
-    #         'bm25': 0.2,
-    #         'word2vec': 0.1,
-    #         'euclidean': 0.1,
-    #         'jaccard': 0.1
-    #     }
-    #     fused_scores = {}
-    #     for doc in docs:
-    #         idx = doc['index']
-    #         if idx not in fused_scores:
-    #             fused_scores[idx] = doc['score'] * method_weights[doc['method']]
-    #         else:
-    #             fused_scores[idx] += doc['score'] * method_weights[doc['method']]
-
-    #     reranked_docs = sorted(docs, key=lambda x: fused_scores[x['index']], reverse=True)
-    #     for doc in reranked_docs:
-    #         doc['wsf_score'] = fused_scores[doc['index']]
-    #     return reranked_docs
-
-    # def semantic_similarity_reranking(self, user_query: str, docs: List[dict]) -> List[dict]:
-    #     query_embedding = np.mean([self.word2vec_model.wv[token] for token in user_query.split() if token in self.word2vec_model.wv], axis=0)
-    #     for doc in docs:
-    #         doc_embedding = np.mean([self.word2vec_model.wv[token] for token in doc['text'].split() if token in self.word2vec_model.wv], axis=0)
-    #         doc_embedding = doc_embedding if doc_embedding.shape == query_embedding.shape else np.zeros(query_embedding.shape)
-    #         doc['semantic_score'] = cosine_similarity([query_embedding], [doc_embedding])[0][0]
-    #     return sorted(docs, key=lambda x: x['semantic_score'], reverse=True)
 
     def build_prompt(self, context: str, user_query: str, response_style: str) -> str:
         styles = {