Knowledge graph entity alignment - TypeSafe AI

Documentation Index

Entity Duplication in Knowledge Graphs

A key problem in knowledge graphs is deciding whether an incoming entity duplicates an
existing one, especially when natural language from disparate sources is all that’s
available. Given potential duplicate pairs, a single TypeSafe Score decides
whether each pair is a duplicate, or whether it deserves a closer look from a curator.

Suppose two data sources describe overlapping sets of the same things, and you need to
know which entry on one side is the same thing as which entry on the other. A knowledge
graph calls those entries entities, and holds the facts recorded about each. Some cheap
but rough first pass has already compared the two sources and picked out 450 pairs worth a
closer look. What remains is to make a judgment call on each pair. Merging two entities inappropriately is the more expensive mistake, since every fact about either entity now describes the merged one, and anything linked to either comes along too. Undoing it later means working out which fact came from where. Missing a match only leaves a duplicate, so the judgment call needs a third option: pairs that are neither safe to merge nor safe to drop. The judgment is a Score question with one level for each of the three outcomes:

We use a Score question because we want to attach a semantic label, the score criteria,
directly to each outcome, including the middle outcome. A Noul question could accomplish
this indirectly through thresholding on its output instead, and a Choice question would
lose the ordered relationship of the three outcomes.

Setup

pip install matplotlib ipython "typesafe-sdk>=0.5.7" cooksafe --extra-index-url https://pypi.typesafe.ai/

then set TYPESAFE_API_KEY. Every call is cached to json_cache.json, which ships with the cookbook, so re-rendering replays the published numbers without calling the API. Delete that file to re-run everything live.

Load the Candidate Pairs

The pairs come from a published benchmark set, the Beer data from the Magellan collection:
two beer catalogues scraped from different websites, already cut down to 450 pairs by that
first rough pass. Each entity carries four fields: name, brewery, style, and alcohol content. Each pair also carries known_same_as, the benchmark’s own answer.

PAIRS = json.loads(Path("candidate_pairs.json").read_text(encoding="utf-8"))
BY_ID = {pair["id"]: pair for pair in PAIRS}

print(f"{len(PAIRS)} candidate pairs. The first one, as the model will see it:")
print(json.dumps({k: PAIRS[0][k] for k in ("entity_a", "entity_b")}, indent=2)[:420])

Ask One Score Question and Three Noul Questions Per Candidate Pair

Both entities go into a single state, as entity_a and entity_b, so the questions are
about the pair and not about either side on its own. All four ride in one request.

The three level descriptions below are the entire decision: each level is one outcome.
There is no threshold constant anywhere in this file. You can also write these descriptions
before you have seen a single score, which is not true of a number you have to fit.
The middle level is the one worth writing carefully. Here it covers variants, special
editions, and names that could plausibly refer to either product, so those reach a curator
instead of being merged or dropped.

LEVELS = [
    "They describe two different products.",
    "They describe closely related products that may or may not be the same one: "
      "a variant, a special edition, or a name that could plausibly refer to either.",
    "They describe one and the same product.",
]
OUTCOME = {0: "leave unlinked", 1: "curator queue", 2: "assert sameAs"}

QUESTIONS = {
    "link_state": Score(
        instructions="How do the two entity descriptions relate as products?",
        criteria=LEVELS,
    ),
    "same_name": Noul(
        instructions="Do the two entities state the same beer name?",
    ),
    "same_brewery": Noul(
        instructions="Are the two entities from the same brewery?",
    ),
    "same_style": Noul(
        instructions="Do the two entities describe the same beer style?",
    ),
}

Route Every Candidate Pair

with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool:
    scored = list(pool.map(lambda pair: score(pair["id"]), PAIRS))

scores = [result["score"] for result in scored]
by_outcome: dict[str, list[str]] = {name: [] for name in OUTCOME.values()}
for pair, s in zip(PAIRS, scores):
    by_outcome[route(s)].append(pair["id"])

for name in ("assert sameAs", "curator queue", "leave unlinked"):
    n = len(by_outcome[name])
    print(f"{name:<16}{n:>5}  ({n / len(PAIRS):>5.1%})")

Output

assert sameAs      40  ( 8.9%)
curator queue      50  (11.1%)
leave unlinked    360  (80.0%)