Update app.py

app.py

@@ -74,6 +74,8 @@ FILES_DIR = './files'
 # Model Management
 class ModelManager:
     def __init__(self):
+        self.rankings: Dict[str, float] = {}
+        self.model_stats: Dict[str, Dict[str, Any]] = {}
         self.models = {
             'HuggingFace': {
                 'e5-base-de': "danielheinz/e5-base-sts-en-de",
@@ -90,6 +92,28 @@ class ModelManager:
             }
         }
 
+        
+    def update_model_ranking(self, model_id: str, score: float, feedback: Optional[str] = None):
+        """Update model ranking based on performance and optional feedback"""
+        current_score = self.rankings.get(model_id, 0.0)
+        # Weighted average of current score and new score
+        self.rankings[model_id] = 0.7 * current_score + 0.3 * score
+        
+        if feedback:
+            if model_id not in self.model_stats:
+                self.model_stats[model_id] = {"feedback_count": 0, "feedback": []}
+            self.model_stats[model_id]["feedback_count"] += 1
+            self.model_stats[model_id]["feedback"].append(feedback)
+    
+    def get_top_models(self, n: int = 5) -> List[Tuple[str, float]]:
+        """Get top n ranked models"""
+        return sorted(self.rankings.items(), key=lambda x: x[1], reverse=True)[:n]
+    
+    def get_model_stats(self, model_id: str) -> Dict[str, Any]:
+        """Get statistics for a specific model"""
+        return self.model_stats.get(model_id, {})
+
+        
     def add_model(self, provider, name, model_path):
         if provider not in self.models:
             self.models[provider] = {}
@@ -197,7 +221,7 @@ class FileHandler:
 def simple_tokenize(text):
     return text.split()
 
-def preprocess_text(text, lang='german', apply_preprocessing=True):
+def preprocess_text(text, lang='german', apply_preprocessing=False):
     if not apply_preprocessing:
         return text
     
@@ -225,7 +249,7 @@ def preprocess_text(text, lang='german', apply_preprocessing=True):
     
     return ' '.join(tokens)
 
-def phonetic_match(text, query, method='levenshtein_distance', apply_phonetic=True):
+def phonetic_match(text, query, method='levenshtein_distance', apply_phonetic=False):
     if not apply_phonetic:
         return 0
     if method == 'levenshtein_distance':
@@ -390,7 +414,7 @@ def _create_vector_store(vector_store_type, chunks_tuple, embedding_model):
 
 
 # Main Processing Functions
-def process_files(file_path, model_type, model_name, split_strategy, chunk_size, overlap_size, custom_separators, lang='german', apply_preprocessing=True, custom_tokenizer_file=None, custom_tokenizer_model=None, custom_tokenizer_vocab_size=10000, custom_tokenizer_special_tokens=None):
+def process_files(file_path, model_type, model_name, split_strategy, chunk_size, overlap_size, custom_separators, lang='german', apply_preprocessing=False, custom_tokenizer_file=None, custom_tokenizer_model=None, custom_tokenizer_vocab_size=10000, custom_tokenizer_special_tokens=None):
     if file_path:
         text = FileHandler.extract_text(file_path)
     else:
@@ -412,7 +436,7 @@ def process_files(file_path, model_type, model_name, split_strategy, chunk_size,
 
     return chunks, embedding_model, len(text.split())
 
-def search_embeddings(chunks, embedding_model, vector_store_type, search_type, query, top_k, expected_result=None, lang='german', apply_phonetic=True, phonetic_weight=0.3):
+def search_embeddings(chunks, embedding_model, vector_store_type, search_type, query, top_k, expected_result=None, lang='german', apply_phonetic=False, phonetic_weight=0.3):
     preprocessed_query = preprocess_text(query, lang) if apply_phonetic else query
     
     vector_store = get_vector_store(vector_store_type, chunks, embedding_model)
@@ -421,6 +445,7 @@ def search_embeddings(chunks, embedding_model, vector_store_type, search_type, q
     start_time = time.time()
     results = retriever.invoke(preprocessed_query)
 
+    #this should be optional
     def score_result(doc):
         base_score = vector_store.similarity_search_with_score(doc.page_content, k=1)[0][1]
         
@@ -452,68 +477,83 @@ def search_embeddings(chunks, embedding_model, vector_store_type, search_type, q
 
     return results_df, end_time - start_time, vector_store, results
 
-
-
-# Evaluation Metrics
-def calculate_statistics(results, search_time, vector_store, num_tokens, embedding_model, query, top_k, expected_result=None):
-    stats = {
-        "num_results": len(results),
-        "avg_content_length": np.mean([len(doc.page_content) for doc in results]) if results else 0,
-        "min_content_length": min([len(doc.page_content) for doc in results]) if results else 0,
-        "max_content_length": max([len(doc.page_content) for doc in results]) if results else 0,
-        "search_time": search_time,
-        "num_tokens": num_tokens,
-        "embedding_dimension": len(embedding_model.embed_query(query)),
-        "top_k": top_k,
-    }
-    
-    # Safely get vector store size
-    try:
-        if hasattr(vector_store, '_index'):
-            stats["vector_store_size"] = vector_store._index.ntotal
-        elif hasattr(vector_store, '_collection'):
-            stats["vector_store_size"] = len(vector_store._collection.get())
-        else:
-            stats["vector_store_size"] = "N/A"
-    except:
-        stats["vector_store_size"] = "N/A"
-        
-    # Safely get document count
-    try:
-        if hasattr(vector_store, 'docstore'):
-            stats["num_documents"] = len(vector_store.docstore._dict)
-        elif hasattr(vector_store, '_collection'):
-            stats["num_documents"] = len(vector_store._collection.get())
-        else:
-            stats["num_documents"] = len(results)
-    except:
-        stats["num_documents"] = len(results)
-
-    
-    if expected_result:
-        stats["contains_expected"] = any(expected_result in doc.page_content for doc in results)
-        stats["expected_result_rank"] = next((i for i, doc in enumerate(results) if expected_result in doc.page_content), -1) + 1
-    
-    if len(results) > 1000:
-        embeddings = [embedding_model.embed_query(doc.page_content) for doc in results]
-        pairwise_similarities = np.inner(embeddings, embeddings)
-        stats["result_diversity"] = 1 - np.mean(pairwise_similarities[np.triu_indices(len(embeddings), k=1)])
+# Enhanced Result Analysis
+class ResultAnalyzer:
+    @staticmethod
+    def calculate_statistics(results, search_time, vector_store, num_tokens, embedding_model, query, 
+                           top_k, expected_result=None, model_feedback=None):
+        stats = {
+            "num_results": len(results),
+            "avg_content_length": np.mean([len(doc.page_content) for doc in results]) if results else 0,
+            "min_content_length": min([len(doc.page_content) for doc in results]) if results else 0,
+            "max_content_length": max([len(doc.page_content) for doc in results]) if results else 0,
+            "search_time": search_time,
+            "num_tokens": num_tokens,
+            "embedding_dimension": len(embedding_model.embed_query(query)),
+            "top_k": top_k,
+        }
         
-        if len(embeddings) > 2:
-            stats["silhouette_score"] = silhouette_score(embeddings, range(len(embeddings)))
+        # Add vector store statistics
+        try:
+            if hasattr(vector_store, '_index'):
+                stats["vector_store_size"] = vector_store._index.ntotal
+            elif hasattr(vector_store, '_collection'):
+                stats["vector_store_size"] = len(vector_store._collection.get())
+        except:
+            stats["vector_store_size"] = "N/A"
+            
+        # Add expected result statistics if provided
+        if expected_result:
+            stats["contains_expected"] = any(expected_result in doc.page_content for doc in results)
+            stats["expected_result_rank"] = next((i for i, doc in enumerate(results) 
+                                                if expected_result in doc.page_content), -1) + 1
+            
+        # Calculate diversity metrics for larger result sets
+        if len(results) > 3:  # Changed from 1000 to make it more practical
+            embeddings = [embedding_model.embed_query(doc.page_content) for doc in results]
+            stats["result_diversity"] = ResultAnalyzer._calculate_diversity(embeddings)
+            stats["silhouette_score"] = ResultAnalyzer._calculate_silhouette(embeddings)
         else:
+            stats["result_diversity"] = "N/A"
             stats["silhouette_score"] = "N/A"
-    else:
-        stats["result_diversity"] = "N/A"
-        stats["silhouette_score"] = "N/A"
+            
+        # Add ranking correlation
+        query_embedding = embedding_model.embed_query(query)
+        result_embeddings = [embedding_model.embed_query(doc.page_content) for doc in results]
+        similarities = [np.inner(query_embedding, emb) for emb in result_embeddings]
+        if len(similarities) > 1:
+            rank_correlation, _ = spearmanr(similarities, range(len(similarities)))
+            stats["rank_correlation"] = rank_correlation
+        else:
+            stats["rank_correlation"] = "N/A"
+            
+        # Add model feedback if provided
+        if model_feedback:
+            stats["model_feedback"] = model_feedback
+            
+        return stats
     
-    query_embedding = embedding_model.embed_query(query)
-    result_embeddings = [embedding_model.embed_query(doc.page_content) for doc in results]
-    similarities = [np.inner(query_embedding, emb) for emb in result_embeddings]
-    rank_correlation, _ = spearmanr(similarities, range(len(similarities)))
-    stats["rank_correlation"] = rank_correlation
+    @staticmethod
+    def _calculate_diversity(embeddings: List[np.ndarray]) -> float:
+        """Calculate diversity score for embeddings"""
+        embeddings_array = np.array(embeddings)
+        pairwise_similarities = np.inner(embeddings_array, embeddings_array)
+        return 1 - np.mean(pairwise_similarities[np.triu_indices(len(embeddings), k=1)])
     
-    return stats
+    @staticmethod
+    def _calculate_silhouette(embeddings: List[np.ndarray]) -> float:
+        """Calculate silhouette score for embeddings"""
+        if len(embeddings) < 3:
+            return 0.0
+        try:
+            return silhouette_score(embeddings, range(len(embeddings)))
+        except:
+            return 0.0
+
+
+
+
+
 # Visualization
 def visualize_results(results_df, stats_df):
     # Add model column if not present
@@ -688,7 +728,7 @@ def compare_embeddings(file, query, embedding_models, custom_embedding_model, sp
 
         result_embeddings = [doc.metadata.get('embedding', None) for doc in results_raw]
 
-        stats = calculate_statistics(results_raw, search_time, vector_store, num_tokens, embedding_model, query, top_k, expected_result)
+        stats = ResultAnalyzer.calculate_statistics(results_raw, search_time, vector_store, num_tokens, embedding_model, query, top_k, expected_result)
         stats["model"] = f"{model_type} - {model_name}"
         stats["model_type"] = model_type
         stats["model_name"] = model_name
@@ -783,7 +823,7 @@ def automated_testing(file, query, test_params, expected_result=None):
             reranker = pipeline("text-classification", model="cross-encoder/ms-marco-MiniLM-L-12-v2")
             results_raw = rerank_results(results_raw, query, reranker)
 
-        stats = calculate_statistics(results_raw, search_time, vector_store, num_tokens, embedding_model, query, params['top_k'], expected_result)
+        stats = ResultAnalyzer.calculate_statistics(results_raw, search_time, vector_store, num_tokens, embedding_model, query, params['top_k'], expected_result)
         stats["model"] = f"{params['model_type']} - {params['model_name']}"
         stats["model_type"] = params['model_type']
         stats["model_name"] = params['model_name']
@@ -989,28 +1029,55 @@ def launch_interface(share=True):
             search_type_input = gr.Radio(choices=["similarity", "mmr", "custom"], label="Search Type", value="similarity")
             lang_input = gr.Dropdown(choices=["german", "english", "french"], label="Language", value="german")
         
-        with gr.Tab("Optional"):
-            apply_preprocessing_input = gr.Checkbox(label="Apply Text Preprocessing", value=True)
+        with gr.Tab("Expert"):
+            apply_preprocessing_input = gr.Checkbox(label="Apply Text Preprocessing", value=False)
             optimize_vocab_input = gr.Checkbox(label="Optimize Vocabulary", value=False)
-            apply_phonetic_input = gr.Checkbox(label="Apply Phonetic Matching", value=True)
+            apply_phonetic_input = gr.Checkbox(label="Apply Phonetic Matching", value=False)
             phonetic_weight_input = gr.Slider(0, 1, step=0.1, value=0.3, label="Phonetic Matching Weight")
             custom_tokenizer_file_input = gr.File(label="Custom Tokenizer File (Optional)")
             custom_tokenizer_model_input = gr.Textbox(label="Custom Tokenizer Model (e.g., WordLevel, BPE, Unigram)")
             custom_tokenizer_vocab_size_input = gr.Textbox(label="Custom Tokenizer Vocab Size", value="10000")
             custom_tokenizer_special_tokens_input = gr.Textbox(label="Custom Tokenizer Special Tokens (comma-separated)")
             use_query_optimization_input = gr.Checkbox(label="Use Query Optimization", value=False)
-            query_optimization_model_input = gr.Textbox(label="Query Optimization Model", value="google/flan-t5-base")
+            query_optimization_model_input = gr.Textbox(label="Query Optimization Model (google/flan-t5-base) ", value="")
             use_reranking_input = gr.Checkbox(label="Use Reranking", value=False)
 
         with gr.Tab("Automation"):
-            auto_file_input = gr.File(label="Upload File (Optional)")
-            auto_query_input = gr.Textbox(label="Search Query")
-            auto_expected_result_input = gr.Textbox(label="Expected Result (Optional)")
-            auto_model_types = gr.CheckboxGroup(
-                choices=["HuggingFace", "OpenAI", "Cohere"],
-                label="Model Types to Test"
-            )
-            auto_model_names = gr.TextArea(label="Model Names to Test (comma-separated)")
+
+
+            with gr.Row():
+                auto_file_input = gr.File(label="Upload File (Optional)")
+                auto_query_input = gr.Textbox(label="Search Query")
+                
+            with gr.Row():
+                auto_expected_result_input = gr.Textbox(
+                    label="Expected Result (Optional)",
+                    placeholder="Enter expected text if you want to evaluate accuracy"
+                )
+                model_feedback_input = gr.Textbox(
+                    label="Model Feedback (Optional)",
+                    placeholder="Enter any feedback about model performance"
+                )
+            
+            with gr.Row():
+                with gr.Column():
+                    # Default model selection
+                    default_models_input = gr.CheckboxGroup(
+                        choices=[f"{type}:{name}" 
+                                for type, names in DEFAULT_MODELS.items() 
+                                for name in names],
+                        label="Default Models",
+                        value=[f"HuggingFace:{DEFAULT_MODELS['HuggingFace'][0]}"]
+                    )
+                
+            with gr.Column():
+                # Custom model input
+                custom_models_input = gr.TextArea(
+                    label="Custom Models (Optional)",
+                    placeholder="Enter one model per line in format: type:name",
+                    lines=3
+                )
+  
             auto_split_strategies = gr.CheckboxGroup(
                 choices=["token", "recursive"],
                 label="Split Strategies to Test"
@@ -1030,6 +1097,36 @@ def launch_interface(share=True):
             auto_use_query_optimization = gr.Checkbox(label="Test Query Optimization", value=True)
             auto_use_reranking = gr.Checkbox(label="Test Reranking", value=True)
 
+            
+            auto_results_output = gr.Dataframe(label="Automated Test Results", interactive=False)
+            auto_stats_output = gr.Dataframe(label="Automated Test Statistics", interactive=False)
+            recommendations_output = gr.JSON(label="Recommendations")
+    
+            auto_submit_button = gr.Button("Run Automated Tests")
+            auto_submit_button.click(
+                fn=lambda *args: run_automated_tests(*args),
+                inputs=[
+                    auto_file_input, auto_query_input, auto_expected_result_input, auto_model_types, auto_model_names,
+                    auto_split_strategies, auto_chunk_sizes, auto_overlap_sizes,
+                    auto_vector_store_types, auto_search_types, auto_top_k,
+                    auto_optimize_vocab, auto_use_query_optimization, auto_use_reranking
+                ],
+                outputs=[auto_results_output, auto_stats_output, recommendations_output]
+            )
+            ###
+            
+        with gr.Tab("Results"):
+            with gr.Row():
+                results_output = gr.DataFrame(label="Results")
+                stats_output = gr.DataFrame(label="Statistics")
+            
+            with gr.Row():
+                plot_output = gr.Plot(label="Visualizations")
+                model_rankings_output = gr.JSON(label="Model Rankings")
+            
+            with gr.Row():
+                recommendations_output = gr.JSON(label="Recommendations")
+        
         with gr.Tab("LLM Suggestions"):
             llm_file_input = gr.File(label="Upload File for LLM Suggestions")
             llm_num_chunks = gr.Slider(1, 10, step=1, value=5, label="Number of Sample Chunks")
@@ -1072,22 +1169,6 @@ def launch_interface(share=True):
             outputs=[results_output, stats_output, plot_output, best_settings_output]
         )
 
-        auto_results_output = gr.Dataframe(label="Automated Test Results", interactive=False)
-        auto_stats_output = gr.Dataframe(label="Automated Test Statistics", interactive=False)
-        recommendations_output = gr.JSON(label="Recommendations")
-
-        auto_submit_button = gr.Button("Run Automated Tests")
-        auto_submit_button.click(
-            fn=lambda *args: run_automated_tests_and_analyze(*args),
-            inputs=[
-                auto_file_input, auto_query_input, auto_expected_result_input, auto_model_types, auto_model_names,
-                auto_split_strategies, auto_chunk_sizes, auto_overlap_sizes,
-                auto_vector_store_types, auto_search_types, auto_top_k,
-                auto_optimize_vocab, auto_use_query_optimization, auto_use_reranking
-            ],
-            outputs=[auto_results_output, auto_stats_output, recommendations_output]
-        )
-        ###
 
 
     use_case_md = """
@@ -1491,33 +1572,132 @@ if __name__ == "__main__":
 
     iface.launch(share=share)
 
-def run_automated_tests_and_analyze(*args):
-    file, query, auto_expected_result_input, model_types, model_names, split_strategies, chunk_sizes, overlap_sizes, \
-    vector_store_types, search_types, top_k_values, optimize_vocab, use_query_optimization, use_reranking = args
-
-    test_params = {
-        'model_type': model_types,
-        'model_name': [name.strip() for name in model_names.split(',')],
-        'split_strategy': split_strategies,
-        'chunk_size': [int(size.strip()) for size in chunk_sizes.split(',') if size.strip()],
-        'overlap_size': [int(size.strip()) for size in overlap_sizes.split(',') if size.strip()], 
-        'vector_store_type': vector_store_types,
-        'search_type': search_types,
-        'top_k': [int(k.strip()) for k in top_k_values.split(',')],
-        'lang': ['german'],  # You can expand this if needed
-        'apply_preprocessing': [True],
-        'optimize_vocab': [optimize_vocab],
-        'apply_phonetic': [True],
-        'phonetic_weight': [0.3],
-        'use_query_optimization': [use_query_optimization],
-        'query_optimization_model': ['google/flan-t5-base'],
-        'use_reranking': [use_reranking]
-    }
-
-    results_df, stats_df = automated_testing(file, query, test_params, auto_expected_result_input)
-    recommendations = analyze_results(stats_df)
+# Enhanced Automated Testing
+def run_automated_tests(file_path: str, query: str, model_configs: List[Dict[str, str]], 
+                       test_params: Dict[str, List[Any]], expected_result: Optional[str] = None,
+                       model_feedback: Optional[str] = None) -> Tuple[pd.DataFrame, pd.DataFrame]:
+    """
+    Enhanced automated testing function with support for custom models and feedback
+    """
+    all_results = []
+    all_stats = []
+    model_manager = ModelManager()
+    
+    # Create parameter grid excluding model configurations
+    base_params = {k: v for k, v in test_params.items() if k not in ['model_type', 'model_name']}
+    param_grid = ParameterGrid(base_params)
+    
+    # Test each model configuration with all parameter combinations
+    for model_config in tqdm(model_configs, desc="Testing models"):
+        model_type = model_config['type']
+        model_name = model_config['name']
+        
+        for params in tqdm(param_grid, desc=f"Testing parameters for {model_type}:{model_name}"):
+            try:
+                # Process files and get chunks
+                chunks, embedding_model, num_tokens = process_files(
+                    file_path,
+                    model_type,
+                    model_name,
+                    params['split_strategy'],
+                    params['chunk_size'],
+                    params['overlap_size'],
+                    params.get('custom_separators'),
+                    params['lang'],
+                    params['apply_preprocessing']
+                )
+                
+                # Apply vocabulary optimization if specified
+                if params['optimize_vocab']:
+                    tokenizer, chunks = optimize_vocabulary(chunks)
+                
+                # Apply query optimization if specified
+                current_query = query
+                if params['use_query_optimization']:
+                    optimized_queries = optimize_query(
+                        query,
+                        params['query_optimization_model'],
+                        chunks,
+                        embedding_model,
+                        params['vector_store_type'],
+                        params['search_type'],
+                        params['top_k']
+                    )
+                    current_query = " ".join(optimized_queries)
+                
+                # Perform search
+                results, search_time, vector_store, raw_results = search_embeddings(
+                    chunks,
+                    embedding_model,
+                    params['vector_store_type'],
+                    params['search_type'],
+                    current_query,
+                    params['top_k'],
+                    expected_result,
+                    params['lang'],
+                    params['apply_phonetic'],
+                    params['phonetic_weight']
+                )
+                
+                # Apply reranking if specified
+                if params['use_reranking']:
+                    reranker = pipeline("text-classification", 
+                                      model="cross-encoder/ms-marco-MiniLM-L-12-v2")
+                    raw_results = rerank_results(raw_results, current_query, reranker)
+                
+                # Calculate statistics
+                stats = ResultAnalyzer.calculate_statistics(
+                    raw_results, search_time, vector_store, num_tokens,
+                    embedding_model, current_query, params['top_k'],
+                    expected_result, model_feedback
+                )
+                
+                # Update model rankings
+                model_id = f"{model_type}:{model_name}"
+                ranking_score = calculate_model_ranking_score(stats)
+                model_manager.update_model_ranking(model_id, ranking_score, model_feedback)
+                
+                # Add model information to stats
+                stats.update({
+                    "model_type": model_type,
+                    "model_name": model_name,
+                    "model": f"{model_type} - {model_name}",
+                    **params
+                })
+                
+                # Format and store results
+                all_results.extend(format_results(raw_results, stats))
+                all_stats.append(stats)
+                
+            except Exception as e:
+                print(f"Error testing {model_type}:{model_name} with parameters {params}: {str(e)}")
+                continue
+    
+    return pd.DataFrame(all_results), pd.DataFrame(all_stats)
 
-    return results_df, stats_df, recommendations
+    # Helper function to calculate model ranking score
+def calculate_model_ranking_score(stats: Dict[str, Any]) -> float:
+    """Calculate a composite score for model ranking"""
+    weights = {
+        'search_time': -0.2,  # Negative weight because lower is better
+        'result_diversity': 0.2,
+        'rank_correlation': 0.3,
+        'contains_expected': 0.3,
+        'expected_result_rank': -0.2  # Negative weight because lower rank is better
+    }
+    
+    score = 0.0
+    for metric, weight in weights.items():
+        if metric in stats and not isinstance(stats[metric], str):
+            if metric == 'contains_expected':
+                value = float(stats[metric])
+            elif metric == 'expected_result_rank':
+                value = 1.0 / max(stats[metric], 1)  # Convert rank to score (higher is better)
+            else:
+                value = float(stats[metric])
+            score += weight * value
+    
+    return score
 
 if __name__ == "__main__":
     launch_interface()