|
| 1 | +"""Integration tests for QKV hook compatibility in TransformerBridge.""" |
| 2 | + |
| 3 | +import torch |
| 4 | + |
| 5 | +from transformer_lens.model_bridge import TransformerBridge |
| 6 | + |
| 7 | + |
| 8 | +class TestQKVHookCompatibility: |
| 9 | + """Test that QKV bridge hooks are compatible with overall model hook access.""" |
| 10 | + |
| 11 | + def test_v_hook_out_equals_blocks_attn_hook_v(self): |
| 12 | + """Test that v_hook_out in QKV bridge equals blocks.0.attn.hook_v on the overall model.""" |
| 13 | + # Load GPT-2 in TransformerBridge |
| 14 | + bridge = TransformerBridge.boot_transformers("gpt2", device="cpu") |
| 15 | + |
| 16 | + # Turn on compatibility mode |
| 17 | + bridge.enable_compatibility_mode(disable_warnings=True) |
| 18 | + |
| 19 | + # Create test input |
| 20 | + test_input = torch.tensor([[1, 2, 3, 4, 5]]) # Simple test sequence |
| 21 | + |
| 22 | + # Get the QKV bridge from the first attention layer |
| 23 | + qkv_bridge = bridge.blocks[0].attn.qkv |
| 24 | + |
| 25 | + # Verify that qkv_bridge is indeed a QKVBridge |
| 26 | + from transformer_lens.model_bridge.generalized_components.qkv_bridge import ( |
| 27 | + QKVBridge, |
| 28 | + ) |
| 29 | + |
| 30 | + assert isinstance(qkv_bridge, QKVBridge), "First attention layer should have a QKVBridge" |
| 31 | + |
| 32 | + # Run a forward pass to populate the hooks |
| 33 | + with torch.no_grad(): |
| 34 | + _ = bridge(test_input) |
| 35 | + |
| 36 | + # Assert that v_hook_out in the QKV bridge is the same object as |
| 37 | + # blocks.0.attn.hook_v on the overall model |
| 38 | + assert ( |
| 39 | + qkv_bridge.v_hook_out is bridge.blocks[0].attn.hook_v |
| 40 | + ), "v_hook_out in QKV bridge should be the same object as blocks.0.attn.hook_v" |
| 41 | + |
| 42 | + # Also test that the hook points have the same properties |
| 43 | + assert ( |
| 44 | + qkv_bridge.v_hook_out.has_hooks() == bridge.blocks[0].attn.hook_v.has_hooks() |
| 45 | + ), "Hook points should have the same hook status" |
| 46 | + |
| 47 | + def test_q_hook_out_equals_blocks_attn_hook_q(self): |
| 48 | + """Test that q_hook_out in QKV bridge equals blocks.0.attn.hook_q on the overall model.""" |
| 49 | + # Load GPT-2 in TransformerBridge |
| 50 | + bridge = TransformerBridge.boot_transformers("gpt2", device="cpu") |
| 51 | + |
| 52 | + # Turn on compatibility mode |
| 53 | + bridge.enable_compatibility_mode(disable_warnings=True) |
| 54 | + |
| 55 | + # Create test input |
| 56 | + test_input = torch.tensor([[1, 2, 3, 4, 5]]) # Simple test sequence |
| 57 | + |
| 58 | + # Get the QKV bridge from the first attention layer |
| 59 | + qkv_bridge = bridge.blocks[0].attn.qkv |
| 60 | + |
| 61 | + # Run a forward pass to populate the hooks |
| 62 | + with torch.no_grad(): |
| 63 | + _ = bridge(test_input) |
| 64 | + |
| 65 | + # Assert that q_hook_out in the QKV bridge is the same object as |
| 66 | + # blocks.0.attn.hook_q on the overall model |
| 67 | + assert ( |
| 68 | + qkv_bridge.q_hook_out is bridge.blocks[0].attn.hook_q |
| 69 | + ), "q_hook_out in QKV bridge should be the same object as blocks.0.attn.hook_q" |
| 70 | + |
| 71 | + def test_k_hook_out_equals_blocks_attn_hook_k(self): |
| 72 | + """Test that k_hook_out in QKV bridge equals blocks.0.attn.hook_k on the overall model.""" |
| 73 | + # Load GPT-2 in TransformerBridge |
| 74 | + bridge = TransformerBridge.boot_transformers("gpt2", device="cpu") |
| 75 | + |
| 76 | + # Turn on compatibility mode |
| 77 | + bridge.enable_compatibility_mode(disable_warnings=True) |
| 78 | + |
| 79 | + # Create test input |
| 80 | + test_input = torch.tensor([[1, 2, 3, 4, 5]]) # Simple test sequence |
| 81 | + |
| 82 | + # Get the QKV bridge from the first attention layer |
| 83 | + qkv_bridge = bridge.blocks[0].attn.qkv |
| 84 | + |
| 85 | + # Run a forward pass to populate the hooks |
| 86 | + with torch.no_grad(): |
| 87 | + _ = bridge(test_input) |
| 88 | + |
| 89 | + # Assert that k_hook_out in the QKV bridge is the same object as |
| 90 | + # blocks.0.attn.hook_k on the overall model |
| 91 | + assert ( |
| 92 | + qkv_bridge.k_hook_out is bridge.blocks[0].attn.hook_k |
| 93 | + ), "k_hook_out in QKV bridge should be the same object as blocks.0.attn.hook_k" |
| 94 | + |
| 95 | + def test_hook_aliases_work_correctly(self): |
| 96 | + """Test that hook aliases work correctly in compatibility mode.""" |
| 97 | + # Load GPT-2 in TransformerBridge |
| 98 | + bridge = TransformerBridge.boot_transformers("gpt2", device="cpu") |
| 99 | + |
| 100 | + # Turn on compatibility mode |
| 101 | + bridge.enable_compatibility_mode(disable_warnings=True) |
| 102 | + |
| 103 | + # Create test input |
| 104 | + test_input = torch.tensor([[1, 2, 3, 4, 5]]) # Simple test sequence |
| 105 | + |
| 106 | + # Get the QKV bridge from the first attention layer |
| 107 | + qkv_bridge = bridge.blocks[0].attn.qkv |
| 108 | + |
| 109 | + # Run a forward pass to populate the hooks |
| 110 | + with torch.no_grad(): |
| 111 | + _ = bridge(test_input) |
| 112 | + |
| 113 | + # Test that hook aliases work correctly |
| 114 | + # These should all reference the same hook points |
| 115 | + assert qkv_bridge.q_hook_out is bridge.blocks[0].attn.hook_q, "Q hook alias should work" |
| 116 | + assert qkv_bridge.k_hook_out is bridge.blocks[0].attn.hook_k, "K hook alias should work" |
| 117 | + assert qkv_bridge.v_hook_out is bridge.blocks[0].attn.hook_v, "V hook alias should work" |
| 118 | + |
| 119 | + # Test that the hook points are accessible through the attention bridge properties |
| 120 | + assert qkv_bridge.q_hook_out is bridge.blocks[0].attn.q.hook_out, "Q property should work" |
| 121 | + assert qkv_bridge.k_hook_out is bridge.blocks[0].attn.k.hook_out, "K property should work" |
| 122 | + assert qkv_bridge.v_hook_out is bridge.blocks[0].attn.v.hook_out, "V property should work" |
| 123 | + |
| 124 | + def test_head_ablation_hook_works_correctly(self): |
| 125 | + """Test that head ablation hook works correctly with TransformerBridge.""" |
| 126 | + # Load GPT-2 in TransformerBridge |
| 127 | + bridge = TransformerBridge.boot_transformers("gpt2", device="cpu") |
| 128 | + |
| 129 | + # Turn on compatibility mode |
| 130 | + bridge.enable_compatibility_mode(disable_warnings=True) |
| 131 | + |
| 132 | + # Create test tokens (same as in the demo) |
| 133 | + gpt2_tokens = torch.tensor([[1, 2, 3, 4, 5, 6, 7, 8, 9, 10]]) |
| 134 | + |
| 135 | + layer_to_ablate = 0 |
| 136 | + head_index_to_ablate = 8 |
| 137 | + |
| 138 | + # Test both hook names |
| 139 | + hook_names_to_test = [ |
| 140 | + "blocks.0.attn.hook_v", # Compatibility mode alias |
| 141 | + "blocks.0.attn.v.hook_out", # Direct property access |
| 142 | + ] |
| 143 | + |
| 144 | + for hook_name in hook_names_to_test: |
| 145 | + print(f"\nTesting hook name: {hook_name}") |
| 146 | + |
| 147 | + # Track if the hook was called |
| 148 | + hook_called = False |
| 149 | + mutation_applied = False |
| 150 | + |
| 151 | + # We define a head ablation hook |
| 152 | + def head_ablation_hook(value, hook): |
| 153 | + nonlocal hook_called, mutation_applied |
| 154 | + hook_called = True |
| 155 | + print(f"Shape of the value tensor: {value.shape}") |
| 156 | + |
| 157 | + # Apply the ablation (out-of-place to avoid view modification error) |
| 158 | + result = value.clone() |
| 159 | + result[:, :, head_index_to_ablate, :] = 0.0 |
| 160 | + |
| 161 | + # Check if the mutation was applied (the result should be zero for the ablated head) |
| 162 | + if torch.all(result[:, :, head_index_to_ablate, :] == 0.0): |
| 163 | + mutation_applied = True |
| 164 | + |
| 165 | + return result |
| 166 | + |
| 167 | + # Get original loss |
| 168 | + original_loss = bridge(gpt2_tokens, return_type="loss") |
| 169 | + |
| 170 | + # Run with head ablation hook |
| 171 | + ablated_loss = bridge.run_with_hooks( |
| 172 | + gpt2_tokens, return_type="loss", fwd_hooks=[(hook_name, head_ablation_hook)] |
| 173 | + ) |
| 174 | + |
| 175 | + print(f"Original Loss: {original_loss.item():.3f}") |
| 176 | + print(f"Ablated Loss: {ablated_loss.item():.3f}") |
| 177 | + |
| 178 | + # Assert that the hook was called |
| 179 | + assert hook_called, f"Head ablation hook should have been called for {hook_name}" |
| 180 | + |
| 181 | + # Assert that the mutation was applied |
| 182 | + assert ( |
| 183 | + mutation_applied |
| 184 | + ), f"Mutation should have been applied to the value tensor for {hook_name}" |
| 185 | + |
| 186 | + # Assert that ablated loss is higher than original loss (ablation should hurt performance) |
| 187 | + assert ( |
| 188 | + ablated_loss.item() > original_loss.item() |
| 189 | + ), f"Ablated loss should be higher than original loss for {hook_name}" |
| 190 | + |
| 191 | + print(f"✅ Hook {hook_name} works correctly!") |
0 commit comments