encoders

AutoEncoderDeepBlockerFrameEncoder

Bases: DeepBlockerFrameEncoder[Tensor]

Autoencoder class for DeepBlocker Frame encoders.

---

hidden_dimensions: Tuple[int, int]: Hidden dimensions
num_epochs: int: Number of epochs if training
batch_size: int: Batch size
learning_rate: float: Learning rate if training
loss_function: Optional[_Loss]: Loss function if training
optimizer: Optional[HintOrType[Optimizer]]: Optimizer if training
optimizer_kwargs: OptionalKwargs: Keyword arguments to inizialize optimizer
frame_encoder: HintOrType[TokenizedFrameEncoder]: Base encoder class
frame_encoder_kwargs: OptionalKwargs: Keyword arguments for initializing frame encoder

Source code in klinker/encoders/deepblocker.py

class AutoEncoderDeepBlockerFrameEncoder(DeepBlockerFrameEncoder[torch.Tensor]):
    """Autoencoder class for DeepBlocker Frame encoders.

    Args:
    ----
        hidden_dimensions: Tuple[int, int]: Hidden dimensions
        num_epochs: int: Number of epochs if training
        batch_size: int: Batch size
        learning_rate: float: Learning rate if training
        loss_function: Optional[_Loss]: Loss function if training
        optimizer: Optional[HintOrType[Optimizer]]: Optimizer if training
        optimizer_kwargs: OptionalKwargs: Keyword arguments to inizialize optimizer
        frame_encoder: HintOrType[TokenizedFrameEncoder]: Base encoder class
        frame_encoder_kwargs: OptionalKwargs: Keyword arguments for initializing frame encoder
    """

    def __init__(
        self,
        hidden_dimensions: Tuple[int, int] = (2 * 150, 150),
        num_epochs: int = 50,
        batch_size: int = 256,
        learning_rate: float = 1e-3,
        frame_encoder: HintOrType[TokenizedFrameEncoder] = None,
        frame_encoder_kwargs: OptionalKwargs = None,
        **kwargs,
    ):
        super().__init__(
            hidden_dimensions=hidden_dimensions,
            num_epochs=num_epochs,
            batch_size=batch_size,
            learning_rate=learning_rate,
            frame_encoder=frame_encoder,
            frame_encoder_kwargs=frame_encoder_kwargs,
            **kwargs,
        )
        self._input_dimension = -1

    @property
    def trainer_cls(self) -> Type[DeepBlockerModelTrainer[torch.Tensor]]:
        return AutoEncoderDeepBlockerModelTrainer

    def create_features(
        self, left: Frame, right: Frame
    ) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
        """Features for AutoEncoder.

        Args:
        ----
          left: Frame: left attributes.
          right: Frame: right attributes.

        Returns:
        -------
            Concatenated left/right encoded, left encoded, right encoded
        """
        left_enc, right_enc = self.inner_encoder._encode_as(
            left, right, return_type="pt"
        )
        left_enc = left_enc.float()
        right_enc = right_enc.float()

        self.input_dimension = left_enc.shape[1]
        return (
            torch.concat([left_enc, right_enc]),
            left_enc,
            right_enc,
        )

create_features(left, right)

Features for AutoEncoder.

---

left: Frame: left attributes. right: Frame: right attributes.

---

Concatenated left/right encoded, left encoded, right encoded

Source code in klinker/encoders/deepblocker.py

def create_features(
    self, left: Frame, right: Frame
) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
    """Features for AutoEncoder.

    Args:
    ----
      left: Frame: left attributes.
      right: Frame: right attributes.

    Returns:
    -------
        Concatenated left/right encoded, left encoded, right encoded
    """
    left_enc, right_enc = self.inner_encoder._encode_as(
        left, right, return_type="pt"
    )
    left_enc = left_enc.float()
    right_enc = right_enc.float()

    self.input_dimension = left_enc.shape[1]
    return (
        torch.concat([left_enc, right_enc]),
        left_enc,
        right_enc,
    )

AverageEmbeddingTokenizedFrameEncoder

Bases: TokenizedFrameEncoder

Averages embeddings of tokenized entity attribute values.

---

tokenized_word_embedder: HintOrType[TokenizedWordEmbedder]: Word Embedding class,
tokenized_word_embedder_kwargs: OptionalKwargs: Keyword arguments for initalizing word embedder

Source code in klinker/encoders/pretrained.py

class AverageEmbeddingTokenizedFrameEncoder(TokenizedFrameEncoder):
    """Averages embeddings of tokenized entity attribute values.

    Args:
    ----
        tokenized_word_embedder: HintOrType[TokenizedWordEmbedder]: Word Embedding class,
        tokenized_word_embedder_kwargs: OptionalKwargs: Keyword arguments for initalizing word embedder
    """

    def __init__(
        self,
        tokenized_word_embedder: HintOrType[TokenizedWordEmbedder] = None,
        tokenized_word_embedder_kwargs: OptionalKwargs = None,
    ):
        self.tokenized_word_embedder = tokenized_word_embedder_resolver.make(
            tokenized_word_embedder, tokenized_word_embedder_kwargs
        )

    @property
    def tokenizer_fn(self) -> Callable[[str], List[str]]:
        return self.tokenized_word_embedder.tokenizer_fn

    def _encode(
        self,
        left: Frame,
        right: Frame,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
    ) -> Tuple[GeneralVector, GeneralVector]:
        if isinstance(left, dd.DataFrame):
            left = left.compute()
            right = right.compute()
        return (
            encode_frame(left, twe=self.tokenized_word_embedder),
            encode_frame(right, twe=self.tokenized_word_embedder),
        )

CrossTupleTrainingDeepBlockerFrameEncoder

Bases: DeepBlockerFrameEncoder

CrossTupleTraining class for DeepBlocker Frame encoders.

---

hidden_dimensions: Tuple[int, int]: Hidden dimensions
num_epochs: int: Number of epochs
batch_size: int: Batch size
learning_rate: float: Learning rate
synth_tuples_per_tuple: int: Synthetic tuples per tuple
pos_to_neg_ratio: float: Ratio of positiv to negative tuples
max_perturbation:float: Degree how much tuples should be corrupted
random_seed: Seed to control randomness
loss_function: Optional[_Loss]: Loss function if training
optimizer: Optional[HintOrType[Optimizer]]: Optimizer if training
optimizer_kwargs: OptionalKwargs: Keyword arguments to inizialize optimizer
frame_encoder: HintOrType[TokenizedFrameEncoder]: Base encoder class
frame_encoder_kwargs: OptionalKwargs: Keyword arguments for initializing frame encoder

Source code in klinker/encoders/deepblocker.py

class CrossTupleTrainingDeepBlockerFrameEncoder(DeepBlockerFrameEncoder):
    """CrossTupleTraining class for DeepBlocker Frame encoders.

    Args:
    ----
        hidden_dimensions: Tuple[int, int]: Hidden dimensions
        num_epochs: int: Number of epochs
        batch_size: int: Batch size
        learning_rate: float: Learning rate
        synth_tuples_per_tuple: int: Synthetic tuples per tuple
        pos_to_neg_ratio: float: Ratio of positiv to negative tuples
        max_perturbation:float: Degree how much tuples should be corrupted
        random_seed: Seed to control randomness
        loss_function: Optional[_Loss]: Loss function if training
        optimizer: Optional[HintOrType[Optimizer]]: Optimizer if training
        optimizer_kwargs: OptionalKwargs: Keyword arguments to inizialize optimizer
        frame_encoder: HintOrType[TokenizedFrameEncoder]: Base encoder class
        frame_encoder_kwargs: OptionalKwargs: Keyword arguments for initializing frame encoder
    """

    def __init__(
        self,
        hidden_dimensions: Tuple[int, int] = (2 * 150, 150),
        num_epochs: int = 50,
        batch_size: int = 256,
        learning_rate: float = 1e-3,
        synth_tuples_per_tuple: int = 5,
        pos_to_neg_ratio: float = 1.0,
        max_perturbation: float = 0.4,
        random_seed=None,
        loss_function: Optional[_Loss] = None,
        optimizer: Optional[HintOrType[Optimizer]] = None,
        optimizer_kwargs: OptionalKwargs = None,
        frame_encoder: HintOrType[TokenizedFrameEncoder] = None,
        frame_encoder_kwargs: OptionalKwargs = None,
    ):
        super().__init__(
            hidden_dimensions=hidden_dimensions,
            num_epochs=num_epochs,
            batch_size=batch_size,
            learning_rate=learning_rate,
            frame_encoder=frame_encoder,
            frame_encoder_kwargs=frame_encoder_kwargs,
            loss_function=loss_function,
            optimizer=optimizer,
            optimizer_kwargs=optimizer_kwargs,
        )
        self.synth_tuples_per_tuple = synth_tuples_per_tuple
        self.pos_to_neg_ratio = pos_to_neg_ratio
        self.max_perturbation = max_perturbation
        self.random_seed = random_seed

    def create_features(
        self, left: Frame, right: Frame
    ) -> Tuple[
        Tuple[torch.Tensor, torch.Tensor, torch.Tensor], torch.Tensor, torch.Tensor
    ]:
        """Create features for cross-tuple training.

        Args:
        ----
          left: Frame: left attributes.
          right: Frame: right attributes.

        Returns:
        -------
            (left_training, right_training, labels), left encoded, right encoded
        """
        if isinstance(left, KlinkerDaskFrame):
            raise NotImplementedError(
                "CrossTupleTrainingDeepBlockerFrameEncoder has not been implemented for dask yet!"
            )

        # TODO refactor this function (copy-pasted from deepblocker repo)
        list_of_tuples = pd.DataFrame(
            np.concatenate([left.values, right.values]), columns=["merged"]
        )["merged"]
        num_positives_per_tuple = self.synth_tuples_per_tuple
        num_negatives_per_tuple = int(
            self.synth_tuples_per_tuple * self.pos_to_neg_ratio
        )
        num_tuples = len(list_of_tuples)
        total_number_of_elems = int(
            num_tuples * (num_positives_per_tuple + num_negatives_per_tuple)
        )

        # We create three lists containing T, T' and L respectively
        # We use the following format: first num_tuples * num_positives_per_tuple correspond to T
        # and the remaining correspond to T'
        left_tuple_list = ["" for _ in range(total_number_of_elems)]
        right_tuple_list = ["" for _ in range(total_number_of_elems)]
        label_list = [0 for _ in range(total_number_of_elems)]

        random.seed(self.random_seed)

        tokenizer = self.inner_encoder.tokenizer_fn
        for index in range(len(list_of_tuples)):
            tokenized_tuple = tokenizer(list_of_tuples[index])
            max_tokens_to_remove = int(len(tokenized_tuple) * self.max_perturbation)

            training_data_index = index * (
                num_positives_per_tuple + num_negatives_per_tuple
            )

            # Create num_positives_per_tuple tuple pairs with positive label
            for _ in range(num_positives_per_tuple):
                tokenized_tuple_copy = tokenized_tuple[:]

                # If the tuple has 10 words and max_tokens_to_remove is 0.5, then we can remove at most 5 words
                # we choose a random number between 0 and 5.
                # suppose it is 3. Then we randomly remove 3 words
                num_tokens_to_remove = random.randint(0, max_tokens_to_remove)
                for _ in range(num_tokens_to_remove):
                    # randint is inclusive. so randint(0, 5) can return 5 also
                    tokenized_tuple_copy.pop(
                        random.randint(0, len(tokenized_tuple_copy) - 1)
                    )

                left_tuple_list[training_data_index] = list_of_tuples[index]
                right_tuple_list[training_data_index] = " ".join(tokenized_tuple_copy)
                label_list[training_data_index] = 1
                training_data_index += 1

            for _ in range(num_negatives_per_tuple):
                left_tuple_list[training_data_index] = list_of_tuples[index]
                right_tuple_list[training_data_index] = random.choice(list_of_tuples)
                label_list[training_data_index] = 0
                training_data_index += 1
        left_train_enc, right_train_enc = self.inner_encoder._encode_as(
            pd.DataFrame(left_tuple_list),
            pd.DataFrame(right_tuple_list),
            return_type="pt",
        )
        self.input_dimension = left_train_enc.shape[1]

        left_enc, right_enc = self.inner_encoder._encode_as(
            left, right, return_type="pt"
        )
        return (
            (left_train_enc.float(), right_train_enc.float(), torch.tensor(label_list)),
            left_enc.float(),
            right_enc.float(),
        )

    def _encode(
        self,
        left: Frame,
        right: Frame,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
    ) -> Tuple[GeneralVector, GeneralVector]:
        self.inner_encoder.prepare(left, right)
        (
            (left_train, right_train, label_list),
            left_enc,
            right_enc,
        ) = self.create_features(left, right)

        assert self.input_dimension is not None
        trainer = CTTDeepBlockerModelTrainer(
            input_dimension=self.input_dimension,
            hidden_dimensions=self.hidden_dimensions,
            learning_rate=self.learning_rate,
        )
        features = (left_train, right_train, torch.tensor(label_list))
        device = resolve_device()
        self.ctt_model = trainer.train(
            features=features,
            num_epochs=self.num_epochs,
            batch_size=self.batch_size,
            device=device,
        )

        return self.ctt_model.encode_side(left_enc, device), self.ctt_model.encode_side(
            right_enc, device
        )

create_features(left, right)

Create features for cross-tuple training.

---

left: Frame: left attributes. right: Frame: right attributes.

---

(left_training, right_training, labels), left encoded, right encoded

Source code in klinker/encoders/deepblocker.py

def create_features(
    self, left: Frame, right: Frame
) -> Tuple[
    Tuple[torch.Tensor, torch.Tensor, torch.Tensor], torch.Tensor, torch.Tensor
]:
    """Create features for cross-tuple training.

    Args:
    ----
      left: Frame: left attributes.
      right: Frame: right attributes.

    Returns:
    -------
        (left_training, right_training, labels), left encoded, right encoded
    """
    if isinstance(left, KlinkerDaskFrame):
        raise NotImplementedError(
            "CrossTupleTrainingDeepBlockerFrameEncoder has not been implemented for dask yet!"
        )

    # TODO refactor this function (copy-pasted from deepblocker repo)
    list_of_tuples = pd.DataFrame(
        np.concatenate([left.values, right.values]), columns=["merged"]
    )["merged"]
    num_positives_per_tuple = self.synth_tuples_per_tuple
    num_negatives_per_tuple = int(
        self.synth_tuples_per_tuple * self.pos_to_neg_ratio
    )
    num_tuples = len(list_of_tuples)
    total_number_of_elems = int(
        num_tuples * (num_positives_per_tuple + num_negatives_per_tuple)
    )

    # We create three lists containing T, T' and L respectively
    # We use the following format: first num_tuples * num_positives_per_tuple correspond to T
    # and the remaining correspond to T'
    left_tuple_list = ["" for _ in range(total_number_of_elems)]
    right_tuple_list = ["" for _ in range(total_number_of_elems)]
    label_list = [0 for _ in range(total_number_of_elems)]

    random.seed(self.random_seed)

    tokenizer = self.inner_encoder.tokenizer_fn
    for index in range(len(list_of_tuples)):
        tokenized_tuple = tokenizer(list_of_tuples[index])
        max_tokens_to_remove = int(len(tokenized_tuple) * self.max_perturbation)

        training_data_index = index * (
            num_positives_per_tuple + num_negatives_per_tuple
        )

        # Create num_positives_per_tuple tuple pairs with positive label
        for _ in range(num_positives_per_tuple):
            tokenized_tuple_copy = tokenized_tuple[:]

            # If the tuple has 10 words and max_tokens_to_remove is 0.5, then we can remove at most 5 words
            # we choose a random number between 0 and 5.
            # suppose it is 3. Then we randomly remove 3 words
            num_tokens_to_remove = random.randint(0, max_tokens_to_remove)
            for _ in range(num_tokens_to_remove):
                # randint is inclusive. so randint(0, 5) can return 5 also
                tokenized_tuple_copy.pop(
                    random.randint(0, len(tokenized_tuple_copy) - 1)
                )

            left_tuple_list[training_data_index] = list_of_tuples[index]
            right_tuple_list[training_data_index] = " ".join(tokenized_tuple_copy)
            label_list[training_data_index] = 1
            training_data_index += 1

        for _ in range(num_negatives_per_tuple):
            left_tuple_list[training_data_index] = list_of_tuples[index]
            right_tuple_list[training_data_index] = random.choice(list_of_tuples)
            label_list[training_data_index] = 0
            training_data_index += 1
    left_train_enc, right_train_enc = self.inner_encoder._encode_as(
        pd.DataFrame(left_tuple_list),
        pd.DataFrame(right_tuple_list),
        return_type="pt",
    )
    self.input_dimension = left_train_enc.shape[1]

    left_enc, right_enc = self.inner_encoder._encode_as(
        left, right, return_type="pt"
    )
    return (
        (left_train_enc.float(), right_train_enc.float(), torch.tensor(label_list)),
        left_enc.float(),
        right_enc.float(),
    )

FrameEncoder

Base class for encoding a KlinkerFrame as embedding.

Source code in klinker/encoders/base.py

class FrameEncoder:
    """Base class for encoding a KlinkerFrame as embedding."""

    def validate(
        self,
        left: Frame,
        right: Frame,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
    ):
        """Check if frames only consist of one column.

        Args:
        ----
          left: Frame: left attributes.
          right: Frame: right attributes.
          left_rel: Optional[Frame]: left relation triples.
          right_rel: Optional[Frame]: right relation triples.

        Raises:
        ------
            ValueError left/right have more than one column.
        """
        if len(left.columns) != 1 or len(right.columns) != 1:
            raise ValueError(
                "Input DataFrames must consist of single column containing all attribute values!"
            )

    def prepare(self, left: Frame, right: Frame) -> Tuple[Frame, Frame]:
        """Prepare for embedding (fill NaNs with empty string).

        Args:
        ----
          left: Frame: left attributes.
          right: Frame: right attributes.

        Returns:
        -------
            left, right
        """
        return left.fillna(""), right.fillna("")

    def _encode(
        self,
        left: Frame,
        right: Frame,
        *,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
    ) -> Tuple[GeneralVector, GeneralVector]:
        raise NotImplementedError

    @overload
    def _encode_as(
        self,
        left: Frame,
        right: Frame,
        *,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
        return_type: Literal["np"],
    ) -> Tuple[np.ndarray, np.ndarray]:
        ...

    @overload
    def _encode_as(
        self,
        left: Frame,
        right: Frame,
        *,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
        return_type: Literal["pt"],
    ) -> Tuple[torch.Tensor, torch.Tensor]:
        ...

    def _encode_as(
        self,
        left: Frame,
        right: Frame,
        *,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
        return_type: GeneralVectorLiteral = "pt",
    ) -> Tuple[GeneralVector, GeneralVector]:
        start = time.time()
        left_enc, right_enc = self._encode(
            left=left, right=right, left_rel=left_rel, right_rel=right_rel
        )
        left_enc = cast_general_vector(left_enc, return_type=return_type)
        right_enc = cast_general_vector(right_enc, return_type=return_type)
        end = time.time()
        self._encoding_time = end - start
        return left_enc, right_enc

    def encode(
        self,
        left: Frame,
        right: Frame,
        *,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
        return_type: GeneralVectorLiteral = "pt",
    ) -> Tuple[NamedVector, NamedVector]:
        """Encode dataframes into named vectors.

        Args:
        ----
          left: Frame: left attribute information.
          right: Frame: right attribute information.
          left_rel: Optional[Frame]: left relation triples.
          right_rel: Optional[Frame]: right relation triples.
          return_type: GeneralVectorLiteral:  Either `pt` or `np` to return as pytorch tensor or numpy array.

        Returns:
        -------
            Embeddings of given left/right dataset.
        """
        self.validate(left, right)
        # TODO check if series can't be used everywhere instead
        # of upgrading in prepare
        left, right = self.prepare(left, right)
        left_enc, right_enc = self._encode_as(
            left=left,
            right=right,
            left_rel=left_rel,
            right_rel=right_rel,
            return_type=return_type,
        )
        if isinstance(left, dd.DataFrame):
            left_names = left.index.compute().tolist()
            right_names = right.index.compute().tolist()
        else:
            left_names = left.index.tolist()
            right_names = right.index.tolist()
        return NamedVector(names=left_names, vectors=left_enc), NamedVector(
            names=right_names, vectors=right_enc
        )

encode(left, right, *, left_rel=None, right_rel=None, return_type='pt')

Encode dataframes into named vectors.

---

left: Frame: left attribute information. right: Frame: right attribute information. left_rel: Optional[Frame]: left relation triples. right_rel: Optional[Frame]: right relation triples. return_type: GeneralVectorLiteral: Either pt or np to return as pytorch tensor or numpy array.

---

Embeddings of given left/right dataset.

Source code in klinker/encoders/base.py

def encode(
    self,
    left: Frame,
    right: Frame,
    *,
    left_rel: Optional[Frame] = None,
    right_rel: Optional[Frame] = None,
    return_type: GeneralVectorLiteral = "pt",
) -> Tuple[NamedVector, NamedVector]:
    """Encode dataframes into named vectors.

    Args:
    ----
      left: Frame: left attribute information.
      right: Frame: right attribute information.
      left_rel: Optional[Frame]: left relation triples.
      right_rel: Optional[Frame]: right relation triples.
      return_type: GeneralVectorLiteral:  Either `pt` or `np` to return as pytorch tensor or numpy array.

    Returns:
    -------
        Embeddings of given left/right dataset.
    """
    self.validate(left, right)
    # TODO check if series can't be used everywhere instead
    # of upgrading in prepare
    left, right = self.prepare(left, right)
    left_enc, right_enc = self._encode_as(
        left=left,
        right=right,
        left_rel=left_rel,
        right_rel=right_rel,
        return_type=return_type,
    )
    if isinstance(left, dd.DataFrame):
        left_names = left.index.compute().tolist()
        right_names = right.index.compute().tolist()
    else:
        left_names = left.index.tolist()
        right_names = right.index.tolist()
    return NamedVector(names=left_names, vectors=left_enc), NamedVector(
        names=right_names, vectors=right_enc
    )

prepare(left, right)

Prepare for embedding (fill NaNs with empty string).

---

left: Frame: left attributes. right: Frame: right attributes.

---

left, right

Source code in klinker/encoders/base.py

def prepare(self, left: Frame, right: Frame) -> Tuple[Frame, Frame]:
    """Prepare for embedding (fill NaNs with empty string).

    Args:
    ----
      left: Frame: left attributes.
      right: Frame: right attributes.

    Returns:
    -------
        left, right
    """
    return left.fillna(""), right.fillna("")

validate(left, right, left_rel=None, right_rel=None)

Check if frames only consist of one column.

---

left: Frame: left attributes. right: Frame: right attributes. left_rel: Optional[Frame]: left relation triples. right_rel: Optional[Frame]: right relation triples.

---

ValueError left/right have more than one column.

Source code in klinker/encoders/base.py

def validate(
    self,
    left: Frame,
    right: Frame,
    left_rel: Optional[Frame] = None,
    right_rel: Optional[Frame] = None,
):
    """Check if frames only consist of one column.

    Args:
    ----
      left: Frame: left attributes.
      right: Frame: right attributes.
      left_rel: Optional[Frame]: left relation triples.
      right_rel: Optional[Frame]: right relation triples.

    Raises:
    ------
        ValueError left/right have more than one column.
    """
    if len(left.columns) != 1 or len(right.columns) != 1:
        raise ValueError(
            "Input DataFrames must consist of single column containing all attribute values!"
        )

GCNFrameEncoder

Bases: RelationFrameEncoder

Use untrained GCN for aggregating neighboring embeddings with self.

---

depth: How many hops of neighbors should be incorporated
edge_weight: Weighting of non-self-loops
self_loop_weight: Weighting of self-loops
layer_dims: Dimensionality of layers if used
bias: Whether to use bias in layers
use_weight_layers: Whether to use randomly initialized layers in aggregation
aggr: Which aggregation to use. Can be :obj:`"sum"`, :obj:`"mean"`, :obj:`"min"` or :obj:`"max"`
attribute_encoder: HintOrType[TokenizedFrameEncoder]: Base encoder class
attribute_encoder_kwargs: OptionalKwargs: Keyword arguments for initializing encoder

Source code in klinker/encoders/gcn.py

class GCNFrameEncoder(RelationFrameEncoder):
    """Use untrained GCN for aggregating neighboring embeddings with self.

    Args:
    ----
        depth: How many hops of neighbors should be incorporated
        edge_weight: Weighting of non-self-loops
        self_loop_weight: Weighting of self-loops
        layer_dims: Dimensionality of layers if used
        bias: Whether to use bias in layers
        use_weight_layers: Whether to use randomly initialized layers in aggregation
        aggr: Which aggregation to use. Can be :obj:`"sum"`, :obj:`"mean"`, :obj:`"min"` or :obj:`"max"`
        attribute_encoder: HintOrType[TokenizedFrameEncoder]: Base encoder class
        attribute_encoder_kwargs: OptionalKwargs: Keyword arguments for initializing encoder
    """

    def __init__(
        self,
        depth: int = 2,
        edge_weight: float = 1.0,
        self_loop_weight: float = 2.0,
        layer_dims: int = 300,
        bias: bool = False,
        use_weight_layers: bool = True,
        aggr: str = "sum",
        attribute_encoder: HintOrType[TokenizedFrameEncoder] = None,
        attribute_encoder_kwargs: OptionalKwargs = None,
    ):
        if not TORCH_SCATTER:
            logger.error("Could not find torch_scatter and/or torch_sparse package!")
        self.depth = depth
        self.edge_weight = edge_weight
        self.self_loop_weight = self_loop_weight
        self.device = resolve_device()
        self.attribute_encoder = tokenized_frame_encoder_resolver.make(
            attribute_encoder, attribute_encoder_kwargs
        )
        layers: List[BasicMessagePassing]
        if use_weight_layers:
            layers = [
                FrozenGCNConv(
                    in_channels=layer_dims,
                    out_channels=layer_dims,
                    edge_weight=edge_weight,
                    self_loop_weight=self_loop_weight,
                    aggr=aggr,
                )
                for _ in range(self.depth)
            ]
        else:
            layers = [
                BasicMessagePassing(
                    edge_weight=edge_weight,
                    self_loop_weight=self_loop_weight,
                    aggr=aggr,
                )
                for _ in range(self.depth)
            ]
        self.layers = layers

    def _encode_rel(
        self,
        rel_triples_left: np.ndarray,
        rel_triples_right: np.ndarray,
        ent_features: NamedVector,
    ) -> GeneralVector:
        full_graph = np.concatenate([rel_triples_left, rel_triples_right])
        edge_index = torch.from_numpy(full_graph[:, [0, 2]]).t()
        x = ent_features.vectors
        for layer in self.layers:
            x = layer.forward(x, edge_index)
        return x

HybridDeepBlockerFrameEncoder

Bases: CrossTupleTrainingDeepBlockerFrameEncoder

Hybrid DeepBlocker class.

Uses both Autoencoder and CrossTupleTraining strategy.

---

frame_encoder: HintOrType[TokenizedFrameEncoder]: Base encoder class
frame_encoder_kwargs: OptionalKwargs: Keyword arguments for initializing frame encoder
hidden_dimensions: Tuple[int, int]: Hidden dimensions
num_epochs: int: Number of epochs if training
batch_size: int: Batch size
learning_rate: float: Learning rate
synth_tuples_per_tuple: int: Synthetic tuples per tuple
pos_to_neg_ratio: float: Ratio of positiv to negative tuples
max_perturbation:float: Degree how much tuples should be corrupted
random_seed: Seed to control randomness
loss_function: Optional[_Loss]: Loss function if training
optimizer: Optional[HintOrType[Optimizer]]: Optimizer if training
optimizer_kwargs: OptionalKwargs: Keyword arguments to inizialize optimizer

Source code in klinker/encoders/deepblocker.py

class HybridDeepBlockerFrameEncoder(CrossTupleTrainingDeepBlockerFrameEncoder):
    """Hybrid DeepBlocker class.

    Uses both Autoencoder and CrossTupleTraining strategy.

    Args:
    ----
        frame_encoder: HintOrType[TokenizedFrameEncoder]: Base encoder class
        frame_encoder_kwargs: OptionalKwargs: Keyword arguments for initializing frame encoder
        hidden_dimensions: Tuple[int, int]: Hidden dimensions
        num_epochs: int: Number of epochs if training
        batch_size: int: Batch size
        learning_rate: float: Learning rate
        synth_tuples_per_tuple: int: Synthetic tuples per tuple
        pos_to_neg_ratio: float: Ratio of positiv to negative tuples
        max_perturbation:float: Degree how much tuples should be corrupted
        random_seed: Seed to control randomness
        loss_function: Optional[_Loss]: Loss function if training
        optimizer: Optional[HintOrType[Optimizer]]: Optimizer if training
        optimizer_kwargs: OptionalKwargs: Keyword arguments to inizialize optimizer
    """

    def __init__(
        self,
        frame_encoder: HintOrType[TokenizedFrameEncoder] = None,
        frame_encoder_kwargs: OptionalKwargs = None,
        hidden_dimensions: Tuple[int, int] = (2 * 150, 150),
        num_epochs: int = 50,
        batch_size: int = 256,
        learning_rate: float = 1e-3,
        synth_tuples_per_tuple: int = 5,
        pos_to_neg_ratio: float = 1.0,
        max_perturbation=0.4,
        random_seed=None,
        loss_function: Optional[_Loss] = None,
        optimizer: Optional[HintOrType[Optimizer]] = None,
        optimizer_kwargs: OptionalKwargs = None,
    ):
        inner_encoder = AutoEncoderDeepBlockerFrameEncoder(
            frame_encoder=frame_encoder,
            frame_encoder_kwargs=frame_encoder_kwargs,
            hidden_dimensions=hidden_dimensions,
            num_epochs=num_epochs,
            batch_size=batch_size,
            learning_rate=learning_rate,
        )
        super().__init__(
            frame_encoder=inner_encoder,
            hidden_dimensions=hidden_dimensions,
            num_epochs=num_epochs,
            batch_size=batch_size,
            learning_rate=learning_rate,
            synth_tuples_per_tuple=synth_tuples_per_tuple,
            pos_to_neg_ratio=pos_to_neg_ratio,
            max_perturbation=max_perturbation,
            random_seed=random_seed,
            loss_function=loss_function,
            optimizer=optimizer,
            optimizer_kwargs=optimizer_kwargs,
        )

LightEAFrameEncoder

Bases: RelationFrameEncoder

Use LightEA algorithm to encode frame.

---

depth: int: Number of hops
mini_dim:int: Mini batching size
rel_dim:int: relation embedding dimensions (same as ent_dim if None)
attribute_encoder: HintOrType[TokenizedFrameEncoder]: Attribute encoder class
attribute_encoder_kwargs: OptionalKwargs: Keyword arguments for initializing attribute encoder class

Reference

Mao et. al.,"LightEA: A Scalable, Robust, and Interpretable Entity Alignment Framework via Three-view Label Propagation", EMNLP 2022 https://aclanthology.org/2022.emnlp-main.52.pdf

Source code in klinker/encoders/light_ea.py

class LightEAFrameEncoder(RelationFrameEncoder):
    """Use LightEA algorithm to encode frame.

    Args:
    ----
        depth: int: Number of hops
        mini_dim:int: Mini batching size
        rel_dim:int: relation embedding dimensions (same as ent_dim if None)
        attribute_encoder: HintOrType[TokenizedFrameEncoder]: Attribute encoder class
        attribute_encoder_kwargs: OptionalKwargs: Keyword arguments for initializing attribute encoder class

    Quote: Reference
        Mao et. al.,"LightEA: A Scalable, Robust, and Interpretable Entity Alignment Framework via Three-view Label Propagation", EMNLP 2022 <https://aclanthology.org/2022.emnlp-main.52.pdf>
    """

    def __init__(
        self,
        depth: int = 2,
        mini_dim: int = 16,
        rel_dim: Optional[int] = None,
        attribute_encoder: HintOrType[TokenizedFrameEncoder] = None,
        attribute_encoder_kwargs: OptionalKwargs = None,
        only_use_neighbor_info: bool = False,
    ):
        self.depth = depth
        self.device = resolve_device()
        self.mini_dim = mini_dim
        self.rel_dim = rel_dim
        self.attribute_encoder = tokenized_frame_encoder_resolver.make(
            attribute_encoder, attribute_encoder_kwargs
        )
        self.only_use_neighbor_info = only_use_neighbor_info

    def _encode_rel(
        self,
        rel_triples_left: np.ndarray,
        rel_triples_right: np.ndarray,
        ent_features: NamedVector,
    ) -> GeneralVector:
        print("Started LightEA")
        (
            node_size,
            rel_size,
            ent_tuple,
            triples_idx,
            ent_ent,
            ent_ent_val,
            rel_ent,
            ent_rel,
        ) = self._transform_graph(rel_triples_left, rel_triples_right)
        return self._get_features(
            node_size,
            rel_size,
            ent_tuple,
            triples_idx,
            ent_ent,
            ent_ent_val,
            rel_ent,
            ent_rel,
            ent_features.vectors,
        )

    def _transform_graph(
        self, rel_triples_left: np.ndarray, rel_triples_right: np.ndarray
    ):
        triples = []
        rel_size = 0
        for line in rel_triples_left:
            h, r, t = line
            triples.append([h, t, 2 * r])
            triples.append([t, h, 2 * r + 1])
            rel_size = max(rel_size, 2 * r + 1)
        for line in rel_triples_right:
            h, r, t = line
            triples.append([h, t, 2 * r])
            triples.append([t, h, 2 * r + 1])
            rel_size = max(rel_size, 2 * r + 1)
        triples = np.unique(triples, axis=0)
        node_size, rel_size = np.max(triples) + 1, np.max(triples[:, 2]) + 1  # type: ignore
        ent_tuple, triples_idx = [], []
        ent_ent_s, rel_ent_s, ent_rel_s = {}, set(), set()
        last, index = (-1, -1), -1

        for i in range(node_size):
            ent_ent_s[(i, i)] = 0

        for h, t, r in triples:
            ent_ent_s[(h, h)] += 1
            ent_ent_s[(t, t)] += 1

            if (h, t) != last:
                last = (h, t)
                index += 1
                ent_tuple.append([h, t])
                ent_ent_s[(h, t)] = 0

            triples_idx.append([index, r])
            ent_ent_s[(h, t)] += 1
            rel_ent_s.add((r, h))
            ent_rel_s.add((t, r))

        ent_tuple = np.array(ent_tuple)  # type: ignore
        triples_idx = np.unique(np.array(triples_idx), axis=0)  # type: ignore

        ent_ent = np.unique(np.array(list(ent_ent_s.keys())), axis=0)
        ent_ent_val = np.array([ent_ent_s[(x, y)] for x, y in ent_ent]).astype(
            "float32"
        )
        rel_ent = np.unique(np.array(list(rel_ent_s)), axis=0)
        ent_rel = np.unique(np.array(list(ent_rel_s)), axis=0)
        return (
            node_size,
            rel_size,
            ent_tuple,
            triples_idx,
            ent_ent,
            ent_ent_val,
            rel_ent,
            ent_rel,
        )

    @torch.no_grad()
    def _get_features(
        self,
        node_size,
        rel_size,
        ent_tuple,
        triples_idx,
        ent_ent,
        ent_ent_val,
        rel_ent,
        ent_rel,
        ent_feature,
    ):
        ent_feature = ent_feature.to(self.device)
        if self.rel_dim is None:
            self.rel_dim = ent_feature.shape[1]
        print(f"ent_feature.shape={ent_feature.shape}")
        rel_feature = torch.zeros((rel_size, ent_feature.shape[-1])).to(self.device)
        print(f"rel_feature.shape={rel_feature.shape}")

        ent_ent, ent_rel, rel_ent, ent_ent_val, triples_idx, ent_tuple = map(
            torch.tensor,
            [ent_ent, ent_rel, rel_ent, ent_ent_val, triples_idx, ent_tuple],
        )

        ent_ent = ent_ent.t()
        ent_rel = ent_rel.t()
        rel_ent = rel_ent.t()
        triples_idx = triples_idx.t()
        ent_tuple = ent_tuple.t()

        ent_ent_graph = torch.sparse_coo_tensor(
            indices=ent_ent, values=ent_ent_val, size=(node_size, node_size)
        ).to(self.device)
        rel_ent_graph = torch.sparse_coo_tensor(
            indices=rel_ent,
            values=torch.ones(rel_ent.shape[1]),
            size=(rel_size, node_size),
        ).to(self.device)
        ent_rel_graph = torch.sparse_coo_tensor(
            indices=ent_rel,
            values=torch.ones(ent_rel.shape[1]),
            size=(node_size, rel_size),
        ).to(self.device)

        # ent_list, rel_list = [ent_feature], [rel_feature]
        ent_list = [ent_feature]
        if self.only_use_neighbor_info:
            ent_list = []
        for dep in trange(self.depth):
            new_rel_feature = torch.from_numpy(
                _batch_sparse_matmul(rel_ent_graph, ent_feature, self.device)
            ).to(self.device)
            new_rel_feature = _my_norm(new_rel_feature)

            new_ent_feature = torch.from_numpy(
                _batch_sparse_matmul(ent_ent_graph, ent_feature, self.device)
            ).to(self.device)
            new_ent_feature += torch.from_numpy(
                _batch_sparse_matmul(ent_rel_graph, rel_feature, self.device)
            ).to(self.device)
            new_ent_feature = _my_norm(new_ent_feature)

            ent_feature = new_ent_feature
            rel_feature = new_rel_feature
            ent_list.append(ent_feature)
            # rel_list.append(rel_feature)
            print(f"dep={dep}, ent_feature.shape={ent_feature.shape}")
            print(f"dep={dep}, rel_feature.shape={rel_feature.shape}")

        ent_feature = torch.cat(ent_list, dim=1)
        print(f"ent_feature.shape={ent_feature.shape}")
        return F.normalize(ent_feature)

RelationFrameEncoder

Bases: FrameEncoder

Base class for Encoders, that also utilize relational information.

Source code in klinker/encoders/base.py

class RelationFrameEncoder(FrameEncoder):
    """Base class for Encoders, that also utilize relational information."""

    attribute_encoder: FrameEncoder

    def validate(
        self,
        left: Frame,
        right: Frame,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
    ):
        """Ensure relation info is provided and attribute frames consist of single column.

        Args:
        ----
          left: Frame: left attribute information.
          right: Frame: right attribute information.
          left_rel: Optional[Frame]: left relation triples.
          right_rel: Optional[Frame]: right relation triples.

        Raises:
        ------
            ValueError: If attribute frames consist of multiple columns or relational frames are missing.
        """
        super().validate(left=left, right=right)
        if left_rel is None or right_rel is None:
            raise ValueError(f"{self.__class__.__name__} needs left_rel and right_rel!")

    def _encode_rel(
        self,
        rel_triples_left: np.ndarray,
        rel_triples_right: np.ndarray,
        ent_features: NamedVector,
    ) -> GeneralVector:
        raise NotImplementedError

    @overload
    def _encode_rel_as(
        self,
        rel_triples_left: np.ndarray,
        rel_triples_right: np.ndarray,
        ent_features: NamedVector,
        return_type: Literal["np"],
    ) -> np.ndarray:
        ...

    @overload
    def _encode_rel_as(
        self,
        rel_triples_left: np.ndarray,
        rel_triples_right: np.ndarray,
        ent_features: NamedVector,
        return_type: Literal["pt"],
    ) -> torch.Tensor:
        ...

    def _encode_rel_as(
        self,
        rel_triples_left: np.ndarray,
        rel_triples_right: np.ndarray,
        ent_features: NamedVector,
        return_type: GeneralVectorLiteral = "pt",
    ) -> GeneralVector:
        enc = self._encode_rel(
            rel_triples_left=rel_triples_left,
            rel_triples_right=rel_triples_right,
            ent_features=ent_features,
        )
        return cast_general_vector(enc, return_type=return_type)

    def encode(
        self,
        left: SeriesType,
        right: SeriesType,
        *,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
        return_type: GeneralVectorLiteral = "pt",
    ) -> Tuple[NamedVector, NamedVector]:
        """Encode dataframes into named vectors.

        Args:
        ----
          left: Frame: left attribute information.
          right: Frame: right attribute information.
          *:
          left_rel: Optional[Frame]: left relation triples.
          right_rel: Optional[Frame]: right relation triples.
          return_type: GeneralVectorLiteral:  Either `pt` or `np` to return as pytorch tensor or numpy array.

        Returns:
        -------
            Embeddings of given left/right dataset.
        """
        self.validate(left=left, right=right, left_rel=left_rel, right_rel=right_rel)
        left, right = self.prepare(left, right)

        start = time.time()
        # encode attributes
        left_attr_enc, right_attr_enc = self.attribute_encoder.encode(
            left, right, return_type=return_type
        )
        all_attr_enc = left_attr_enc.concat(right_attr_enc)

        # map string based triples to int
        entity_mapping = all_attr_enc.entity_id_mapping
        if isinstance(left_rel, dd.DataFrame):
            left_rel = left_rel.compute()
            right_rel = right_rel.compute()
        rel_triples_left, entity_mapping, rel_mapping = id_map_rel_triples(
            left_rel, entity_mapping=entity_mapping
        )
        rel_triples_right, entity_mapping, rel_mapping = id_map_rel_triples(
            right_rel,
            entity_mapping=entity_mapping,
            rel_mapping=rel_mapping,
        )

        # initialize entity features randomly and replace with
        # attribute features where known
        ent_features = initialize_and_fill(known=all_attr_enc, all_names=entity_mapping)
        left_ids = list(_get_ids(left, left_rel))
        right_ids = list(_get_ids(right, right_rel))

        # encode relations
        features = self._encode_rel_as(
            rel_triples_left=rel_triples_left,
            rel_triples_right=rel_triples_right,
            ent_features=ent_features,
            return_type=return_type,
        )
        named_features = NamedVector(names=entity_mapping, vectors=features)  # type: ignore

        end = time.time()
        self._encoding_time = end - start
        return named_features.subset(list(left_ids)), named_features.subset(
            list(right_ids)
        )

encode(left, right, *, left_rel=None, right_rel=None, return_type='pt')

Encode dataframes into named vectors.

---

left: Frame: left attribute information. right: Frame: right attribute information. *: left_rel: Optional[Frame]: left relation triples. right_rel: Optional[Frame]: right relation triples. return_type: GeneralVectorLiteral: Either pt or np to return as pytorch tensor or numpy array.

---

Embeddings of given left/right dataset.

Source code in klinker/encoders/base.py

def encode(
    self,
    left: SeriesType,
    right: SeriesType,
    *,
    left_rel: Optional[Frame] = None,
    right_rel: Optional[Frame] = None,
    return_type: GeneralVectorLiteral = "pt",
) -> Tuple[NamedVector, NamedVector]:
    """Encode dataframes into named vectors.

    Args:
    ----
      left: Frame: left attribute information.
      right: Frame: right attribute information.
      *:
      left_rel: Optional[Frame]: left relation triples.
      right_rel: Optional[Frame]: right relation triples.
      return_type: GeneralVectorLiteral:  Either `pt` or `np` to return as pytorch tensor or numpy array.

    Returns:
    -------
        Embeddings of given left/right dataset.
    """
    self.validate(left=left, right=right, left_rel=left_rel, right_rel=right_rel)
    left, right = self.prepare(left, right)

    start = time.time()
    # encode attributes
    left_attr_enc, right_attr_enc = self.attribute_encoder.encode(
        left, right, return_type=return_type
    )
    all_attr_enc = left_attr_enc.concat(right_attr_enc)

    # map string based triples to int
    entity_mapping = all_attr_enc.entity_id_mapping
    if isinstance(left_rel, dd.DataFrame):
        left_rel = left_rel.compute()
        right_rel = right_rel.compute()
    rel_triples_left, entity_mapping, rel_mapping = id_map_rel_triples(
        left_rel, entity_mapping=entity_mapping
    )
    rel_triples_right, entity_mapping, rel_mapping = id_map_rel_triples(
        right_rel,
        entity_mapping=entity_mapping,
        rel_mapping=rel_mapping,
    )

    # initialize entity features randomly and replace with
    # attribute features where known
    ent_features = initialize_and_fill(known=all_attr_enc, all_names=entity_mapping)
    left_ids = list(_get_ids(left, left_rel))
    right_ids = list(_get_ids(right, right_rel))

    # encode relations
    features = self._encode_rel_as(
        rel_triples_left=rel_triples_left,
        rel_triples_right=rel_triples_right,
        ent_features=ent_features,
        return_type=return_type,
    )
    named_features = NamedVector(names=entity_mapping, vectors=features)  # type: ignore

    end = time.time()
    self._encoding_time = end - start
    return named_features.subset(list(left_ids)), named_features.subset(
        list(right_ids)
    )

validate(left, right, left_rel=None, right_rel=None)

Ensure relation info is provided and attribute frames consist of single column.

---

left: Frame: left attribute information. right: Frame: right attribute information. left_rel: Optional[Frame]: left relation triples. right_rel: Optional[Frame]: right relation triples.

---

ValueError: If attribute frames consist of multiple columns or relational frames are missing.

Source code in klinker/encoders/base.py

def validate(
    self,
    left: Frame,
    right: Frame,
    left_rel: Optional[Frame] = None,
    right_rel: Optional[Frame] = None,
):
    """Ensure relation info is provided and attribute frames consist of single column.

    Args:
    ----
      left: Frame: left attribute information.
      right: Frame: right attribute information.
      left_rel: Optional[Frame]: left relation triples.
      right_rel: Optional[Frame]: right relation triples.

    Raises:
    ------
        ValueError: If attribute frames consist of multiple columns or relational frames are missing.
    """
    super().validate(left=left, right=right)
    if left_rel is None or right_rel is None:
        raise ValueError(f"{self.__class__.__name__} needs left_rel and right_rel!")

SIFEmbeddingTokenizedFrameEncoder

Bases: TokenizedFrameEncoder

Use Smooth Inverse Frequency weighting scheme to aggregate token embeddings.

---

sif_weighting_param: float: weighting parameter
remove_pc:bool: remove first principal component
min_freq: int: minimum frequency of occurence
tokenized_word_embedder: HintOrType[TokenizedWordEmbedder]: Word Embedding class,
tokenized_word_embedder_kwargs: OptionalKwargs: Keyword arguments for initalizing word embedder

Reference

Arora et. al.,"A Simple but Tough-to-Beat Baseline for Sentence Embeddings", ICLR 2017 https://openreview.net/pdf?id=SyK00v5xx

Source code in klinker/encoders/pretrained.py

class SIFEmbeddingTokenizedFrameEncoder(TokenizedFrameEncoder):
    """Use Smooth Inverse Frequency weighting scheme to aggregate token embeddings.

    Args:
    ----
        sif_weighting_param: float: weighting parameter
        remove_pc:bool: remove first principal component
        min_freq: int: minimum frequency of occurence
        tokenized_word_embedder: HintOrType[TokenizedWordEmbedder]: Word Embedding class,
        tokenized_word_embedder_kwargs: OptionalKwargs: Keyword arguments for initalizing word embedder

    Quote: Reference
        Arora et. al.,"A Simple but Tough-to-Beat Baseline for Sentence Embeddings", ICLR 2017 <https://openreview.net/pdf?id=SyK00v5xx>
    """

    def __init__(
        self,
        sif_weighting_param: float = 1e-3,
        remove_pc: bool = True,
        min_freq: int = 0,
        tokenized_word_embedder: HintOrType[TokenizedWordEmbedder] = None,
        tokenized_word_embedder_kwargs: OptionalKwargs = None,
        reduce_dim_to: Optional[int] = None,
        umap_n_neighbors: int = 15,
        umap_min_dist: int = 0.1,
    ):
        self.tokenized_word_embedder = tokenized_word_embedder_resolver.make(
            tokenized_word_embedder, tokenized_word_embedder_kwargs
        )

        self.sif_weighting_param = sif_weighting_param
        self.remove_pc = remove_pc
        self.min_freq = min_freq
        self.token_weight_dict: Optional[Dict[str, float]] = None
        self.reduce_dim_to = reduce_dim_to
        self.umap_n_neighbors = umap_n_neighbors
        self.umap_min_dist = umap_min_dist

    @property
    def tokenizer_fn(self) -> Callable[[str], List[str]]:
        """ """
        return self.tokenized_word_embedder.tokenizer_fn

    def prepare(self, left: Frame, right: Frame) -> Tuple[Frame, Frame]:
        """Prepare value counts.

        Args:
        ----
          left: Frame: left attribute frame.
          right: Frame: right attribute frame.

        Returns:
        -------
            left, right
        """
        left, right = super().prepare(left, right)
        merged_col = "merged"
        left.columns = [merged_col]
        right.columns = [merged_col]
        all_values = concat_frames([left, right])

        value_counts = (
            all_values[merged_col]
            .apply(self.tokenized_word_embedder.tokenizer_fn)
            .explode()
            .value_counts()
        )

        def sif_weighting(x, a: float, min_freq: int, total_tokens: int):
            if x >= min_freq:
                return a / (a + x / total_tokens)
            else:
                return 1.0

        total_tokens = value_counts.sum()
        if isinstance(left, KlinkerDaskFrame):
            total_tokens = total_tokens.compute()

        token_weight_dict = value_counts.apply(
            sif_weighting,
            a=self.sif_weighting_param,
            min_freq=self.min_freq,
            total_tokens=total_tokens,
        )

        if isinstance(left, KlinkerDaskFrame):
            token_weight_dict = token_weight_dict.compute()

        self.token_weight_dict = token_weight_dict.to_dict()
        return left, right

    def _postprocess(self, left, right) -> Tuple[GeneralVector, GeneralVector]:
        # From the code of the SIF paper at
        # https://github.com/PrincetonML/SIF/blob/master/src/SIF_embedding.py
        if self.remove_pc:
            concat_fn = (
                np.concatenate if isinstance(left, np.ndarray) else torch.concatenate
            )
            embeddings = concat_fn([left, right])
            svd = TruncatedSVD(n_components=1, n_iter=7, random_state=0)
            svd.fit(embeddings)
            pc = svd.components_
            sif_embeddings = embeddings - embeddings.dot(pc.transpose()) * pc
            return sif_embeddings[: len(left)], sif_embeddings[len(left) :]
        return left, right

    def _reduce_dim(
        self, left_emb: GeneralVector, right_emb: GeneralVector
    ) -> Tuple[GeneralVector, GeneralVector]:
        if self.reduce_dim_to:
            initial_dim = left_emb.shape[0]
            if self.reduce_dim_to == initial_dim:
                logger.info(
                    f"Can't reduce to the same dimensionality ({initial_dim}) so returning"
                )
                return left_emb, right_emb
            if self.reduce_dim_to > initial_dim:
                raise ValueError(
                    f"Cannot reduce embeddings of dimensionality {initial_dim} to higher dimensionality of {self.reduce_dim_to}!"
                )
            logger.info(f"Reducing embedding dim to {self.reduce_dim_to}")
            umap = UMAP(
                n_components=self.reduce_dim_to,
                n_neighbors=self.umap_n_neighbors,
                min_dist=self.umap_min_dist,
            )
            all_vec = (
                np.concatenate([left_emb, right_emb])
                if isinstance(left_emb, np.ndarray)
                else torch.concat([left_emb, right_emb])
            )
            reduced_vec = umap.fit_transform(all_vec)
            left_emb = reduced_vec[: len(left_emb)]
            right_emb = reduced_vec[len(left_emb) :]
        return left_emb, right_emb

    def _encode(
        self,
        left: Frame,
        right: Frame,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
    ) -> Tuple[GeneralVector, GeneralVector]:
        if self.token_weight_dict is None:
            self.prepare(left, right)
        if isinstance(left, KlinkerDaskFrame):
            left_enc = left.map_partitions(
                encode_frame,
                twe=self.tokenized_word_embedder,
                weight_dict=self.token_weight_dict,
            ).compute()
            right_enc = right.map_partitions(
                encode_frame,
                twe=self.tokenized_word_embedder,
                weight_dict=self.token_weight_dict,
            ).compute()
        else:
            left_enc = encode_frame(
                left,
                twe=self.tokenized_word_embedder,
                weight_dict=self.token_weight_dict,
            )
            right_enc = encode_frame(
                right,
                twe=self.tokenized_word_embedder,
                weight_dict=self.token_weight_dict,
            )
        if self.remove_pc:
            left_enc, right_enc = self._postprocess(left_enc, right_enc)
        return self._reduce_dim(left_enc, right_enc)

tokenizer_fn: Callable[[str], List[str]] property

prepare(left, right)

Prepare value counts.

---

left: Frame: left attribute frame. right: Frame: right attribute frame.

---

left, right

Source code in klinker/encoders/pretrained.py

def prepare(self, left: Frame, right: Frame) -> Tuple[Frame, Frame]:
    """Prepare value counts.

    Args:
    ----
      left: Frame: left attribute frame.
      right: Frame: right attribute frame.

    Returns:
    -------
        left, right
    """
    left, right = super().prepare(left, right)
    merged_col = "merged"
    left.columns = [merged_col]
    right.columns = [merged_col]
    all_values = concat_frames([left, right])

    value_counts = (
        all_values[merged_col]
        .apply(self.tokenized_word_embedder.tokenizer_fn)
        .explode()
        .value_counts()
    )

    def sif_weighting(x, a: float, min_freq: int, total_tokens: int):
        if x >= min_freq:
            return a / (a + x / total_tokens)
        else:
            return 1.0

    total_tokens = value_counts.sum()
    if isinstance(left, KlinkerDaskFrame):
        total_tokens = total_tokens.compute()

    token_weight_dict = value_counts.apply(
        sif_weighting,
        a=self.sif_weighting_param,
        min_freq=self.min_freq,
        total_tokens=total_tokens,
    )

    if isinstance(left, KlinkerDaskFrame):
        token_weight_dict = token_weight_dict.compute()

    self.token_weight_dict = token_weight_dict.to_dict()
    return left, right

SentenceTransformerTokenizedFrameEncoder

Bases: TokenizedFrameEncoder

Uses sentencetransformer library to encode frames.

See https://www.sbert.net/docs/pretrained_models.html for a list of models.

---

model_name: str: pretrained model name
max_length: int: max number of tokens per row
batch_size: int: size of batch for encoding

Examples:

---

>>> # doctest: +SKIP
>>> import pandas as pd

>>> from klinker.data import KlinkerPandasFrame
>>> from klinker.encoders import SentenceTransformerTokenizedFrameEncoder

>>> left = KlinkerPandasFrame.from_df(
         pd.DataFrame(
             [("a1", "John Doe"), ("a2", "Jane Doe")], columns=["id", "values"]
         ),
         table_name="A",
         id_col="id",
    ).set_index("id")
>>> right = KlinkerPandasFrame.from_df(
        pd.DataFrame(
            [("b1", "Johnny Doe"), ("b2", "Jane Doe")], columns=["id", "values"]
        ),
        table_name="B",
        id_col="id",
    ).set_index("id")
>>> ttfe = SentenceTransformerTokenizedFrameEncoder(
        model_name="st5",
        max_length=10,
        batch_size=2
    )
>>> left_enc, right_enc = ttfe.encode(left=left, right=right)

Source code in klinker/encoders/pretrained.py

class SentenceTransformerTokenizedFrameEncoder(TokenizedFrameEncoder):
    """Uses sentencetransformer library to encode frames.

    See <https://www.sbert.net/docs/pretrained_models.html> for a list of models.

    Args:
    ----
        model_name: str: pretrained model name
        max_length: int: max number of tokens per row
        batch_size: int: size of batch for encoding

    Examples:
    --------
        >>> # doctest: +SKIP
        >>> import pandas as pd

        >>> from klinker.data import KlinkerPandasFrame
        >>> from klinker.encoders import SentenceTransformerTokenizedFrameEncoder

        >>> left = KlinkerPandasFrame.from_df(
                 pd.DataFrame(
                     [("a1", "John Doe"), ("a2", "Jane Doe")], columns=["id", "values"]
                 ),
                 table_name="A",
                 id_col="id",
            ).set_index("id")
        >>> right = KlinkerPandasFrame.from_df(
                pd.DataFrame(
                    [("b1", "Johnny Doe"), ("b2", "Jane Doe")], columns=["id", "values"]
                ),
                table_name="B",
                id_col="id",
            ).set_index("id")
        >>> ttfe = SentenceTransformerTokenizedFrameEncoder(
                model_name="st5",
                max_length=10,
                batch_size=2
            )
        >>> left_enc, right_enc = ttfe.encode(left=left, right=right)

    """

    def __init__(
        self,
        model_name: str = "gtr-t5-base",
        max_length: int = 128,
        batch_size: int = 512,
        reduce_dim_to: Optional[int] = None,
        reduce_sample_perc: float = 0.3,
    ):
        if SentenceTransformer is None:
            raise ImportError("Please install the sentence-transformers library!")
        self.model = SentenceTransformer(model_name)
        logger.info("Loaded model")
        self.model.max_seq_length = max_length
        self.batch_size = batch_size
        self.reduce_dim_to = reduce_dim_to
        self.reduce_sample_perc = reduce_sample_perc
        self._added_reduce_layer = False

    @property
    def tokenizer_fn(self) -> Callable[[str], List[str]]:
        return self.model.tokenizer.tokenize

    @torch.no_grad()
    def _encode_side(self, df: Frame, convert_to_tensor: bool = True) -> GeneralVector:
        vals = df[df.columns[0]].values
        if isinstance(df, KlinkerDaskFrame):
            vals = vals.compute()
        if convert_to_tensor:
            return self.model.encode(
                vals, batch_size=self.batch_size, convert_to_tensor=True
            )
        return self.model.encode(
            vals, batch_size=self.batch_size, convert_to_numpy=True
        )

    def _add_dimensionality_reduction_layer(self, left: Frame, right: Frame):
        # see https://github.com/UKPLab/sentence-transformers/blob/master/examples/training/distillation/dimensionality_reduction.py
        logger.info(
            f"Using PCA to output embeddings with dimensionality of {self.reduce_dim_to}. Training on {self.reduce_sample_perc * 100} % of the data."
        )
        lt_embeddings = self._encode_side(
            left.sample(frac=self.reduce_sample_perc), convert_to_tensor=False
        )
        rt_embeddings = self._encode_side(
            right.sample(frac=self.reduce_sample_perc), convert_to_tensor=False
        )
        train_embeddings = np.concatenate([lt_embeddings, rt_embeddings])
        # Compute PCA on the train embeddings matrix
        pca = PCA(n_components=self.reduce_dim_to)
        pca.fit(train_embeddings)
        pca_comp = np.asarray(pca.components_)

        # We add a dense layer to the model, so that it will produce directly embeddings with the new size
        dense = models.Dense(
            in_features=self.model.get_sentence_embedding_dimension(),
            out_features=self.reduce_dim_to,
            bias=False,
            activation_function=torch.nn.Identity(),
        )
        dense.linear.weight = torch.nn.Parameter(torch.tensor(pca_comp))
        self.model.add_module("dense", dense)
        logger.info(
            f"Done! Added a dense layer with shape ({dense.in_features}, {dense.out_features}) to the model"
        )
        self._added_reduce_layer = True

    def _encode(
        self,
        left: Frame,
        right: Frame,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
    ) -> Tuple[GeneralVector, GeneralVector]:
        logger.info("Started encode")
        if self.reduce_dim_to and not self._added_reduce_layer:
            self._add_dimensionality_reduction_layer(left, right)
        return self._encode_side(left), self._encode_side(right)

TokenizedFrameEncoder

Bases: FrameEncoder

FrameEncoder that uses tokenization of attribute values.

Source code in klinker/encoders/base.py

class TokenizedFrameEncoder(FrameEncoder):
    """FrameEncoder that uses tokenization of attribute values."""

    @property
    def tokenizer_fn(self) -> Callable[[str], List[str]]:
        """ """
        raise NotImplementedError

tokenizer_fn: Callable[[str], List[str]] property

TransformerTokenizedFrameEncoder

Bases: TokenizedFrameEncoder

Encode frames using pre-trained transformer.

See https://huggingface.co/docs/transformers/main/en/model_doc/auto#transformers.AutoModel.from_pretrained for more information on pretrained models.

---

model_name: str: Transformer name or path
max_length: int: max number of tokens per row
batch_size: int: size of batch for encoding

Examples:

---

>>> # doctest: +SKIP
>>> import pandas as pd

>>> from klinker.data import KlinkerPandasFrame
>>> from klinker.encoders import TransformerTokenizedFrameEncoder

>>> left = KlinkerPandasFrame.from_df(
         pd.DataFrame(
             [("a1", "John Doe"), ("a2", "Jane Doe")], columns=["id", "values"]
         ),
         table_name="A",
         id_col="id",
    ).set_index("id")
>>> right = KlinkerPandasFrame.from_df(
        pd.DataFrame(
            [("b1", "Johnny Doe"), ("b2", "Jane Doe")], columns=["id", "values"]
        ),
        table_name="B",
        id_col="id",
    ).set_index("id")
>>> ttfe = TransformerTokenizedFrameEncoder(
        model_name="bert-base-cased",
        max_length=10,
        batch_size=2
    )
>>> left_enc, right_enc = ttfe.encode(left=left, right=right)

Source code in klinker/encoders/pretrained.py

class TransformerTokenizedFrameEncoder(TokenizedFrameEncoder):
    """Encode frames using pre-trained transformer.

    See <https://huggingface.co/docs/transformers/main/en/model_doc/auto#transformers.AutoModel.from_pretrained> for more information on pretrained models.

    Args:
    ----
        model_name: str: Transformer name or path
        max_length: int: max number of tokens per row
        batch_size: int: size of batch for encoding

    Examples:
    --------
        >>> # doctest: +SKIP
        >>> import pandas as pd

        >>> from klinker.data import KlinkerPandasFrame
        >>> from klinker.encoders import TransformerTokenizedFrameEncoder

        >>> left = KlinkerPandasFrame.from_df(
                 pd.DataFrame(
                     [("a1", "John Doe"), ("a2", "Jane Doe")], columns=["id", "values"]
                 ),
                 table_name="A",
                 id_col="id",
            ).set_index("id")
        >>> right = KlinkerPandasFrame.from_df(
                pd.DataFrame(
                    [("b1", "Johnny Doe"), ("b2", "Jane Doe")], columns=["id", "values"]
                ),
                table_name="B",
                id_col="id",
            ).set_index("id")
        >>> ttfe = TransformerTokenizedFrameEncoder(
                model_name="bert-base-cased",
                max_length=10,
                batch_size=2
            )
        >>> left_enc, right_enc = ttfe.encode(left=left, right=right)

    """

    def __init__(
        self,
        model_name: str = "bert-base-cased",
        max_length: int = 128,
        batch_size: int = 512,
    ):
        if AutoModel is None:
            raise ImportError("Please install the transformers library!")
        self.model = AutoModel.from_pretrained(model_name)
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.max_length = max_length
        self.batch_size = batch_size

    @property
    def tokenizer_fn(self) -> Callable[[str], List[str]]:
        return self.tokenizer.tokenize

    @torch.no_grad()
    def _encode_side(self, df: Frame) -> GeneralVector:
        encoded = []
        for batch in _batch_generator(df, self.batch_size):
            tok = self.tokenizer(
                list(batch),
                return_tensors="pt",
                padding=True,
                truncation=True,
                max_length=self.max_length,
            )
            encoded.append(self.model(**tok).pooler_output.detach())
        return torch.vstack(encoded)

    def _encode(
        self,
        left: Frame,
        right: Frame,
        left_rel: Optional[Frame] = None,
        right_rel: Optional[Frame] = None,
    ) -> Tuple[GeneralVector, GeneralVector]:
        return self._encode_side(left), self._encode_side(right)