multihop_graph_interdependence

Bases: graph_interdependence

A multihop graph-based interdependence function.

Notes

To model multi-hop dependency relationships among data instances, we introduce the multi-hop graph interdependence function and the accumulative multi-hop graph interdependence function as follows:

\[
    \begin{equation}
    \xi(\mathbf{x} | h) = \mathbf{A}^h \in R^{m \times m} \text{, and } \xi(\mathbf{x} | 0: h) = \sum_{i=0}^h \mathbf{A}^i \in R^{m \times m}.
    \end{equation}
\]

Attributes:

Name	Type	Description
h	int	Number of hops to consider in the graph.
accumulative	bool	Whether to accumulate interdependence over multiple hops.

Methods:

Name	Description
__init__	Initializes the multihop graph interdependence function.
calculate_A	Computes the multihop interdependence matrix.

Source code in tinybig/interdependence/topological_interdependence.py

class multihop_graph_interdependence(graph_interdependence):
    r"""
        A multihop graph-based interdependence function.

        Notes
        ----------

        To model multi-hop dependency relationships among data instances, we introduce the multi-hop graph interdependence function and the accumulative multi-hop graph interdependence function as follows:

        $$
            \begin{equation}
            \xi(\mathbf{x} | h) = \mathbf{A}^h \in R^{m \times m} \text{, and } \xi(\mathbf{x} | 0: h) = \sum_{i=0}^h \mathbf{A}^i \in R^{m \times m}.
            \end{equation}
        $$

        Attributes
        ----------
        h : int
            Number of hops to consider in the graph.
        accumulative : bool
            Whether to accumulate interdependence over multiple hops.

        Methods
        -------
        __init__(...)
            Initializes the multihop graph interdependence function.
        calculate_A(...)
            Computes the multihop interdependence matrix.
    """

    def __init__(self, h: int = 1, accumulative: bool = False, name: str = 'multihop_graph_interdependence', *args, **kwargs):
        """
            Initializes the multihop graph interdependence function.

            Parameters
            ----------
            h : int, optional
                Number of hops to consider. Defaults to 1.
            accumulative : bool, optional
                Whether to accumulate interdependence over multiple hops. Defaults to False.
            name : str, optional
                Name of the interdependence function. Defaults to 'multihop_graph_interdependence'.
            *args : tuple
                Additional positional arguments.
            **kwargs : dict
                Additional keyword arguments.
        """
        super().__init__(name=name, *args, **kwargs)
        self.h = h
        self.accumulative = accumulative

    def calculate_A(self, x: torch.Tensor = None, w: torch.nn.Parameter = None, device: str = 'cpu', *args, **kwargs):
        """
            Computes the multihop interdependence matrix.

            Parameters
            ----------
            x : torch.Tensor, optional
                Input tensor of shape `(batch_size, num_features)`. Defaults to None.
            w : torch.nn.Parameter, optional
                Parameter tensor. Defaults to None.
            device : str, optional
                Device for computation ('cpu', 'cuda'). Defaults to 'cpu'.
            *args : tuple
                Additional positional arguments.
            **kwargs : dict
                Additional keyword arguments.

            Returns
            -------
            torch.Tensor
                The computed multihop interdependence matrix.

            Raises
            ------
            AssertionError
                If the computed matrix shape is invalid.
        """
        if not self.require_data and not self.require_parameters and self.A is not None:
            return self.A
        else:
            adj, mappings = self.graph.to_matrix(self_dependence=self.self_dependence, normalization=self.normalization, normalization_mode=self.normalization_mode, device=device)

            self.node_id_index_map = mappings['node_id_index_map']
            self.node_index_id_map = mappings['node_index_id_map']

            if self.accumulative:
                A = accumulative_matrix_power(adj, self.h)
            else:
                A = matrix_power(adj, self.h)

            A = self.post_process(x=A, device=device)

            if self.interdependence_type in ['column', 'right', 'attribute', 'attribute_interdependence']:
                assert A.shape == (self.m, self.calculate_m_prime())
            elif self.interdependence_type in ['row', 'left', 'instance', 'instance_interdependence']:
                assert A.shape == (self.b, self.calculate_b_prime())

            if not self.require_data and not self.require_parameters and self.A is None:
                self.A = A
            return A

__init__(h=1, accumulative=False, name='multihop_graph_interdependence', *args, **kwargs)

Initializes the multihop graph interdependence function.

Parameters:

Name	Type	Description	Default
h	int	Number of hops to consider. Defaults to 1.	1
accumulative	bool	Whether to accumulate interdependence over multiple hops. Defaults to False.	False
name	str	Name of the interdependence function. Defaults to 'multihop_graph_interdependence'.	'multihop_graph_interdependence'
*args	tuple	Additional positional arguments.	()
**kwargs	dict	Additional keyword arguments.	{}

Source code in tinybig/interdependence/topological_interdependence.py

def __init__(self, h: int = 1, accumulative: bool = False, name: str = 'multihop_graph_interdependence', *args, **kwargs):
    """
        Initializes the multihop graph interdependence function.

        Parameters
        ----------
        h : int, optional
            Number of hops to consider. Defaults to 1.
        accumulative : bool, optional
            Whether to accumulate interdependence over multiple hops. Defaults to False.
        name : str, optional
            Name of the interdependence function. Defaults to 'multihop_graph_interdependence'.
        *args : tuple
            Additional positional arguments.
        **kwargs : dict
            Additional keyword arguments.
    """
    super().__init__(name=name, *args, **kwargs)
    self.h = h
    self.accumulative = accumulative

calculate_A(x=None, w=None, device='cpu', *args, **kwargs)

Computes the multihop interdependence matrix.

Parameters:

Name	Type	Description	Default
x	Tensor	Input tensor of shape (batch_size, num_features). Defaults to None.	None
w	Parameter	Parameter tensor. Defaults to None.	None
device	str	Device for computation ('cpu', 'cuda'). Defaults to 'cpu'.	'cpu'
*args	tuple	Additional positional arguments.	()
**kwargs	dict	Additional keyword arguments.	{}

Returns:

Type	Description
Tensor	The computed multihop interdependence matrix.

Raises:

Type	Description
AssertionError	If the computed matrix shape is invalid.

Source code in tinybig/interdependence/topological_interdependence.py

def calculate_A(self, x: torch.Tensor = None, w: torch.nn.Parameter = None, device: str = 'cpu', *args, **kwargs):
    """
        Computes the multihop interdependence matrix.

        Parameters
        ----------
        x : torch.Tensor, optional
            Input tensor of shape `(batch_size, num_features)`. Defaults to None.
        w : torch.nn.Parameter, optional
            Parameter tensor. Defaults to None.
        device : str, optional
            Device for computation ('cpu', 'cuda'). Defaults to 'cpu'.
        *args : tuple
            Additional positional arguments.
        **kwargs : dict
            Additional keyword arguments.

        Returns
        -------
        torch.Tensor
            The computed multihop interdependence matrix.

        Raises
        ------
        AssertionError
            If the computed matrix shape is invalid.
    """
    if not self.require_data and not self.require_parameters and self.A is not None:
        return self.A
    else:
        adj, mappings = self.graph.to_matrix(self_dependence=self.self_dependence, normalization=self.normalization, normalization_mode=self.normalization_mode, device=device)

        self.node_id_index_map = mappings['node_id_index_map']
        self.node_index_id_map = mappings['node_index_id_map']

        if self.accumulative:
            A = accumulative_matrix_power(adj, self.h)
        else:
            A = matrix_power(adj, self.h)

        A = self.post_process(x=A, device=device)

        if self.interdependence_type in ['column', 'right', 'attribute', 'attribute_interdependence']:
            assert A.shape == (self.m, self.calculate_m_prime())
        elif self.interdependence_type in ['row', 'left', 'instance', 'instance_interdependence']:
            assert A.shape == (self.b, self.calculate_b_prime())

        if not self.require_data and not self.require_parameters and self.A is None:
            self.A = A
        return A