Sources

Concrete EntitySource wrappers over external targets too large, or already indexed elsewhere, to enumerate as a list[Entity].

wn

EntitySource wrapper for WordNet via the wn library.

Wraps wn -- a query-driven interface to locally-installed wordnets (e.g. Open English WordNet) -- as an EntitySource, so WSD/WSA can target a full dictionary without materializing every synset as an Entity up front. wn itself still requires a lexicon to have been downloaded once (python -m wn download oewn:2021, or wn.download("oewn:2021")) -- this module only avoids loading its senses into memory ahead of time, not the download step.

WnEntitySource

WnEntitySource(
    lexicon: str = "oewn:2021", lang: str | None = None
)

Bases: EntitySource

Queries a locally-installed wn lexicon as an EntitySource.

See the module docstring for the one-time lexicon download wn itself requires.

Parameters:

Name	Type	Description	Default
lexicon	str	A wn lexicon specifier (e.g. the default "oewn:2021", Open English WordNet 2021). Passed straight through to wn.synsets/wn.synset.	'oewn:2021'
lang	str | None	Restrict to a single BCP-47 language code (e.g. "en"). Left as None (the default), an entity's labels carry a per-word language tag instead of a plain string, since a multilingual/multi-lexicon query can return words from more than one language on a single synset.	None

Raises:

Type	Description
OptionalDependencyError	If wn isn't installed.

Source code in src/linkingtk/sources/wn.py

def __init__(self, lexicon: str = "oewn:2021", lang: str | None = None) -> None:
    try:
        import wn
    except ImportError as exc:
        raise OptionalDependencyError("WnEntitySource", "wn") from exc
    self._wn = wn
    self.lexicon = lexicon
    self.lang = lang

get

get(entity_id: str) -> Entity | None

Look up a synset by its wn id, or None if it doesn't exist.

Source code in src/linkingtk/sources/wn.py

def get(self, entity_id: str) -> Entity | None:
    """Look up a synset by its `wn` id, or ``None`` if it doesn't exist."""
    try:
        synset = self._wn.synset(entity_id, lexicon=self.lexicon, lang=self.lang)
    except self._wn.Error:
        return None
    return self._to_entity(synset)

search

search(query: str, top_k: int = 10) -> list[Entity]

Return up to top_k synsets containing query as a lemma.

Ordering follows wn.synsets' own (the lexicon's stored sense order -- for Open English WordNet, most-frequent-sense first).

Source code in src/linkingtk/sources/wn.py

def search(self, query: str, top_k: int = 10) -> list[Entity]:
    """Return up to ``top_k`` synsets containing ``query`` as a lemma.

    Ordering follows ``wn.synsets``' own (the lexicon's stored sense
    order -- for Open English WordNet, most-frequent-sense first).
    """
    synsets = self._wn.synsets(query, lexicon=self.lexicon, lang=self.lang)
    if not synsets and "_" in query:
        synsets = self._wn.synsets(
            query.replace("_", " "), lexicon=self.lexicon, lang=self.lang
        )
    return [self._to_entity(synset) for synset in synsets[:top_k]]

sensekey_to_synset_id

sensekey_to_synset_id(
    sense_key: str, lexicon: str = "omw-en:1.4"
) -> str | None

Resolve a Princeton WordNet sense key (e.g. "group%1:03:00::") to its synset id.

A sense key isn't a wn id by itself -- wn has no lookup-by-sense-key API, so this instead lists every sense of the key's lemma in lexicon and matches by each sense's own recorded identifier metadata (the sense key wn was built from). Also tries sense_key's adjective/adjective-satellite variant (its part-of-speech digit swapped between 3 and 5) if the exact one doesn't match -- without it, every satellite-adjective instance in a real corpus silently fails to resolve. Used by UfsacDataset, whose corpora are tagged with WordNet 3.0 sense keys rather than synset ids.

Parameters:

Name	Type	Description	Default
sense_key	str	A WordNet sense key, e.g. "group%1:03:00::".	required
lexicon	str	A wn lexicon specifier whose senses carry the original sense key as their identifier metadata -- true of "omw-en:1.4" (the default, OMW's English WordNet based on WordNet 3.0) but not of every lexicon.	'omw-en:1.4'

Returns:

Type	Description
str | None	The matching synset's wn id, or None if no sense in lexicon
str | None	carries that sense key.

Raises:

Type	Description
OptionalDependencyError	If wn isn't installed.

Source code in src/linkingtk/sources/wn.py

def sensekey_to_synset_id(sense_key: str, lexicon: str = "omw-en:1.4") -> str | None:
    """Resolve a Princeton WordNet sense key (e.g. ``"group%1:03:00::"``) to its synset id.

    A sense key isn't a `wn` id by itself -- `wn` has no lookup-by-sense-key
    API, so this instead lists every sense of the key's lemma in `lexicon`
    and matches by each sense's own recorded ``identifier`` metadata (the
    sense key `wn` was built from). Also tries `sense_key`'s
    adjective/adjective-satellite variant (its part-of-speech digit
    swapped between ``3`` and ``5``) if the exact one doesn't match --
    without it, every satellite-adjective instance in a real corpus
    silently fails to resolve. Used by
    [UfsacDataset][linkingtk.datasets.ufsac.UfsacDataset], whose corpora
    are tagged with WordNet 3.0 sense keys rather than synset ids.

    Args:
        sense_key: A WordNet sense key, e.g. ``"group%1:03:00::"``.
        lexicon: A `wn` lexicon specifier whose senses carry the original
            sense key as their ``identifier`` metadata -- true of
            ``"omw-en:1.4"`` (the default, OMW's English WordNet based on
            WordNet 3.0) but not of every lexicon.

    Returns:
        The matching synset's `wn` id, or ``None`` if no sense in `lexicon`
        carries that sense key.

    Raises:
        OptionalDependencyError: If `wn` isn't installed.
    """
    try:
        import wn
    except ImportError as exc:
        raise OptionalDependencyError("sensekey_to_synset_id", "wn") from exc

    lemma = sense_key.split("%", 1)[0]
    senses = wn.senses(lemma, lexicon=lexicon)
    if not senses and "_" in lemma:
        senses = wn.senses(lemma.replace("_", " "), lexicon=lexicon)
    candidates = {sense_key}
    variant = _adjective_satellite_variant(sense_key)
    if variant is not None:
        candidates.add(variant)
    for sense in senses:
        if sense.metadata().get("identifier") in candidates:
            return sense.synset().id
    return None

synset_id_to_sensekey

synset_id_to_sensekey(
    synset_id: str, lemma: str, lexicon: str = "omw-en:1.4"
) -> str | None

The inverse of sensekey_to_synset_id: given a synset id and the lemma it was reached through, return that lemma's sense key on that synset.

A synset can lexicalize several lemmas, each with its own sense key on that synset -- lemma picks which one, the same way UFSAC/SemEval's own gold-standard sense keys are always relative to a specific instance's lemma, not just its synset. Useful for converting this package's predictions back to sense-key form to cross-check against an external, sense-key-based scorer (e.g. the Raganato et al. WSD evaluation framework's own Scorer.java).

Returns lexicon's own stored form as-is, unlike sensekey_to_synset_id's adjective/adjective-satellite fallback -- there's no target key here to fall back towards. In practice this is omw-en:1.4's own convention (always digit 5, adjective satellite), which for adjectives can differ from what a specific external gold file uses for that word (the official Raganato et al. gold files use a genuine, per-word mix of both 3 and 5, not one uniformly) -- a real, known source of mismatched-but-not-wrong predictions when cross-checking adjective instances against an external scorer.

Parameters:

Name	Type	Description	Default
synset_id	str	A wn synset id, e.g. "omw-en-00031264-n".	required
lemma	str	The lemma to resolve the sense key for.	required
lexicon	str	A wn lexicon specifier whose senses carry the original sense key as their identifier metadata -- see sensekey_to_synset_id.	'omw-en:1.4'

Returns:

Type	Description
str | None	The matching sense key, or None if lemma has no sense on
str | None	synset_id in lexicon.

Raises:

Type	Description
OptionalDependencyError	If wn isn't installed.

Source code in src/linkingtk/sources/wn.py

def synset_id_to_sensekey(synset_id: str, lemma: str, lexicon: str = "omw-en:1.4") -> str | None:
    """The inverse of [sensekey_to_synset_id][linkingtk.sources.wn.sensekey_to_synset_id]:
    given a synset id and the lemma it was reached through, return that
    lemma's sense key on that synset.

    A synset can lexicalize several lemmas, each with its own sense key on
    that synset -- `lemma` picks which one, the same way UFSAC/SemEval's
    own gold-standard sense keys are always relative to a specific
    instance's lemma, not just its synset. Useful for converting this
    package's predictions back to sense-key form to cross-check against an
    external, sense-key-based scorer (e.g. the Raganato et al. WSD
    evaluation framework's own `Scorer.java`).

    Returns `lexicon`'s own stored form as-is, unlike
    `sensekey_to_synset_id`'s adjective/adjective-satellite fallback --
    there's no target key here to fall back *towards*. In practice this
    is `omw-en:1.4`'s own convention (always digit `5`, adjective
    satellite), which for adjectives can differ from what a specific
    external gold file uses for that word (the official Raganato et al.
    gold files use a genuine, per-word mix of both `3` and `5`, not one
    uniformly) -- a real, known source of mismatched-but-not-wrong
    predictions when cross-checking adjective instances against an
    external scorer.

    Args:
        synset_id: A `wn` synset id, e.g. ``"omw-en-00031264-n"``.
        lemma: The lemma to resolve the sense key for.
        lexicon: A `wn` lexicon specifier whose senses carry the original
            sense key as their ``identifier`` metadata -- see
            `sensekey_to_synset_id`.

    Returns:
        The matching sense key, or ``None`` if `lemma` has no sense on
        `synset_id` in `lexicon`.

    Raises:
        OptionalDependencyError: If `wn` isn't installed.
    """
    try:
        import wn
    except ImportError as exc:
        raise OptionalDependencyError("synset_id_to_sensekey", "wn") from exc

    senses = wn.senses(lemma, lexicon=lexicon)
    if not senses and "_" in lemma:
        senses = wn.senses(lemma.replace("_", " "), lexicon=lexicon)
    for sense in senses:
        if sense.synset().id == synset_id:
            identifier = sense.metadata().get("identifier")
            return identifier if isinstance(identifier, str) else None
    return None

synset_id_to_wn30_offset

synset_id_to_wn30_offset(synset_id: str) -> str

Convert an omw-en:1.4 synset id to its WordNet 3.0 byte-offset form.

E.g. "omw-en-02084071-n" -> "wn:02084071n". Pure string manipulation, not a wn lookup -- confirmed directly that omw-en:1.4's own synset ids already embed the PWN 3.0 offset (wn.synsets("dog", pos="n", lexicon="omw-en:1.4")[0].id == "omw-en-02084071-n", matching EWISER's own checkpoint dictionary entry "wn:02084071n"). Used to look up a candidate sense's index in EwiserEncoder's checkpoint-loaded output vocabulary.

This does not hold for oewn:* lexicons, whose offsets have since diverged from PWN 3.0 (confirmed: oewn:2021's id for the same synset is "oewn-01930264-v" for an unrelated example -- a different number, not just a different prefix).

Parameters:

Name	Type	Description	Default
synset_id	str	An omw-en:1.4 synset id, e.g. "omw-en-02084071-n".	required

Returns:

Type	Description
str	The WordNet 3.0 offset form, e.g. "wn:02084071n".

Raises:

Type	Description
ValueError	If synset_id isn't in omw-en:1.4's "omw-en-########-{n,v,a,r,s}" shape.

Source code in src/linkingtk/sources/wn.py

def synset_id_to_wn30_offset(synset_id: str) -> str:
    """Convert an ``omw-en:1.4`` synset id to its WordNet 3.0 byte-offset form.

    E.g. ``"omw-en-02084071-n"`` -> ``"wn:02084071n"``. Pure string
    manipulation, not a `wn` lookup -- confirmed directly that
    ``omw-en:1.4``'s own synset ids already embed the PWN 3.0 offset
    (``wn.synsets("dog", pos="n", lexicon="omw-en:1.4")[0].id ==
    "omw-en-02084071-n"``, matching EWISER's own checkpoint dictionary
    entry ``"wn:02084071n"``). Used to look up a candidate sense's index in
    [EwiserEncoder][linkingtk.algorithms.wsd.ewiser.EwiserEncoder]'s
    checkpoint-loaded output vocabulary.

    This does **not** hold for ``oewn:*`` lexicons, whose offsets have
    since diverged from PWN 3.0 (confirmed: ``oewn:2021``'s id for the same
    synset is ``"oewn-01930264-v"`` for an unrelated example -- a different
    number, not just a different prefix).

    Args:
        synset_id: An ``omw-en:1.4`` synset id, e.g. ``"omw-en-02084071-n"``.

    Returns:
        The WordNet 3.0 offset form, e.g. ``"wn:02084071n"``.

    Raises:
        ValueError: If `synset_id` isn't in ``omw-en:1.4``'s
            ``"omw-en-########-{n,v,a,r,s}"`` shape.
    """
    match = _OMW_EN_SYNSET_ID_RE.match(synset_id)
    if match is None:
        raise ValueError(
            f"{synset_id!r} isn't an omw-en:1.4 synset id (expected "
            '"omw-en-########-{n,v,a,r,s}") -- the WordNet 3.0 offset '
            "conversion only applies to omw-en:1.4, whose synset ids already "
            "embed the PWN 3.0 byte offset directly; other lexicons (e.g. "
            "oewn:*) have since diverged and have no such mapping."
        )
    offset, pos = match.groups()
    return f"wn:{offset}{pos}"

synset_id_via_ili

synset_id_via_ili(
    synset_id: str, lexicon: str, target_lexicon: str
) -> str | None

Map synset_id (a synset id in lexicon) to its counterpart in target_lexicon, via their shared Interlingual Index (ILI) entry.

Unlike wn30_offset_to_synset_id, this works across any two lexicons with ILI links, not just ones that happen to share omw-en:1.4's byte-offset-embedding id convention -- in particular it's what lets EwiserLinker resolve candidates from a WnEntitySource lexicon that differs from the one a checkpoint's SenseVocabulary was built from (e.g. oewn:2025+ candidates against an omw-en:1.4 vocabulary): confirmed directly that oewn:2025+'s "oewn-00319912-v" and omw-en:1.4's "omw-en-00319111-v" are the same synset (different byte offsets -- oewn's have diverged from PWN 3.0, see synset_id_to_wn30_offset's own docstring -- but the same ILI entry, "i23324"), so a plain id-string comparison would never find this match, while this does (see #67).

Parameters:

Name	Type	Description	Default
synset_id	str	A synset id in lexicon, e.g. "oewn-00319912-v".	required
lexicon	str	The wn lexicon synset_id belongs to.	required
target_lexicon	str	The wn lexicon to resolve the matching synset in.	required

Returns:

Type	Description
str | None	The matching synset id in target_lexicon, or None if
str | None	synset_id doesn't resolve in lexicon, has no ILI entry, or
str | None	target_lexicon has no synset attached to that same ILI entry
str | None	(any of which are real, unremarkable outcomes -- not every synset
str | None	is cross-lexicon-linked). The first match is returned if more than
str | None	one synset in target_lexicon shares the ILI entry.

Raises:

Type	Description
OptionalDependencyError	If wn isn't installed.

Source code in src/linkingtk/sources/wn.py

def synset_id_via_ili(synset_id: str, lexicon: str, target_lexicon: str) -> str | None:
    """Map `synset_id` (a synset id in `lexicon`) to its counterpart in
    `target_lexicon`, via their shared Interlingual Index (ILI) entry.

    Unlike [wn30_offset_to_synset_id][linkingtk.sources.wn.wn30_offset_to_synset_id],
    this works across *any* two lexicons with ILI links, not just ones that
    happen to share `omw-en:1.4`'s byte-offset-embedding id convention --
    in particular it's what lets
    [EwiserLinker][linkingtk.algorithms.wsd.ewiser.EwiserLinker] resolve
    candidates from a `WnEntitySource` lexicon that differs from the one a
    checkpoint's `SenseVocabulary` was built from (e.g. `oewn:2025+`
    candidates against an `omw-en:1.4` vocabulary): confirmed directly that
    `oewn:2025+`'s ``"oewn-00319912-v"`` and `omw-en:1.4`'s
    ``"omw-en-00319111-v"`` are the *same* synset (different byte offsets
    -- `oewn`'s have diverged from PWN 3.0, see
    `synset_id_to_wn30_offset`'s own docstring -- but the same ILI entry,
    ``"i23324"``), so a plain id-string comparison would never find this
    match, while this does (see #67).

    Args:
        synset_id: A synset id in `lexicon`, e.g. ``"oewn-00319912-v"``.
        lexicon: The `wn` lexicon `synset_id` belongs to.
        target_lexicon: The `wn` lexicon to resolve the matching synset in.

    Returns:
        The matching synset id in `target_lexicon`, or ``None`` if
        `synset_id` doesn't resolve in `lexicon`, has no ILI entry, or
        `target_lexicon` has no synset attached to that same ILI entry
        (any of which are real, unremarkable outcomes -- not every synset
        is cross-lexicon-linked). The first match is returned if more than
        one synset in `target_lexicon` shares the ILI entry.

    Raises:
        OptionalDependencyError: If `wn` isn't installed.
    """
    try:
        import wn
    except ImportError as exc:
        raise OptionalDependencyError("synset_id_via_ili", "wn") from exc

    try:
        synset = wn.synset(synset_id, lexicon=lexicon)
    except wn.Error:
        return None
    ili = synset.ili
    if ili is None:
        return None
    matches = wn.synsets(ili=ili, lexicon=target_lexicon)
    return matches[0].id if matches else None

wn30_offset_to_synset_id

wn30_offset_to_synset_id(
    offset: str, lexicon: str = "omw-en:1.4"
) -> str | None

The inverse of synset_id_to_wn30_offset: given a WordNet 3.0 offset (e.g. "wn:02084071n"), return the matching synset id in lexicon.

Unlike synset_id_to_wn30_offset, this validates existence via a real wn lookup rather than pure string construction, since offset may come from an external source (e.g. an EWISER checkpoint's own dictionary file) that doesn't necessarily resolve in the installed lexicon.

Parameters:

Name	Type	Description	Default
offset	str	A WordNet 3.0 offset, e.g. "wn:02084071n".	required
lexicon	str	A wn lexicon specifier whose synset ids follow omw-en:1.4's offset-embedding convention (see synset_id_to_wn30_offset). Defaults to "omw-en:1.4".	'omw-en:1.4'

Returns:

Type	Description
str | None	The matching synset id, or None if offset doesn't resolve in
str | None	lexicon.

Raises:

Type	Description
ValueError	If offset isn't in "wn:########pos" shape.
OptionalDependencyError	If wn isn't installed.

Source code in src/linkingtk/sources/wn.py

def wn30_offset_to_synset_id(offset: str, lexicon: str = "omw-en:1.4") -> str | None:
    """The inverse of
    [synset_id_to_wn30_offset][linkingtk.sources.wn.synset_id_to_wn30_offset]:
    given a WordNet 3.0 offset (e.g. ``"wn:02084071n"``), return the
    matching synset id in `lexicon`.

    Unlike `synset_id_to_wn30_offset`, this validates existence via a real
    `wn` lookup rather than pure string construction, since `offset` may
    come from an external source (e.g. an EWISER checkpoint's own
    dictionary file) that doesn't necessarily resolve in the installed
    lexicon.

    Args:
        offset: A WordNet 3.0 offset, e.g. ``"wn:02084071n"``.
        lexicon: A `wn` lexicon specifier whose synset ids follow
            ``omw-en:1.4``'s offset-embedding convention (see
            `synset_id_to_wn30_offset`). Defaults to ``"omw-en:1.4"``.

    Returns:
        The matching synset id, or ``None`` if `offset` doesn't resolve in
        `lexicon`.

    Raises:
        ValueError: If `offset` isn't in ``"wn:########pos"`` shape.
        OptionalDependencyError: If `wn` isn't installed.
    """
    match = _WN30_OFFSET_RE.match(offset)
    if match is None:
        raise ValueError(f'{offset!r} isn\'t a WordNet 3.0 offset (expected "wn:########pos")')

    try:
        import wn
    except ImportError as exc:
        raise OptionalDependencyError("wn30_offset_to_synset_id", "wn") from exc

    digits, pos = match.groups()
    prefix = lexicon.split(":", 1)[0]
    synset_id = f"{prefix}-{digits}-{pos}"
    try:
        wn.synset(synset_id, lexicon=lexicon)
    except wn.Error:
        return None
    return synset_id

wikipedia

EntitySource wrapper for Wikipedia, via the public MediaWiki API.

Wraps the MediaWiki API as a query-driven EntitySource, so EL can target live Wikipedia directly -- without downloading or materializing a dump -- the same way WnEntitySource targets a local WordNet.

WikipediaEntitySource

WikipediaEntitySource(
    lang: str = "en", session: Session | None = None
)

Bases: EntitySource

Queries live Wikipedia, via the MediaWiki API, as an EntitySource.

Entity ids are Wikipedia page titles, kept exactly as Wikipedia has them (including any parenthetical disambiguator, e.g. "Paris (mythology)") so get can round-trip them. Labels, however, have that disambiguator stripped (e.g. "Paris") with the topic moved to the front of the description instead (e.g. "(mythology) ...") -- see _split_title -- so that a bare-word mention can actually reach a disambiguated page as a candidate through ExactMatch's exact-label blocking.

Parameters:

Name	Type	Description	Default
lang	str	The Wikipedia language edition to query (e.g. "en" for en.wikipedia.org).	'en'
session	Session | None	A requests.Session to issue requests through. Defaults to a fresh one carrying a descriptive User-Agent (MediaWiki API etiquette requires one, and requests.Session's own default User-Agent gets rejected outright -- confirmed directly: the live API 403s a request left at requests' default "python-requests/*" header). A caller-supplied session is used exactly as given, headers untouched -- set your own User-Agent on it. Pass a shared session to reuse connections, or a test double to avoid real network calls.	None

Raises:

Type	Description
OptionalDependencyError	If requests isn't installed.

Source code in src/linkingtk/sources/wikipedia.py

def __init__(self, lang: str = "en", session: requests.Session | None = None) -> None:
    try:
        import requests
    except ImportError as exc:
        raise OptionalDependencyError("WikipediaEntitySource", "wikipedia") from exc
    self.lang = lang
    self._api_url = f"https://{lang}.wikipedia.org/w/api.php"
    if session is not None:
        self._session = session
    else:
        self._session = requests.Session()
        self._session.headers["User-Agent"] = _USER_AGENT

get

get(entity_id: str) -> Entity | None

Look up a Wikipedia page by title, or None if it doesn't exist.

Uses action=query&prop=extracts to fetch the page's intro paragraph as plain text.

Source code in src/linkingtk/sources/wikipedia.py

def get(self, entity_id: str) -> Entity | None:
    """Look up a Wikipedia page by title, or ``None`` if it doesn't exist.

    Uses `action=query&prop=extracts` to fetch the page's intro
    paragraph as plain text.
    """
    params: dict[str, str | int] = {
        "action": "query",
        "prop": "extracts|pageprops",
        "titles": entity_id,
        "exintro": 1,
        "explaintext": 1,
        "format": "json",
    }
    response = self._session.get(self._api_url, params=params, timeout=10)
    response.raise_for_status()
    pages = response.json()["query"]["pages"]
    page = next(iter(pages.values()))
    if "missing" in page:
        return None
    label, topic = _split_title(page["title"])
    return Entity(
        id=page["title"],
        labels=[label],
        description=_prefix_topic(page.get("extract") or None, topic),
    )

search

search(query: str, top_k: int = 10) -> list[Entity]

Full-text search Wikipedia for query, best match first.

Uses action=query&list=search -- each hit's snippet (an HTML-highlighted excerpt around the matched terms) becomes the returned entity's description, with the highlighting markup stripped.

Source code in src/linkingtk/sources/wikipedia.py

def search(self, query: str, top_k: int = 10) -> list[Entity]:
    """Full-text search Wikipedia for `query`, best match first.

    Uses `action=query&list=search` -- each hit's ``snippet`` (an
    HTML-highlighted excerpt around the matched terms) becomes the
    returned entity's description, with the highlighting markup
    stripped.
    """
    params: dict[str, str | int] = {
        "action": "query",
        "list": "search",
        "srsearch": query,
        "srlimit": top_k,
        "format": "json",
    }
    response = self._session.get(self._api_url, params=params, timeout=10)
    response.raise_for_status()
    hits = response.json()["query"]["search"]
    entities = []
    for hit in hits:
        label, topic = _split_title(hit["title"])
        entities.append(
            Entity(
                id=hit["title"],
                labels=[label],
                description=_prefix_topic(_clean_snippet(hit["snippet"]), topic),
            )
        )
    return entities

wikidata

EntitySource wrapper for Wikidata, via the public action API.

Wraps Wikidata's action API as a query-driven EntitySource, the same way WikipediaEntitySource targets live Wikipedia. Live per-mention Wikidata queries don't scale well (see linkingtk.sources.vector_index's module docstring) -- pass a prebuilt VectorIndexEntitySource as vector_index to search/get entirely locally instead. WikidataDumpEntities in this module streams Entity objects straight from a downloaded Wikidata dump to build one.

WikidataDumpEntities

WikidataDumpEntities(
    path: str | Path,
    lang: str = "en",
    limit: int | None = None,
    progress: bool = True,
)

Streams Entity objects from an official Wikidata JSON dump.

Wikidata publishes its full dump (https://www.wikidata.org/wiki/Wikidata:Database_download) as latest-all.json.gz: a JSON array of entity objects, one compact object per line (plus bracket-only first/last lines) -- written that way specifically so it can be streamed line by line rather than parsed as one huge array, which is exactly what this class does. A truncated or filtered dump in the same line-delimited shape works too.

Generalizes (reads the dump directly, no intermediate file needed) the dump-processing step of wn-wd-entity-align's own FAISS-index pipeline (wikidata_faiss/extract_dump_labels.py, which instead expects an already-filtered, already-sorted N-Triples export).

Re-iterable -- __iter__ reopens (re-requests the URL, or re-opens the local file) and re-reads from the start every call -- so it can be passed straight to VectorIndexEntitySource.build even with reduced_dim set (which needs two passes over its input); a plain generator can't do that, since it's exhausted after one pass.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the dump file, or an http(s):// URL to stream it from directly (e.g. Wikidata's own dump mirrors) -- fetched fresh, never cached to disk, since the whole point is not having to hold a multi-hundred-GB file locally. Either way, gzip-decompressed on the fly if the name ends in .gz (as the official dump does).	required
lang	str	Which language's label/description/aliases to read -- see WikidataEntitySource.	'en'
limit	int | None	Stop after yielding this many entities, e.g. for a quick local test against a full multi-hundred-GB dump.	None
progress	bool	Show a tqdm progress bar (lines read, entities yielded) while iterating -- this can run for hours over a full dump, so some feedback matters. Pass False to disable.	True

Source code in src/linkingtk/sources/wikidata.py

def __init__(
    self,
    path: str | Path,
    lang: str = "en",
    limit: int | None = None,
    progress: bool = True,
) -> None:
    self.path = path if isinstance(path, str) and _is_url(path) else Path(path)
    self.lang = lang
    self.limit = limit
    self.progress = progress

WikidataEntitySource

WikidataEntitySource(
    lang: str = "en",
    session: Session | None = None,
    vector_index: VectorIndexEntitySource | None = None,
)

Bases: EntitySource

Queries live Wikidata, via its action API, as an EntitySource.

Entity ids are QIDs (e.g. "Q42"). A P31 ("instance of") claim, when present, is surfaced into Entity.properties["instance_of"] as space-joined QIDs.

Parameters:

Name	Type	Description	Default
lang	str	The Wikidata language to request labels/descriptions in (e.g. "en").	'en'
session	Session | None	A requests.Session to issue requests through. Defaults to a fresh one carrying a descriptive User-Agent -- see WikipediaEntitySource for why that matters. A caller-supplied session is used exactly as given, headers untouched.	None
vector_index	VectorIndexEntitySource | None	An optional prebuilt local index (see linkingtk.sources.vector_index) to search/get against instead of the live API. search delegates to it entirely; get tries it first and only falls back to a live wbgetentities call on a miss (the index may be a partial or sampled build).	None

Raises:

Type	Description
OptionalDependencyError	If requests isn't installed.

Source code in src/linkingtk/sources/wikidata.py

def __init__(
    self,
    lang: str = "en",
    session: requests.Session | None = None,
    vector_index: VectorIndexEntitySource | None = None,
) -> None:
    try:
        import requests
    except ImportError as exc:
        raise OptionalDependencyError("WikidataEntitySource", "wikipedia") from exc
    self.lang = lang
    self.vector_index = vector_index
    self._api_url = "https://www.wikidata.org/w/api.php"
    if session is not None:
        self._session = session
    else:
        self._session = requests.Session()
        self._session.headers["User-Agent"] = _USER_AGENT

get

get(entity_id: str) -> Entity | None

Look up a QID, or None if it doesn't resolve.

Tries vector_index first (if given); falls back to a live action=wbgetentities call on a miss.

Source code in src/linkingtk/sources/wikidata.py

def get(self, entity_id: str) -> Entity | None:
    """Look up a QID, or ``None`` if it doesn't resolve.

    Tries `vector_index` first (if given); falls back to a live
    ``action=wbgetentities`` call on a miss.
    """
    if self.vector_index is not None:
        cached = self.vector_index.get(entity_id)
        if cached is not None:
            return cached

    params: dict[str, str | int] = {
        "action": "wbgetentities",
        "ids": entity_id,
        "languages": self.lang,
        "format": "json",
    }
    response = self._session.get(self._api_url, params=params, timeout=10)
    response.raise_for_status()
    payload = response.json()
    # Wikidata reports a not-found entity two different ways: a
    # syntactically plausible but nonexistent QID comes back inside
    # "entities" with a "missing" marker, while a QID outside any
    # plausible id range comes back as a top-level API error instead --
    # confirmed directly against the live API for "Q999999999" (missing
    # marker) vs. "Q999999999999" (top-level error).
    if "error" in payload:
        return None
    entity = next(iter(payload["entities"].values()))
    if "missing" in entity:
        return None
    return _entity_from_wikidata_json(entity, self.lang)

search

search(query: str, top_k: int = 10) -> list[Entity]

Nearest entities to query, best match first.

Delegates to vector_index (no network) if one was given at construction; otherwise uses action=wbsearchentities.

Source code in src/linkingtk/sources/wikidata.py

def search(self, query: str, top_k: int = 10) -> list[Entity]:
    """Nearest entities to `query`, best match first.

    Delegates to `vector_index` (no network) if one was given at
    construction; otherwise uses ``action=wbsearchentities``.
    """
    if self.vector_index is not None:
        return self.vector_index.search(query, top_k)

    params: dict[str, str | int] = {
        "action": "wbsearchentities",
        "search": query,
        "language": self.lang,
        "limit": top_k,
        "format": "json",
    }
    response = self._session.get(self._api_url, params=params, timeout=10)
    response.raise_for_status()
    hits = response.json()["search"]
    return [
        Entity(id=hit["id"], labels=[hit["label"]], description=hit.get("description"))
        for hit in hits
        if "label" in hit
    ]

vector_index

A local, FAISS-backed EntitySource -- generalized from a project-specific index.

Querying a huge external source (e.g. Wikidata) live, per mention, doesn't scale: wn-wd-entity-align hit exactly this and built a one-off FAISS label-similarity index over a bulk Wikidata label dump to search entirely offline instead. This module generalizes that approach into a reusable EntitySource: build a VectorIndexEntitySource once, from any Iterable[Entity] (a Wikidata dump, a large WnEntitySource export, anything), save() it, and search()/get() it later with no network calls at all -- e.g. as WikidataEntitySource's vector_index argument.

Embedder

Bases: Protocol

Structural type for a batch text encoder, e.g. sentence_transformers.SentenceTransformer's own encode(list[str], batch_size=...) -> np.ndarray (with its default convert_to_numpy=True) matches this directly -- no wrapper class needed, the same way Vectorizer matches scikit-learn's fit_transform/transform interface.

batch_size must be accepted (even if a concrete Embedder ignores it): VectorIndexEntitySource.build always passes it explicitly, since SentenceTransformer.encode otherwise silently re-chunks whatever's handed to it into its own default sub-batches of 32 for the actual forward passes -- regardless of how many texts the call itself received. That single silent default was confirmed (profiling a real GPU) to cap throughput at ~7k texts/s no matter the caller's batch size, vs. ~39k texts/s once a sane batch size (512+) is actually threaded through -- a >5x slowdown, and the real reason a full Wikidata-dump build was taking over two days (issue #68), not the low CPU utilization that made it look like a parallelism problem.

VectorIndexEntitySource

VectorIndexEntitySource(
    index: Index,
    embedder: Embedder,
    ids_path: Path,
    entities_db_path: Path,
    vh: NDArray[floating[Any]] | None,
)

Bases: EntitySource

A local FAISS approximate-nearest-neighbor index over embedded entity text, queried as an EntitySource.

Build once via build (optionally save/ load to reuse across processes); search/get afterwards touch no network and no external service.

Raises:

Type	Description
OptionalDependencyError	If faiss isn't installed.

Source code in src/linkingtk/sources/vector_index.py

def __init__(
    self,
    index: faiss.Index,
    embedder: Embedder,
    ids_path: Path,
    entities_db_path: Path,
    vh: npt.NDArray[np.floating[Any]] | None,
) -> None:
    self._index = index
    self._embedder = embedder
    self._ids_path = ids_path
    self._offsets = _build_or_load_offsets(ids_path)
    self._entities_db_path = entities_db_path
    self._entities_db = dbm.open(str(entities_db_path), "r")
    self._vh = vh

build classmethod

build(
    entities: Iterable[Entity],
    embedder: Embedder,
    path: Path,
    field: Field = "label",
    reduced_dim: int | None = 28,
    sample_size: int = 100000,
    sample_scan_limit: int | None = None,
    batch_size: int = 4096,
    encode_batch_size: int = 32,
) -> VectorIndexEntitySource

Build a fresh index over entities, persisted under path.

Streams entities rather than materializing it, so it scales to a source too large to fit in memory at once -- e.g. WikidataDumpEntities, reading directly from a downloaded Wikidata dump.

Parameters:

Name	Type	Description	Default
entities	Iterable[Entity]	The entities to index, embedded on the text(s) field resolves (default: one index row per label). Must be re-iterable (e.g. a list, or an object like WikidataDumpEntities whose __iter__ starts fresh each call) when reduced_dim is set, since fitting the SVD projection and indexing are then two separate passes over entities -- a single-use iterator/generator only works with reduced_dim=None (one pass).	required
embedder	Embedder	A batch text encoder (see Embedder), e.g. a real sentence_transformers.SentenceTransformer(...).	required
path	Path	Directory to write the index bundle to (created if missing) -- see save for its contents.	required
field	Field	Which entity text to embed: "label" (each distinct label embedded as its own index row, all mapping back to the same entity -- so an exact match on any one label isn't diluted by the others), "description", or "context" (one row per entity). A callable taking an Entity and returning str may be passed instead, for fields not covered above (one row per entity) -- e.g. lambda e: " ".join(label_texts(e)) for the old space-joined-labels behavior. An entity with no text for field gets no rows (so is never a search hit), but is still stored for get.	'label'
reduced_dim	int | None	If set (the default, 28, matching wn-wd-entity-align's own build), embeddings are projected down to this many dimensions via a truncated SVD fit on a sample of entities' texts, shrinking the index -- pass None to index full-dimensional embeddings instead.	28
sample_size	int	Max texts sampled (reservoir sampling, so not biased toward entities' start) to fit the SVD projection. Ignored if reduced_dim is None.	100000
sample_scan_limit	int | None	If set, the SVD-fitting pass stops after reading this many entities, reservoir-sampling sample_size texts from just that prefix rather than from all of entities -- e.g. to avoid streaming a whole Wikidata dump twice when the first few million entities already give a representative sample. None (the default) scans everything. Ignored if reduced_dim is None.	None
batch_size	int	Texts buffered per embedder.encode call (and per FAISS add).	4096
encode_batch_size	int	The batch_size passed to embedder.encode itself -- for SentenceTransformer, what sizes each GPU forward pass, and so the GPU memory a build holds. The default of 32 keeps that to roughly the model's own weights (LaBSE: ~1.9 GB reserved at 32 vs. ~6.4 GB at 4096) while still embedding thousands of labels a second -- far faster than streaming a Wikidata dump over the network supplies them, which is the real bottleneck there. Raise it (e.g. 512) only when embedding is the bottleneck, such as building from a local file (issue #68).	32

Raises:

Type	Description
OptionalDependencyError	If faiss isn't installed.
TypeError	If reduced_dim is set and entities is a single-use iterator rather than a re-iterable object.

Source code in src/linkingtk/sources/vector_index.py

@classmethod
def build(
    cls,
    entities: Iterable[Entity],
    embedder: Embedder,
    path: Path,
    field: Field = "label",
    reduced_dim: int | None = 28,
    sample_size: int = 100_000,
    sample_scan_limit: int | None = None,
    batch_size: int = 4096,
    encode_batch_size: int = 32,
) -> VectorIndexEntitySource:
    """Build a fresh index over `entities`, persisted under `path`.

    Streams `entities` rather than materializing it, so it scales to a
    source too large to fit in memory at once -- e.g.
    [WikidataDumpEntities][linkingtk.sources.wikidata.WikidataDumpEntities],
    reading directly from a downloaded Wikidata dump.

    Args:
        entities: The entities to index, embedded on the text(s) `field`
            resolves (default: one index row per label).
            Must be re-iterable (e.g. a `list`, or an object like
            `WikidataDumpEntities` whose `__iter__` starts fresh each
            call) when `reduced_dim` is set, since fitting the SVD
            projection and indexing are then two separate passes over
            `entities` -- a single-use iterator/generator only works
            with `reduced_dim=None` (one pass).
        embedder: A batch text encoder (see `Embedder`), e.g. a real
            `sentence_transformers.SentenceTransformer(...)`.
        path: Directory to write the index bundle to (created if
            missing) -- see `save` for its contents.
        field: Which entity text to embed: ``"label"`` (each distinct
            label embedded as its own index row, all mapping back to
            the same entity -- so an exact match on any one label isn't
            diluted by the others), ``"description"``, or
            ``"context"`` (one row per entity). A callable taking an
            ``Entity`` and returning ``str`` may be passed instead, for
            fields not covered above (one row per entity) -- e.g.
            ``lambda e: " ".join(label_texts(e))`` for the old
            space-joined-labels behavior. An entity with no text for
            `field` gets no rows (so is never a search hit), but is
            still stored for `get`.
        reduced_dim: If set (the default, 28, matching
            `wn-wd-entity-align`'s own build), embeddings are projected
            down to this many dimensions via a truncated SVD fit on a
            sample of `entities`' texts, shrinking the index -- pass
            `None` to index full-dimensional embeddings instead.
        sample_size: Max texts sampled (reservoir sampling, so not
            biased toward `entities`' start) to fit the SVD projection.
            Ignored if `reduced_dim` is `None`.
        sample_scan_limit: If set, the SVD-fitting pass stops after
            reading this many entities, reservoir-sampling
            `sample_size` texts from just that prefix rather than from
            all of `entities` -- e.g. to avoid streaming a whole
            Wikidata dump twice when the first few million entities
            already give a representative sample. `None` (the
            default) scans everything. Ignored if `reduced_dim` is
            `None`.
        batch_size: Texts buffered per `embedder.encode` call (and per
            FAISS add).
        encode_batch_size: The `batch_size` passed to `embedder.encode`
            itself -- for `SentenceTransformer`, what sizes each GPU
            forward pass, and so the GPU memory a build holds. The
            default of 32 keeps that to roughly the model's own weights
            (LaBSE: ~1.9 GB reserved at 32 vs. ~6.4 GB at 4096) while
            still embedding thousands of labels a second -- far faster
            than streaming a Wikidata dump over the network supplies
            them, which is the real bottleneck there. Raise it (e.g.
            512) only when embedding is the bottleneck, such as
            building from a local file (issue #68).

    Raises:
        OptionalDependencyError: If `faiss` isn't installed.
        TypeError: If `reduced_dim` is set and `entities` is a
            single-use iterator rather than a re-iterable object.
    """
    try:
        import faiss
    except ImportError as exc:
        raise OptionalDependencyError("VectorIndexEntitySource", "vector-index") from exc

    extract = _row_texts(field)

    vh: npt.NDArray[np.floating[Any]] | None = None
    if reduced_dim is not None:
        if iter(entities) is entities:
            raise TypeError(
                "entities must be re-iterable when reduced_dim is set -- fitting "
                "the SVD projection and indexing are two separate passes over "
                "entities. Pass a list, or an object like WikidataDumpEntities "
                "whose __iter__ starts fresh each call, not a single-use "
                "iterator/generator -- or pass reduced_dim=None for a one-shot "
                "iterator."
            )
        sample = _reservoir_sample_texts(
            entities, sample_size, extract, scan_limit=sample_scan_limit
        )
        if sample:
            sample_vectors = np.asarray(embedder.encode(sample, batch_size=encode_batch_size))
            _, _, vh_full = np.linalg.svd(sample_vectors, full_matrices=False)
            # A too-small/low-rank sample (fewer distinct texts than
            # reduced_dim) can't fit a projection of the requested size --
            # silently slicing past vh_full's actual row count would instead
            # produce a smaller-than-recorded matrix, corrupting save/load.
            effective_dim = min(reduced_dim, vh_full.shape[0])
            vh = vh_full[:effective_dim, :].T.astype(np.float32)
            reduced_dim = effective_dim

    index: faiss.Index | None = None

    path.mkdir(parents=True, exist_ok=True)
    ids_path = path / "ids.txt"
    entities_db_path = path / "entities.db"

    with (
        ids_path.open("w", encoding="utf-8") as ids_file,
        dbm.open(str(entities_db_path), "n") as entities_db,
    ):
        # One entry per index row: an entity contributes one row per text.
        batch_ids: list[str] = []
        batch_texts: list[str] = []

        def flush_batch() -> None:
            nonlocal index
            if not batch_texts:
                return
            vectors = _project_and_normalize(
                np.asarray(embedder.encode(batch_texts, batch_size=encode_batch_size)), vh
            )
            if index is None:
                index = faiss.IndexFlatIP(vectors.shape[1])
            index.add(vectors)
            for entity_id in batch_ids:
                ids_file.write(f"{entity_id}\n")

        for entity in entities:
            entities_db[entity.id] = _entity_to_json(entity)
            for text in extract(entity):
                batch_ids.append(entity.id)
                batch_texts.append(text)
            if len(batch_texts) >= batch_size:
                flush_batch()
                batch_ids.clear()
                batch_texts.clear()
        flush_batch()

    if index is None:
        # No entities given -- still need an index of the right dimensionality.
        probe = _project_and_normalize(np.asarray(embedder.encode([""])), vh)
        index = faiss.IndexFlatIP(probe.shape[1])

    meta = {"reduced_dim": reduced_dim}
    (path / "meta.json").write_text(json.dumps(meta))
    faiss.write_index(index, str(path / "index.faiss"))
    if vh is not None:
        vh.tofile(path / "vh.bin")

    return cls(
        index=index,
        embedder=embedder,
        ids_path=ids_path,
        entities_db_path=entities_db_path,
        vh=vh,
    )

get

get(entity_id: str) -> Entity | None

Look up an indexed entity by id, or None if it isn't in the index.

Source code in src/linkingtk/sources/vector_index.py

def get(self, entity_id: str) -> Entity | None:
    """Look up an indexed entity by id, or `None` if it isn't in the index."""
    raw = self._entities_db.get(entity_id)
    if raw is None:
        return None
    return _entity_from_json(raw.decode("utf-8") if isinstance(raw, bytes) else raw)

load classmethod

load(
    path: Path, embedder: Embedder
) -> VectorIndexEntitySource

Reload an index bundle written by build/save from path.

embedder must be re-supplied (it isn't serialized) and should be the same one -- or one producing embeddings in the same space -- used to build the index.

Source code in src/linkingtk/sources/vector_index.py

@classmethod
def load(cls, path: Path, embedder: Embedder) -> VectorIndexEntitySource:
    """Reload an index bundle written by `build`/`save` from `path`.

    `embedder` must be re-supplied (it isn't serialized) and should be
    the same one -- or one producing embeddings in the same space --
    used to build the index.
    """
    try:
        import faiss
    except ImportError as exc:
        raise OptionalDependencyError("VectorIndexEntitySource", "vector-index") from exc

    index = faiss.read_index(str(path / "index.faiss"))
    meta = json.loads((path / "meta.json").read_text())
    vh: npt.NDArray[np.floating[Any]] | None = None
    vh_path = path / "vh.bin"
    if meta.get("reduced_dim") is not None and vh_path.exists():
        vh = np.fromfile(vh_path, dtype=np.float32).reshape(-1, meta["reduced_dim"])

    return cls(
        index=index,
        embedder=embedder,
        ids_path=path / "ids.txt",
        entities_db_path=path / "entities.db",
        vh=vh,
    )

save

save(path: Path) -> None

Write this index's bundle to path (a directory, created if missing).

Writes index.faiss (the FAISS index itself), vh.bin (the SVD projection matrix, only if one was used), ids.txt (+ a cached ids.txt.offsets.npy, the entity id of each index row, one per line in row order -- an id repeats once per label it was indexed under), entities.db (a dbm store, entity id -> JSON), and meta.json.

Source code in src/linkingtk/sources/vector_index.py

def save(self, path: Path) -> None:
    """Write this index's bundle to `path` (a directory, created if missing).

    Writes `index.faiss` (the FAISS index itself), `vh.bin` (the SVD
    projection matrix, only if one was used), `ids.txt` (+ a cached
    `ids.txt.offsets.npy`, the entity id of each index row, one per line
    in row order -- an id repeats once per label it was indexed under),
    `entities.db` (a `dbm` store, entity id -> JSON), and `meta.json`.
    """
    try:
        import faiss
    except ImportError as exc:
        raise OptionalDependencyError("VectorIndexEntitySource", "vector-index") from exc

    path.mkdir(parents=True, exist_ok=True)
    faiss.write_index(self._index, str(path / "index.faiss"))
    if self._vh is not None:
        self._vh.tofile(path / "vh.bin")
    reduced_dim = self._vh.shape[1] if self._vh is not None else None
    (path / "meta.json").write_text(json.dumps({"reduced_dim": reduced_dim}))

    if (path / "ids.txt").resolve() != self._ids_path.resolve():
        ids_text = self._ids_path.read_text(encoding="utf-8")
        (path / "ids.txt").write_text(ids_text, encoding="utf-8")
    if (path / "entities.db").resolve() != self._entities_db_path.resolve():
        with dbm.open(str(path / "entities.db"), "n") as dst:
            for key in self._entities_db.keys():  # noqa: SIM118
                dst[key] = self._entities_db[key]

search

search(query: str, top_k: int = 10) -> list[Entity]

Nearest entities to query by embedding cosine similarity, best match first.

Source code in src/linkingtk/sources/vector_index.py

def search(self, query: str, top_k: int = 10) -> list[Entity]:
    """Nearest entities to `query` by embedding cosine similarity, best match first."""
    return self.search_batch([query], top_k)[0]

search_batch

search_batch(
    queries: list[str], top_k: int = 10
) -> list[list[Entity]]

Real batched search: one FAISS query for all of queries at once.

Several index rows can map to the same entity (one per label), so rows are deduplicated by entity -- each ranked by its best-matching row -- and the search is retried with a doubled row count until every query has top_k distinct entities or the index is exhausted.

Source code in src/linkingtk/sources/vector_index.py

def search_batch(self, queries: list[str], top_k: int = 10) -> list[list[Entity]]:
    """Real batched search: one FAISS query for all of `queries` at once.

    Several index rows can map to the same entity (one per label), so
    rows are deduplicated by entity -- each ranked by its best-matching
    row -- and the search is retried with a doubled row count until
    every query has `top_k` distinct entities or the index is exhausted.
    """
    if not queries:
        return []
    vectors = _project_and_normalize(np.asarray(self._embedder.encode(queries)), self._vh)
    total_rows = int(self._index.ntotal)
    k = top_k
    while True:
        _distances, indices = self._index.search(vectors, min(k, total_rows) or top_k)
        ranked_ids = [self._distinct_ids(row_indices, top_k) for row_indices in indices]
        if k >= total_rows or all(len(ids) >= top_k for ids in ranked_ids):
            break
        k *= 2
    results = []
    for ids in ranked_ids:
        hits = [entity for entity_id in ids if (entity := self.get(entity_id)) is not None]
        results.append(hits)
    return results