dicee.abstracts =============== .. py:module:: dicee.abstracts Attributes ---------- .. autoapisummary:: dicee.abstracts.logger Classes ------- .. autoapisummary:: dicee.abstracts.AbstractTrainer dicee.abstracts.BaseInteractiveKGE dicee.abstracts.InteractiveQueryDecomposition dicee.abstracts.AbstractCallback dicee.abstracts.AbstractPPECallback dicee.abstracts.BaseInteractiveTrainKGE Module Contents --------------- .. py:data:: logger .. py:class:: AbstractTrainer(args, callbacks) Abstract base class for KGE model trainers. Provides the callback dispatch mechanism shared by all concrete trainer implementations (TorchTrainer, TorchDDPTrainer, etc.). Sub-classes call the ``on_*`` hooks at the appropriate points in the training loop so that any registered :class:`AbstractCallback` can react. :param args: Processed configuration object. Must expose at least ``random_seed`` (int). :type args: argparse.Namespace or similar :param callbacks: Ordered list of callback instances to invoke at each lifecycle hook. :type callbacks: list of AbstractCallback .. py:attribute:: attributes .. py:attribute:: callbacks .. py:attribute:: is_global_zero :value: True .. py:attribute:: global_rank :value: 0 .. py:attribute:: local_rank :value: 0 .. py:attribute:: strategy :value: None .. py:method:: on_fit_start(*args, **kwargs) Dispatch ``on_fit_start`` to all registered callbacks. Called once before the first training epoch begins. .. py:method:: on_fit_end(*args, **kwargs) Dispatch ``on_fit_end`` to all registered callbacks. Called once after the last training epoch completes. .. py:method:: on_train_epoch_start(*args, **kwargs) Dispatch ``on_train_epoch_start`` to all registered callbacks. Called at the beginning of every epoch. .. py:method:: on_train_epoch_end(*args, **kwargs) Dispatch ``on_train_epoch_end`` to all registered callbacks. Called at the end of every epoch after the loss has been accumulated. .. py:method:: on_train_batch_end(*args, **kwargs) Dispatch ``on_train_batch_end`` to all registered callbacks. Called after each mini-batch gradient update. .. py:method:: save_checkpoint(full_path: str, model) -> None :staticmethod: Persist model weights to disk. :param full_path: Absolute or relative file path (including filename) where the ``state_dict`` will be written, e.g. ``'Experiments/run1/model.pt'``. :type full_path: str :param model: The model whose ``state_dict`` is to be saved. :type model: torch.nn.Module .. py:class:: BaseInteractiveKGE(path: str = None, url: str = None, construct_ensemble: bool = False, model_name: str = None, apply_semantic_constraint: bool = False) Base class for interactive, post-training use of KGE models. Loads a pre-trained model from disk (or a remote URL) together with its entity/relation index mappings and exposes the prediction API used by :class:`~dicee.knowledge_graph_embeddings.KGE`. :param path: Path to the experiment directory produced by :class:`Execute`. Must contain ``model.pt``, ``configuration.json``, ``entity_to_idx.csv`` and ``relation_to_idx.csv``. :type path: str, optional :param url: Remote URL of a pre-trained model to download. Mutually exclusive with *path*. :type url: str, optional :param construct_ensemble: When ``True``, load all checkpoint files in *path* and average their weights to form an ensemble model. Defaults to ``False``. :type construct_ensemble: bool, optional :param model_name: Filename (without extension) of the checkpoint to load when multiple ``.pt`` files exist in *path*. :type model_name: str, optional :param apply_semantic_constraint: Reserved for future use. Defaults to ``False``. :type apply_semantic_constraint: bool, optional .. py:attribute:: construct_ensemble :value: False .. py:attribute:: apply_semantic_constraint :value: False .. py:attribute:: configs .. py:method:: get_eval_report() -> dict .. py:method:: get_bpe_token_representation(str_entity_or_relation: Union[List[str], str]) -> Union[List[List[int]], List[int]] :param str_entity_or_relation: :type str_entity_or_relation: corresponds to a str or a list of strings to be tokenized via BPE and shaped. :rtype: A list integer(s) or a list of lists containing integer(s) .. py:method:: get_padded_bpe_triple_representation(triples: List[List[str]]) -> Tuple[List, List, List] :param triples: .. py:method:: set_model_train_mode() -> None Switch the underlying model to training mode. Calls ``model.train()`` and re-enables gradient computation for all parameters so that subsequent calls to optimisation steps work correctly after a period of inference. .. py:method:: set_model_eval_mode() -> None Switch the underlying model to evaluation mode. Calls ``model.eval()`` and freezes all parameters (``requires_grad = False``) so that dropout and batch-norm layers behave deterministically during inference. .. py:property:: name .. py:method:: sample_entity(n: int) -> List[str] Return *n* random entity strings without replacement. :param n: Number of entities to sample. Must be non-negative and at most ``num_entities``. :type n: int :returns: Randomly selected entity string labels. :rtype: List[str] .. py:method:: sample_relation(n: int) -> List[str] Return *n* random relation strings without replacement. :param n: Number of relations to sample. Must be non-negative and at most ``num_relations``. :type n: int :returns: Randomly selected relation string labels. :rtype: List[str] .. py:method:: is_seen(entity: str = None, relation: str = None) -> bool Check whether an entity or relation was present in the training set. Exactly one of *entity* or *relation* should be provided. :param entity: Entity string label to look up. :type entity: str, optional :param relation: Relation string label to look up. :type relation: str, optional :returns: ``True`` if the given string is in the respective index mapping, ``False`` otherwise. :rtype: bool .. py:method:: save() -> None Persist the current model weights to the experiment directory. The checkpoint filename encodes the current timestamp so successive calls do not overwrite each other. Ensemble models are saved with an ``_ensemble_`` infix in the filename. .. py:method:: get_entity_index(x: str) -> int Return the integer index for a given entity string. :param x: Entity string label (must have been seen during training). :type x: str :returns: Corresponding row index in the entity embedding matrix. :rtype: int .. py:method:: get_relation_index(x: str) -> int Return the integer index for a given relation string. :param x: Relation string label (must have been seen during training). :type x: str :returns: Corresponding row index in the relation embedding matrix. :rtype: int .. py:method:: index_triple(head_entity: List[str], relation: List[str], tail_entity: List[str]) -> Tuple[torch.LongTensor, torch.LongTensor, torch.LongTensor] Convert string triple lists to integer index tensors. :param head_entity: Head entity string labels. :type head_entity: List[str] :param relation: Relation string labels. :type relation: List[str] :param tail_entity: Tail entity string labels. :type tail_entity: List[str] :returns: **idx_head_entity, idx_relation, idx_tail_entity** -- Each has shape ``(n, 1)`` containing the integer indices for the corresponding strings. :rtype: torch.LongTensor .. py:method:: add_new_entity_embeddings(entity_name: str = None, embeddings: torch.FloatTensor = None) -> None Extend the entity embedding table with a new entity at inference time. The new entity is appended to both ``entity_to_idx`` / ``idx_to_entity`` mappings and the ``entity_embeddings`` weight tensor so that subsequent calls to prediction methods can reference it by name. :param entity_name: String label for the new entity. If the entity already exists in the index no modification is made. :type entity_name: str :param embeddings: 1-D float tensor of length ``embedding_dim`` containing the pre-computed embedding for the new entity. :type embeddings: torch.FloatTensor .. py:method:: get_entity_embeddings(items: List[str]) -> torch.FloatTensor Return the embedding vectors for the given entity strings. For standard (non-BPE) models the method looks up each string in ``entity_to_idx`` and returns the corresponding rows of the entity embedding matrix. For BPE models subword token embeddings are fetched and flattened into a single vector per entity. :param items: Entity string labels to retrieve. :type items: List[str] :returns: Shape ``(len(items), embedding_dim)``. :rtype: torch.FloatTensor .. py:method:: get_relation_embeddings(items: List[str]) -> torch.FloatTensor Return the embedding vectors for the given relation strings. :param items: Relation string labels to retrieve. :type items: List[str] :returns: Shape ``(len(items), embedding_dim)``. :rtype: torch.FloatTensor .. py:method:: construct_input_and_output(head_entity: List[str], relation: List[str], tail_entity: List[str], labels) -> Tuple[torch.LongTensor, torch.FloatTensor] Build an indexed triple tensor and a label tensor from string inputs. :param head_entity: Head entity string labels. :type head_entity: List[str] :param relation: Relation string labels. :type relation: List[str] :param tail_entity: Tail entity string labels. :type tail_entity: List[str] :param labels: Binary or soft labels (one per triple) used as training targets. :type labels: list or array-like :returns: * **x** (*torch.LongTensor*) -- Shape ``(n, 3)`` integer-indexed triples. * **labels** (*torch.FloatTensor*) -- Shape ``(n,)`` float label tensor. .. py:method:: parameters() .. py:class:: InteractiveQueryDecomposition Mixin that provides fuzzy-logic operators for multi-hop EPFO query answering. The three families of operators — T-norm, T-conorm, and negation norm — are applied element-wise over entity score tensors to compose complex queries from atomic link-prediction results (e.g. 2p, 3p, 2i, ip, up). .. py:method:: t_norm(tens_1: torch.Tensor, tens_2: torch.Tensor, tnorm: str = 'min') -> torch.Tensor Apply a T-norm to combine two entity score distributions. :param tens_1: Score tensors of identical shape, values in ``[0, 1]``. :type tens_1: torch.Tensor :param tens_2: Score tensors of identical shape, values in ``[0, 1]``. :type tens_2: torch.Tensor :param tnorm: Operator to use. ``'min'`` applies the Gödel (min) T-norm; ``'prod'`` applies the product T-norm. :type tnorm: str :returns: Element-wise combined scores of the same shape as the inputs. :rtype: torch.Tensor .. py:method:: tensor_t_norm(subquery_scores: torch.FloatTensor, tnorm: str = 'min') -> torch.FloatTensor Compute T-norm over [0,1] ^{n imes d} where n denotes the number of hops and d denotes number of entities .. py:method:: t_conorm(tens_1: torch.Tensor, tens_2: torch.Tensor, tconorm: str = 'min') -> torch.Tensor Apply a T-conorm (S-norm) to combine two score distributions (union). :param tens_1: Score tensors of identical shape, values in ``[0, 1]``. :type tens_1: torch.Tensor :param tens_2: Score tensors of identical shape, values in ``[0, 1]``. :type tens_2: torch.Tensor :param tconorm: Operator to use. ``'min'`` applies the Gödel (max) T-conorm; ``'prod'`` applies the probabilistic sum T-conorm. :type tconorm: str :returns: Element-wise combined scores of the same shape as the inputs. :rtype: torch.Tensor .. py:method:: negnorm(tens_1: torch.Tensor, lambda_: float, neg_norm: str = 'standard') -> torch.Tensor Apply a negation norm (complement) to an entity score distribution. :param tens_1: Input score tensor, values in ``[0, 1]``. :type tens_1: torch.Tensor :param lambda_: Shape parameter used by the Sugeno and Yager negation norms. Ignored for the standard complement. :type lambda_: float :param neg_norm: Which negation to apply: ``'standard'`` (``1 - x``), ``'sugeno'``, or ``'yager'``. :type neg_norm: str :returns: Complemented score tensor of the same shape as *tens_1*. :rtype: torch.Tensor .. py:class:: AbstractCallback Bases: :py:obj:`abc.ABC`, :py:obj:`lightning.pytorch.callbacks.Callback` Abstract base class for KGE training lifecycle callbacks. Concrete sub-classes override one or more hook methods to perform custom actions at specific points during training (e.g. weight averaging, periodic evaluation, model checkpointing). All hooks have empty default implementations so sub-classes only need to override the hooks they care about. Callbacks are registered by passing them to the trainer's *callbacks* list. They are also compatible with PyTorch Lightning trainers because this class extends ``lightning.pytorch.callbacks.Callback``. .. py:method:: on_init_start(*args, **kwargs) Called when the trainer is about to be constructed. Override to perform setup that must happen before any trainer state is initialised. .. py:method:: on_init_end(*args, **kwargs) Called immediately after the trainer has been constructed. Override to perform setup that requires a fully initialised trainer. .. py:method:: on_fit_start(trainer, model) Called once before the first training epoch. :param trainer: The active trainer instance. :type trainer: AbstractTrainer or pl.Trainer :param model: The model about to be trained. :type model: BaseKGE .. py:method:: on_train_epoch_end(trainer, model) Called at the end of each training epoch. :param trainer: The active trainer instance. :type trainer: AbstractTrainer or pl.Trainer :param model: The model being trained. ``model.loss_history`` contains the per-epoch average losses accumulated so far. :type model: BaseKGE .. py:method:: on_train_batch_end(*args, **kwargs) Called after each mini-batch gradient update. Override to inspect or modify the model at a finer granularity than epoch-level hooks. .. py:method:: on_fit_end(*args, **kwargs) Called once after the final training epoch completes. Override to perform post-training actions such as saving the final model state, computing evaluation metrics, or cleaning up resources. .. py:class:: AbstractPPECallback(num_epochs, path, epoch_to_start, last_percent_to_consider) Bases: :py:obj:`AbstractCallback` Abstract base class for Post-training Parameter Ensembling (PPE) callbacks. Sub-classes implement weight-averaging strategies (SWA, EMA, SWAG, …) by overriding :meth:`on_train_epoch_end` and :meth:`on_fit_end`. Common book-keeping (epoch counter, sample counter, alpha weights) is managed here. :param num_epochs: Total number of training epochs. :type num_epochs: int :param path: Experiment directory where averaged checkpoints will be written. :type path: str :param epoch_to_start: First epoch at which the averaging procedure should begin. :type epoch_to_start: int :param last_percent_to_consider: Fraction of the total training epochs (counted from the end) whose checkpoints are included in the ensemble. :type last_percent_to_consider: float .. py:attribute:: num_epochs .. py:attribute:: path .. py:attribute:: sample_counter :value: 0 .. py:attribute:: epoch_count :value: 0 .. py:attribute:: alphas :value: None .. py:method:: on_fit_start(trainer, model) Called once before the first training epoch. :param trainer: The active trainer instance. :type trainer: AbstractTrainer or pl.Trainer :param model: The model about to be trained. :type model: BaseKGE .. py:method:: on_fit_end(trainer, model) Called once after the final training epoch completes. Override to perform post-training actions such as saving the final model state, computing evaluation metrics, or cleaning up resources. .. py:method:: store_ensemble(param_ensemble) -> None .. py:class:: BaseInteractiveTrainKGE Abstract/base class for training knowledge graph embedding models interactively. This class provides methods for re-training KGE models and also Literal Embedding model. .. py:method:: train_triples(h: List[str], r: List[str], t: List[str], labels: List[float], iteration=2, optimizer=None) .. py:method:: train_k_vs_all(h, r, iteration=1, lr=0.001) Train k vs all :param head_entity: :param relation: :param iteration: :param lr: :return: .. py:method:: train(kg, lr=0.1, epoch=10, batch_size=32, neg_sample_ratio=10, num_workers=1) -> None Retrained a pretrain model on an input KG via negative sampling. .. py:method:: train_literals(train_file_path: str = None, num_epochs: int = 100, lit_lr: float = 0.001, lit_normalization_type: str = 'z-norm', batch_size: int = 1024, sampling_ratio: float = None, random_seed=1, loader_backend: str = 'pandas', freeze_entity_embeddings: bool = True, gate_residual: bool = True, device: str = None, suffle_data: bool = True) Trains the Literal Embeddings model using literal data. :param train_file_path: Path to the training data file. :type train_file_path: str :param num_epochs: Number of training epochs. :type num_epochs: int :param lit_lr: Learning rate for the literal model. :type lit_lr: float :param norm_type: Normalization type to use ('z-norm', 'min-max', or None). :type norm_type: str :param batch_size: Batch size for training. :type batch_size: int :param sampling_ratio: Ratio of training triples to use. :type sampling_ratio: float :param loader_backend: Backend for loading the dataset ('pandas' or 'rdflib'). :type loader_backend: str :param freeze_entity_embeddings: If True, freeze the entity embeddings during training. :type freeze_entity_embeddings: bool :param gate_residual: If True, use gate residual connections in the model. :type gate_residual: bool :param device: Device to use for training ('cuda' or 'cpu'). If None, will use available GPU or CPU. :type device: str :param suffle_data: If True, shuffle the dataset before training. :type suffle_data: bool