Coverage for node / src / stigmem_node / recall / entity_resolver.py: 94%
97 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""Fuzzy entity resolver — spec §2.6.6 (v0.8, Track F).
3Three-layer resolution pipeline:
5 Layer 1 — Canonical normalisation (entity_normalizer.py).
6 Deterministic, idempotent. Collapses case/whitespace/hyphen variants.
8 Layer 2 — Alias table lookup (entity_aliases, Migration 003).
9 Explicit pre-registered mappings for known legacy URIs.
11 Layer 3 — Token-fuzzy scoring over the live fact graph.
12 For informal URIs (type:id) with the same type prefix, scores
13 known entity candidates by token overlap and SequenceMatcher
14 similarity on the id segment. Returns ranked candidates.
15 Threshold: FUZZY_SCORE_THRESHOLD (default 0.5).
17The full Kompl-style resolver (phonetic matching, NLP-based entity linking)
18is deferred to Phase 7. Layer 3 here covers the common cases:
19 - Abbreviated names: user:alice ≡ user:a.smith (partial token match)
20 - Whitespace/hyphen: user:alice-smith ≡ user:alice_smith (Layer 1)
21 - Legacy aliases: user:alice → user:a.smith (Layer 2 explicit mapping)
22"""
24from __future__ import annotations
26import re
27import sqlite3
28from dataclasses import dataclass, field
29from difflib import SequenceMatcher
30from typing import Any
32from ..utility.entity_normalizer import NormalizationError, normalize_entity_uri
34_TOKEN_SPLIT_RE = re.compile(r"[.\-_/\s]+")
35_FORMAL_PREFIX = "stigmem://"
36FUZZY_SCORE_THRESHOLD = 0.5
39@dataclass
40class ResolveCandidate:
41 uri: str
42 score: float
43 layer: int # 1=canonical, 2=alias, 3=fuzzy
44 match_note: str = ""
47@dataclass
48class ResolveResult:
49 query: str
50 canonical: str | None = None # None if normalisation failed
51 layer1_match: bool = False
52 layer2_match: str | None = None # alias target if found
53 layer3_candidates: list[ResolveCandidate] = field(default_factory=list)
55 @property
56 def best(self) -> str | None:
57 """Return the highest-confidence resolved URI, or None."""
58 if self.layer2_match:
59 return self.layer2_match
60 if self.canonical and self.layer1_match:
61 return self.canonical
62 if self.layer3_candidates: 62 ↛ 63line 62 didn't jump to line 63 because the condition on line 62 was never true
63 return self.layer3_candidates[0].uri
64 return self.canonical
67def _tokenise(id_segment: str) -> list[str]:
68 """Split an id segment into lowercase tokens."""
69 return [t for t in _TOKEN_SPLIT_RE.split(id_segment.lower()) if t]
72def _type_prefix(uri: str) -> str | None:
73 """Extract 'type' from 'type:id' informal URIs. Returns None for formal or bare ids."""
74 if uri.startswith(_FORMAL_PREFIX) or ":" not in uri:
75 return None
76 return uri.split(":", 1)[0]
79def _id_segment(uri: str) -> str:
80 """Extract the id part (after the type: prefix) for informal URIs."""
81 if ":" in uri and not uri.startswith(_FORMAL_PREFIX):
82 return uri.split(":", 1)[1]
83 return uri
86def _token_score(a_tokens: list[str], b_tokens: list[str]) -> float:
87 """
88 Similarity score [0, 1] between two token lists.
90 Combines:
91 - Jaccard overlap on full tokens
92 - Initial/prefix match: token in a is a prefix of a token in b (covers a.smith ↔ alice)
93 - Best SequenceMatcher ratio across all pair combinations
94 """
95 if not a_tokens or not b_tokens:
96 return 0.0
98 a_set = set(a_tokens)
99 b_set = set(b_tokens)
101 # Jaccard on exact tokens
102 jaccard = len(a_set & b_set) / len(a_set | b_set)
104 # Prefix/initial match: score +0.3 if any token in a is a prefix of any token in b
105 prefix_bonus = 0.0
106 for ta in a_tokens:
107 for tb in b_tokens:
108 if len(ta) >= 1 and tb.startswith(ta[0]):
109 prefix_bonus = max(prefix_bonus, 0.3)
110 if len(ta) > 1 and tb.startswith(ta[:2]):
111 prefix_bonus = max(prefix_bonus, 0.5)
113 # SequenceMatcher on concatenated id strings
114 seq_score = SequenceMatcher(None, "".join(a_tokens), "".join(b_tokens)).ratio()
116 return min(1.0, max(jaccard, prefix_bonus, seq_score))
119def resolve_entity(
120 raw: str,
121 conn: sqlite3.Connection,
122 *,
123 tenant_id: str = "default",
124 top_k: int = 5,
125 threshold: float = FUZZY_SCORE_THRESHOLD,
126) -> ResolveResult:
127 """Resolve raw entity URI through 3 layers.
129 Args:
130 raw: Raw entity URI string (may be non-canonical, aliased, or abbreviated).
131 conn: Open SQLite connection (read-only queries only).
132 top_k: Maximum Layer 3 candidates to return.
133 threshold: Minimum fuzzy score to include in Layer 3 candidates.
135 Returns:
136 ResolveResult with layer-by-layer findings and a `.best` shorthand.
137 """
138 result = ResolveResult(query=raw)
140 # -------------------------------------------------------------------
141 # Layer 1 — canonical normalisation
142 # -------------------------------------------------------------------
143 try:
144 canonical = normalize_entity_uri(raw)
145 except NormalizationError:
146 return result # malformed input; no resolution possible
148 result.canonical = canonical
150 # Check if the canonical form itself appears in the fact graph.
151 live_check = conn.execute(
152 "SELECT 1 FROM facts WHERE entity = ? AND tenant_id = ? AND confidence > 0.0 LIMIT 1",
153 (canonical, tenant_id),
154 ).fetchone()
155 if live_check:
156 result.layer1_match = True
157 return result # exact hit; no further layers needed
159 # -------------------------------------------------------------------
160 # Layer 2 — alias table lookup
161 # -------------------------------------------------------------------
162 alias_row = conn.execute(
163 "SELECT canonical_uri FROM entity_aliases WHERE raw_uri = ? AND tenant_id = ?",
164 (canonical, tenant_id),
165 ).fetchone()
166 if alias_row is None:
167 # Also try the raw input (pre-normalisation alias)
168 alias_row = conn.execute(
169 "SELECT canonical_uri FROM entity_aliases WHERE raw_uri = ? AND tenant_id = ?",
170 (raw, tenant_id),
171 ).fetchone()
173 if alias_row:
174 result.layer2_match = str(alias_row["canonical_uri"])
175 return result # explicit alias; authoritative
177 # -------------------------------------------------------------------
178 # Layer 3 — token-fuzzy scoring over same-type entities
179 # -------------------------------------------------------------------
180 type_prefix = _type_prefix(canonical)
181 if type_prefix is None: 181 ↛ 183line 181 didn't jump to line 183 because the condition on line 181 was never true
182 # Formal or bare URIs: skip Layer 3 (no type-scoped candidates).
183 return result
185 id_seg = _id_segment(canonical)
186 query_tokens = _tokenise(id_seg)
187 if not query_tokens: 187 ↛ 188line 187 didn't jump to line 188 because the condition on line 187 was never true
188 return result
190 # Fetch all distinct entity URIs with the same type prefix from the fact graph.
191 # Escape LIKE metacharacters in the (caller-derived) type prefix so it cannot
192 # inject wildcards into the scan (consistent with intents; audit F-2).
193 safe_prefix = type_prefix.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
194 prefix_pattern = f"{safe_prefix}:%"
195 candidate_rows: list[Any] = conn.execute(
196 """SELECT DISTINCT entity FROM facts
197 WHERE entity LIKE ? ESCAPE '\\' AND tenant_id = ? AND confidence > 0.0
198 LIMIT 2000""",
199 (prefix_pattern, tenant_id),
200 ).fetchall()
202 scored: list[ResolveCandidate] = []
203 for row in candidate_rows:
204 candidate_uri: str = row["entity"]
205 if candidate_uri == canonical: 205 ↛ 206line 205 didn't jump to line 206 because the condition on line 205 was never true
206 continue # already checked in Layer 1
208 cand_id_seg = _id_segment(candidate_uri)
209 cand_tokens = _tokenise(cand_id_seg)
210 score = _token_score(query_tokens, cand_tokens)
212 if score >= threshold:
213 scored.append(
214 ResolveCandidate(
215 uri=candidate_uri,
216 score=score,
217 layer=3,
218 match_note=f"token_score={score:.2f} ({id_seg!r} ~ {cand_id_seg!r})",
219 )
220 )
222 scored.sort(key=lambda c: c.score, reverse=True)
223 result.layer3_candidates = scored[:top_k]
224 return result