xbe.at

Miscellany / AI applications and systems

Ensuring Useful AI-Generated Summaries

17 min read, 14 sections

Contents
  1. Implementing G-Eval Framework Base
  2. Coherence Evaluation Module
  3. Consistency Checker Implementation
  4. Fluency Analysis Engine
  5. Relevance Scoring System
  6. Chain-of-Thought Implementation
  7. Probability Weighting System
  8. Real-world Application: News Article Summarization
  9. Results Visualization and Reporting
  10. Quality Score Benchmarking
  11. Performance Optimization Module
  12. Automated Testing and Validation
  13. Integration Framework
  14. Additional Resources

Implementing G-Eval Framework Base

The G-Eval framework requires a robust foundation to evaluate AI-generated summaries across multiple dimensions. This implementation creates the core evaluation class with methods to handle basic scoring functionality and metric calculations.

import numpy as np
from typing import List, Dict, Union

class GEvalBase:
    def __init__(self, source_text: str, summary_text: str):
        self.source = source_text
        self.summary = summary_text
        self.metrics = {
            'coherence': {'weight': 0.3, 'score': 0},
            'consistency': {'weight': 0.3, 'score': 0},
            'fluency': {'weight': 0.2, 'score': 0},
            'relevance': {'weight': 0.2, 'score': 0}
        }
    
    def calculate_weighted_score(self) -> float:
        return sum(metric['weight'] * metric['score'] 
                  for metric in self.metrics.values())
    
    def validate_score_range(self, score: float, 
                           metric: str) -> bool:
        ranges = {
            'fluency': (1, 3),
            'default': (1, 5)
        }
        min_val, max_val = ranges.get(metric, ranges['default'])
        return min_val <= score <= max_val

# Example usage
source = "Original document text here..."
summary = "Summary to evaluate..."
evaluator = GEvalBase(source, summary)

Coherence Evaluation Module

The coherence evaluation component analyzes the logical flow and structural integrity of summaries using natural language processing techniques and custom scoring algorithms that consider sentence transitions and topic consistency.

from nltk import sent_tokenize
import numpy as np

class CoherenceEvaluator:
    def __init__(self):
        self.transition_weights = {
            'perfect': 1.0,
            'good': 0.8,
            'moderate': 0.6,
            'poor': 0.3
        }
    
    def evaluate_transitions(self, text: str) -> float:
        sentences = sent_tokenize(text)
        if len(sentences) < 2:
            return self.transition_weights['moderate']
            
        transition_scores = []
        for i in range(len(sentences) - 1):
            current = sentences[i].lower()
            next_sent = sentences[i + 1].lower()
            
            # Simple transition analysis
            common_words = len(set(current.split()) & 
                             set(next_sent.split()))
            score = min(common_words / len(set(current.split())), 1.0)
            transition_scores.append(score)
            
        return np.mean(transition_scores)

# Example usage
evaluator = CoherenceEvaluator()
text = "This is a test summary. It contains multiple sentences."
coherence_score = evaluator.evaluate_transitions(text)
print(f"Coherence Score: {coherence_score:.2f}")

Consistency Checker Implementation

A critical component that validates factual consistency between source and summary texts using semantic similarity measurements and fact extraction techniques to identify potential hallucinations or misrepresentations.

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import spacy

class ConsistencyChecker:
    def __init__(self):
        self.nlp = spacy.load('en_core_web_sm')
        self.vectorizer = TfidfVectorizer(stop_words='english')
        
    def extract_facts(self, text: str) -> List[str]:
        doc = self.nlp(text)
        facts = []
        for sent in doc.sents:
            if self._is_factual_statement(sent):
                facts.append(sent.text)
        return facts
    
    def _is_factual_statement(self, sent) -> bool:
        return any(token.dep_ in ['nsubj', 'dobj'] 
                  for token in sent)
    
    def calculate_consistency(self, source: str, 
                            summary: str) -> float:
        source_facts = self.extract_facts(source)
        summary_facts = self.extract_facts(summary)
        
        if not summary_facts:
            return 0.0
            
        vectors = self.vectorizer.fit_transform(
            source_facts + summary_facts)
        similarity_matrix = cosine_similarity(vectors)
        
        # Calculate average similarity of summary facts 
        # to source facts
        n_source = len(source_facts)
        n_summary = len(summary_facts)
        fact_similarities = similarity_matrix[
            n_source:, :n_source]
        
        return float(np.mean(np.max(fact_similarities, axis=1)))

# Example usage
checker = ConsistencyChecker()
source = "The company reported 20% growth in Q2 2023."
summary = "The company saw significant growth in Q2 2023."
consistency_score = checker.calculate_consistency(source, summary)
print(f"Consistency Score: {consistency_score:.2f}")

Fluency Analysis Engine

This component evaluates the linguistic quality of summaries by analyzing grammar, punctuation, and natural language flow using advanced NLP techniques and statistical measures of text quality.

import language_tool_python
from textblob import TextBlob
import re

class FluencyAnalyzer:
    def __init__(self):
        self.tool = language_tool_python.LanguageTool('en-US')
        
    def analyze_fluency(self, text: str) -> Dict[str, float]:
        # Grammar check
        grammar_errors = len(self.tool.check(text))
        
        # Readability analysis
        blob = TextBlob(text)
        sentiment_polarity = abs(blob.sentiment.polarity)
        
        # Sentence structure variety
        sentences = [str(sent) for sent in blob.sentences]
        sent_lengths = [len(str(sent).split()) 
                       for sent in blob.sentences]
        length_variety = np.std(sent_lengths)
        
        # Calculate normalized scores
        grammar_score = max(0, 1 - (grammar_errors / 
                                  len(text.split())))
        structure_score = min(1, length_variety / 10)
        
        return {
            'grammar': grammar_score,
            'structure': structure_score,
            'final_score': (grammar_score * 0.6 + 
                          structure_score * 0.4)
        }

# Example usage
analyzer = FluencyAnalyzer()
text = "This is a well-written summary. It flows naturally."
scores = analyzer.analyze_fluency(text)
print(f"Fluency Scores: {scores}")

Relevance Scoring System

The relevance evaluation system implements sophisticated algorithms to measure how well a summary captures essential information from the source text while maintaining appropriate information density and coverage.

from sklearn.feature_extraction.text import CountVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

class RelevanceScorer:
    def __init__(self):
        self.vectorizer = CountVectorizer(
            ngram_range=(1, 2),
            stop_words='english'
        )
        
    def calculate_relevance(self, source: str, 
                           summary: str) -> float:
        # Extract key phrases
        key_phrases = self._extract_key_phrases(source)
        
        # Calculate coverage
        coverage_score = self._calculate_coverage(
            key_phrases, summary)
        
        # Calculate information density
        density_score = self._calculate_density(summary)
        
        # Combined weighted score
        return 0.7 * coverage_score + 0.3 * density_score
    
    def _extract_key_phrases(self, text: str) -> List[str]:
        # Simple key phrase extraction based on TF-IDF
        vectorizer = TfidfVectorizer(ngram_range=(1, 2))
        tfidf_matrix = vectorizer.fit_transform([text])
        feature_names = vectorizer.get_feature_names_out()
        
        # Get top scoring phrases
        scores = zip(feature_names,
                    np.asarray(tfidf_matrix.sum(axis=0)
                    ).ravel())
        sorted_scores = sorted(scores, 
                             key=lambda x: x[1], 
                             reverse=True)
        return [phrase for phrase, score 
                in sorted_scores[:10]]
    
    def _calculate_coverage(self, key_phrases: List[str], 
                          summary: str) -> float:
        covered = sum(1 for phrase in key_phrases 
                     if phrase.lower() in summary.lower())
        return covered / len(key_phrases)
    
    def _calculate_density(self, summary: str) -> float:
        words = summary.split()
        unique_words = len(set(words))
        return min(1.0, unique_words / len(words))

# Example usage
scorer = RelevanceScorer()
source = "Long source text with important information..."
summary = "Concise summary capturing key points..."
relevance_score = scorer.calculate_relevance(source, summary)
print(f"Relevance Score: {relevance_score:.2f}")

Chain-of-Thought Implementation

The Chain-of-Thought (CoT) module implements a step-by-step reasoning process to guide the evaluation process, ensuring thorough and systematic assessment of each dimension through structured prompting and analysis.

from dataclasses import dataclass
from typing import List, Dict, Optional

@dataclass
class ThoughtStep:
    reasoning: str
    score: float
    confidence: float

class ChainOfThought:
    def __init__(self):
        self.thought_chain: List[ThoughtStep] = []
        self.final_score: Optional[float] = None
        
    def add_thought(self, reasoning: str, 
                   score: float, 
                   confidence: float = 1.0) -> None:
        self.thought_chain.append(
            ThoughtStep(reasoning, score, confidence))
    
    def analyze_dimension(self, 
                         text: str, 
                         dimension: str) -> float:
        self.thought_chain.clear()
        
        # Initial assessment
        self.add_thought(
            f"Analyzing {dimension} - Initial reading",
            self._initial_score(text),
            0.7
        )
        
        # Detailed analysis
        detailed_score = self._detailed_analysis(
            text, dimension)
        self.add_thought(
            "Detailed analysis completed",
            detailed_score,
            0.9
        )
        
        # Calculate weighted final score
        self.final_score = self._calculate_final_score()
        return self.final_score
    
    def _initial_score(self, text: str) -> float:
        # Basic scoring based on text characteristics
        return min(5.0, len(text.split()) / 100)
    
    def _detailed_analysis(self, 
                          text: str, 
                          dimension: str) -> float:
        # Implement dimension-specific analysis
        analysis_methods = {
            'coherence': self._analyze_coherence,
            'consistency': self._analyze_consistency,
            'fluency': self._analyze_fluency,
            'relevance': self._analyze_relevance
        }
        return analysis_methods.get(
            dimension, self._default_analysis)(text)
    
    def _calculate_final_score(self) -> float:
        if not self.thought_chain:
            return 0.0
        
        weighted_scores = [
            step.score * step.confidence 
            for step in self.thought_chain
        ]
        weights = [step.confidence 
                  for step in self.thought_chain]
        
        return sum(weighted_scores) / sum(weights)

# Example usage
cot = ChainOfThought()
text = "Sample text for analysis..."
score = cot.analyze_dimension(text, "coherence")
print(f"Final Score: {score:.2f}")
print("Thought Chain:")
for step in cot.thought_chain:
    print(f"- {step.reasoning}: {step.score:.2f} "
          f"(confidence: {step.confidence:.2f})")

Probability Weighting System

This sophisticated scoring mechanism implements probability-based weighting to refine evaluation scores, accounting for uncertainty and confidence levels in different aspects of the assessment process.

import numpy as np
from scipy.stats import beta
from typing import List, Tuple

class ProbabilityWeighting:
    def __init__(self):
        self.confidence_threshold = 0.85
        self.min_samples = 3
        
    def calculate_weighted_score(self, 
                               scores: List[float],
                               confidences: List[float]) -> float:
        if len(scores) < self.min_samples:
            return np.mean(scores)
            
        # Convert scores to beta distribution parameters
        alpha, beta_param = self._scores_to_beta_params(
            scores, confidences)
        
        # Calculate weighted mean using beta distribution
        weighted_mean = alpha / (alpha + beta_param)
        
        # Adjust based on confidence
        confidence = self._calculate_confidence(
            alpha, beta_param)
        
        if confidence < self.confidence_threshold:
            # Blend with simple mean if confidence is low
            simple_mean = np.mean(scores)
            blend_factor = confidence / self.confidence_threshold
            return (blend_factor * weighted_mean + 
                   (1 - blend_factor) * simple_mean)
            
        return weighted_mean
    
    def _scores_to_beta_params(self, 
                             scores: List[float],
                             confidences: List[float]) -> Tuple[float, float]:
        # Normalize scores to [0,1] range
        normalized_scores = np.array(scores) / 5.0
        
        # Calculate weighted mean and variance
        weights = np.array(confidences)
        weighted_mean = np.average(
            normalized_scores, weights=weights)
        weighted_var = np.average(
            (normalized_scores - weighted_mean) ** 2, 
            weights=weights)
        
        # Convert to beta parameters
        alpha = weighted_mean * (
            (weighted_mean * (1 - weighted_mean) / 
             weighted_var) - 1)
        beta_param = (1 - weighted_mean) * (
            (weighted_mean * (1 - weighted_mean) / 
             weighted_var) - 1)
        
        return max(0.01, alpha), max(0.01, beta_param)
    
    def _calculate_confidence(self, 
                            alpha: float, 
                            beta_param: float) -> float:
        # Calculate confidence based on distribution width
        variance = (alpha * beta_param) / (
            (alpha + beta_param) ** 2 * 
            (alpha + beta_param + 1))
        return 1 - min(1, np.sqrt(variance) * 4)

# Example usage
weighter = ProbabilityWeighting()
scores = [4.2, 3.8, 4.5, 4.0]
confidences = [0.9, 0.8, 0.95, 0.85]
final_score = weighter.calculate_weighted_score(
    scores, confidences)
print(f"Final Weighted Score: {final_score:.2f}")

Real-world Application: News Article Summarization

This implementation demonstrates the practical application of G-Eval for evaluating AI-generated news article summaries, including preprocessing, evaluation, and detailed scoring across all dimensions.

import pandas as pd
from typing import Dict, Any
from datetime import datetime

class NewsArticleEvaluator:
    def __init__(self):
        self.coherence_eval = CoherenceEvaluator()
        self.consistency_check = ConsistencyChecker()
        self.fluency_analyzer = FluencyAnalyzer()
        self.relevance_scorer = RelevanceScorer()
        self.cot = ChainOfThought()
        self.prob_weighter = ProbabilityWeighting()
        
    def evaluate_summary(self, 
                        article: str, 
                        summary: str) -> Dict[str, Any]:
        # Preprocess texts
        cleaned_article = self._preprocess_text(article)
        cleaned_summary = self._preprocess_text(summary)
        
        # Evaluate all dimensions
        scores = {
            'coherence': self.coherence_eval.evaluate_transitions(
                cleaned_summary),
            'consistency': self.consistency_check.calculate_consistency(
                cleaned_article, cleaned_summary),
            'fluency': self.fluency_analyzer.analyze_fluency(
                cleaned_summary)['final_score'],
            'relevance': self.relevance_scorer.calculate_relevance(
                cleaned_article, cleaned_summary)
        }
        
        # Apply CoT reasoning
        cot_scores = []
        confidences = []
        for dimension, score in scores.items():
            cot_score = self.cot.analyze_dimension(
                cleaned_summary, dimension)
            cot_scores.append(cot_score)
            confidences.append(
                self.cot.thought_chain[-1].confidence)
        
        # Calculate final weighted score
        final_score = self.prob_weighter.calculate_weighted_score(
            cot_scores, confidences)
        
        return {
            'dimension_scores': scores,
            'cot_analysis': self.cot.thought_chain,
            'final_score': final_score,
            'timestamp': datetime.now().isoformat(),
            'summary_length': len(cleaned_summary.split()),
            'compression_ratio': len(cleaned_summary.split()) / 
                               len(cleaned_article.split())
        }
    
    def _preprocess_text(self, text: str) -> str:
        # Basic preprocessing
        text = text.strip()
        text = re.sub(r'\s+', ' ', text)
        text = re.sub(r'[^\w\s.,!?-]', '', text)
        return text

# Example usage
evaluator = NewsArticleEvaluator()
article = """Long news article text..."""
summary = """AI-generated summary..."""
results = evaluator.evaluate_summary(article, summary)
print(f"Final Score: {results['final_score']:.2f}")
print("\nDimension Scores:")
for dim, score in results['dimension_scores'].items():
    print(f"{dim}: {score:.2f}")

Results Visualization and Reporting

A comprehensive visualization and reporting system that transforms G-Eval metrics into actionable insights through interactive charts, comparative analysis, and detailed performance breakdowns.

import matplotlib.pyplot as plt
import seaborn as sns
from typing import List, Dict
import pandas as pd

class GEvalVisualizer:
    def __init__(self):
        plt.style.use('seaborn')
        self.dimension_colors = {
            'coherence': '#2ecc71',
            'consistency': '#3498db',
            'fluency': '#e74c3c',
            'relevance': '#f1c40f'
        }
        
    def create_summary_report(self, 
                            eval_results: Dict) -> None:
        fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(
            2, 2, figsize=(15, 12))
        
        # Dimension scores radar chart
        self._plot_radar_chart(
            eval_results['dimension_scores'], ax1)
        
        # Score distribution histogram
        self._plot_score_distribution(
            eval_results['cot_analysis'], ax2)
        
        # Confidence levels
        self._plot_confidence_levels(
            eval_results['cot_analysis'], ax3)
        
        # Timeline of scores
        self._plot_score_timeline(
            eval_results['dimension_scores'], ax4)
        
        plt.tight_layout()
        return fig
    
    def _plot_radar_chart(self, 
                         scores: Dict[str, float], 
                         ax: plt.Axes) -> None:
        categories = list(scores.keys())
        values = list(scores.values())
        
        # Create radar chart
        angles = np.linspace(0, 2*np.pi, len(categories), 
                           endpoint=False)
        values = np.concatenate((values, [values[0]]))
        angles = np.concatenate((angles, [angles[0]]))
        
        ax.plot(angles, values)
        ax.fill(angles, values, alpha=0.25)
        ax.set_xticks(angles[:-1])
        ax.set_xticklabels(categories)
        ax.set_title('Dimension Scores Analysis')
    
    def _plot_score_distribution(self, 
                               cot_analysis: List, 
                               ax: plt.Axes) -> None:
        scores = [step.score for step in cot_analysis]
        sns.histplot(scores, ax=ax, bins=10)
        ax.set_title('Score Distribution')
        ax.set_xlabel('Score')
        ax.set_ylabel('Frequency')
    
    def _plot_confidence_levels(self, 
                              cot_analysis: List, 
                              ax: plt.Axes) -> None:
        confidences = [step.confidence 
                      for step in cot_analysis]
        reasoning = [step.reasoning 
                    for step in cot_analysis]
        
        ax.barh(reasoning, confidences)
        ax.set_title('Confidence Levels by Analysis Step')
        ax.set_xlabel('Confidence')
    
    def _plot_score_timeline(self, 
                           scores: Dict[str, float], 
                           ax: plt.Axes) -> None:
        dimensions = list(scores.keys())
        values = list(scores.values())
        
        ax.plot(dimensions, values, marker='o')
        ax.set_title('Score Timeline')
        ax.set_xlabel('Dimension')
        ax.set_ylabel('Score')
        plt.xticks(rotation=45)

# Example usage
visualizer = GEvalVisualizer()
eval_results = {
    'dimension_scores': {
        'coherence': 4.2,
        'consistency': 3.8,
        'fluency': 2.5,
        'relevance': 4.0
    },
    'cot_analysis': [
        ThoughtStep("Initial analysis", 3.5, 0.7),
        ThoughtStep("Detailed review", 4.0, 0.9),
        ThoughtStep("Final assessment", 4.2, 0.95)
    ]
}
fig = visualizer.create_summary_report(eval_results)
plt.show()

Quality Score Benchmarking

The benchmarking module implements sophisticated comparison mechanisms to evaluate summary quality against established standards and peer performances, providing contextual scoring and performance metrics.

from sklearn.preprocessing import MinMaxScaler
import numpy as np
from typing import List, Dict, Optional

class QualityBenchmarker:
    def __init__(self):
        self.benchmark_scores = {
            'coherence': {'min': 3.0, 'max': 4.5, 
                         'target': 4.0},
            'consistency': {'min': 3.5, 'max': 4.8, 
                          'target': 4.2},
            'fluency': {'min': 2.0, 'max': 3.0, 
                       'target': 2.5},
            'relevance': {'min': 3.2, 'max': 4.6, 
                         'target': 4.0}
        }
        self.scaler = MinMaxScaler()
        
    def benchmark_summary(self, 
                         scores: Dict[str, float],
                         peer_scores: Optional[
                             List[Dict[str, float]]] = None
                         ) -> Dict[str, Any]:
        # Calculate normalized scores
        normalized_scores = self._normalize_scores(scores)
        
        # Compare against benchmarks
        benchmark_comparison = self._compare_to_benchmarks(
            normalized_scores)
        
        # Calculate peer percentiles if available
        peer_analysis = None
        if peer_scores:
            peer_analysis = self._analyze_peer_comparison(
                scores, peer_scores)
        
        # Calculate quality index
        quality_index = self._calculate_quality_index(
            normalized_scores, benchmark_comparison)
        
        return {
            'normalized_scores': normalized_scores,
            'benchmark_comparison': benchmark_comparison,
            'peer_analysis': peer_analysis,
            'quality_index': quality_index,
            'recommendations': self._generate_recommendations(
                benchmark_comparison)
        }
    
    def _normalize_scores(self, 
                         scores: Dict[str, float]
                         ) -> Dict[str, float]:
        normalized = {}
        for dimension, score in scores.items():
            benchmark = self.benchmark_scores[dimension]
            normalized[dimension] = (score - benchmark['min']) / (
                benchmark['max'] - benchmark['min'])
        return normalized
    
    def _compare_to_benchmarks(self, 
                             normalized_scores: Dict[str, float]
                             ) -> Dict[str, Dict[str, float]]:
        comparison = {}
        for dimension, score in normalized_scores.items():
            benchmark = self.benchmark_scores[dimension]
            target_normalized = (
                benchmark['target'] - benchmark['min']) / (
                benchmark['max'] - benchmark['min'])
            
            comparison[dimension] = {
                'score': score,
                'target': target_normalized,
                'gap': target_normalized - score
            }
        return comparison
    
    def _analyze_peer_comparison(self, 
                               scores: Dict[str, float],
                               peer_scores: List[Dict[str, float]]
                               ) -> Dict[str, float]:
        percentiles = {}
        for dimension in scores.keys():
            peer_values = [ps[dimension] 
                          for ps in peer_scores]
            percentile = percentileofscore(
                peer_values, scores[dimension])
            percentiles[dimension] = percentile
        return percentiles
    
    def _calculate_quality_index(self, 
                               normalized_scores: Dict[str, float],
                               benchmark_comparison: Dict[
                                   str, Dict[str, float]]
                               ) -> float:
        # Weighted average based on benchmark gaps
        weights = {
            'coherence': 0.3,
            'consistency': 0.3,
            'fluency': 0.2,
            'relevance': 0.2
        }
        
        weighted_scores = []
        total_weight = 0
        
        for dimension, score in normalized_scores.items():
            gap = benchmark_comparison[dimension]['gap']
            weight = weights[dimension] * (1 - abs(gap))
            weighted_scores.append(score * weight)
            total_weight += weight
        
        return sum(weighted_scores) / total_weight

# Example usage
benchmarker = QualityBenchmarker()
scores = {
    'coherence': 4.1,
    'consistency': 3.9,
    'fluency': 2.4,
    'relevance': 3.8
}
peer_scores = [
    {'coherence': 3.8, 'consistency': 4.0, 
     'fluency': 2.3, 'relevance': 3.9},
    {'coherence': 4.2, 'consistency': 3.7, 
     'fluency': 2.5, 'relevance': 4.0}
]
results = benchmarker.benchmark_summary(scores, peer_scores)
print(f"Quality Index: {results['quality_index']:.2f}")

Performance Optimization Module

This module implements advanced optimization techniques to enhance G-Eval's processing efficiency and scoring accuracy through parallel processing, caching mechanisms, and adaptive threshold adjustments.

from functools import lru_cache
import multiprocessing as mp
from typing import List, Dict, Tuple
import numpy as np

class PerformanceOptimizer:
    def __init__(self, n_processes: int = None):
        self.n_processes = n_processes or mp.cpu_count()
        self.cache_size = 1024
        self.score_history = []
        self.threshold_adjustments = {
            'coherence': 1.0,
            'consistency': 1.0,
            'fluency': 1.0,
            'relevance': 1.0
        }
        
    @lru_cache(maxsize=1024)
    def optimize_evaluation(self, 
                          text: str, 
                          dimension: str) -> float:
        # Parallel processing for large texts
        if len(text.split()) > 1000:
            return self._parallel_process(text, dimension)
        return self._single_process(text, dimension)
    
    def _parallel_process(self, 
                         text: str, 
                         dimension: str) -> float:
        # Split text into chunks for parallel processing
        chunks = self._split_text(text)
        
        with mp.Pool(self.n_processes) as pool:
            results = pool.starmap(
                self._process_chunk,
                [(chunk, dimension) for chunk in chunks]
            )
        
        # Combine results using weighted average
        return self._combine_results(results)
    
    def _split_text(self, text: str) -> List[str]:
        words = text.split()
        chunk_size = max(100, len(words) // self.n_processes)
        return [' '.join(words[i:i + chunk_size])
                for i in range(0, len(words), chunk_size)]
    
    def _process_chunk(self, 
                      chunk: str, 
                      dimension: str) -> Tuple[float, float]:
        # Process individual chunk with adaptive thresholds
        base_score = self._calculate_base_score(
            chunk, dimension)
        confidence = self._calculate_confidence(chunk)
        
        # Apply threshold adjustments
        adjusted_score = base_score * self.threshold_adjustments[
            dimension]
        
        return adjusted_score, confidence
    
    def _calculate_base_score(self, 
                            text: str, 
                            dimension: str) -> float:
        # Dimension-specific scoring logic
        scoring_methods = {
            'coherence': self._score_coherence,
            'consistency': self._score_consistency,
            'fluency': self._score_fluency,
            'relevance': self._score_relevance
        }
        return scoring_methods[dimension](text)
    
    def _combine_results(self, 
                        results: List[Tuple[float, float]]
                        ) -> float:
        scores, confidences = zip(*results)
        return np.average(scores, weights=confidences)
    
    def update_thresholds(self, 
                         new_scores: Dict[str, float]) -> None:
        self.score_history.append(new_scores)
        if len(self.score_history) > 100:
            self.score_history.pop(0)
            self._adjust_thresholds()
    
    def _adjust_thresholds(self) -> None:
        for dimension in self.threshold_adjustments:
            scores = [score[dimension] 
                     for score in self.score_history]
            mean = np.mean(scores)
            std = np.std(scores)
            
            # Adjust thresholds based on statistical analysis
            if std > 0.5:  # High variance
                self.threshold_adjustments[dimension] *= 0.95
            elif std < 0.2:  # Low variance
                self.threshold_adjustments[dimension] *= 1.05
            
            # Ensure thresholds stay within reasonable bounds
            self.threshold_adjustments[dimension] = np.clip(
                self.threshold_adjustments[dimension],
                0.5, 2.0
            )

# Example usage
optimizer = PerformanceOptimizer()
text = "Long text for analysis..." * 100  # Large text
dimension = "coherence"
optimized_score = optimizer.optimize_evaluation(text, dimension)
print(f"Optimized Score: {optimized_score:.2f}")

# Update thresholds with new scores
new_scores = {
    'coherence': 4.2,
    'consistency': 3.9,
    'fluency': 2.5,
    'relevance': 4.0
}
optimizer.update_thresholds(new_scores)

Automated Testing and Validation

This module implements comprehensive testing procedures for validating G-Eval's performance across different types of summaries and content domains, ensuring consistent and reliable evaluation results.

import pytest
from typing import Dict, List, Any
import json
import numpy as np

class GEvalTester:
    def __init__(self):
        self.test_cases = self._load_test_cases()
        self.tolerance = 0.1
        self.min_required_score = 3.0
        self.evaluator = NewsArticleEvaluator()
        
    def run_comprehensive_tests(self) -> Dict[str, Any]:
        test_results = {
            'dimension_tests': self._test_dimensions(),
            'edge_cases': self._test_edge_cases(),
            'consistency_check': self._test_consistency(),
            'performance_metrics': self._test_performance()
        }
        
        return {
            'results': test_results,
            'summary': self._generate_test_summary(test_results),
            'recommendations': self._generate_recommendations(
                test_results)
        }
    
    def _test_dimensions(self) -> Dict[str, List[Dict]]:
        results = {}
        for dimension in ['coherence', 'consistency', 
                         'fluency', 'relevance']:
            dimension_results = []
            for test_case in self.test_cases[dimension]:
                result = self._run_dimension_test(
                    dimension, test_case)
                dimension_results.append(result)
            results[dimension] = dimension_results
        return results
    
    def _run_dimension_test(self, 
                           dimension: str,
                           test_case: Dict) -> Dict:
        try:
            score = self.evaluator.evaluate_summary(
                test_case['article'],
                test_case['summary']
            )['dimension_scores'][dimension]
            
            passed = abs(score - test_case['expected_score']
                        ) <= self.tolerance
            
            return {
                'test_id': test_case['id'],
                'score': score,
                'expected': test_case['expected_score'],
                'passed': passed,
                'error': None if passed else 'Score deviation'
            }
        except Exception as e:
            return {
                'test_id': test_case['id'],
                'score': None,
                'expected': test_case['expected_score'],
                'passed': False,
                'error': str(e)
            }
    
    def _test_edge_cases(self) -> List[Dict]:
        edge_cases = [
            {'type': 'empty', 'content': ''},
            {'type': 'very_long', 
             'content': 'word ' * 10000},
            {'type': 'special_chars', 
             'content': '!@#$%^&*()\n\t'},
            {'type': 'multilingual', 
             'content': 'English текст 中文'}
        ]
        
        results = []
        for case in edge_cases:
            try:
                score = self.evaluator.evaluate_summary(
                    case['content'], case['content'])
                results.append({
                    'type': case['type'],
                    'passed': True,
                    'score': score,
                    'error': None
                })
            except Exception as e:
                results.append({
                    'type': case['type'],
                    'passed': False,
                    'score': None,
                    'error': str(e)
                })
        return results
    
    def _test_consistency(self) -> Dict[str, float]:
        # Test same input multiple times
        test_case = self.test_cases['coherence'][0]
        scores = []
        
        for _ in range(10):
            score = self.evaluator.evaluate_summary(
                test_case['article'],
                test_case['summary']
            )['final_score']
            scores.append(score)
        
        return {
            'mean': np.mean(scores),
            'std': np.std(scores),
            'max_deviation': max(scores) - min(scores)
        }
    
    def _test_performance(self) -> Dict[str, float]:
        import time
        
        start_time = time.time()
        for test_case in self.test_cases['coherence'][:5]:
            self.evaluator.evaluate_summary(
                test_case['article'],
                test_case['summary']
            )
        end_time = time.time()
        
        return {
            'avg_processing_time': (
                end_time - start_time) / 5,
            'memory_usage': self._get_memory_usage()
        }

# Example usage
tester = GEvalTester()
test_results = tester.run_comprehensive_tests()
print(json.dumps(test_results['summary'], indent=2))

Integration Framework

This module provides a robust integration framework for incorporating G-Eval into existing systems and workflows, with support for different input formats, APIs, and output specifications.

from typing import Dict, Any, Optional
import json
import asyncio
from dataclasses import dataclass, asdict

@dataclass
class GEvalRequest:
    source_text: str
    summary_text: str
    config: Optional[Dict] = None
    metadata: Optional[Dict] = None

@dataclass
class GEvalResponse:
    scores: Dict[str, float]
    analysis: Dict[str, Any]
    metadata: Dict[str, Any]

class GEvalIntegration:
    def __init__(self):
        self.evaluator = NewsArticleEvaluator()
        self.optimizer = PerformanceOptimizer()
        self.queue = asyncio.Queue()
        
    async def process_request(self, 
                            request: GEvalRequest
                            ) -> GEvalResponse:
        # Validate and preprocess request
        processed_request = self._preprocess_request(request)
        
        # Evaluate summary
        scores = await self._evaluate_async(processed_request)
        
        # Optimize and analyze results
        analysis = self._analyze_results(scores)
        
        # Prepare response
        return GEvalResponse(
            scores=scores,
            analysis=analysis,
            metadata=self._generate_metadata(request)
        )
    
    def _preprocess_request(self, 
                          request: GEvalRequest
                          ) -> GEvalRequest:
        # Apply configuration settings
        if request.config:
            self._configure_evaluator(request.config)
        
        # Clean and validate input texts
        cleaned_request = GEvalRequest(
            source_text=self._clean_text(request.source_text),
            summary_text=self._clean_text(request.summary_text),
            config=request.config,
            metadata=request.metadata
        )
        
        return cleaned_request
    
    async def _evaluate_async(self, 
                            request: GEvalRequest
                            ) -> Dict[str, float]:
        # Perform evaluation asynchronously
        evaluation_task = asyncio.create_task(
            self._run_evaluation(request))
        
        # Add to queue for processing
        await self.queue.put(evaluation_task)
        
        # Wait for results
        return await evaluation_task
    
    async def _run_evaluation(self, 
                            request: GEvalRequest
                            ) -> Dict[str, float]:
        try:
            results = self.evaluator.evaluate_summary(
                request.source_text,
                request.summary_text
            )
            
            # Optimize scores
            for dimension, score in results[
                'dimension_scores'].items():
                results['dimension_scores'][
                    dimension] = self.optimizer.optimize_evaluation(
                    request.summary_text, dimension)
            
            return results['dimension_scores']
            
        except Exception as e:
            raise IntegrationError(
                f"Evaluation failed: {str(e)}")
    
    def _analyze_results(self, 
                        scores: Dict[str, float]
                        ) -> Dict[str, Any]:
        return {
            'statistical_analysis': self._run_statistics(
                scores),
            'quality_assessment': self._assess_quality(
                scores),
            'recommendations': self._generate_recommendations(
                scores)
        }
    
    @staticmethod
    def _generate_metadata(request: GEvalRequest
                         ) -> Dict[str, Any]:
        return {
            'timestamp': datetime.now().isoformat(),
            'request_config': request.config,
            'request_metadata': request.metadata,
            'processing_info': {
                'version': '1.0.0',
                'processing_time': time.time()
            }
        }

# Example usage
async def main():
    integrator = GEvalIntegration()
    request = GEvalRequest(
        source_text="Original article text...",
        summary_text="Generated summary...",
        config={'threshold': 0.8},
        metadata={'user_id': '123'}
    )
    
    response = await integrator.process_request(request)
    print(json.dumps(asdict(response), indent=2))

# Run the example
if __name__ == "__main__":
    asyncio.run(main())

Additional Resources

Note: Always verify these URLs and papers as they may have changed or been updated.

More in AI applications and systems

  1. Evolving AI Retrieval and Generation Techniques8 min

    Retrieval-Augmented Generation (RAG) is a powerful approach in AI that combines information retrieval with text generation. However, the field is rapidly...

  2. Enhancing AI Efficiency with Chunking13 min

    Chunking is a fundamental technique in Retrieval-Augmented Generation (RAG) systems that significantly improves AI efficiency. By breaking down large texts...

  3. Explainable AI Demystifying Complex Models8 min

    SHAP (SHapley Additive exPlanations) values provide a unified measure of feature importance based on game theoretic principles. These values represent the...

  4. Developing AI Technology Literacy! Learn, Unlearn, and Relearn8 min

    AI technology literacy is crucial in today's digital world. It involves understanding, evaluating, and using AI systems ethically and safely. This...

  5. Exploring the Building Blocks of Modern AI12 min

    A Multi-layer Perceptron forms the foundation of modern neural networks, implementing forward and backward propagation to learn patterns in data through...

  6. Detecting AI-Generated Texts with GigaCheck16 min

    Text classification task implementation using the Mistral-7B model for AI-generated content detection. This approach utilizes fine-tuning techniques on a...