pyRDF2Vec

In pyRDF2Vec, a walking strategy is the process of generating graph walks for each vertex in a Knowledge Graph up to a certain depth (number of hops). Strategies are categorized into two types:

1. Extraction strategies: Used to generate the initial walks.
2. Transformation strategies: Used to modify or relabel the walks to create different representations.

Common strategies include:

• anonymous walks: Transforms label information into positional information.
• community walks: An extraction strategy that allows for community hops (jumping to a node in the same community rather than a direct neighbor) based on a specific probability.
• Hierarchical Random Walks (HALK): A transformation strategy that removes rare entities from random walks.
• N-Gram walks: A transformation strategy that creates N-grams from N consecutive hops in a walk, which are then relabeled.
• walklets: A transformation strategy using walks of length two, consisting of the root of the original walk and one of the hops.

To generate embeddings for a list of entities from a Knowledge Graph (KG), follow these steps:

1. Load entities: Prepare a list of entity URIs (e.g., from a CSV/TSV).
2. Define the KG: Initialize a KG object using a SPARQL endpoint or a local file.
3. Configure the Transformer: Create an RDF2VecTransformer by specifying an embedder (e.g., Word2Vec) and a walking strategy (e.g., RandomWalker).
4. Run transformation: Call fit_transform(knowledge_graph, entities) to obtain the embeddings and associated literals.

Tip: For better accuracy in most datasets, set with_reverse=True in your walking strategy (like RandomWalker) to allow the model to learn from both parents and children.

import pandas as pd
from pyrdf2vec import RDF2VecTransformer
from pyrdf2vec.embedders import Word2Vec
from pyrdf2vec.graphs import KG
from pyrdf2vec.walkers import RandomWalker

# 1. Load entities
data = pd.read_csv("samples/countries-cities/entities.tsv", sep="\t")
entities = [entity for entity in data["location"]]

# 2. Define Knowledge Graph (SPARQL endpoint)
knowledge_graph = KG(
    "https://dbpedia.org/sparql",
    skip_predicates={"www.w3.org/1999/02/22-rdf-syntax-ns#type"},
    literals=[
        ["http://dbpedia.org/ontology/wikiPageWikiLink", "http://www.w3.org/2004/02/skos/core#prefLabel"],
        ["http://dbpedia.org/ontology/humanDevelopmentIndex"],
    ],
)

# 3. Create transformer
transformer = RDF2VecTransformer(
    Word2Vec(epochs=10),
    walkers=[RandomWalker(4, 10, with_reverse=True, n_jobs=2)],
)

# 4. Get embeddings and literals
embeddings, literals = transformer.fit_transform(knowledge_graph, entities)

To implement a custom sampling strategy for assigning weights to links in a Knowledge Graph, follow these steps:

1. Create the sampler: Define your new class (e.g., FooSampler) in pyrdf2vec/samplers/.
2. Register the sampler: Import your class in pyrdf2vec/samplers/__init__.py and add it to the __all__ list.
3. Implement the logic: Extend the Sampler class and implement the fit and get_weight methods.
4. Test: Create a unit test in tests/samplers/foo.py and run it using pytest.
5. Validate: Run tox -e lint,docs to ensure code style and documentation compliance.

import attr
from pyrdf2vec.graph import KG
from pyrdf2vec.samplers import Sampler
from pyrdf2vec.typings import Hop

@attr.s
class FooSampler(Sampler):
    def fit(self, kg: KG) -> None:
        # Implement fitting logic
        pass

    def get_weight(self, hop: Hop) -> int:
        # Implement weight calculation logic
        return 1

To implement a custom walking strategy for extracting walks from a Knowledge Graph, follow these steps:

1. Create the walker: Define your new class (e.g., FooWalker) in pyrdf2vec/walkers/.
2. Register the walker: Import your class in pyrdf2vec/walkers/__init__.py and add it to the __all__ list.
3. Implement the logic: Extend the Walker class and implement the _extract method.
4. Test: Create a unit test in tests/walkers/foo.py and run it using pytest.
5. Validate: Run tox -e lint,docs to ensure code style and documentation compliance.

from pyrdf2vec.graphs import KG, Vertex
from pyrdf2vec.typings import EntityWalks, SWalk, Walk
from pyrdf2vec.walkers import Walker
import attr

@attr.s
class FooWalker(Walker):
    def _extract(self, kg: KG, instance: Vertex) -> EntityWalks:
        # Implement walk extraction logic here
        canonical_walks: Set[SWalk] = set()
        return {instance.name: list(canonical_walks)}

To implement a new embedding technique (e.g., for training a corpus of walks), follow these five steps:

1. Create the embedder: Define your new class (e.g., Foo) in pyrdf2vec/embedders.
2. Register the embedder: Import your new class in pyrdf2vec/embedders/__init__.py and add it to the __all__ list.
3. Implement the Embedder interface: Extend the Embedder class and implement the fit and transform methods.
4. Add unit tests: Create a test file in tests/embedders/ (e.g., tests/embedders/foo.py) following the existing Word2Vec tests as a template.
5. Verify quality: Run linting and documentation checks using tox.

Required Method Signatures:

from typing import List
import attr
from pyrdf2vec.embedders import Embedder
from pyrdf2vec.typings import Embeddings, Entities

@attr.s
class Foo(Embedder):
    def fit(self, walks: List[List[SWalk]], is_updated: bool = False) -> Embedder:
        # Implementation logic
        pass

    def transform(self, entities: Entities) -> Embeddings:
        # Implementation logic
        pass

from typing import List

import attr

from pyrdf2vec.embedders import Embedder
from pyrdf2vec.typings import Embeddings, Entities

@attr.s
class Foo(Embedder):
    """Defines Foo embedding technique."""

    def fit(self,  walks: List[List[SWalk]], is_updated: bool = False) -> Embedder:
        """Fits the model based on provided walks.

        Args:
            walks: The walks to create the corpus to to fit the model.
            is_update: True if the new walks should be added to old model's
                walks, False otherwise.
            Defaults to False.

        Returns:
            The fitted Foo model.

        """
        # TODO: to be implemented

    def transform(self, entities: Entities) -> Embeddings:
        """The features vector of the provided entities.

        Args:
            entities: The entities including test entities to create the
                embeddings. Since RDF2Vec is unsupervised, there is no label
                leakage.

        Returns:
            The features vector of the provided entities.

        """
        # TODO: to be implemented