app.py

import gradio as gr
import pandas as pd
import numpy as np
import re
import unicodedata
from typing import Dict, List, Tuple
import ftfy
import nltk
from bert_score import score as bert_score
from rouge_score import rouge_scorer
from nltk.translate.bleu_score import sentence_bleu, SmoothingFunction
from nltk.translate.meteor_score import meteor_score
from deepeval.test_case import LLMTestCase
from deepeval.metrics import AnswerRelevancyMetric, FaithfulnessMetric, GEval
from deepeval.models import DeepEvalBaseLLM
import google.generativeai as genai
from groq import Groq
import os
from io import StringIO

# Download required NLTK data
nltk.download('punkt', quiet=True)
nltk.download('wordnet', quiet=True)

# Configuration
GEMINI_API_KEY = "your_gemini_api_key"  # Replace with your key
GROQ_API_KEY = "your_groq_api_key"      # Replace with your key

# Initialize APIs
genai.configure(api_key=GEMINI_API_KEY)
groq_client = Groq(api_key=GROQ_API_KEY)

class LLMProvider:
    """Abstract base class for LLM providers"""
    def __init__(self, model_name: str):
        self.model_name = model_name
    
    def generate(self, prompt: str) -> str:
        raise NotImplementedError
    
    def get_model_name(self) -> str:
        return self.model_name

class GeminiProvider(LLMProvider):
    """Gemini implementation"""
    def __init__(self, model_name: str = "gemini-1.5-flash"):
        super().__init__(model_name)
        self.model = genai.GenerativeModel(model_name)
    
    def generate(self, prompt: str) -> str:
        try:
            response = self.model.generate_content(prompt)
            return response.text.strip()
        except Exception as e:
            return f"Error generating with Gemini: {str(e)}"

class GroqProvider(LLMProvider):
    """Groq implementation for LLaMA models"""
    def __init__(self, model_name: str = "llama3-70b-8192"):
        super().__init__(model_name)
    
    def generate(self, prompt: str) -> str:
        try:
            chat_completion = groq_client.chat.completions.create(
                messages=[
                    {"role": "user", "content": prompt}
                ],
                model=self.model_name,
                temperature=0.7,
                max_tokens=2048
            )
            return chat_completion.choices[0].message.content.strip()
        except Exception as e:
            return f"Error generating with Groq: {str(e)}"

class DeepEvalLLMWrapper(DeepEvalBaseLLM):
    """Wrapper for DeepEval to work with our providers"""
    def __init__(self, provider: LLMProvider):
        self.provider = provider
    
    def load_model(self):
        return self.provider
    
    def generate(self, prompt: str) -> str:
        return self.provider.generate(prompt)
    
    def get_model_name(self) -> str:
        return self.provider.get_model_name()

def clean_text(text: str) -> str:
    """Clean text by fixing encoding and normalizing"""
    if not text or not isinstance(text, str):
        return ""
    
    # Fix encoding artifacts
    text = ftfy.fix_text(text)
    text = unicodedata.normalize('NFKD', text)
    
    # Fix quotes and other common issues
    text = text.replace('“', '"').replace('â€', '"')
    text = text.replace('–', '-').replace('—', '-')
    text = text.replace('‘', "'").replace('’', "'")
    
    # Remove non-ASCII characters
    text = re.sub(r'[^\x00-\x7F]+', ' ', text)
    
    # Normalize whitespace
    text = ' '.join(text.split())
    
    return text.strip()

def evaluate_metrics(input_text: str, candidate_text: str, reference_text: str) -> Dict:
    """Run comprehensive evaluation on the generated text"""
    
    # Clean the texts
    cleaned_input = clean_text(input_text)
    cleaned_candidate = clean_text(candidate_text)
    cleaned_reference = clean_text(reference_text)
    
    results = {}
    
    # Traditional metrics
    try:
        # BLEU Score
        smooth = SmoothingFunction().method4
        bleu_score = sentence_bleu(
            [cleaned_reference.split()], 
            cleaned_candidate.split(), 
            smoothing_function=smooth
        )
        results["BLEU"] = bleu_score
        
        # ROUGE Score
        rouge_scorer_obj = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)
        rouge_scores = rouge_scorer_obj.score(cleaned_reference, cleaned_candidate)
        rouge_avg = (rouge_scores['rouge1'].fmeasure + 
                    rouge_scores['rouge2'].fmeasure + 
                    rouge_scores['rougeL'].fmeasure) / 3
        results["ROUGE"] = rouge_avg
        
        # METEOR Score
        meteor = meteor_score([cleaned_reference.split()], cleaned_candidate.split())
        results["METEOR"] = meteor
        
        # BERT Score
        P, R, F1 = bert_score([cleaned_candidate], [cleaned_reference], lang="en", verbose=False)
        results["BERTScore"] = F1.item()
        
    except Exception as e:
        results["Error"] = f"Traditional metrics error: {str(e)}"
    
    # LLM-as-judge metrics (using Gemini for consistency)
    try:
        judge_provider = GeminiProvider("gemini-1.5-flash")
        judge_wrapper = DeepEvalLLMWrapper(judge_provider)
        
        test_case = LLMTestCase(
            input=cleaned_input,
            actual_output=cleaned_candidate,
            expected_output=cleaned_reference
        )
        
        # Answer Relevancy
        answer_rel = AnswerRelevancyMetric(model=judge_wrapper)
        answer_rel.measure(test_case)
        results["AnswerRelevancy"] = answer_rel.score
        
        # Faithfulness
        faith = FaithfulnessMetric(model=judge_wrapper)
        faith.measure(test_case)
        results["Faithfulness"] = faith.score
        
        # GEval
        geval = GEval(
            name="OverallQuality",
            criteria="Evaluate if the candidate response is accurate, complete, and well-written.",
            evaluation_params=[
                "input", "actual_output", "expected_output"
            ],
            model=judge_wrapper
        )
        geval.measure(test_case)
        results["GEval"] = geval.score
        
    except Exception as e:
        results["LLM_Judge_Error"] = f"LLM-as-judge metrics error: {str(e)}"
    
    # Normalization and Hybrid Score
    normalization_ranges = {
        "AnswerRelevancy": (0.0, 1.0),
        "Faithfulness": (0.0, 1.0),
        "GEval": (0.0, 1.0),
        "BERTScore": (0.7, 0.95),
        "ROUGE": (0.0, 0.6),
        "BLEU": (0.0, 0.4),
        "METEOR": (0.0, 0.6)
    }
    
    weights = {
        "AnswerRelevancy": 0.10,
        "Faithfulness": 0.10,
        "GEval": 0.025,
        "BERTScore": 0.20,
        "ROUGE": 0.15,
        "BLEU": 0.025,
        "METEOR": 0.15
    }
    
    # Normalize scores
    normalized_scores = {}
    for metric, value in results.items():
        if metric in normalization_ranges and isinstance(value, (int, float)):
            min_v, max_v = normalization_ranges[metric]
            if max_v > min_v:  # Avoid division by zero
                norm = max(min((value - min_v) / (max_v - min_v), 1.0), 0.0)
                normalized_scores[metric] = norm
            else:
                normalized_scores[metric] = 0.5
        elif isinstance(value, (int, float)):
            normalized_scores[metric] = value
    
    # Calculate weighted average
    if normalized_scores:
        weighted_sum = sum(normalized_scores.get(m, 0) * w for m, w in weights.items())
        total_weight = sum(w for m, w in weights.items() if m in normalized_scores)
        results["WeightedAverage"] = weighted_sum / total_weight if total_weight > 0 else 0.0
    else:
        results["WeightedAverage"] = 0.0
    
    return results

def process_single_text(input_text: str, model_choice: str) -> Tuple[str, str, Dict]:
    """Process a single text input"""
    if not input_text or len(input_text.strip()) < 10:
        return "", "", {"Error": "Input text too short"}
    
    # Choose model
    if model_choice == "Gemini":
        provider = GeminiProvider("gemini-1.5-flash")
    elif model_choice == "LLaMA-3-70b":
        provider = GroqProvider("llama3-70b-8192")
    else:  # LLaMA-3-8b
        provider = GroqProvider("llama3-8b-8192")
    
    # Generate candidate
    prompt = f"""Rewrite the following paragraph in a fresh, concise, and professional style while preserving its full meaning and key information:

{input_text}

Provide only the rewritten text without any additional commentary."""
    
    candidate = provider.generate(prompt)
    
    # Use cleaned input as reference (simulating human-quality standard)
    reference = clean_text(input_text)
    
    # Evaluate
    scores = evaluate_metrics(input_text, candidate, reference)
    
    return candidate, reference, scores

def process_file(file_obj, model_choice: str) -> Tuple[pd.DataFrame, str]:
    """Process a CSV file with multiple articles"""
    try:
        # Read the file
        content = file_obj.read().decode('utf-8')
        df = pd.read_csv(StringIO(content))
        
        # Assume first column is the text
        text_column = df.columns[0]
        
        results = []
        
        for idx, row in df.iterrows():
            text = str(row[text_column])
            candidate, reference, scores = process_single_text(text, model_choice)
            
            result_row = {
                'Original_Text': text,
                'Generated_Candidate': candidate,
                'Reference_Text': reference
            }
            result_row.update(scores)
            results.append(result_row)
        
        results_df = pd.DataFrame(results)
        return results_df, "File processed successfully!"
        
    except Exception as e:
        return pd.DataFrame(), f"Error processing file: {str(e)}"

def create_gradio_interface():
    """Create the Gradio interface"""
    
    with gr.Blocks(title="LLM Evaluation Framework") as demo:
        gr.Markdown("# 📊 LLM Evaluation Framework for Professional Content Rewriting")
        gr.Markdown("Evaluate and compare LLM-generated content using multiple metrics. Choose between Gemini and LLaMA models.")
        
        with gr.Tabs():
            with gr.Tab("Single Text Processing"):
                with gr.Row():
                    with gr.Column(scale=2):
                        input_text = gr.Textbox(
                            label="Input Text",
                            placeholder="Enter the text you want to rewrite...",
                            lines=10
                        )
                        
                        model_choice_single = gr.Radio(
                            ["Gemini", "LLaMA-3-70b", "LLaMA-3-8b"],
                            label="Choose Model",
                            value="Gemini"
                        )
                        
                        submit_btn = gr.Button("Generate & Evaluate", variant="primary")
                    
                    with gr.Column(scale=3):
                        gr.Markdown("### Results")
                        
                        with gr.Tabs():
                            with gr.Tab("Generated Text"):
                                candidate_output = gr.Textbox(
                                    label="Generated Candidate",
                                    lines=10,
                                    show_copy_button=True
                                )
                                reference_output = gr.Textbox(
                                    label="Reference Text (Cleaned Input)",
                                    lines=5,
                                    show_copy_button=True
                                )
                            
                            with gr.Tab("Evaluation Scores"):
                                scores_output = gr.JSON(label="Detailed Scores")
                                
                                weighted_avg = gr.Number(
                                    label="Weighted Average Score (0-1)",
                                    precision=4
                                )
                                
                                interpretation = gr.Textbox(
                                    label="Interpretation",
                                    interactive=False
                                )
        
            with gr.Tab("Batch Processing (CSV File)"):
                with gr.Row():
                    with gr.Column(scale=1):
                        file_input = gr.File(
                            label="Upload CSV File",
                            file_types=['.csv']
                        )
                        
                        model_choice_file = gr.Radio(
                            ["Gemini", "LLaMA-3-70b", "LLaMA-3-8b"],
                            label="Choose Model for Batch Processing",
                            value="Gemini"
                        )
                        
                        process_file_btn = gr.Button("Process File", variant="primary")
                    
                    with gr.Column(scale=2):
                        gr.Markdown("### Results")
                        file_results = gr.Dataframe(
                            label="Evaluation Results",
                            interactive=False
                        )
                        file_status = gr.Textbox(label="Status")
        
        # Examples
        gr.Examples(
            examples=[
                ["The immune system plays a crucial role in protecting the human body from pathogens such as bacteria, viruses, and other harmful invaders. It is composed of innate and adaptive components that work together to detect and eliminate foreign threats.", "Gemini"],
                ["Climate change is one of the most pressing challenges facing humanity today. Rising global temperatures have led to severe weather patterns, including more intense storms, droughts, and heatwaves.", "LLaMA-3-70b"]
            ],
            inputs=[input_text, model_choice_single],
            outputs=[candidate_output, reference_output, scores_output, weighted_avg, interpretation]
        )
        
        # Event handlers
        def handle_single_process(text, model):
            if not text:
                return "", "", {}, 0, "Please enter some text."
            
            candidate, reference, scores = process_single_text(text, model)
            
            # Get weighted average
            weighted_avg_val = scores.get("WeightedAverage", 0)
            
            # Interpretation
            if weighted_avg_val >= 0.85:
                interpretation_text = "✅ Outstanding performance (A) - ready for professional use"
            elif weighted_avg_val >= 0.70:
                interpretation_text = "✅ Strong performance (B) - good quality with minor improvements"
            elif weighted_avg_val >= 0.50:
                interpretation_text = "⚠️ Adequate performance (C) - usable but needs refinement"
            elif weighted_avg_val >= 0.30:
                interpretation_text = "❌ Weak performance (D) - requires significant revision"
            else:
                interpretation_text = "❌ Poor performance (F) - likely needs complete rewriting"
            
            return candidate, reference, scores, weighted_avg_val, interpretation_text
        
        def handle_file_process(file, model):
            if file is None:
                return pd.DataFrame(), "Please upload a file."
            return process_file(file, model)
        
        submit_btn.click(
            fn=handle_single_process,
            inputs=[input_text, model_choice_single],
            outputs=[candidate_output, reference_output, scores_output, weighted_avg, interpretation]
        )
        
        process_file_btn.click(
            fn=handle_file_process,
            inputs=[file_input, model_choice_file],
            outputs=[file_results, file_status]
        )
        
        gr.Markdown("""
        ## 📝 How to Use
        
        1. **Single Text Processing**: Enter your text and choose a model to generate a professional rewrite.
        2. **Batch Processing**: Upload a CSV file with one article per row in the first column.
        3. **Model Options**:
           - **Gemini**: Google's advanced language model
           - **LLaMA-3-70b**: Large Meta model (70B parameters)
           - **LLaMA-3-8b**: Smaller Meta model (8B parameters)
        
        ## 📊 Evaluation Metrics
        
        The system evaluates performance using multiple metrics:
        - **Traditional**: BLEU, ROUGE, METEOR (n-gram overlap)
        - **Semantic**: BERTScore (embedding similarity)
        - **LLM-as-Judge**: AnswerRelevancy, Faithfulness, GEval
        - **Final Score**: Weighted average of all metrics (0-1 scale)
        """)
    
    return demo

# Launch the app
if __name__ == "__main__":
    app = create_gradio_interface()
    app.launch(share=True)