Skip to content

base

FrameEncoder

Base class for encoding a KlinkerFrame as embedding.

Source code in klinker/encoders/base.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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]:
        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)
        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)
        start = time.time()
        left_enc, right_enc = self._encode_as(
            left=left,
            right=right,
            left_rel=left_rel,
            right_rel=right_rel,
            return_type=return_type,
        )
        end = time.time()
        self._encoding_time = end - start
        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.

Parameters:

Name Type Description Default
left Frame

Frame: left attribute information.

 required
right Frame

Frame: right attribute information.

 required
left_rel Optional[Frame]

Optional[Frame]: left relation triples.

 None
right_rel Optional[Frame]

Optional[Frame]: right relation triples.

 None
return_type GeneralVectorLiteral

GeneralVectorLiteral: Either pt or np to return as pytorch tensor or numpy array.

 'pt'

Returns:

Type Description
Tuple[NamedVector, NamedVector]

Embeddings of given left/right dataset.
Source code in klinker/encoders/base.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
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)
    start = time.time()
    left_enc, right_enc = self._encode_as(
        left=left,
        right=right,
        left_rel=left_rel,
        right_rel=right_rel,
        return_type=return_type,
    )
    end = time.time()
    self._encoding_time = end - start
    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).

Parameters:

Name Type Description Default
left Frame

Frame: left attributes.

 required
right Frame

Frame: right attributes.

 required

Returns:

Type Description
Tuple[Frame, Frame]

left, right
Source code in klinker/encoders/base.py 
45
46
47
48
49
50
51
52
53
54
55
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.

Parameters:

Name Type Description Default
left Frame

Frame: left attributes.

 required
right Frame

Frame: right attributes.

 required
left_rel Optional[Frame]

Optional[Frame]: left relation triples.

 None
right_rel Optional[Frame]

Optional[Frame]: right relation triples.

 None
Source code in klinker/encoders/base.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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!"
        )

RelationFrameEncoder

Bases: FrameEncoder

Base class for Encoders, that also utilize relational information.

Source code in klinker/encoders/base.py 
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
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
        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.

Parameters:

Name Type Description Default
left SeriesType

Frame: left attribute information.

 required
right SeriesType

Frame: right attribute information.

 required
*
required
left_rel Optional[Frame]

Optional[Frame]: left relation triples.

 None
right_rel Optional[Frame]

Optional[Frame]: right relation triples.

 None
return_type GeneralVectorLiteral

GeneralVectorLiteral: Either pt or np to return as pytorch tensor or numpy array.

 'pt'

Returns:

Type Description
Tuple[NamedVector, NamedVector]

Embeddings of given left/right dataset.
Source code in klinker/encoders/base.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
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
    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.

Parameters:

Name Type Description Default
left Frame

Frame: left attribute information.

 required
right Frame

Frame: right attribute information.

 required
left_rel Optional[Frame]

Optional[Frame]: left relation triples.

 None
right_rel Optional[Frame]

Optional[Frame]: right relation triples.

 None

Raises:

Type Description
ValueError

If attribute frames consist of multiple columns or relational frames are missing.
Source code in klinker/encoders/base.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
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!")

TokenizedFrameEncoder

Bases: FrameEncoder

FrameEncoder that uses tokenization of attribute values.

Source code in klinker/encoders/base.py 
153
154
155
156
157
158
159
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

initialize_and_fill(known, all_names, initializer=nn.init.xavier_normal_, initializer_kwargs=None)

Use initalizer and set known values from NamedVector

Parameters:

Name Type Description Default
known NamedVector[Tensor]

NamedVector[torch.Tensor]: Known Embeddings.

 required
all_names Union[List[str], Dict[str, int]]

Union[List[str], Dict[str, int]]: All entity names.

 required
initializer

Torch initializer.

 xavier_normal_
initializer_kwargs OptionalKwargs

Keyword args passed to initializer.

 None

Returns:

Type Description
NamedVector[Tensor]

Named Vector filled with known values and others from initializer.

Examples:

>>> from klinker.encoders.base import initialize_and_fill
>>> from klinker.data import NamedVector
>>> import torch
>>> known = NamedVector(["e1","e2"], torch.rand(2,2))
>>> known # doctest: +SKIP
NamedVector(0|"e1": [0.1697, 0.6516],
        1|"e2": [0.2020, 0.8281],
        dtype=torch.float32)
>>> initialize_and_fill(known, ["e0","e1","e2","e3"]) # doctest: +SKIP
NamedVector(0|"e0": [-0.1876,  1.0404],
        1|"e1": [0.1697, 0.6516],
        2|"e2": [0.2020, 0.8281],
        3|"e3": [ 0.0254, -0.8697],
        dtype=torch.float32)
Source code in klinker/encoders/base.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def initialize_and_fill(
    known: NamedVector[torch.Tensor],
    all_names: Union[List[str], Dict[str, int]],
    initializer=nn.init.xavier_normal_,
    initializer_kwargs: OptionalKwargs = None,
) -> NamedVector[torch.Tensor]:
    """Use initalizer and set known values from NamedVector

    Args:
      known: NamedVector[torch.Tensor]: Known Embeddings.
      all_names: Union[List[str], Dict[str, int]]: All entity names.
      initializer: Torch initializer.
      initializer_kwargs: Keyword args passed to initializer.

    Returns:
        Named Vector filled with known values and others from initializer.

    Examples:

        >>> from klinker.encoders.base import initialize_and_fill
        >>> from klinker.data import NamedVector
        >>> import torch
        >>> known = NamedVector(["e1","e2"], torch.rand(2,2))
        >>> known # doctest: +SKIP
        NamedVector(0|"e1": [0.1697, 0.6516],
                1|"e2": [0.2020, 0.8281],
                dtype=torch.float32)
        >>> initialize_and_fill(known, ["e0","e1","e2","e3"]) # doctest: +SKIP
        NamedVector(0|"e0": [-0.1876,  1.0404],
                1|"e1": [0.1697, 0.6516],
                2|"e2": [0.2020, 0.8281],
                3|"e3": [ 0.0254, -0.8697],
                dtype=torch.float32)

    """
    if not set(known.names).union(set(all_names)) == set(all_names):
        raise ValueError("Known vector must be subset of all_names!")
    initializer = initializer_resolver.lookup(initializer)
    initializer_kwargs = initializer_kwargs or {}

    # get same shape for single vector
    empty = torch.empty_like(known[0])

    # create full lengthy empty matrix with correct shape
    vector = torch.stack([empty] * len(all_names))
    vector = initializer(vector, **initializer_kwargs)
    nv = NamedVector(names=all_names, vectors=vector)
    nv[known.names] = known.vectors
    return nv