File size: 3,761 Bytes
dff2db9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
"""
Perplexity analysis module for measuring model uncertainty.
Mathematical formulation: PPL = exp(-1/N * Σ log p(w_i | w_{<i}))
"""

from typing import List, Optional
import math
import numpy as np
import torch


class PerplexityAnalyzer:
    """Computes perplexity of generated texts under the generating model."""
    
    def __init__(self, tokenizer, model):
        """
        Initialize perplexity analyzer.
        
        Args:
            tokenizer: Model tokenizer
            model: Language model for perplexity calculation
        """
        self.tokenizer = tokenizer
        self.model = model
    
    def compute_perplexity(self, texts: List[str]) -> Optional[float]:
        """
        Compute perplexity of generated texts.
        
        PPL = exp(-1/N * Σ log p(w_i | w_{<i}))
        
        Args:
            texts: List of generated texts
            
        Returns:
            Perplexity value (lower is usually better)
        """
        clean = self._filter_valid_texts(texts)
        if not clean:
            return None
        
        device = next(self.model.parameters()).device
        losses = []
        
        for text in clean:
            try:
                enc = self.tokenizer(
                    text, 
                    return_tensors="pt", 
                    truncation=True, 
                    max_length=512
                )
                input_ids = enc["input_ids"].to(device)
                
                if input_ids.shape[1] < 2:
                    continue
                
                with torch.no_grad():
                    outputs = self.model(input_ids=input_ids, labels=input_ids)
                
                if outputs.loss is not None and torch.isfinite(outputs.loss):
                    losses.append(float(outputs.loss.detach().cpu()))
            except Exception:
                continue
        
        if not losses:
            return None
        
        mean_loss = float(np.mean(losses))
        
        # Prevent overflow
        if mean_loss > 20:
            return float("inf")
        
        return float(math.exp(mean_loss))
    
    def compute_cross_entropy(self, texts: List[str]) -> Optional[float]:
        """
        Compute mean cross-entropy loss.
        
        Args:
            texts: List of generated texts
            
        Returns:
            Mean cross-entropy value
        """
        clean = self._filter_valid_texts(texts)
        if not clean:
            return None
        
        device = next(self.model.parameters()).device
        losses = []
        
        for text in clean:
            try:
                enc = self.tokenizer(text, return_tensors="pt", truncation=True, max_length=512)
                input_ids = enc["input_ids"].to(device)
                
                if input_ids.shape[1] < 2:
                    continue
                
                with torch.no_grad():
                    outputs = self.model(input_ids=input_ids, labels=input_ids)
                
                if outputs.loss is not None:
                    losses.append(float(outputs.loss.detach().cpu()))
            except Exception:
                continue
        
        return float(np.mean(losses)) if losses else None
    
    @staticmethod
    def _filter_valid_texts(texts: List[str]) -> List[str]:
        """Filter out empty or error texts."""
        return [t.strip() for t in texts if t and not t.startswith("ERROR")]
    
    def compute_all(self, texts: List[str]) -> dict:
        """Compute both perplexity and cross-entropy."""
        return {
            "Perplexity": self.compute_perplexity(texts),
            "Cross-Entropy": self.compute_cross_entropy(texts),
        }