app.py

import warnings
import time
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 до 100:
Оригинальный запрос: {original_prompt}
Ответ: {response}
Оцените по критериям:
1. Креативность (уникальность и оригинальность ответа)
2. Разнообразие (использование разных языковых средств)
3. Релевантность (соответствие запросу)
Дайте только числовые оценки в формате:
Креативность: [число]
Разнообразие: [число]
Релевантность: [число]"""

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

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

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

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

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()
        
    def evaluate_model(self, df, model_name, prompt_col='rus_prompt'):
        """Evaluate a single model's responses"""
        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
        
        results = {
            'model': model_name,
            'creativity_score': creative_score,
            'stability_score': stability_score,
            'combined_score': combined_score,
            'creative_details': {
                'creativity': creative_df["Креативность"].mean(),
                'diversity': creative_df["Разнообразие"].mean(),
                'relevance': creative_df["Релевантность"].mean(),
            },
            'stability_details': stability_results
        }
        
        # Save detailed results
        output_file = f'evaluated_responses_{model_name}.csv'
        creative_df.to_csv(output_file, index=False)
        print(f"Detailed results saved to {output_file}")
        
        return results
    
    def evaluate_all_models(self, df, models=None, prompt_col='rus_prompt'):
        """Evaluate multiple models from the dataframe"""
        if models is None:
            # Find all columns ending with _answers
            answer_cols = [col for col in df.columns if col.endswith('_answers')]
            models = [col.replace('_answers', '') for col in answer_cols]
        
        results = []
        for model in models:
            try:
                model_results = self.evaluate_model(df, model, prompt_col)
                results.append(model_results)
                print(f"Completed evaluation for {model}")
            except Exception as e:
                print(f"Error evaluating {model}: {str(e)}")
        
        benchmark_df = pd.DataFrame(results)
        benchmark_df.to_csv('benchmark_results.csv', index=False)
        print("Benchmark completed. Results saved to benchmark_results.csv")
        
        return benchmark_df


def create_gradio_interface():
    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")
            models_input = gr.Textbox(label="Model names (comma-separated, leave blank for auto-detection)")
        
        evaluate_btn = gr.Button("Run Benchmark")
        
        with gr.Row():
            benchmark_output = gr.DataFrame(label="Benchmark Results")
        
        def evaluate_batch(api_key, file, prompt_column, models_text):
            try:
                # Load the CSV file
                file_path = file.name
                df = pd.read_csv(file_path)
                
                # Process model names if provided
                models = None
                if models_text.strip():
                    models = [m.strip() for m in models_text.split(',')]
                
                # Run the evaluation
                evaluator = BenchmarkEvaluator(api_key)
                results = evaluator.evaluate_all_models(df, models, prompt_column)
                
                return results
            except Exception as e:
                return pd.DataFrame({'Error': [str(e)]})
        
        evaluate_btn.click(
            evaluate_batch,
            inputs=[gemini_api_key, csv_file, prompt_col, models_input],
            outputs=benchmark_output
        )

    return app


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


if __name__ == "__main__":
    main()