Upload 15 files

.gitattributes

@@ -1,35 +1,3 @@
-*.7z filter=lfs diff=lfs merge=lfs -text
-*.arrow filter=lfs diff=lfs merge=lfs -text
-*.bin filter=lfs diff=lfs merge=lfs -text
-*.bz2 filter=lfs diff=lfs merge=lfs -text
-*.ckpt filter=lfs diff=lfs merge=lfs -text
-*.ftz filter=lfs diff=lfs merge=lfs -text
-*.gz filter=lfs diff=lfs merge=lfs -text
-*.h5 filter=lfs diff=lfs merge=lfs -text
-*.joblib filter=lfs diff=lfs merge=lfs -text
-*.lfs.* filter=lfs diff=lfs merge=lfs -text
-*.mlmodel filter=lfs diff=lfs merge=lfs -text
-*.model filter=lfs diff=lfs merge=lfs -text
-*.msgpack filter=lfs diff=lfs merge=lfs -text
-*.npy filter=lfs diff=lfs merge=lfs -text
-*.npz filter=lfs diff=lfs merge=lfs -text
-*.onnx filter=lfs diff=lfs merge=lfs -text
-*.ot filter=lfs diff=lfs merge=lfs -text
-*.parquet filter=lfs diff=lfs merge=lfs -text
-*.pb filter=lfs diff=lfs merge=lfs -text
-*.pickle filter=lfs diff=lfs merge=lfs -text
-*.pkl filter=lfs diff=lfs merge=lfs -text
-*.pt filter=lfs diff=lfs merge=lfs -text
-*.pth filter=lfs diff=lfs merge=lfs -text
-*.rar filter=lfs diff=lfs merge=lfs -text
-*.safetensors filter=lfs diff=lfs merge=lfs -text
-saved_model/**/* filter=lfs diff=lfs merge=lfs -text
-*.tar.* filter=lfs diff=lfs merge=lfs -text
-*.tar filter=lfs diff=lfs merge=lfs -text
-*.tflite filter=lfs diff=lfs merge=lfs -text
-*.tgz filter=lfs diff=lfs merge=lfs -text
-*.wasm filter=lfs diff=lfs merge=lfs -text
-*.xz filter=lfs diff=lfs merge=lfs -text
-*.zip filter=lfs diff=lfs merge=lfs -text
-*.zst filter=lfs diff=lfs merge=lfs -text
-*tfevents* filter=lfs diff=lfs merge=lfs -text
+# Auto detect text files and perform LF normalization
+* text=auto
+poster/BanglaAutoKG.png filter=lfs diff=lfs merge=lfs -text

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2024 Azmine Toushik Wasi
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1,3 +1,54 @@
----
-license: mit
----
+# ***BanglaAutoKG*: Automatic Bangla Knowledge Graph Construction with Semantic Neural Graph Filtering**
+- **Authors:** Azmine Toushik Wasi, Taki Hasan Rafi,  Raima Islam and Dong-Kyu Chae
+- **ACL Anthology**: https://aclanthology.org/2024.lrec-main.189/
+- **arXiv**: https://arxiv.org/abs/2404.03528
+
+---
+**Abstract:** Knowledge Graphs (KGs) have proven essential in information processing and reasoning applications because they link related entities and give context-rich information, supporting efficient information retrieval and knowledge discovery; presenting information flow in a very effective manner. Despite being widely used globally, Bangla is relatively underrepresented in KGs due to a lack of comprehensive datasets, encoders, NER (named entity recognition) models, POS (part-of-speech) taggers, and lemmatizers, hindering efficient information processing and reasoning applications in the language. Addressing the KG scarcity in Bengali, we propose BanglaAutoKG, a pioneering framework that is able to automatically construct Bengali KGs from any Bangla text. We utilize multilingual LLMs to understand various languages and correlate entities and relations universally. By employing a translation dictionary to identify English equivalents and extracting word features from pre-trained BERT models, we construct the foundational KG. To reduce noise and align word embeddings with our goal, we employ graph-based polynomial filters. Lastly, we implement a GNN-based semantic filter, which elevates contextual understanding and trims unnecessary edges, culminating in the formation of the definitive KG. Empirical findings and case studies demonstrate the universal effectiveness of our model, capable of autonomously constructing semantically enriched KGs from any text.
+
+
+## Architecture
+ Pipeline of *BanglaAutoKG*.
+
+<p align="center">
+  <img src="fig/BanglaAutoKG.png" width="1000"/>
+</p>
+
+## Example Case Studies
+Example uses of *BanglaAutoKG*.
+
+<p align="center">
+  <img src="fig/KGs.png" width="1000"/>
+</p>
+
+Texts are available in `.\case-studies\` folder.
+
+---
+
+## Citation
+- **ACL** : 
+```
+Azmine Toushik Wasi, Taki Hasan Rafi, Raima Islam, and Dong-Kyu Chae. 2024. BanglaAutoKG: Automatic Bangla Knowledge Graph Construction with Semantic Neural Graph Filtering. In Proceedings of the 2024 Joint International Conference on Computational Linguistics, Language Resources and Evaluation (LREC-COLING 2024), pages 2100–2106, Torino, Italia. ELRA and ICCL.
+```
+- **BIBTEX** : 
+```
+@inproceedings{wasi-etal-2024-banglaautokg-automatic,
+    title = "{B}angla{A}uto{KG}: Automatic {B}angla Knowledge Graph Construction with Semantic Neural Graph Filtering",
+    author = "Wasi, Azmine Toushik  and Rafi, Taki Hasan  and Islam, Raima  and Chae, Dong-Kyu",
+    booktitle = "Proceedings of the 2024 Joint International Conference on Computational Linguistics, Language Resources and Evaluation (LREC-COLING 2024)",
+    month = may,
+    year = "2024",
+    address = "Torino, Italia",
+    publisher = "ELRA and ICCL",
+    url = "https://aclanthology.org/2024.lrec-main.189",
+    pages = "2100--2106",
+    code = "https://github.com/azminewasi/BanglaAutoKG",
+}
+```
+
+---
+
+# Poster
+<p align="center">
+  <img src="poster/BanglaAutoKG.png" width="1000"/>
+</p>

case-studies/Wiki-text.md

@@ -0,0 +1,9 @@
+# Wikipedia Article Part: বাংলাদেশ জাতীয় ফুটবল দল
+- Collected from: [URL](https://bn.wikipedia.org/s/1cmz)
+
+---
+---
+
+বাংলাদেশ জাতীয় ফুটবল দল হচ্ছে আন্তর্জাতিক ফুটবলে বাংলাদেশের প্রতিনিধিত্বকারী পুরুষদের জাতীয় দল, যার সকল কার্যক্রম বাংলাদেশের ফুটবলের সর্বোচ্চ নিয়ন্ত্রক সংস্থা বাংলাদেশ ফুটবল ফেডারেশন দ্বারা নিয়ন্ত্রিত হয়। এই দলটি ১৯৭৬ সাল হতে ফুটবলের সর্বোচ্চ সংস্থা ফিফার এবং ১৯৭৪ সাল হতে তাদের আঞ্চলিক সংস্থা এশিয়ান ফুটবল কনফেডারেশনের সদস্য হিসেবে রয়েছে। ১৯৭৩ সালের ২৬শে জুলাই তারিখে, বাংলাদেশ প্রথমবারের মতো আন্তর্জাতিক খেলায় অংশগ্রহণ করেছে; মালয়েশিয়ার কুয়ালালামপুরে অনুষ্ঠিত বাংলাদেশ এবং থাইল্যান্ডের মধ্যকার উক্ত ম্যাচটি ২–২ গোলে ড্র হয়েছে।
+
+৩৬,০০০ ধারণক্ষমতাবিশিষ্ট বঙ্গবন্ধু জাতীয় স্টেডিয়ামে বাংলার বাঘ নামে পরিচিত এই দলটি তাদের সকল হোম ম্যাচ আয়োজন করে থাকে। এই দলের প্রধান কার্যালয় বাংলাদেশের রাজধানী ঢাকার মতিঝিলের বঙ্গবন্ধু জাতীয় স্টেডিয়ামের নিকটবর্তী বিএফএফ ভবনে অবস্থিত। বর্তমানে এই দলের ম্যানেজারের দায়িত্ব পালন করছেন হাভিয়ের কাবরেরা এবং অধিনায়কের দায়িত্ব পালন করছেন কলকাতা মোহামেডানের মধ্যমাঠের খেলোয়াড় জামাল ভূইয়া।

case-studies/poem.md

@@ -0,0 +1,44 @@
+# কবিতা (Poem): স্বাধীনতার খুশি
+- কবি (Poet): অমরেশ বিশ্বাস
+- Collected from: [URL](https://www.poetrystate.com/amaresh/%E0%A6%B8%E0%A7%8D%E0%A6%AC%E0%A6%BE%E0%A6%A7%E0%A7%80%E0%A6%A8%E0%A6%A4%E0%A6%BE%E0%A6%B0-%E0%A6%96%E0%A7%81%E0%A6%B6%E0%A6%BF/)
+
+---
+---
+
+স্বাধীনতা কষ্টে পাওয়া
+
+অনেক লড়াই করে
+
+আমরা সবাই আজকে স্বাধীন
+
+শ্বাস নিই প্রাণ ভরে।
+
+---
+
+শপথ নিল দেশের মানুষ
+
+২৫শে মার্চ রাতে
+
+সইবে না দেশ অত্যাচার আর
+
+রাজাকারের হাতে।
+
+---
+
+উঠল ধ্বনি জয়বাংলার
+
+পাকিস্তান আর নয়
+
+বাঙালীরা ফেলল মুছে
+
+প্রাণের থেকে ভয়।
+
+---
+
+নয় নটা মাস চলল লড়াই
+
+স্বাধীন হল দেশ
+
+উড়ছে সবুজ লাল পতাকা
+
+নেই আজ খুশির শেষ।

model/components/denoiser/FeatureDN.py

@@ -0,0 +1,221 @@
+import math
+from typing import Optional
+
+import torch
+import torch.nn.functional as F
+from torch import Tensor
+from torch.nn import Parameter
+
+from torch_geometric.nn.conv import MessagePassing
+from torch_geometric.nn.dense.linear import Linear
+from torch_geometric.nn.inits import glorot, zeros
+from torch_geometric.typing import Adj, OptTensor, SparseTensor, torch_sparse
+from torch_geometric.utils import (
+    add_self_loops,
+    batched_negative_sampling,
+    dropout_edge,
+    is_undirected,
+    negative_sampling,
+    remove_self_loops,
+    softmax,
+    to_undirected,
+)
+
+
+class FeatureDeNoisingLayer(MessagePassing):
+
+    att_x: OptTensor
+    att_y: OptTensor
+
+    def __init__(self, in_channels: int, out_channels: int, heads: int = 1,
+                 concat: bool = True, negative_slope: float = 0.2,
+                 dropout: float = 0.0, add_self_loops: bool = True,
+                 bias: bool = True, attention_type: str = 'MX',
+                 neg_sample_ratio: float = 0.5, edge_sample_ratio: float = 1.0,
+                 is_undirected: bool = False, **kwargs):
+        kwargs.setdefault('aggr', 'add')
+        super().__init__(node_dim=0, **kwargs)
+
+        self.in_channels = in_channels
+        self.out_channels = out_channels
+        self.heads = heads
+        self.concat = concat
+        self.negative_slope = negative_slope
+        self.dropout = dropout
+        self.add_self_loops = add_self_loops
+        self.attention_type = attention_type
+        self.neg_sample_ratio = neg_sample_ratio
+        self.edge_sample_ratio = edge_sample_ratio
+        self.is_undirected = is_undirected
+
+        assert attention_type in ['MX', 'SD']
+        assert 0.0 < neg_sample_ratio and 0.0 < edge_sample_ratio <= 1.0
+
+        self.lin = Linear(in_channels, heads * out_channels, bias=False,
+                          weight_initializer='glorot')
+
+        if self.attention_type == 'MX':
+            self.att_l = Parameter(torch.empty(1, heads, out_channels))
+            self.att_r = Parameter(torch.empty(1, heads, out_channels))
+        else:  # self.attention_type == 'SD'
+            self.register_parameter('att_l', None)
+            self.register_parameter('att_r', None)
+
+        self.att_x = self.att_y = None  # x/y for self-supervision
+
+        if bias and concat:
+            self.bias = Parameter(torch.empty(heads * out_channels))
+        elif bias and not concat:
+            self.bias = Parameter(torch.empty(out_channels))
+        else:
+            self.register_parameter('bias', None)
+
+        self.reset_parameters()
+
+    def reset_parameters(self):
+        super().reset_parameters()
+        self.lin.reset_parameters()
+        glorot(self.att_l)
+        glorot(self.att_r)
+        zeros(self.bias)
+
+    def forward(
+        self,
+        x: Tensor,
+        edge_index: Adj,
+        neg_edge_index: OptTensor = None,
+        batch: OptTensor = None,
+    ) -> Tensor:
+        r"""Runs the forward pass of the module.
+
+        Args:
+            x (torch.Tensor): The input node features.
+            edge_index (torch.Tensor or SparseTensor): The edge indices.
+            neg_edge_index (torch.Tensor, optional): The negative edges to
+                train against. If not given, uses negative sampling to
+                calculate negative edges. (default: :obj:`None`)
+            batch (torch.Tensor, optional): The batch vector
+                :math:`\mathbf{b} \in {\{ 0, \ldots, B-1\}}^N`, which assigns
+                each element to a specific example.
+                Used when sampling negatives on-the-fly in mini-batch
+                scenarios. (default: :obj:`None`)
+        """
+        N, H, C = x.size(0), self.heads, self.out_channels
+
+        if self.add_self_loops:
+            if isinstance(edge_index, SparseTensor):
+                edge_index = torch_sparse.fill_diag(edge_index, 1.)
+            else:
+                edge_index, _ = remove_self_loops(edge_index)
+                edge_index, _ = add_self_loops(edge_index, num_nodes=N)
+
+        x = self.lin(x).view(-1, H, C)
+
+        # propagate_type: (x: Tensor)
+        out = self.propagate(edge_index, x=x)
+
+        if self.training:
+            if isinstance(edge_index, SparseTensor):
+                col, row, _ = edge_index.coo()
+                edge_index = torch.stack([row, col], dim=0)
+            pos_edge_index = self.positive_sampling(edge_index)
+
+            pos_att = self.get_attention(
+                edge_index_i=pos_edge_index[1],
+                x_i=x[pos_edge_index[1]],
+                x_j=x[pos_edge_index[0]],
+                num_nodes=x.size(0),
+                return_logits=True,
+            )
+
+            if neg_edge_index is None:
+                neg_edge_index = self.negative_sampling(edge_index, N, batch)
+
+            neg_att = self.get_attention(
+                edge_index_i=neg_edge_index[1],
+                x_i=x[neg_edge_index[1]],
+                x_j=x[neg_edge_index[0]],
+                num_nodes=x.size(0),
+                return_logits=True,
+            )
+
+            self.att_x = torch.cat([pos_att, neg_att], dim=0)
+            self.att_y = self.att_x.new_zeros(self.att_x.size(0))
+            self.att_y[:pos_edge_index.size(1)] = 1.
+
+        if self.concat is True:
+            out = out.view(-1, self.heads * self.out_channels)
+        else:
+            out = out.mean(dim=1)
+
+        if self.bias is not None:
+            out = out + self.bias
+
+        return out
+
+    def message(self, edge_index_i: Tensor, x_i: Tensor, x_j: Tensor,
+                size_i: Optional[int]) -> Tensor:
+        alpha = self.get_attention(edge_index_i, x_i, x_j, num_nodes=size_i)
+        alpha = F.dropout(alpha, p=self.dropout, training=self.training)
+        return x_j * alpha.view(-1, self.heads, 1)
+
+    def negative_sampling(self, edge_index: Tensor, num_nodes: int,
+                          batch: OptTensor = None) -> Tensor:
+
+        num_neg_samples = int(self.neg_sample_ratio * self.edge_sample_ratio *
+                              edge_index.size(1))
+
+        if not self.is_undirected and not is_undirected(
+                edge_index, num_nodes=num_nodes):
+            edge_index = to_undirected(edge_index, num_nodes=num_nodes)
+
+        if batch is None:
+            neg_edge_index = negative_sampling(edge_index, num_nodes,
+                                               num_neg_samples=num_neg_samples)
+        else:
+            neg_edge_index = batched_negative_sampling(
+                edge_index, batch, num_neg_samples=num_neg_samples)
+
+        return neg_edge_index
+
+    def positive_sampling(self, edge_index: Tensor) -> Tensor:
+        pos_edge_index, _ = dropout_edge(edge_index,
+                                         p=1. - self.edge_sample_ratio,
+                                         training=self.training)
+        return pos_edge_index
+
+    def get_attention(self, edge_index_i: Tensor, x_i: Tensor, x_j: Tensor,
+                      num_nodes: Optional[int],
+                      return_logits: bool = False) -> Tensor:
+
+        if self.attention_type == 'MX':
+            logits = (x_i * x_j).sum(dim=-1)
+            if return_logits:
+                return logits
+
+            alpha = (x_j * self.att_l).sum(-1) + (x_i * self.att_r).sum(-1)
+            alpha = alpha * logits.sigmoid()
+
+        else:  # self.attention_type == 'SD'
+            alpha = (x_i * x_j).sum(dim=-1) / math.sqrt(self.out_channels)
+            if return_logits:
+                return alpha
+
+        alpha = F.leaky_relu(alpha, self.negative_slope)
+        alpha = softmax(alpha, edge_index_i, num_nodes=num_nodes)
+        return alpha
+
+    def get_attention_loss(self) -> Tensor:
+        r"""Computes the self-supervised graph attention loss."""
+        if not self.training:
+            return torch.tensor([0], device=self.lin.weight.device)
+
+        return F.binary_cross_entropy_with_logits(
+            self.att_x.mean(dim=-1),
+            self.att_y,
+        )
+
+    def __repr__(self) -> str:
+        return (f'{self.__class__.__name__}({self.in_channels}, '
+                f'{self.out_channels}, heads={self.heads}, '
+                f'type={self.attention_type})')

model/components/relation_extractors/LocalRE.py

@@ -0,0 +1,130 @@
+from typing import Optional
+
+import torch
+from torch import Tensor
+from torch.nn import Parameter
+
+from torch_geometric.nn.conv import MessagePassing
+from torch_geometric.nn.dense.linear import Linear
+from torch_geometric.nn.inits import zeros
+from torch_geometric.typing import OptTensor
+from torch_geometric.utils import get_laplacian
+
+
+class LocalRelationExtractionLayer(MessagePassing):
+
+    def __init__(
+        self,
+        in_channels: int,
+        out_channels: int,
+        K: int,
+        normalization: Optional[str] = 'sym',
+        bias: bool = True,
+        **kwargs,
+    ):
+        kwargs.setdefault('aggr', 'add')
+        super().__init__(**kwargs)
+
+        assert K > 0
+        assert normalization in [None, 'sym', 'rw'], 'Invalid normalization'
+
+        self.in_channels = in_channels
+        self.out_channels = out_channels
+        self.normalization = normalization
+        self.lins = torch.nn.ModuleList([
+            Linear(in_channels, out_channels, bias=False,
+                   weight_initializer='glorot') for _ in range(K)
+        ])
+
+        if bias:
+            self.bias = Parameter(Tensor(out_channels))
+        else:
+            self.register_parameter('bias', None)
+
+        self.reset_parameters()
+
+    def reset_parameters(self):
+        super().reset_parameters()
+        for lin in self.lins:
+            lin.reset_parameters()
+        zeros(self.bias)
+
+    def __norm__(
+        self,
+        edge_index: Tensor,
+        num_nodes: Optional[int],
+        edge_weight: OptTensor,
+        normalization: Optional[str],
+        lambda_max: OptTensor = None,
+        dtype: Optional[int] = None,
+        batch: OptTensor = None,
+    ):
+        edge_index, edge_weight = get_laplacian(edge_index, edge_weight,
+                                                normalization, dtype,
+                                                num_nodes)
+        assert edge_weight is not None
+
+        if lambda_max is None:
+            lambda_max = 2.0 * edge_weight.max()
+        elif not isinstance(lambda_max, Tensor):
+            lambda_max = torch.tensor(lambda_max, dtype=dtype,
+                                      device=edge_index.device)
+        assert lambda_max is not None
+
+        if batch is not None and lambda_max.numel() > 1:
+            lambda_max = lambda_max[batch[edge_index[0]]]
+
+        edge_weight = (2.0 * edge_weight) / lambda_max
+        edge_weight.masked_fill_(edge_weight == float('inf'), 0)
+
+        loop_mask = edge_index[0] == edge_index[1]
+        edge_weight[loop_mask] -= 1
+
+        return edge_index, edge_weight
+
+    def forward(
+        self,
+        x: Tensor,
+        edge_index: Tensor,
+        edge_weight: OptTensor = None,
+        batch: OptTensor = None,
+        lambda_max: OptTensor = None,
+    ) -> Tensor:
+
+        edge_index, norm = self.__norm__(
+            edge_index,
+            x.size(self.node_dim),
+            edge_weight,
+            self.normalization,
+            lambda_max,
+            dtype=x.dtype,
+            batch=batch,
+        )
+
+        Tx_0 = x
+        Tx_1 = x  # Dummy.
+        out = self.lins[0](Tx_0)
+
+        # propagate_type: (x: Tensor, norm: Tensor)
+        if len(self.lins) > 1:
+            Tx_1 = self.propagate(edge_index, x=x, norm=norm)
+            out = out + self.lins[1](Tx_1)
+
+        for lin in self.lins[2:]:
+            Tx_2 = self.propagate(edge_index, x=Tx_1, norm=norm)
+            Tx_2 = 2. * Tx_2 - Tx_0
+            out = out + lin.forward(Tx_2)
+            Tx_0, Tx_1 = Tx_1, Tx_2
+
+        if self.bias is not None:
+            out = out + self.bias
+
+        return out
+
+    def message(self, x_j: Tensor, norm: Tensor) -> Tensor:
+        return norm.view(-1, 1) * x_j
+
+    def __repr__(self) -> str:
+        return (f'{self.__class__.__name__}({self.in_channels}, '
+                f'{self.out_channels}, K={len(self.lins)}, '
+                f'normalization={self.normalization})')

model/components/relation_extractors/TopologicalRE.py

@@ -0,0 +1,77 @@
+from typing import Final, Tuple, Union
+
+import torch
+from torch import Tensor
+
+from torch_geometric import EdgeIndex
+from torch_geometric.nn.conv import MessagePassing
+from torch_geometric.nn.dense.linear import Linear
+from torch_geometric.typing import Adj, OptPairTensor, OptTensor, Size
+from torch_geometric.utils import spmm
+
+
+class TopologicalRelationExtractionLayer(MessagePassing):
+    SUPPORTS_FUSED_EDGE_INDEX: Final[bool] = True
+
+    def __init__(
+        self,
+        in_channels: Union[int, Tuple[int, int]],
+        out_channels: int,
+        aggr: str = 'add',
+        bias: bool = True,
+        **kwargs,
+    ):
+        super().__init__(aggr=aggr, **kwargs)
+
+        self.in_channels = in_channels
+        self.out_channels = out_channels
+
+        if isinstance(in_channels, int):
+            in_channels = (in_channels, in_channels)
+
+        self.lin_rel = Linear(in_channels[0], out_channels, bias=bias)
+        self.lin_root = Linear(in_channels[1], out_channels, bias=False)
+
+        self.reset_parameters()
+
+    def reset_parameters(self):
+        super().reset_parameters()
+        self.lin_rel.reset_parameters()
+        self.lin_root.reset_parameters()
+
+    def forward(self, x: Union[Tensor, OptPairTensor], edge_index: Adj,
+                edge_weight: OptTensor = None, size: Size = None) -> Tensor:
+
+        if isinstance(x, Tensor):
+            x = (x, x)
+
+        # propagate_type: (x: OptPairTensor, edge_weight: OptTensor)
+        out = self.propagate(edge_index, x=x, edge_weight=edge_weight,
+                             size=size)
+        out = self.lin_rel(out)
+
+        x_r = x[1]
+        if x_r is not None:
+            out = out + self.lin_root(x_r)
+
+        return out
+
+    def message(self, x_j: Tensor, edge_weight: OptTensor) -> Tensor:
+        return x_j if edge_weight is None else edge_weight.view(-1, 1) * x_j
+
+    def message_and_aggregate(
+        self,
+        edge_index: Adj,
+        x: OptPairTensor,
+        edge_weight: OptTensor,
+    ) -> Tensor:
+
+        if not torch.jit.is_scripting() and isinstance(edge_index, EdgeIndex):
+            return edge_index.matmul(
+                other=x[0],
+                input_value=edge_weight,
+                reduce=self.aggr,
+                transpose=True,
+            )
+
+        return spmm(edge_index, x[0], reduce=self.aggr)

model/components/semantic_info_conv/Semantic_IC.py

@@ -0,0 +1,298 @@
+import typing
+from typing import Optional, Tuple, Union
+
+import torch
+import torch.nn.functional as F
+from torch import Tensor
+from torch.nn import Parameter
+
+from torch_geometric.nn.conv import MessagePassing
+from torch_geometric.nn.dense.linear import Linear
+from torch_geometric.nn.inits import glorot, zeros
+from torch_geometric.typing import (
+    Adj,
+    NoneType,
+    OptPairTensor,
+    OptTensor,
+    Size,
+    SparseTensor,
+    torch_sparse,
+)
+from torch_geometric.utils import (
+    add_self_loops,
+    is_torch_sparse_tensor,
+    remove_self_loops,
+    softmax,
+)
+from torch_geometric.utils.sparse import set_sparse_value
+
+if typing.TYPE_CHECKING:
+    from typing import overload
+else:
+    from torch.jit import _overload_method as overload
+
+
+class SemanticInformationConvolutionLayer(MessagePassing):
+
+    def __init__(
+        self,
+        in_channels: Union[int, Tuple[int, int]],
+        out_channels: int,
+        heads: int = 1,
+        concat: bool = True,
+        negative_slope: float = 0.2,
+        dropout: float = 0.0,
+        add_self_loops: bool = True,
+        edge_dim: Optional[int] = None,
+        fill_value: Union[float, Tensor, str] = 'mean',
+        bias: bool = True,
+        **kwargs,
+    ):
+        kwargs.setdefault('aggr', 'add')
+        super().__init__(node_dim=0, **kwargs)
+
+        self.in_channels = in_channels
+        self.out_channels = out_channels
+        self.heads = heads
+        self.concat = concat
+        self.negative_slope = negative_slope
+        self.dropout = dropout
+        self.add_self_loops = add_self_loops
+        self.edge_dim = edge_dim
+        self.fill_value = fill_value
+
+        # In case we are operating in bipartite graphs, we apply separate
+        # transformations 'lin_src' and 'lin_dst' to source and target nodes:
+        self.lin = self.lin_src = self.lin_dst = None
+        if isinstance(in_channels, int):
+            self.lin = Linear(in_channels, heads * out_channels, bias=False,
+                              weight_initializer='glorot')
+        else:
+            self.lin_src = Linear(in_channels[0], heads * out_channels, False,
+                                  weight_initializer='glorot')
+            self.lin_dst = Linear(in_channels[1], heads * out_channels, False,
+                                  weight_initializer='glorot')
+
+        # The learnable parameters to compute attention coefficients:
+        self.att_src = Parameter(torch.empty(1, heads, out_channels))
+        self.att_dst = Parameter(torch.empty(1, heads, out_channels))
+
+        if edge_dim is not None:
+            self.lin_edge = Linear(edge_dim, heads * out_channels, bias=False,
+                                   weight_initializer='glorot')
+            self.att_edge = Parameter(torch.empty(1, heads, out_channels))
+        else:
+            self.lin_edge = None
+            self.register_parameter('att_edge', None)
+
+        if bias and concat:
+            self.bias = Parameter(torch.empty(heads * out_channels))
+        elif bias and not concat:
+            self.bias = Parameter(torch.empty(out_channels))
+        else:
+            self.register_parameter('bias', None)
+
+        self.reset_parameters()
+
+    def reset_parameters(self):
+        super().reset_parameters()
+        if self.lin is not None:
+            self.lin.reset_parameters()
+        if self.lin_src is not None:
+            self.lin_src.reset_parameters()
+        if self.lin_dst is not None:
+            self.lin_dst.reset_parameters()
+        if self.lin_edge is not None:
+            self.lin_edge.reset_parameters()
+        glorot(self.att_src)
+        glorot(self.att_dst)
+        glorot(self.att_edge)
+        zeros(self.bias)
+
+    @overload
+    def forward(
+        self,
+        x: Union[Tensor, OptPairTensor],
+        edge_index: Adj,
+        edge_attr: OptTensor = None,
+        size: Size = None,
+        return_attention_weights: NoneType = None,
+    ) -> Tensor:
+        pass
+
+    @overload
+    def forward(  # noqa: F811
+        self,
+        x: Union[Tensor, OptPairTensor],
+        edge_index: Tensor,
+        edge_attr: OptTensor = None,
+        size: Size = None,
+        return_attention_weights: bool = None,
+    ) -> Tuple[Tensor, Tuple[Tensor, Tensor]]:
+        pass
+
+    @overload
+    def forward(  # noqa: F811
+        self,
+        x: Union[Tensor, OptPairTensor],
+        edge_index: SparseTensor,
+        edge_attr: OptTensor = None,
+        size: Size = None,
+        return_attention_weights: bool = None,
+    ) -> Tuple[Tensor, SparseTensor]:
+        pass
+
+    def forward(  # noqa: F811
+        self,
+        x: Union[Tensor, OptPairTensor],
+        edge_index: Adj,
+        edge_attr: OptTensor = None,
+        size: Size = None,
+        return_attention_weights: Optional[bool] = None,
+    ) -> Union[
+            Tensor,
+            Tuple[Tensor, Tuple[Tensor, Tensor]],
+            Tuple[Tensor, SparseTensor],
+    ]:
+        r"""Runs the forward pass of the module.
+
+        Args:
+            x (torch.Tensor or (torch.Tensor, torch.Tensor)): The input node
+                features.
+            edge_index (torch.Tensor or SparseTensor): The edge indices.
+            edge_attr (torch.Tensor, optional): The edge features.
+                (default: :obj:`None`)
+            size ((int, int), optional): The shape of the adjacency matrix.
+                (default: :obj:`None`)
+            return_attention_weights (bool, optional): If set to :obj:`True`,
+                will additionally return the tuple
+                :obj:`(edge_index, attention_weights)`, holding the computed
+                attention weights for each edge. (default: :obj:`None`)
+        """
+        # NOTE: attention weights will be returned whenever
+        # `return_attention_weights` is set to a value, regardless of its
+        # actual value (might be `True` or `False`). This is a current somewhat
+        # hacky workaround to allow for TorchScript support via the
+        # `torch.jit._overload` decorator, as we can only change the output
+        # arguments conditioned on type (`None` or `bool`), not based on its
+        # actual value.
+
+        H, C = self.heads, self.out_channels
+
+        # We first transform the input node features. If a tuple is passed, we
+        # transform source and target node features via separate weights:
+        if isinstance(x, Tensor):
+            assert x.dim() == 2, "Static graphs not supported in 'GATConv'"
+
+            if self.lin is not None:
+                x_src = x_dst = self.lin(x).view(-1, H, C)
+            else:
+                # If the module is initialized as bipartite, transform source
+                # and destination node features separately:
+                assert self.lin_src is not None and self.lin_dst is not None
+                x_src = self.lin_src(x).view(-1, H, C)
+                x_dst = self.lin_dst(x).view(-1, H, C)
+
+        else:  # Tuple of source and target node features:
+            x_src, x_dst = x
+            assert x_src.dim() == 2, "Static graphs not supported in 'GATConv'"
+
+            if self.lin is not None:
+                # If the module is initialized as non-bipartite, we expect that
+                # source and destination node features have the same shape and
+                # that they their transformations are shared:
+                x_src = self.lin(x_src).view(-1, H, C)
+                if x_dst is not None:
+                    x_dst = self.lin(x_dst).view(-1, H, C)
+            else:
+                assert self.lin_src is not None and self.lin_dst is not None
+
+                x_src = self.lin_src(x_src).view(-1, H, C)
+                if x_dst is not None:
+                    x_dst = self.lin_dst(x_dst).view(-1, H, C)
+
+        x = (x_src, x_dst)
+
+        # Next, we compute node-level attention coefficients, both for source
+        # and target nodes (if present):
+        alpha_src = (x_src * self.att_src).sum(dim=-1)
+        alpha_dst = None if x_dst is None else (x_dst * self.att_dst).sum(-1)
+        alpha = (alpha_src, alpha_dst)
+
+        if self.add_self_loops:
+            if isinstance(edge_index, Tensor):
+                # We only want to add self-loops for nodes that appear both as
+                # source and target nodes:
+                num_nodes = x_src.size(0)
+                if x_dst is not None:
+                    num_nodes = min(num_nodes, x_dst.size(0))
+                num_nodes = min(size) if size is not None else num_nodes
+                edge_index, edge_attr = remove_self_loops(
+                    edge_index, edge_attr)
+                edge_index, edge_attr = add_self_loops(
+                    edge_index, edge_attr, fill_value=self.fill_value,
+                    num_nodes=num_nodes)
+            elif isinstance(edge_index, SparseTensor):
+                if self.edge_dim is None:
+                    edge_index = torch_sparse.set_diag(edge_index)
+                else:
+                    raise NotImplementedError(
+                        "The usage of 'edge_attr' and 'add_self_loops' "
+                        "simultaneously is currently not yet supported for "
+                        "'edge_index' in a 'SparseTensor' form")
+
+        # edge_updater_type: (alpha: OptPairTensor, edge_attr: OptTensor)
+        alpha = self.edge_updater(edge_index, alpha=alpha, edge_attr=edge_attr,
+                                  size=size)
+
+        # propagate_type: (x: OptPairTensor, alpha: Tensor)
+        out = self.propagate(edge_index, x=x, alpha=alpha, size=size)
+
+        if self.concat:
+            out = out.view(-1, self.heads * self.out_channels)
+        else:
+            out = out.mean(dim=1)
+
+        if self.bias is not None:
+            out = out + self.bias
+
+        if isinstance(return_attention_weights, bool):
+            if isinstance(edge_index, Tensor):
+                if is_torch_sparse_tensor(edge_index):
+                    # TODO TorchScript requires to return a tuple
+                    adj = set_sparse_value(edge_index, alpha)
+                    return out, (adj, alpha)
+                else:
+                    return out, (edge_index, alpha)
+            elif isinstance(edge_index, SparseTensor):
+                return out, edge_index.set_value(alpha, layout='coo')
+        else:
+            return out
+
+    def edge_update(self, alpha_j: Tensor, alpha_i: OptTensor,
+                    edge_attr: OptTensor, index: Tensor, ptr: OptTensor,
+                    dim_size: Optional[int]) -> Tensor:
+        # Given edge-level attention coefficients for source and target nodes,
+        # we simply need to sum them up to "emulate" concatenation:
+        alpha = alpha_j if alpha_i is None else alpha_j + alpha_i
+        if index.numel() == 0:
+            return alpha
+        if edge_attr is not None and self.lin_edge is not None:
+            if edge_attr.dim() == 1:
+                edge_attr = edge_attr.view(-1, 1)
+            edge_attr = self.lin_edge(edge_attr)
+            edge_attr = edge_attr.view(-1, self.heads, self.out_channels)
+            alpha_edge = (edge_attr * self.att_edge).sum(dim=-1)
+            alpha = alpha + alpha_edge
+
+        alpha = F.leaky_relu(alpha, self.negative_slope)
+        alpha = softmax(alpha, index, ptr, dim_size)
+        alpha = F.dropout(alpha, p=self.dropout, training=self.training)
+        return alpha
+
+    def message(self, x_j: Tensor, alpha: Tensor) -> Tensor:
+        return alpha.unsqueeze(-1) * x_j
+
+    def __repr__(self) -> str:
+        return (f'{self.__class__.__name__}({self.in_channels}, '
+                f'{self.out_channels}, heads={self.heads})')

model/metric/A_SFAS.py

@@ -0,0 +1,34 @@
+import numpy as np
+
+def calculate_A_SFAS(tensor1, tensor2):
+    """
+    Computes the Average Semantic Feature Alignment scores between two n-dimensional tensors.
+
+    Parameters:
+    tensor1 (numpy.ndarray): The first tensor.
+    tensor2 (numpy.ndarray): The second tensor.
+
+    Returns:
+    float: The Average Semantic Feature Alignment scores between tensor1 and tensor2.
+    """
+    # Flatten the tensors to 1-dimensional arrays
+    flat_tensor1 = tensor1.flatten()
+    flat_tensor2 = tensor2.flatten()
+    
+    # Compute the dot product
+    dot_product = np.dot(flat_tensor1, flat_tensor2)
+    
+    # Compute the magnitudes (norms) of the tensors
+    norm_tensor1 = np.linalg.norm(flat_tensor1)
+    norm_tensor2 = np.linalg.norm(flat_tensor2)
+    
+    # Compute the cosine similarity
+    cosine_sim = dot_product / (norm_tensor1 * norm_tensor2)
+    
+    return cosine_sim
+
+# Example usage:
+tensor1 = np.array([[1, 2, 3], [4, 5, 6]])
+tensor2 = np.array([[1, 0, 0], [0, 1, 0]])
+similarity = cosine_similarity(tensor1, tensor2)
+print("Cosine Similarity:", similarity)