Tests for self attention using CumConcatLayer

Author: Zettelkasten

tests/test_TFNetworkRecLayer.py

@@ -6837,6 +6837,150 @@ def test_RelativePositionalEncodingLayer():
     print(out)  # random...
 
 
+def _build_self_attention_layer(d, input, output, inside_rec_layer, query_axis, num_heads=8, key_dim=64,
+                                value_dim=64, dropout=0.0):
+  """
+  Essentially this does
+    d[output + '_att'] = {"class": "self_attention", "num_heads": num_heads,
+      "total_key_dim": num_heads * key_dim,
+      "n_out": num_heads * value_dim, "from": [input],
+      "attention_left_only": inside_rec_layer,
+      "attention_dropout": dropout, "forward_weights_init": self.ff_init}
+  But using multiple layers.
+  """
+  # Create (non-accumulated) query, key and value
+  d[output + '_qkv0'] = {
+    'class': 'linear', 'activation': None, 'with_bias': False, 'from': [input],
+    'n_out': num_heads * (2 * key_dim + value_dim)}  # [B,T?,F|n*(2d_k+d_v)]
+  d[output + '_qkv'] = {
+    'class': 'split_dims', 'axis': 'F', 'dims': (num_heads, 2 * key_dim + value_dim),
+    'from': [output + '_qkv0']}  # [B,T?,n,F|2d_k+d_v]
+  d[output + '_qkv_split'] = {
+    'class': 'split', 'axis': 'F', 'size_splits': (key_dim, key_dim, value_dim),
+    'from': [output + '_qkv']}
+  d[output + '_query'] = {
+    'class': 'copy', 'from': [output + '_qkv_split/0']}  # [B,T?,n,F|d_k]
+  d[output + '_key'] = {
+    'class': 'copy', 'from': [output + '_qkv_split/1']}  # [B,T?,n,F|d_k]
+  d[output + '_value'] = {
+    'class': 'copy', 'from': [output + '_qkv_split/2']}  # [B,T?,n,F|d_v]
+
+  # Accumulate keys/values or rename the axis
+  key_dim_tag = DimensionTag(kind=DimensionTag.Types.Time, description='self-att-keys')
+  key_axis = 'stag:' + key_dim_tag.description
+  if inside_rec_layer:
+    d[output + '_key_accum'] = {
+      'class': 'cum_concat', 'from': [output + '_key'], 'new_dim': key_dim_tag}  # [B,T|rec-history,n,F|d_k]
+    d[output + '_value_accum'] = {
+      'class': 'cum_concat', 'from': [output + '_value'], 'new_dim': key_dim_tag}  # [B,T|rec-history,n,F|d_v]
+  else:
+    d[output + '_key_accum'] = {
+      'class': 'reinterpret_data', 'set_dim_tags': {query_axis: key_dim_tag},
+      'from': [output + '_key']}  # [B,T|keys,n,F|d_k]
+    d[output + '_value_accum'] = {
+      'class': 'reinterpret_data', 'set_dim_tags': {query_axis: key_dim_tag},
+      'from': [output + '_value']}  # [B,T|keys,n,F|d_v]
+
+  # Calculate the energies
+  d[output + '_energy'] = {
+    'class': 'dot', 'from': [output + '_query', output + '_key_accum'],
+    'red1': 'static:-1', 'red2': 'static:-1', 'common': ['B', 'static:0']}  # [B,n,T?,T|rec-history]
+
+  d[output + '_weights'] = {
+    'class': 'softmax_over_spatial', 'from': [output + '_energy'], 'axis': key_axis,
+    'energy_factor': key_dim ** -0.5}  # [B,n,T?,T|rec-history]
+  d[output + '_weights_drop'] = {
+    'class': 'dropout', 'dropout_noise_shape': {'*': None}, 'from': [output + '_weights'],
+    'dropout': dropout}  # [B,n,T?,T|rec-history]
+
+  d[output + '_output'] = {
+    'class': 'dot', 'from': [output + '_weights_drop', output + '_value_accum'],
+    'red1': key_axis, 'red2': key_axis, 'common': ['B', query_axis, 'static:0']}  # [B,n,T?,F|d_v]
+  d[output + '_att'] = {
+    'class': 'merge_dims', 'axes': 'static', 'from': [output + '_output']}  # [B,T?,F|n*d_v]
+
+
+def test_CumConcatLayer_self_attention_equal_to_SelfAttentionLayer():
+  n_time = 13
+  num_heads, key_dim, value_dim = 2, 3, 3
+  for inside_rec_layer in [False, True]:
+    with make_scope() as session:
+      print('Testing inside_rec_layer=%s' % inside_rec_layer)
+
+      # build net dict
+      single_layer_net_dict = {
+        "class": "self_attention", "from": "data", "num_heads": num_heads, "total_key_dim": num_heads * key_dim,
+        "n_out": num_heads * value_dim, "attention_left_only": inside_rec_layer, 'is_output_layer': True}  # [B,T,F]
+      if inside_rec_layer:
+        net_dict = {
+          "output": {
+            "class": "rec", "target": "classes",
+            "unit": {
+              "single_layer_att": single_layer_net_dict, # [B,T,F]
+              "multi_layer_att": None  # [B,T,F], added below.
+            }}}
+        _build_self_attention_layer(
+          net_dict["output"], 'data', 'multi_layer', inside_rec_layer=False, query_axis='stag:extern_data:classes',
+          num_heads=num_heads, key_dim=key_dim, value_dim=value_dim)
+        net_dict["output"]["multi_layer_att"]["is_output_layer"] = True
+      else:
+        net_dict = {
+          "single_layer_att": single_layer_net_dict, # [B,T,F]
+          "multi_layer_att": None  # [B,T,F], added below.
+        }
+        _build_self_attention_layer(
+          net_dict, 'data', 'multi_layer', inside_rec_layer=False, query_axis='stag:extern_data:data',
+          num_heads=num_heads, key_dim=key_dim, value_dim=value_dim)
+        net_dict["multi_layer_att"]["is_output_layer"] = True
+
+      config = Config({"debug_print_layer_output_template": True, "debug_add_check_numerics_ops": True})
+      config.update(dict(num_inputs=num_heads*key_dim, num_outputs=num_heads*value_dim))
+      network = TFNetwork(config=config, train_flag=True)
+      network.construct_from_dict(net_dict)
+
+      if inside_rec_layer:
+        single_layer = network.get_layer("output/single_layer_att")
+        multi_layer = network.get_layer("output/multi_layer_att")
+      else:
+        single_layer = network.get_layer("single_layer_att")
+        multi_layer = network.get_layer("multi_layer_att")
+
+      assert_equal(single_layer.output.shape, (None, num_heads * value_dim))
+      assert_equal(multi_layer.output.shape, (None, num_heads * value_dim))
+
+      # set weights equal.
+      single_weights = single_layer.params["QKV"]
+      multi_weights = multi_layer.params["W"]
+      assert_equal(single_weights.shape, multi_weights.shape)
+      weights = numpy.random.rand(*single_weights.shape)
+      session.run(tf.assign(single_weights, weights))
+      session.run(tf.assign(multi_weights, weights))
+
+      # fetch/compare outputs
+      from tests.test_TFNetworkLayer import make_feed_dict
+      feed_dict = make_feed_dict(network.extern_data.data.values(), same_time=True, n_time=n_time)
+      single, multi = session.run(
+          [single_layer.output.placeholder, multi_layer.output.placeholder], feed_dict=feed_dict)
+      print('single layer output:')
+      pprint(single)
+      print('multi layer output:')
+      pprint(multi)
+      numpy.testing.assert_almost_equal(single, multi, decimal=5)
+      print('They are equal!')
+
+
+def test_self_attention_optimize_out():
+  num_heads, key_dim, value_dim = 2, 3, 3
+  network = {}
+  _build_self_attention_layer(
+    network, 'data:source', 'att', inside_rec_layer=True, query_axis='stag:extern_data:data',
+    num_heads=num_heads, key_dim=key_dim, value_dim=value_dim)
+
+  check_reclayer_optimize_out(
+    {'class': 'copy', 'from': 'att_att', 'n_out': value_dim * num_heads},
+    other_subnet_layers=network)
+
+
 if __name__ == "__main__":
   try:
     better_exchook.install()