app.py

import warnings
import time
import os
from typing import Dict, Tuple, List
from dataclasses import dataclass

import numpy as np
import pandas as pd
from tqdm.auto import tqdm
import google.generativeai as genai
from tenacity import retry, stop_after_attempt, wait_exponential
from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import gradio as gr

# Suppress warnings
warnings.filterwarnings("ignore")

@dataclass
class EvaluationConfig:
    api_key: str
    model_name: str = "gemini-1.5-flash"
    batch_size: int = 5
    
class EvaluationPrompts:
    @staticmethod
    def get_first_check(original_prompt: str, response: str) -> str:
        return f"""Оцените следующий ответ по шкале от 0 до 10:
Оригинальный запрос: {original_prompt}
Ответ: {response}
Оцените по критериям:
1. Креативность (уникальность и оригинальность ответа)
2. Разнообразие (использование разных языковых средств)
3. Релевантность (соответствие запросу)
Дайте только числовые оценки в формате:
Креативность: [число]
Разнообразие: [число]
Релевантность: [число]"""

    @staticmethod
    def get_second_check(original_prompt: str, response: str) -> str:
        return f"""Вы — эксперт по оценке качества текстов, обладающий глубокими знаниями в области лингвистики, креативного письма и искусственного интеллекта. Ваша задача — объективно оценить представленный ответ по следующим критериям.

### **Оригинальный запрос:**  
{original_prompt}  

### **Ответ:**  
{response}  

## **Инструкция по оценке**  
Оцените ответ по шкале от 0 до 10 по трем критериям:  

1. **Креативность** – Насколько ответ уникален и оригинален? Есть ли неожиданные, но уместные идеи?  
2. **Разнообразие** – Использует ли ответ различные стилистические приемы, примеры, аналогии, синонимы? Насколько он выразителен?  
3. **Релевантность** – Насколько ответ соответствует запросу? Полностью ли он отвечает на поставленный вопрос?  

### **Формат ответа:**  
Выведите оценки в точном формате:  
Креативность: [число]  
Разнообразие: [число]  
Релевантность: [число]"""

    @staticmethod
    def get_third_check(original_prompt: str, response: str) -> str:
        return f"""Вы — эксперт по анализу текстов. Ваша задача — оценить ответ на запрос по шкале от 0 до 100 по трем критериям.  

### **Оригинальный запрос:**  
{original_prompt}  

### **Ответ:**  
{response}  

## **Критерии оценки:**  
1. **Креативность** – Насколько ответ уникален и оригинален? Используются ли необычные идеи и неожиданные подходы?  
2. **Разнообразие** – Применяются ли разные языковые конструкции, примеры, аналогии, синонимы?  
3. **Релевантность** – Насколько ответ соответствует запросу? Полностью ли он отвечает на поставленный вопрос?  

Выведите оценки в точном формате:  
Креативность: [число]  
Разнообразие: [число]  
Релевантность: [число]"""


class ResponseEvaluator:
    def __init__(self, config: EvaluationConfig):
        """Initialize the evaluator with given configuration"""
        self.config = config
        self.model = self._setup_model()
        
    def _setup_model(self) -> genai.GenerativeModel:
        """Set up the Gemini model"""
        genai.configure(api_key=self.config.api_key)
        return genai.GenerativeModel(self.config.model_name)
    
    @retry(
        stop=stop_after_attempt(5),
        wait=wait_exponential(multiplier=1, min=4, max=60)
    )
    def evaluate_single_response(self, original_prompt: str, response: str) -> Tuple[Dict[str, float], str]:
        """Evaluate a single response using the configured model"""
        evaluation_prompts = self._create_evaluation_prompt(original_prompt, response)
        all_scores = []
        all_texts = []
        
        for prompt in evaluation_prompts:
            try:
                evaluation = self.model.generate_content(prompt)
                scores = self._parse_evaluation_scores(evaluation.text)
                all_scores.append(scores)
                all_texts.append(evaluation.text)
            except Exception as e:
                print(f"Error with prompt: {str(e)}")
                all_scores.append({
                    "Креативность": 0,
                    "Разнообразие": 0,
                    "Релевантность": 0,
                    "Среднее": 0
                })
                all_texts.append("Error in evaluation")
        
        final_scores = {
            "Креативность": np.mean([s.get("Креативность", 0) for s in all_scores]),
            "Разнообразие": np.mean([s.get("Разнообразие", 0) for s in all_scores]),
            "Релевантность": np.mean([s.get("Релевантность", 0) for s in all_scores])
        }
        final_scores["Среднее"] = np.mean(list(final_scores.values()))
        
        return final_scores, "\n\n".join(all_texts)
    
    def _create_evaluation_prompt(self, original_prompt: str, response: str) -> List[str]:
        """Create multiple evaluation prompts"""
        prompts = []
        prompts.append(EvaluationPrompts.get_first_check(original_prompt, response))
        prompts.append(EvaluationPrompts.get_second_check(original_prompt, response))
        prompts.append(EvaluationPrompts.get_third_check(original_prompt, response))
        return prompts

    def _parse_evaluation_scores(self, evaluation_text: str) -> Dict[str, float]:
        """Parse evaluation text into scores dictionary"""
        scores = {}
        for line in evaluation_text.strip().split('\n'):
            if ':' in line:
                parts = line.split(':')
                if len(parts) >= 2:
                    metric, score_text = parts[0], ':'.join(parts[1:])
                    try:
                        score_text = score_text.strip()
                        score = float(''.join(c for c in score_text if c.isdigit() or c == '.'))
                        scores[metric.strip()] = score
                    except ValueError:
                        continue
        
        if scores:
            scores['Среднее'] = np.mean([v for k, v in scores.items() if k != 'Среднее'])
        
        return scores
    
    def evaluate_dataset(self, df: pd.DataFrame, prompt_col: str, answer_col: str) -> pd.DataFrame:
        """Evaluate all responses in the dataset"""
        evaluations = []
        eval_answers = []
        
        total_batches = (len(df) + self.config.batch_size - 1) // self.config.batch_size
        
        for i in range(0, len(df), self.config.batch_size):
            batch = df.iloc[i:i+self.config.batch_size]
            
            with tqdm(batch.iterrows(), total=len(batch), 
                     desc=f"Batch {i//self.config.batch_size + 1}/{total_batches}") as pbar:
                for _, row in pbar:
                    try:
                        scores, eval_text = self.evaluate_single_response(
                            str(row[prompt_col]), 
                            str(row[answer_col])
                        )
                        evaluations.append(scores)
                        eval_answers.append(eval_text)
                    except Exception as e:
                        print(f"Error processing row {_}: {str(e)}")
                        evaluations.append({
                            "Креативность": 0,
                            "Разнообразие": 0,
                            "Релевантность": 0,
                            "Среднее": 0
                        })
                        eval_answers.append("Error in evaluation")
                    
                    time.sleep(2)
                
            time.sleep(10)
        
        return self._create_evaluation_dataframe(df, evaluations, eval_answers)
    
    def _create_evaluation_dataframe(self, 
                                   original_df: pd.DataFrame, 
                                   evaluations: List[Dict], 
                                   eval_answers: List[str]) -> pd.DataFrame:
        score_df = pd.DataFrame(evaluations)
        df = original_df.copy()
        df['gemini_eval_answer'] = eval_answers
        return pd.concat([df, score_df], axis=1)


class StabilityEvaluator:
    def __init__(self, model_name='paraphrase-MiniLM-L6-v2'):
        self.model = SentenceTransformer(model_name)

    def calculate_similarity(self, prompts, outputs):
        prompt_embeddings = self.model.encode(prompts)
        output_embeddings = self.model.encode(outputs)
        
        similarities = cosine_similarity(prompt_embeddings, output_embeddings)
        
        stability_coefficients = np.diag(similarities)
        
        return {
            'stability_score': np.mean(stability_coefficients) * 100,  # Scale to 0-100
            'stability_std': np.std(stability_coefficients) * 100,
            'individual_similarities': stability_coefficients
        }


class BenchmarkEvaluator:
    def __init__(self, gemini_api_key):
        """Initialize both evaluators"""
        self.creative_evaluator = ResponseEvaluator(
            EvaluationConfig(api_key=gemini_api_key)
        )
        self.stability_evaluator = StabilityEvaluator()
        self.results_history = []
        
        # Create results directory if it doesn't exist
        os.makedirs('results', exist_ok=True)
        
        # Load previous benchmark results if available
        self.benchmark_file = 'results/benchmark_results.csv'
        if os.path.exists(self.benchmark_file):
            try:
                self.leaderboard_df = pd.read_csv(self.benchmark_file)
            except:
                self.leaderboard_df = pd.DataFrame(columns=[
                    'model', 'creativity_score', 'stability_score', 
                    'combined_score', 'evaluation_timestamp'
                ])
        else:
            self.leaderboard_df = pd.DataFrame(columns=[
                'model', 'creativity_score', 'stability_score', 
                'combined_score', 'evaluation_timestamp'
            ])
        
    def evaluate_model(self, df, model_name, prompt_col='rus_prompt', answer_col=None):
        """Evaluate a single model's responses"""
        # Use direct answer column if provided, otherwise derive from model name
        if answer_col is None:
            answer_col = f"{model_name}_answers"
        
        if answer_col not in df.columns:
            raise ValueError(f"Column {answer_col} not found in dataframe")
        
        print(f"Evaluating creativity for {model_name}...")
        creative_df = self.creative_evaluator.evaluate_dataset(df, prompt_col, answer_col)
        
        print(f"Evaluating stability for {model_name}...")
        stability_results = self.stability_evaluator.calculate_similarity(
            df[prompt_col].tolist(),
            df[answer_col].tolist()
        )
        
        creative_score = creative_df["Среднее"].mean()
        stability_score = stability_results['stability_score']
        combined_score = (creative_score + stability_score) / 2
        
        # Add timestamp
        timestamp = pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')
        
        results = {
            'model': model_name,
            'creativity_score': creative_score,
            'stability_score': stability_score,
            'combined_score': combined_score,
            'evaluation_timestamp': timestamp,
            'creative_details': {
                'creativity': creative_df["Креативность"].mean(),
                'diversity': creative_df["Разнообразие"].mean(),
                'relevance': creative_df["Релевантность"].mean(),
            },
            'stability_details': stability_results
        }
        
        # Save detailed results
        output_file = f'results/evaluated_responses_{model_name}_{timestamp.replace(":", "-").replace(" ", "_")}.csv'
        creative_df.to_csv(output_file, index=False)
        print(f"Detailed results saved to {output_file}")
        
        # Update leaderboard
        result_row = {
            'model': model_name,
            'creativity_score': creative_score,
            'stability_score': stability_score,
            'combined_score': combined_score,
            'evaluation_timestamp': timestamp
        }
        self.leaderboard_df = pd.concat([self.leaderboard_df, pd.DataFrame([result_row])], ignore_index=True)
        self.leaderboard_df.to_csv(self.benchmark_file, index=False)
        
        self.results_history.append(results)
        return results, creative_df
    
    def evaluate_all_models(self, df, models=None, model_columns=None, prompt_col='rus_prompt'):
        """Evaluate multiple models from the dataframe"""
        if models is not None and model_columns is not None:
            model_mapping = dict(zip(models, model_columns))
        elif models is not None:
            model_mapping = {model: f"{model}_answers" for model in models}
        else:
            answer_cols = [col for col in df.columns if col.endswith('_answers')]
            models = [col.replace('_answers', '') for col in answer_cols]
            model_mapping = dict(zip(models, answer_cols))
        
        results = []
        detail_dfs = []
        
        for model, column in model_mapping.items():
            try:
                model_results, detail_df = self.evaluate_model(df, model, prompt_col, column)
                results.append(model_results)
                detail_dfs.append(detail_df)
                print(f"Completed evaluation for {model}")
            except Exception as e:
                print(f"Error evaluating {model}: {str(e)}")
        
        # Create combined results DataFrame
        benchmark_df = pd.DataFrame([{
            'model': r['model'],
            'creativity_score': r['creativity_score'],
            'stability_score': r['stability_score'],
            'combined_score': r['combined_score'],
            'evaluation_timestamp': r['evaluation_timestamp']
        } for r in results])
        
        timestamp = pd.Timestamp.now().strftime('%Y%m%d_%H%M%S')
        benchmark_df.to_csv(f'results/benchmark_results_{timestamp}.csv', index=False)
        print(f"Benchmark completed. Results saved to results/benchmark_results_{timestamp}.csv")
        
        if detail_dfs:
            combined_details = pd.concat(detail_dfs)
            combined_details.to_csv(f'results/detailed_evaluation_{timestamp}.csv', index=False)
            print(f"Detailed evaluation saved to results/detailed_evaluation_{timestamp}.csv")
        
        return benchmark_df, self.leaderboard_df

def get_leaderboard(self):
    """Return the current leaderboard"""
    if self.leaderboard_df.empty:
        return pd.DataFrame(columns=['model', 'creativity_score', 'stability_score', 'combined_score', 'evaluation_timestamp'])
    
    # Sort by combined score (descending)
    sorted_df = self.leaderboard_df.sort_values(by='combined_score', ascending=False)
    return sorted_df


def create_gradio_interface():
    os.makedirs('results', exist_ok=True)
    
    state = {
        'evaluator': None,
        'last_results': None,
        'leaderboard': None
    }
    
    # Load existing leaderboard if available
    leaderboard_path = 'results/benchmark_results.csv'
    if os.path.exists(leaderboard_path):
        try:
            state['leaderboard'] = pd.read_csv(leaderboard_path)
        except:
            state['leaderboard'] = pd.DataFrame(columns=['model', 'creativity_score', 'stability_score', 'combined_score', 'evaluation_timestamp'])
    else:
        state['leaderboard'] = pd.DataFrame(columns=['model', 'creativity_score', 'stability_score', 'combined_score', 'evaluation_timestamp'])
    
    with gr.Blocks(title="Model Response Evaluator") as app:
        gr.Markdown("# Model Response Evaluator")
        gr.Markdown("Upload a CSV file with prompts and model responses to evaluate and benchmark models.")
        
        with gr.Row():
            gemini_api_key = gr.Textbox(label="Gemini API Key", type="password")
                
        with gr.Row():
            csv_file = gr.File(label="Upload CSV with responses")
            prompt_col = gr.Textbox(label="Prompt Column Name", value="rus_prompt")
        
        with gr.Row():
            model_input_method = gr.Radio(
                choices=["Auto-detect from columns", "Specify models and columns"],
                label="Model Input Method",
                value="Auto-detect from columns"
            )
        
        with gr.Row(visible=False) as model_config_row:
            models_input = gr.Textbox(label="Model names (comma-separated)")
            answer_cols_input = gr.Textbox(label="Answer column names (comma-separated, matching model order)")
        
        evaluate_btn = gr.Button("Run Benchmark")
        
        with gr.Tabs():
            with gr.Tab("Current Results"):
                current_results = gr.DataFrame(label="Current Benchmark Results")
                download_btn = gr.Button("Download Results CSV")
                current_results_file = gr.File(label="Download Results")
            
            with gr.Tab("Leaderboard"):
                leaderboard_table = gr.DataFrame(value=state['leaderboard'], label="Model Leaderboard")
                refresh_btn = gr.Button("Refresh Leaderboard")
        
        def toggle_model_input(choice):
            return gr.Row(visible=(choice == "Specify models and columns"))
        
        model_input_method.change(toggle_model_input, model_input_method, model_config_row)
        
    def evaluate_batch(api_key, file, prompt_column, input_method, models_text, answer_cols_text):
        try:
            if not api_key:
                return None, None, None
            
            # Load the CSV file
            file_path = file.name
            df = pd.read_csv(file_path)
            
            # Initialize evaluator
            state['evaluator'] = BenchmarkEvaluator(api_key)
            
            # Process model names and columns if provided
            if input_method == "Specify models and columns":
                if not models_text.strip() or not answer_cols_text.strip():
                    return None, None, None
                
                models = [m.strip() for m in models_text.split(',')]
                answer_cols = [c.strip() for c in answer_cols_text.split(',')]
                
                if len(models) != len(answer_cols):
                    return pd.DataFrame({'Error': ['Number of models and answer columns must match']}), state['leaderboard'], None
                
                results_df, leaderboard_df = state['evaluator'].evaluate_all_models(
                    df, models=models, model_columns=answer_cols, prompt_col=prompt_column
                )
            else:
                # Auto-detect mode
                results_df, leaderboard_df = state['evaluator'].evaluate_all_models(
                    df, prompt_col=prompt_column
                )
            
            timestamp = pd.Timestamp.now().strftime('%Y%m%d_%H%M%S')
            results_path = f'results/benchmark_results_{timestamp}.csv'
            results_df.to_csv(results_path, index=False)
            
            # Update state
            state['last_results'] = results_df
            state['leaderboard'] = leaderboard_df
            
            return results_df, leaderboard_df, results_path
        except Exception as e:
            error_df = pd.DataFrame({'Error': [str(e)]})
            return error_df, state['leaderboard'], None
            
            def download_results():
                if state['last_results'] is not None:
                    timestamp = pd.Timestamp.now().strftime('%Y%m%d_%H%M%S')
                    file_path = f'results/benchmark_download_{timestamp}.csv'
                    state['last_results'].to_csv(file_path, index=False)
                    return file_path
                return None
        
        def refresh_leaderboard():
            # Reload leaderboard from file
            if os.path.exists('results/benchmark_results.csv'):
                state['leaderboard'] = pd.read_csv('results/benchmark_results.csv')
            return state['leaderboard']
        
        evaluate_btn.click(
            evaluate_batch,
            inputs=[gemini_api_key, csv_file, prompt_col, model_input_method, models_input, answer_cols_input],
            outputs=[current_results, leaderboard_table, current_results_file]
        )
        
        download_btn.click(download_results, inputs=[], outputs=[current_results_file])
        refresh_btn.click(refresh_leaderboard, inputs=[], outputs=[leaderboard_table])
        
        # Initialize the leaderboard
        leaderboard_table.value = state['leaderboard']

    return app


def main():
    app = create_gradio_interface()
    app.launch(share=True)


if __name__ == "__main__":
    main()