fix: enforce canonical hard non-match constraints across ER clusters
This commit is contained in:
@@ -4,6 +4,10 @@ Python no longer scores identity. It sends opaque records to
|
||||
/api/0pt/internal/er/candidates, then only applies the returned canonical
|
||||
`match` decisions. If the canonical engine is unavailable, records are
|
||||
returned UNMERGED (fail closed against false identity merges).
|
||||
|
||||
A canonical hard non-match is also a cluster-wide constraint: transitive
|
||||
matches are never allowed to pull two records with conflicting stable identity
|
||||
evidence into the same entity.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -37,21 +41,36 @@ def _record_id(record: dict[str, Any]) -> str:
|
||||
return "erp_" + hashlib.sha256(raw).hexdigest()[:32]
|
||||
|
||||
|
||||
def _unique_record_ids(records: list[dict[str, Any]]) -> list[str]:
|
||||
seen: dict[str, int] = {}
|
||||
unique_ids: list[str] = []
|
||||
for record in records:
|
||||
rid = _record_id(record)
|
||||
n = seen.get(rid, 0)
|
||||
seen[rid] = n + 1
|
||||
unique_ids.append(rid if n == 0 else f"{rid}_{n}")
|
||||
return unique_ids
|
||||
|
||||
|
||||
def _apply_matches(records: list[dict[str, Any]], pairs: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
if not records:
|
||||
return []
|
||||
|
||||
ids = [_record_id(r) for r in records]
|
||||
# Exact duplicate payloads are indistinguishable; deterministic suffixes
|
||||
# make request ids unique without changing merge semantics.
|
||||
seen: dict[str, int] = {}
|
||||
unique_ids: list[str] = []
|
||||
for rid in ids:
|
||||
n = seen.get(rid, 0)
|
||||
seen[rid] = n + 1
|
||||
unique_ids.append(rid if n == 0 else f"{rid}_{n}")
|
||||
|
||||
unique_ids = _unique_record_ids(records)
|
||||
known_ids = set(unique_ids)
|
||||
parent = {rid: rid for rid in unique_ids}
|
||||
members = {rid: {rid} for rid in unique_ids}
|
||||
|
||||
# Only explicit hard-rule non-matches are cluster-wide negative evidence.
|
||||
# A normal score-based non_match merely means "insufficient similarity" and
|
||||
# must not become a universal cannot-link rule.
|
||||
hard_non_matches: set[frozenset[str]] = set()
|
||||
for pair in pairs:
|
||||
a, b = pair.get("record_a"), pair.get("record_b")
|
||||
if a not in known_ids or b not in known_ids:
|
||||
continue
|
||||
if pair.get("status") == "non_match" and pair.get("decided_by") not in (None, "score"):
|
||||
hard_non_matches.add(frozenset((a, b)))
|
||||
|
||||
def find(x: str) -> str:
|
||||
while parent[x] != x:
|
||||
@@ -59,42 +78,72 @@ def _apply_matches(records: list[dict[str, Any]], pairs: list[dict[str, Any]]) -
|
||||
x = parent[x]
|
||||
return x
|
||||
|
||||
def union(a: str, b: str) -> None:
|
||||
ra, rb = find(a), find(b)
|
||||
if ra != rb:
|
||||
parent[max(ra, rb)] = min(ra, rb)
|
||||
def components_conflict(left: set[str], right: set[str]) -> bool:
|
||||
return any(frozenset((a, b)) in hard_non_matches for a in left for b in right)
|
||||
|
||||
match_pairs = [p for p in pairs if p.get("status") == "match"]
|
||||
def union(a: str, b: str) -> bool:
|
||||
ra, rb = find(a), find(b)
|
||||
if ra == rb:
|
||||
return True
|
||||
left, right = members[ra], members[rb]
|
||||
if components_conflict(left, right):
|
||||
return False
|
||||
keep, drop = (ra, rb) if ra < rb else (rb, ra)
|
||||
parent[drop] = keep
|
||||
members[keep] = members[ra] | members[rb]
|
||||
members.pop(drop, None)
|
||||
return True
|
||||
|
||||
# Hard-rule matches are applied before probabilistic/score matches; within
|
||||
# each class, stronger scores go first. No later edge may violate a hard
|
||||
# negative constraint established above.
|
||||
match_pairs = [
|
||||
p for p in pairs
|
||||
if p.get("status") == "match"
|
||||
and p.get("record_a") in known_ids
|
||||
and p.get("record_b") in known_ids
|
||||
]
|
||||
match_pairs.sort(
|
||||
key=lambda p: (
|
||||
0 if p.get("decided_by") not in (None, "score") else 1,
|
||||
-float(p.get("score") or 0.0),
|
||||
str(p.get("candidate_id") or ""),
|
||||
)
|
||||
)
|
||||
accepted_edges: list[dict[str, Any]] = []
|
||||
for pair in match_pairs:
|
||||
a, b = pair.get("record_a"), pair.get("record_b")
|
||||
if a in parent and b in parent:
|
||||
union(a, b)
|
||||
a, b = str(pair["record_a"]), str(pair["record_b"])
|
||||
if union(a, b):
|
||||
accepted_edges.append(pair)
|
||||
|
||||
groups: dict[str, list[tuple[str, dict[str, Any]]]] = {}
|
||||
for rid, record in zip(unique_ids, records):
|
||||
groups.setdefault(find(rid), []).append((rid, copy.deepcopy(record)))
|
||||
|
||||
out: list[dict[str, Any]] = []
|
||||
for members in groups.values():
|
||||
members.sort(key=lambda item: (-float(item[1].get("confidence", 0.0)), item[0]))
|
||||
anchor_id, anchor = members[0]
|
||||
sources = sorted({str(r.get("source")) for _, r in members if r.get("source")})
|
||||
for group in groups.values():
|
||||
group.sort(key=lambda item: (-float(item[1].get("confidence", 0.0)), item[0]))
|
||||
_, anchor = group[0]
|
||||
sources = sorted({str(r.get("source")) for _, r in group if r.get("source")})
|
||||
anchor["agreed_sources"] = len(sources) or 1
|
||||
anchor["confidence"] = min(0.99, float(anchor.get("confidence", 0.0)) + 0.1 * max(0, len(sources) - 1))
|
||||
anchor["merged_from"] = [r.get("source") for _, r in members[1:] if r.get("source")]
|
||||
anchor["confidence"] = min(
|
||||
0.99,
|
||||
float(anchor.get("confidence", 0.0)) + 0.1 * max(0, len(sources) - 1),
|
||||
)
|
||||
anchor["merged_from"] = [r.get("source") for _, r in group[1:] if r.get("source")]
|
||||
|
||||
provenance = list(anchor.get("provenance", []))
|
||||
keys = {(p.get("field_name"), p.get("source_id")) for p in provenance}
|
||||
for _, record in members[1:]:
|
||||
for _, record in group[1:]:
|
||||
for p in record.get("provenance", []):
|
||||
key = (p.get("field_name"), p.get("source_id"))
|
||||
if key not in keys:
|
||||
provenance.append(copy.deepcopy(p))
|
||||
keys.add(key)
|
||||
|
||||
member_ids = {rid for rid, _ in members}
|
||||
member_ids = {rid for rid, _ in group}
|
||||
edges = [
|
||||
p for p in match_pairs
|
||||
p for p in accepted_edges
|
||||
if p.get("record_a") in member_ids and p.get("record_b") in member_ids
|
||||
]
|
||||
er_info = {
|
||||
@@ -109,7 +158,7 @@ def _apply_matches(records: list[dict[str, Any]], pairs: list[dict[str, Any]]) -
|
||||
for p in edges
|
||||
],
|
||||
}
|
||||
if len(members) > 1:
|
||||
if len(group) > 1:
|
||||
for p in provenance:
|
||||
p["entity_resolution"] = er_info
|
||||
anchor["provenance"] = provenance
|
||||
@@ -131,19 +180,15 @@ async def resolve(
|
||||
|
||||
url = _url()
|
||||
token = os.getenv("INTERNAL_TOKEN", "")
|
||||
if not url or not token:
|
||||
if not url or not token or not tenant_id or not project_id:
|
||||
return [copy.deepcopy(r) for r in records]
|
||||
|
||||
ids = [_record_id(r) for r in records]
|
||||
seen: dict[str, int] = {}
|
||||
unique_ids = _unique_record_ids(records)
|
||||
er_records = []
|
||||
for rid, record in zip(ids, records):
|
||||
n = seen.get(rid, 0)
|
||||
seen[rid] = n + 1
|
||||
unique = rid if n == 0 else f"{rid}_{n}"
|
||||
for rid, record in zip(unique_ids, records):
|
||||
name = str(record.get("name") or "")
|
||||
er_records.append({
|
||||
"record_id": unique,
|
||||
"record_id": rid,
|
||||
"fields": {
|
||||
"market": record.get("market"),
|
||||
"block_initial": name[:1].upper(),
|
||||
|
||||
Reference in New Issue
Block a user