tf2_gnn/layers/gnn.py
def build(self, tensor_shapes: GNNInput):
        """Build the various layers in the model.

        Args:
            tensor_shapes: A GNNInput of tensor shapes.

        Returns:
            Nothing, but initialises the layers in the model based on the tensor shapes given.
        """
        # First, we go through the input shapes and make sure that anything which might vary batch
        # to batch (number of nodes / number of edges) is set to None.
        initial_node_features_shape: tf.TensorShape = tensor_shapes.node_features
        variable_node_features_shape = tf.TensorShape((None, initial_node_features_shape[1]))
        adjacency_list_shapes = tensor_shapes.adjacency_lists
        embedded_shape = tf.TensorShape((None, self._hidden_dim))

        with tf.name_scope(f"{self._message_passing_class.__name__}_GNN"):
            # Then we construct the layers themselves:
            with tf.name_scope("gnn_initial_node_projection"):
                self._initial_projection_layer = tf.keras.layers.Dense(
                    units=self._hidden_dim,
                    use_bias=False,
                    activation=self._initial_node_representation_activation_fn,
                )
                self._initial_projection_layer.build(variable_node_features_shape)

            # Construct the graph message passing layers.
            for layer_idx in range(self._num_layers):
                with tf.name_scope(f"Layer_{layer_idx}"):
                    with tf.name_scope("MessagePassing"):
                        self._mp_layers.append(
                            self._message_passing_class(self._params)
                        )
                        self._mp_layers[-1].build(
                            MessagePassingInput(embedded_shape, adjacency_list_shapes)
                        )

                    # If required, prepare for a LayerNorm:
                    if self._use_inter_layer_layernorm:
                        with tf.name_scope(f"LayerNorm"):
                            self._inter_layer_layernorms.append(
                                tf.keras.layers.LayerNormalization()
                            )
                            self._inter_layer_layernorms[-1].build(embedded_shape)

                    # Construct the per-node dense layers.
                    if layer_idx % self._dense_every_num_layers == 0:
                        with tf.name_scope(f"Dense"):
                            self._dense_layers[str(layer_idx)] = tf.keras.layers.Dense(
                                units=self._hidden_dim,
                                use_bias=False,
                                activation=self._dense_intermediate_layer_activation_fn,
                            )
                            self._dense_layers[str(layer_idx)].build(embedded_shape)

                    if (
                        layer_idx
                        and layer_idx % self._global_exchange_every_num_layers == 0
                    ):
                        with tf.name_scope(f"Global_Exchange"):
                            if self._global_exchange_mode.lower() == "mean":
                                exchange_layer_class = GraphGlobalMeanExchange
                            elif self._global_exchange_mode.lower() == "gru":
                                exchange_layer_class = GraphGlobalGRUExchange
                            elif self._global_exchange_mode.lower() == "mlp":
                                exchange_layer_class = GraphGlobalMLPExchange
                            exchange_layer = exchange_layer_class(
                                hidden_dim=self._hidden_dim,
                                weighting_fun=self._global_exchange_weighting_fun,
                                num_heads=self._global_exchange_num_heads,
                                dropout_rate=self._global_exchange_dropout_rate,
                            )
                            exchange_layer.build(
                                GraphGlobalExchangeInput(
                                    node_embeddings=tf.TensorShape(
                                        (None, self._hidden_dim)
                                    ),
                                    node_to_graph_map=tf.TensorShape((None,)),
                                    num_graphs=tf.TensorShape(()),
                                )
                            )
                            self._global_exchange_layers[
                                str(layer_idx)
                            ] = exchange_layer

        super().build(tensor_shapes)

        # The following is needed to work around a limitation in the @tf.function annotation.
        # (See https://github.com/tensorflow/tensorflow/issues/32457 for a related issue,
        #  though there are many more).
        # Our aim is to trace the `call` function once and for all. However, as the first
        # dimension of node features and adjacency lists keeps changing between batches (with
        # the number of nodes/edges in the batch), generalisation doesn't work automatically.
        # Instead, we have to specify the input spec explicitly; but as this depends on a
        # build-time constant (the number of edges), we cannot do that by just using @tf.function.
        # Instead, we construct the TensorSpec explicitly, and then use setattr to wrap
        # our function using tf.function.
        #
        # Finally, the `return_all_representations` option changes the shape of the return values,
        # but a tf.function-traced function must return the same shape on all code paths. To
        # handle this, we let the core function _always_ return all representations (and trace
        # that for performance reasons), and then use a thin wrapper `call` function to drop
        # the unneeded return value if needed.
        internal_call_input_spec = (
            GNNInput(
                node_features=tf.TensorSpec(shape=variable_node_features_shape, dtype=tf.float32),
                adjacency_lists=tuple(
                    tf.TensorSpec(shape=(None, 2), dtype=tf.int32)
                    for _ in range(len(adjacency_list_shapes))
                ),
                node_to_graph_map=tf.TensorSpec(shape=(None,), dtype=tf.int32),
                num_graphs=tf.TensorSpec(shape=(), dtype=tf.int32),
            ),
            tf.TensorSpec(shape=(), dtype=tf.bool)
        )
        setattr(self, "_internal_call", tf.function(func=self._internal_call, input_signature=internal_call_input_spec))
Build the various layers in the model.

        Args:
            tensor_shapes: A GNNInput of tensor shapes.

        Returns:
            Nothing, but initialises the layers in the model based on the tensor shapes given.
        
Build the various layers in the model.
Build the various layers in the model .
def build(self, tensor_shapes: GNNInput):
        initial_node_features_shape: tf.TensorShape = tensor_shapes.node_features
        variable_node_features_shape = tf.TensorShape((None, initial_node_features_shape[1]))
        adjacency_list_shapes = tensor_shapes.adjacency_lists
        embedded_shape = tf.TensorShape((None, self._hidden_dim))
        with tf.name_scope(f"{self._message_passing_class.__name__}_GNN"):
            with tf.name_scope("gnn_initial_node_projection"):
                self._initial_projection_layer = tf.keras.layers.Dense(
                    units=self._hidden_dim,
                    use_bias=False,
                    activation=self._initial_node_representation_activation_fn,
                )
                self._initial_projection_layer.build(variable_node_features_shape)
            for layer_idx in range(self._num_layers):
                with tf.name_scope(f"Layer_{layer_idx}"):
                    with tf.name_scope("MessagePassing"):
                        self._mp_layers.append(
                            self._message_passing_class(self._params)
                        )
                        self._mp_layers[-1].build(
                            MessagePassingInput(embedded_shape, adjacency_list_shapes)
                        )
                    if self._use_inter_layer_layernorm:
                        with tf.name_scope(f"LayerNorm"):
                            self._inter_layer_layernorms.append(
                                tf.keras.layers.LayerNormalization()
                            )
                            self._inter_layer_layernorms[-1].build(embedded_shape)
                    if layer_idx % self._dense_every_num_layers == 0:
                        with tf.name_scope(f"Dense"):
                            self._dense_layers[str(layer_idx)] = tf.keras.layers.Dense(
                                units=self._hidden_dim,
                                use_bias=False,
                                activation=self._dense_intermediate_layer_activation_fn,
                            )
                            self._dense_layers[str(layer_idx)].build(embedded_shape)
                    if (
                        layer_idx
                        and layer_idx % self._global_exchange_every_num_layers == 0
                    ):
                        with tf.name_scope(f"Global_Exchange"):
                            if self._global_exchange_mode.lower() == "mean":
                                exchange_layer_class = GraphGlobalMeanExchange
                            elif self._global_exchange_mode.lower() == "gru":
                                exchange_layer_class = GraphGlobalGRUExchange
                            elif self._global_exchange_mode.lower() == "mlp":
                                exchange_layer_class = GraphGlobalMLPExchange
                            exchange_layer = exchange_layer_class(
                                hidden_dim=self._hidden_dim,
                                weighting_fun=self._global_exchange_weighting_fun,
                                num_heads=self._global_exchange_num_heads,
                                dropout_rate=self._global_exchange_dropout_rate,
                            )
                            exchange_layer.build(
                                GraphGlobalExchangeInput(
                                    node_embeddings=tf.TensorShape(
                                        (None, self._hidden_dim)
                                    ),
                                    node_to_graph_map=tf.TensorShape((None,)),
                                    num_graphs=tf.TensorShape(()),
                                )
                            )
                            self._global_exchange_layers[
                                str(layer_idx)
                            ] = exchange_layer
        super().build(tensor_shapes)
         Finally, the `return_all_representations` option changes the shape of the return values,
         but a tf.function-traced function must return the same shape on all code paths. To
         handle this, we let the core function _always_ return all representations (and trace
         that for performance reasons), and then use a thin wrapper `call` function to drop
         the unneeded return value if needed.
        internal_call_input_spec = (
            GNNInput(
                node_features=tf.TensorSpec(shape=variable_node_features_shape, dtype=tf.float32),
                adjacency_lists=tuple(
                    tf.TensorSpec(shape=(None, 2), dtype=tf.int32)
                    for _ in range(len(adjacency_list_shapes))
                ),
                node_to_graph_map=tf.TensorSpec(shape=(None,), dtype=tf.int32),
                num_graphs=tf.TensorSpec(shape=(), dtype=tf.int32),
            ),
            tf.TensorSpec(shape=(), dtype=tf.bool)
        )
        setattr(self, "_internal_call", tf.function(func=self._internal_call, input_signature=internal_call_input_spec))
def build ( self , tensor_shapes : GNNInput ) : initial_node_features_shape : tf . TensorShape = tensor_shapes . node_features variable_node_features_shape = tf . TensorShape ( ( None , initial_node_features_shape [ 1 ] ) ) adjacency_list_shapes = tensor_shapes . adjacency_lists embedded_shape = tf . TensorShape ( ( None , self . _hidden_dim ) ) with tf . name_scope ( f"{self._message_passing_class.__name__}_GNN" ) : with tf . name_scope ( "gnn_initial_node_projection" ) : self . _initial_projection_layer = tf . keras . layers . Dense ( units = self . _hidden_dim , use_bias = False , activation = self . _initial_node_representation_activation_fn , ) self . _initial_projection_layer . build ( variable_node_features_shape ) for layer_idx in range ( self . _num_layers ) : with tf . name_scope ( f"Layer_{layer_idx}" ) : with tf . name_scope ( "MessagePassing" ) : self . _mp_layers . append ( self . _message_passing_class ( self . _params ) ) self . _mp_layers [ - 1 ] . build ( MessagePassingInput ( embedded_shape , adjacency_list_shapes ) ) if self . _use_inter_layer_layernorm : with tf . name_scope ( f"LayerNorm" ) : self . _inter_layer_layernorms . append ( tf . keras . layers . LayerNormalization ( ) ) self . _inter_layer_layernorms [ - 1 ] . build ( embedded_shape ) if layer_idx % self . _dense_every_num_layers == 0 : with tf . name_scope ( f"Dense" ) : self . _dense_layers [ str ( layer_idx ) ] = tf . keras . layers . Dense ( units = self . _hidden_dim , use_bias = False , activation = self . _dense_intermediate_layer_activation_fn , ) self . _dense_layers [ str ( layer_idx ) ] . build ( embedded_shape ) if ( layer_idx and layer_idx % self . _global_exchange_every_num_layers == 0 ) : with tf . name_scope ( f"Global_Exchange" ) : if self . _global_exchange_mode . lower ( ) == "mean" : exchange_layer_class = GraphGlobalMeanExchange elif self . _global_exchange_mode . lower ( ) == "gru" : exchange_layer_class = GraphGlobalGRUExchange elif self . _global_exchange_mode . lower ( ) == "mlp" : exchange_layer_class = GraphGlobalMLPExchange exchange_layer = exchange_layer_class ( hidden_dim = self . _hidden_dim , weighting_fun = self . _global_exchange_weighting_fun , num_heads = self . _global_exchange_num_heads , dropout_rate = self . _global_exchange_dropout_rate , ) exchange_layer . build ( GraphGlobalExchangeInput ( node_embeddings = tf . TensorShape ( ( None , self . _hidden_dim ) ) , node_to_graph_map = tf . TensorShape ( ( None , ) ) , num_graphs = tf . TensorShape ( ( ) ) , ) ) self . _global_exchange_layers [ str ( layer_idx ) ] = exchange_layer super ( ) . build ( tensor_shapes ) internal_call_input_spec = ( GNNInput ( node_features = tf . TensorSpec ( shape = variable_node_features_shape , dtype = tf . float32 ) , adjacency_lists = tuple ( tf . TensorSpec ( shape = ( None , 2 ) , dtype = tf . int32 ) for _ in range ( len ( adjacency_list_shapes ) ) ) , node_to_graph_map = tf . TensorSpec ( shape = ( None , ) , dtype = tf . int32 ) , num_graphs = tf . TensorSpec ( shape = ( ) , dtype = tf . int32 ) , ) , tf . TensorSpec ( shape = ( ) , dtype = tf . bool ) ) setattr ( self , "_internal_call" , tf . function ( func = self . _internal_call , input_signature = internal_call_input_spec ) )
Build the various layers in the model.
Build the various layers in the model .
"""Build the various layers in the model.

        Args:
            tensor_shapes: A GNNInput of tensor shapes.

        Returns:
            Nothing, but initialises the layers in the model based on the tensor shapes given.
        """ # First, we go through the input shapes and make sure that anything which might vary batch # to batch (number of nodes / number of edges) is set to None. # Then we construct the layers themselves: # Construct the graph message passing layers. # If required, prepare for a LayerNorm: # Construct the per-node dense layers. # The following is needed to work around a limitation in the @tf.function annotation. # (See https://github.com/tensorflow/tensorflow/issues/32457 for a related issue, #  though there are many more). # Our aim is to trace the `call` function once and for all. However, as the first # dimension of node features and adjacency lists keeps changing between batches (with # the number of nodes/edges in the batch), generalisation doesn't work automatically. # Instead, we have to specify the input spec explicitly; but as this depends on a # build-time constant (the number of edges), we cannot do that by just using @tf.function. # Instead, we construct the TensorSpec explicitly, and then use setattr to wrap # our function using tf.function. # # Finally, the `return_all_representations` option changes the shape of the return values, # but a tf.function-traced function must return the same shape on all code paths. To # handle this, we let the core function _always_ return all representations (and trace # that for performance reasons), and then use a thin wrapper `call` function to drop # the unneeded return value if needed.
Nothing, but initialises the layers in the model based on the tensor shapes given.
Nothing but initialises the layers in the model based on the tensor shapes given .
self
tensor_shapes
GNNInput
A GNNInput of tensor shapes.
A GNNInput of tensor shapes .