Feat: Add LIME and SHAP model interpretability tools

evaluation/model_interpretability.py

@@ -0,0 +1,289 @@
+"""
+模型可解释性工具：LIME 和 SHAP（简化版）
+
+LIME: Local Interpretable Model-agnostic Explanations
+SHAP: SHapley Additive exPlanations
+
+注意：这是简化版实现，用于演示目的
+实际使用时建议安装 lime 和 shap 包：pip install lime shap
+"""
+import sys
+import math
+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 = hash(t) % 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 = hash(t) % 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)