File size: 9,532 Bytes
f221b4b
 
 
 
 
 
 
 
 
 
 
5e7eab2
f221b4b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
5e7eab2
f221b4b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
5e7eab2
f221b4b
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
"""
模型可解释性工具:LIME 和 SHAP(简化版)

LIME: Local Interpretable Model-agnostic Explanations
SHAP: SHapley Additive exPlanations

注意:这是简化版实现,用于演示目的
实际使用时建议安装 lime 和 shap 包:pip install lime shap
"""
import sys
import math
import hashlib
import torch
import numpy as np
from pathlib import Path
from typing import List, Optional, Dict
from collections import defaultdict

sys.path.insert(0, '.')


class SimplifiedLIME:
    """
    简化版 LIME 解释器
    
    通过遮蔽输入 token 来衡量每个 token 对输出的贡献
    """
    
    def __init__(self, model, tokenizer=None, num_samples: int = 100):
        self.model = model
        self.tokenizer = tokenizer
        self.num_samples = num_samples
    
    def explain(self, text: str, top_k: int = 10) -> Dict:
        """
        解释单个文本的预测
        
        Args:
            text: 输入文本
            top_k: 返回最重要的 top_k token
        
        Returns:
            dict: {token_importances: [{token, importance}], prediction, confidence}
        """
        self.model.eval()
        
        # Tokenize
        if self.tokenizer:
            tokens = text.split()  # 简化分词
        else:
            tokens = text.split()
        
        if not tokens:
            return {'token_importances': [], 'num_tokens': 0, 'original_score': 0.0}
        
        # 获取原始预测
        original_score = self._predict_score(tokens)
        
        # 遮蔽每个 token 并计算重要性
        importances = []
        for i in range(len(tokens)):
            masked_tokens = tokens[:i] + ['[MASK]'] + tokens[i+1:]
            masked_score = self._predict_score(masked_tokens)
            
            # 重要性 = 原始分数 - 遮蔽后分数
            importance = original_score - masked_score
            importances.append({
                'token': tokens[i],
                'index': i,
                'importance': importance,
            })
        
        # 排序
        importances.sort(key=lambda x: abs(x['importance']), reverse=True)
        
        return {
            'token_importances': importances[:top_k],
            'num_tokens': len(tokens),
            'original_score': original_score,
        }
    
    def _predict_score(self, tokens):
        """使用模型预测分数(简化版)"""
        # 将 token 转为 input_ids
        vocab_size = getattr(self.model.config, 'vocab_size', 100) if hasattr(self.model, 'config') else 100
        
        input_ids = []
        for t in tokens:
            # 简单 hash 到 vocab 范围
            token_id = int(hashlib.md5(t.encode()).hexdigest(), 16) % vocab_size
            input_ids.append(token_id)
        
        input_ids = torch.tensor([input_ids])
        attention_mask = torch.ones_like(input_ids)
        
        with torch.no_grad():
            try:
                outputs = self.model(input_ids=input_ids, attention_mask=attention_mask)
                if isinstance(outputs, dict):
                    logits = outputs.get('logits', outputs.get('output', torch.tensor([[0.0]])))
                else:
                    logits = outputs[0] if isinstance(outputs, (list, tuple)) else outputs
                
                # 返回最大 logit 作为分数
                score = logits[0, -1].max().item()
            except:
                score = 0.0
        
        return score


class SimplifiedSHAP:
    """
    简化版 SHAP 解释器
    
    使用 Shapley 值的近似计算来衡量每个 token 的贡献
    """
    
    def __init__(self, model, tokenizer=None, max_coalitions: int = 50):
        self.model = model
        self.tokenizer = tokenizer
        self.max_coalitions = max_coalitions
    
    def explain(self, text: str, top_k: int = 10) -> Dict:
        """
        使用 Shapley 值解释单个文本
        
        Args:
            text: 输入文本
            top_k: 返回最重要的 top_k token
        
        Returns:
            dict: {shap_values: [{token, shap_value}], base_value, prediction}
        """
        self.model.eval()
        
        tokens = text.split()
        n = len(tokens)
        
        if n == 0:
            return {'shap_values': [], 'base_value': 0.0}
        
        # 计算 base value(空输入的预测)
        base_value = self._predict_score([])
        
        # 计算 Shapley 值
        shap_values = []
        
        for i in range(n):
            # 简化 Shapley 值计算:随机采样联盟
            marginal_contributions = []
            
            for _ in range(min(self.max_coalitions, 2 ** n)):
                # 随机生成联盟(不包含 token i)
                coalition = [j for j in range(n) if j != i and np.random.random() > 0.5]
                
                # 有 token i 的联盟
                coalition_with_i = coalition + [i]
                
                # 计算边际贡献
                score_without = self._predict_score([tokens[j] for j in coalition])
                score_with = self._predict_score([tokens[j] for j in coalition_with_i])
                
                marginal_contributions.append(score_with - score_without)
            
            # Shapley 值 = 边际贡献的平均
            shap_value = np.mean(marginal_contributions) if marginal_contributions else 0.0
            shap_values.append({
                'token': tokens[i],
                'index': i,
                'shap_value': shap_value,
            })
        
        # 排序
        shap_values.sort(key=lambda x: abs(x['shap_value']), reverse=True)
        
        return {
            'shap_values': shap_values[:top_k],
            'base_value': base_value,
            'num_tokens': n,
        }
    
    def _predict_score(self, tokens):
        """使用模型预测分数"""
        vocab_size = getattr(self.model.config, 'vocab_size', 100) if hasattr(self.model, 'config') else 100
        
        if not tokens:
            return 0.0
        
        input_ids = []
        for t in tokens:
            token_id = int(hashlib.md5(t.encode()).hexdigest(), 16) % vocab_size
            input_ids.append(token_id)
        
        input_ids = torch.tensor([input_ids])
        attention_mask = torch.ones_like(input_ids)
        
        with torch.no_grad():
            try:
                outputs = self.model(input_ids=input_ids, attention_mask=attention_mask)
                if isinstance(outputs, dict):
                    logits = outputs.get('logits', outputs.get('output', torch.tensor([[0.0]])))
                else:
                    logits = outputs[0] if isinstance(outputs, (list, tuple)) else outputs
                
                score = logits[0, -1].max().item()
            except:
                score = 0.0
        
        return score


class AttentionVisualizer:
    """
    基于注意力权重的解释工具
    """
    
    def __init__(self, model):
        self.model = model
    
    def get_attention_weights(self, input_ids, attention_mask=None):
        """
        获取注意力权重(简化版)
        
        注意:需要模型支持输出注意力权重
        """
        self.model.eval()
        
        with torch.no_grad():
            try:
                outputs = self.model(input_ids=input_ids, attention_mask=attention_mask, output_attentions=True)
                if isinstance(outputs, dict) and 'attentions' in outputs:
                    return outputs['attentions']
            except:
                pass
        
        return None


if __name__ == "__main__":
    print("=" * 60)
    print("Fusion-LLM 模型可解释性工具测试")
    print("=" * 60)
    print()
    
    # 创建模型
    print("[1] 创建模型...")
    from models.fusion_mini import FusionMini, FusionMiniConfig
    config = FusionMiniConfig(vocab_size=100, hidden_size=32, num_hidden_layers=1)
    model = FusionMini(config)
    print("   模型已创建")
    print()
    
    # 测试 LIME
    print("[2] 测试 LIME 解释器...")
    lime = SimplifiedLIME(model, num_samples=10)
    explanation = lime.explain("the cat sat on the mat", top_k=5)
    print(f"   Token 数量: {explanation['num_tokens']}")
    print(f"   原始分数: {explanation['original_score']:.4f}")
    print(f"   Top-5 重要 token:")
    for item in explanation['token_importances']:
        print(f"      '{item['token']}' (index={item['index']}): {item['importance']:.4f}")
    print("   LIME 测试通过")
    print()
    
    # 测试 SHAP
    print("[3] 测试 SHAP 解释器...")
    np.random.seed(42)
    shap = SimplifiedSHAP(model, max_coalitions=10)
    shap_result = shap.explain("the cat sat on the mat", top_k=5)
    print(f"   Base value: {shap_result['base_value']:.4f}")
    print(f"   Top-5 SHAP token:")
    for item in shap_result['shap_values']:
        print(f"      '{item['token']}' (index={item['index']}): {item['shap_value']:.4f}")
    print("   SHAP 测试通过")
    print()
    
    # 边缘情况
    print("[4] 边缘情况测试...")
    empty_explain = lime.explain("", top_k=5)
    assert empty_explain['num_tokens'] == 0
    print("   空输入测试通过")
    
    single_explain = lime.explain("hello", top_k=5)
    assert single_explain['num_tokens'] == 1
    print("   单 token 测试通过")
    print()
    
    print("[PASS] 模型可解释性工具测试全部通过")
    sys.exit(0)