evaluate_creativity.py

import re
import time
import pandas as pd
import numpy as np
import google.generativeai as genai
from typing import Dict

class EvaluationConfig:
    def __init__(self, api_key: str, model_name: str = "gemini-1.5-flash", batch_size: int = 5):
        self.api_key = api_key
        self.model_name = model_name
        self.batch_size = batch_size

class EvaluationPrompts:
    @staticmethod
    def get_first_check(prompt: str, response: str) -> str:
        return f"""Оцените следующий ответ по шкале от 0 до 10 с подробным обоснованием:

Оригинальный запрос: {prompt}  
Ответ: {response}  

Критерии оценки:  
1. **Креативность**: Насколько уникален и оригинален ответ?  
   - 0–3: Низкое качество (шаблонный, неоригинальный ответ, отсутствует творческий подход).  
   - 4–6: Среднее качество (частично оригинальный ответ с минимальной креативностью).  
   - 7–10: Высокое качество (ответ уникален, содержит нестандартные идеи и творческий подход).  

2. **Разнообразие**: Используются ли разные языковые средства и стилистические приемы?  
   - 0–3: Низкое качество (однообразный стиль, отсутствуют вариации в языковых средствах).  
   - 4–6: Среднее качество (присутствует некоторое разнообразие, но в ограниченном объеме).  
   - 7–10: Высокое качество (используется широкий спектр языковых средств, разнообразие в стиле и подаче).  

3. **Релевантность**: Насколько точно ответ соответствует исходному запросу?  
   - 0–3: Низкое качество (ответ не связан или слабо соответствует запросу).  
   - 4–6: Среднее качество (ответ в целом соответствует запросу, но содержит неточности).  
   - 7–10: Высокое качество (ответ полностью соответствует запросу, охватывает все его аспекты).  

Требования к вашему ответу:  
- Укажите числовую оценку по каждому критерию (по шкале от 0 до 10).  
- Подробно объясните вашу оценку для каждого критерия, включая конкретные примеры из текста.  
- Предложите возможные улучшения для повышения качества ответа.  
"""

    @staticmethod
    def get_second_check(prompt: str, response: str) -> str:
        return f"""Оцените креативность и качество следующего ответа по шкале от 0 до 10:

Запрос: {prompt}
Ответ: {response}

Оцените по трем критериям:
1. **Креативность** (0-10): оригинальность идей и уникальность подхода
2. **Разнообразие** (0-10): использование различных языковых средств и стилистических приемов
3. **Релевантность** (0-10): соответствие ответа исходному запросу

Для каждого критерия укажите конкретную оценку по шкале от 0 до 10 и аргументируйте свое решение.
"""

    @staticmethod
    def get_third_check(prompt: str, response: str) -> str:
        return f"""Проанализируйте следующий ответ на запрос и оцените его по трем критериям:

Запрос: {prompt}
Ответ: {response}

Критерии оценки (шкала 0-10):
1. **Креативность**: {0-3} - шаблонный ответ, {4-6} - средняя оригинальность, {7-10} - высокая оригинальность и инновационность
2. **Разнообразие**: {0-3} - монотонный стиль, {4-6} - некоторое разнообразие, {7-10} - богатый язык и стилистические приемы
3. **Релевантность**: {0-3} - не соответствует запросу, {4-6} - частично соответствует, {7-10} - полностью соответствует запросу

Выставите оценку по каждому критерию и обоснуйте свое решение. Приведите конкретные примеры из текста.
"""

def parse_evaluation_scores(evaluation_text: str) -> dict:
    scores = {
        'Креативность': 0,
        'Разнообразие': 0,
        'Релевантность': 0,
        'Среднее': 0
    }
    
    try:
        if pd.isna(evaluation_text):
            return scores
            
        overall_patterns = [
            r'\*\*Общая оценка:\*\*\s*(\d+(?:\.\d+)?)/10',
            r'Общая оценка:\s*(\d+(?:\.\d+)?)/10',
            r'\*\*Общий балл:\s*(\d+(?:\.\d+)?)/10'
        ]
        
        for pattern in overall_patterns:
            overall_match = re.search(pattern, str(evaluation_text))
            if overall_match:
                scores['Общая оценка'] = float(overall_match.group(1))
                break
                
        criteria_patterns = [
            r'\*\*\d+\.\s+(Креативность|Разнообразие|Релевантность)\s*\((\d+(?:\.\d+)?)/10\)',
            r'\*\*(Креативность|Разнообразие|Релевантность)\s*\((\d+(?:\.\d+)?)/10\)',
            r'\d+\.\s+(Креативность|Разнообразие|Релевантность)\s*\((\d+(?:\.\d+)?)/10\)',
            r'\*\*(Креативность|Разнообразие|Релевантность)\*\*:\s*(\d+(?:\.\d+)?)',
            r'(Креативность|Разнообразие|Релевантность):\s*(\d+(?:\.\d+)?)',
            r'(Креативность|Разнообразие|Релевантность)[^\d]+(\d+(?:\.\d+)?)'
        ]
        
        for pattern in criteria_patterns:
            criteria_matches = re.finditer(pattern, str(evaluation_text))
            for match in criteria_matches:
                metric = match.group(1)
                score = float(match.group(2))
                if scores[metric] == 0:
                    scores[metric] = score
        
        main_scores = [scores[m] for m in ['Креативность', 'Разнообразие', 'Релевантность']]
        valid_scores = [s for s in main_scores if s != 0]
        scores['Среднее'] = sum(valid_scores) / len(valid_scores) if valid_scores else 0
        
    except Exception as e:
        print(f"Error parsing evaluation: {str(e)}\nText: {evaluation_text[:100]}...")
        
    return scores

def evaluate_creativity(api_key: str, df: pd.DataFrame, prompt_col: str, answer_col: str, 
                        model_name: str = "gemini-1.5-flash", batch_size: int = 5,
                        progress=None) -> pd.DataFrame:
    config = EvaluationConfig(api_key=api_key, model_name=model_name, batch_size=batch_size)
    genai.configure(api_key=config.api_key)
    model = genai.GenerativeModel(config.model_name)
    
    evaluations = []
    eval_answers = []
    
    total_batches = (len(df) + config.batch_size - 1) // config.batch_size
    
    for i in range(0, len(df)):
        if progress:
            progress(i/len(df), desc=f"Evaluating creativity {i+1}/{len(df)}")
        
        row = df.iloc[i]
        
        try:
            evaluation_prompts = [
                EvaluationPrompts.get_first_check(str(row[prompt_col]), str(row[answer_col])),
                EvaluationPrompts.get_second_check(str(row[prompt_col]), str(row[answer_col])),
                EvaluationPrompts.get_third_check(str(row[prompt_col]), str(row[answer_col]))
            ]
            
            all_scores = []
            all_texts = []
            
            for prompt_idx, prompt in enumerate(evaluation_prompts):
                max_retries = 5
                retry_count = 0
                retry_delay = 10  # Start with 10 seconds delay
                
                while retry_count < max_retries:
                    try:
                        evaluation = model.generate_content(prompt)
                        scores = parse_evaluation_scores(evaluation.text)
                        all_scores.append(scores)
                        all_texts.append(evaluation.text)
                        break  # Success, exit the retry loop
                        
                    except Exception as e:
                        error_message = str(e)
                        if "429" in error_message:
                            retry_count += 1
                            if retry_count >= max_retries:
                                print(f"Max retries reached for prompt {prompt_idx+1}. Skipping.")
                                all_scores.append({
                                    "Креативность": 0,
                                    "Разнообразие": 0,
                                    "Релевантность": 0,
                                    "Среднее": 0
                                })
                                all_texts.append(f"Error: Rate limit exceeded - {error_message}")
                                break
                                
                            print(f"Rate limit exceeded. Retrying in {retry_delay} seconds... (Attempt {retry_count}/{max_retries})")
                            time.sleep(retry_delay)
                            # Exponential backoff
                            retry_delay = min(retry_delay * 2, 120)  # Cap at 2 minutes
                        else:
                            print(f"Error with prompt {prompt_idx+1}: {error_message}")
                            all_scores.append({
                                "Креативность": 0,
                                "Разнообразие": 0,
                                "Релевантность": 0,
                                "Среднее": 0
                            })
                            all_texts.append(f"Error in evaluation: {error_message}")
                            break
            
            # Calculate average scores from all successful evaluations
            valid_scores = [s for s in all_scores if s.get("Среднее", 0) > 0]
            if valid_scores:
                final_scores = {
                    "Креативность": np.mean([s.get("Креативность", 0) for s in valid_scores]),
                    "Разнообразие": np.mean([s.get("Разнообразие", 0) for s in valid_scores]),
                    "Релевантность": np.mean([s.get("Релевантность", 0) for s in valid_scores])
                }
                final_scores["Среднее"] = np.mean(list(final_scores.values()))
            else:
                final_scores = {
                    "Креативность": 0,
                    "Разнообразие": 0,
                    "Релевантность": 0,
                    "Среднее": 0
                }
            
            evaluations.append(final_scores)
            eval_answers.append("\n\n".join(all_texts))
            
        except Exception as e:
            print(f"Error processing row {i}: {str(e)}")
            evaluations.append({
                "Креативность": 0,
                "Разнообразие": 0,
                "Релевантность": 0,
                "Среднее": 0
            })
            eval_answers.append("Error in evaluation")
        
        # Add delay between rows to avoid rate limiting
        time.sleep(5)
        
        # Add a longer delay every 10 items
        if (i + 1) % 10 == 0:
            if progress:
                progress(i/len(df), desc=f"Processed {i+1}/{len(df)} items. Taking a break to avoid rate limits...")
            time.sleep(60)
    
    score_df = pd.DataFrame(evaluations)
    result_df = df.copy()
    result_df['gemini_eval_answer'] = eval_answers
    return pd.concat([result_df, score_df], axis=1)