Case-sensitive exact-match retrieval for notation-bearing ontology terms.
Vocabularies keyed by a literal token in source text — unit symbols, chemical
formulae, gene symbols, CAS numbers — are a poor fit for dense semantic search.
This module indexes those tokens at atomization time and matches them directly
against raw chunk text at query time, outside the semantic atom budget.
Bases: Tool
In-memory case-sensitive token → atom_id index.
Source code in ontocast/tool/vector_store/lexical_trigger.py
| class LexicalTriggerIndex(Tool):
"""In-memory case-sensitive token → atom_id index."""
max_match_atoms: int = Field(
default=16,
ge=0,
description="Maximum atom IDs returned per match call.",
)
_token_to_atom_ids: dict[str, list[str]] = PrivateAttr(default_factory=dict)
_atom_to_tokens: dict[str, list[str]] = PrivateAttr(default_factory=dict)
_ontology_to_atoms: dict[str, set[str]] = PrivateAttr(default_factory=dict)
_substring_triggers: dict[str, list[str]] = PrivateAttr(default_factory=dict)
def clear(self) -> None:
self._token_to_atom_ids = {}
self._atom_to_tokens = {}
self._ontology_to_atoms = {}
self._substring_triggers = {}
def register_atoms(self, atoms: Iterable[GraphAtom]) -> None:
for atom in atoms:
self.register_atom(atom)
def register_atom(self, atom: GraphAtom) -> None:
triggers = dedupe_preserve_case(atom.lexical_triggers)
if not triggers:
return
atom_id = atom.atom_id
self._atom_to_tokens[atom_id] = triggers
if atom.ontology_iri:
self._ontology_to_atoms.setdefault(atom.ontology_iri, set()).add(atom_id)
for trigger in triggers:
self._token_to_atom_ids.setdefault(trigger, []).append(atom_id)
if _needs_substring_scan(trigger):
self._substring_triggers.setdefault(trigger, []).append(atom_id)
def unregister_ontology(self, ontology_iri: str) -> None:
atom_ids = self._ontology_to_atoms.pop(ontology_iri, set())
for atom_id in atom_ids:
self._unregister_atom(atom_id)
def _unregister_atom(self, atom_id: str) -> None:
triggers = self._atom_to_tokens.pop(atom_id, [])
for trigger in triggers:
bucket = self._token_to_atom_ids.get(trigger)
if bucket is not None:
self._token_to_atom_ids[trigger] = [
aid for aid in bucket if aid != atom_id
]
if not self._token_to_atom_ids[trigger]:
del self._token_to_atom_ids[trigger]
sub_bucket = self._substring_triggers.get(trigger)
if sub_bucket is not None:
self._substring_triggers[trigger] = [
aid for aid in sub_bucket if aid != atom_id
]
if not self._substring_triggers[trigger]:
del self._substring_triggers[trigger]
def match(self, text: str, *, max_atoms: int | None = None) -> list[str]:
"""Return atom IDs whose lexical triggers appear in ``text``."""
limit = self.max_match_atoms if max_atoms is None else max_atoms
if limit <= 0 or not text:
return []
hits: list[str] = []
seen: set[str] = set()
def add(atom_id: str) -> None:
if atom_id in seen:
return
seen.add(atom_id)
hits.append(atom_id)
for token in tokenize_for_lexical_match(text):
for atom_id in self._token_to_atom_ids.get(token, ()):
add(atom_id)
if len(hits) >= limit:
return hits[:limit]
for trigger, atom_ids in sorted(
self._substring_triggers.items(), key=lambda item: -len(item[0])
):
if _substring_match_with_boundaries(trigger, text):
for atom_id in atom_ids:
add(atom_id)
if len(hits) >= limit:
return hits[:limit]
return hits[:limit]
|
Return atom IDs whose lexical triggers appear in text.
Source code in ontocast/tool/vector_store/lexical_trigger.py
| def match(self, text: str, *, max_atoms: int | None = None) -> list[str]:
"""Return atom IDs whose lexical triggers appear in ``text``."""
limit = self.max_match_atoms if max_atoms is None else max_atoms
if limit <= 0 or not text:
return []
hits: list[str] = []
seen: set[str] = set()
def add(atom_id: str) -> None:
if atom_id in seen:
return
seen.add(atom_id)
hits.append(atom_id)
for token in tokenize_for_lexical_match(text):
for atom_id in self._token_to_atom_ids.get(token, ()):
add(atom_id)
if len(hits) >= limit:
return hits[:limit]
for trigger, atom_ids in sorted(
self._substring_triggers.items(), key=lambda item: -len(item[0])
):
if _substring_match_with_boundaries(trigger, text):
for atom_id in atom_ids:
add(atom_id)
if len(hits) >= limit:
return hits[:limit]
return hits[:limit]
|
Deduplicate trigger strings without normalizing case.
Source code in ontocast/tool/vector_store/lexical_trigger.py
| def dedupe_preserve_case(values: Iterable[str]) -> list[str]:
"""Deduplicate trigger strings without normalizing case."""
out: list[str] = []
seen: set[str] = set()
for raw in values:
value = raw.strip()
if not value or value in seen:
continue
seen.add(value)
out.append(value)
return out
|
True when a bare label/altLabel plausibly denotes a formal code.
Source code in ontocast/tool/vector_store/lexical_trigger.py
| def looks_like_lexical_code(
value: str,
*,
min_len: int,
max_len: int,
) -> bool:
"""True when a bare label/altLabel plausibly denotes a formal code."""
stripped = value.strip()
if not stripped or any(ch.isspace() for ch in stripped):
return False
if len(stripped) < min_len or len(stripped) > max_len:
return False
if stripped.lower() in _HEURISTIC_STOPWORDS:
return False
has_digit = any(ch.isdigit() for ch in stripped)
has_mixed_case = any(ch.isupper() for ch in stripped) and any(
ch.islower() for ch in stripped
)
has_symbol_punct = any(not ch.isalnum() for ch in stripped)
if stripped.isalpha() and stripped.islower() and len(stripped) > 4:
return False
return (
has_digit
or has_mixed_case
or has_symbol_punct
or (stripped.isalpha() and len(stripped) <= 4 and stripped[0].isupper())
)
|
Extract case-preserved candidate tokens from raw source text.
Source code in ontocast/tool/vector_store/lexical_trigger.py
| def tokenize_for_lexical_match(text: str) -> list[str]:
"""Extract case-preserved candidate tokens from raw source text."""
if not text:
return []
return _TOKEN_PATTERN.findall(text)
|