文档索引
在以下地址获取完整的文档索引：https://docs.typesafe.ai/llms.txt 使用此文件可以在进一步探索之前发现所有可用页面。

知识图谱实体对齐

决定来自两个啤酒目录的 450 个候选配对中哪些描述了相同的产品。一个 TypeSafe Score 问题承担了整个决策，因为其三个级别对应你可以对配对执行的三种操作：合并它、保持其未链接状态，或将其交给策展人。没有需要拟合的阈值，并且有三个 Noul 问题包含在同一个请求中，以告知策展人两个来源在哪个字段上存在分歧。

知识图谱中的一个关键问题是决定传入的实体是否重复现有实体，尤其是在只有来自不同来源的自然语言可用的情况下。给定潜在重复配对，单个 TypeSafe Score 决定每对是否是重复项，或者是否值得策展人进一步查看。

假设两个数据源描述了相同事物的重叠集合，你需要知道一方中的哪个条目与另一方中的哪个条目是同一事物。知识图谱将这些条目称为实体，并保存关于每个实体记录的事实。一些廉价但粗略的初步处理已经比较了两个来源，并挑出了 450 对值得仔细查看的配对。剩下的就是对这些配对做出判断。

不当合并两个实体是代价更高的错误，因为有关任一实体的所有事实现在都描述了合并后的实体，任何指向任一实体的链接也会随之而来。稍后撤销意味着要理清每个事实的来源。错过匹配只会产生重复项，因此判断需要第三个选项：既不安全合并也不安全丢弃的对。

判断是一个 Score 问题，每种结果对应一个级别：

不同产品 — 保持两个实体未链接
相关，但可能不是同一个 — 交由策展人决定
相同产品 — 合并它们

我们使用 Score 问题，因为我们想将语义标签（即评分标准）直接附加到每个结果上，包括中间结果。Noul 问题可以通过对其输出进行阈值处理间接实现此目的，而 Choice 问题会失去三个结果之间的有序关系。

接下来，对于我们要考虑的实体的每个字段，Noul 都会询问这些字段是否匹配的问题，这些问题可以包含在同一个请求中。如果分数既不属于 “相同产品”也不属于“不同产品”级别，这些 nouls 将为编辑提供更详细的信息。

最终你会得到一个 route() 函数，它接收一个候选配对并返回三种结果之一，无需你根据自己数据拟合任何阈值。

flowchart LR
    PAIR["one candidate pair<br/><i>both entities, one state</i>"] --> CALL

    subgraph CALL["one request, four questions"]
        direction TB
        S["<b>Score:</b> how do the two relate?<br/>· different product<br/>· related, but possibly not the same<br/>· same product"]
        N["<b>Nouls:</b> one per compared field<br/>· same name?<br/>· same brewery?<br/>· same style?"]
        %% invisible link: without an edge these two share a rank, which in a TB
        %% subgraph puts them side by side instead of stacked
        S ~~~ N
    end

    S --> R{"round to the<br/>nearest level"}
    R -->|"different"| DROP["leave unlinked"]
    R -->|"same"| M["assert sameAs"]
    %% the queue is last so the dotted edge below reaches it without crossing
    %% the arrow into `assert sameAs`
    R -->|"related"| Q["curator queue"]
    N -.->|"which field<br/>they disagree on"| Q

Setup

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

然后设置 TYPESAFE_API_KEY。每次调用都会缓存到 json_cache.json 中，该文件随 cookbook 提供，因此重新渲染时会重放已发布的数字，而无需调用 API。删除该文件即可重新进行实时运行。

以下数字来自 2026 年 8 月 11 日的 jev-1.12。

import json
import os
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path

import matplotlib
import matplotlib.pyplot as plt
from cooksafe import JsonCache, make_playground_link
from IPython.display import Markdown, display
from typesafe_sdk import Noul, Score, TypeSafeClient

matplotlib.use("Agg")  # headless render

TYPESAFE_MODEL = "jev-1.12"
MAX_WORKERS = 6  # small pool; the public endpoint rate-limits above roughly eight

client = TypeSafeClient(
    api_key=os.environ.get(
        "TYPESAFE_API_KEY", "cache-only"
    ),  # keyless kernels replay the cache
    base_url=os.environ.get("TYPESAFE_ENDPOINT"),
    timeout=120.0,
)
json_cache = JsonCache(Path("json_cache.json"))

加载候选配对

这些配对来自已发布的基准数据集，即 Magellan 集合中的 Beer 数据：从不同网站抓取的 Beer 目录，已经通过初步粗略处理缩减到 450 对。每个实体包含四个字段：名称、酿酒厂、风格和酒精含量。每对还包含 known_same_as，即基准自己的答案。

文本保持发布时的原样，未经预处理：HTML 实体从未转换回字符，撇号被拆分为单独的单词，少数字符解码错误。

每个配对发出一个请求，因此花费取决于你收到的配对数量，而非任一源的大小。

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])

450 candidate pairs. The first one, as the model will see it:
{
  "entity_a": {
    "name": "C N Red Imperial Red Ale",
    "brewery": "Redwood Lodge",
    "style": "American Amber / Red Ale",
    "abv": "8.10 %"
  },
  "entity_b": {
    "name": "Kinetic Infrared Imperial Red Ale",
    "brewery": "Kinetic Brewing Company",
    "style": "American Strong Ale",
    "abv": "9.30 %"
  }
}

为每个候选配对提出一个 Score 问题和三个 Noul 问题

两个实体都进入一个状态，分别为 entity_a 和 entity_b，因此问题是关于配对的，而不是关于各自一方的。所有四个问题都在一个请求中传输。

下方的三个级别描述构成了完整的决策：每个级别对应一个结果。此文件中没有设置任何阈值常量。你也可以在看到任何分数之前编写这些描述，这对于必须拟合的数字来说是不可能的。

中间级别是值得仔细撰写的一个。这里它涵盖变体、特别版以及可能合理指代任一产品的名称，因此这些会由策展人处理，而非合并或丢弃。

OUTCOME 命名了三种结果。合并结果称为 assert sameAs，因为 sameAs 是记录两个实体是同一事物的标准方式，而编写一个就是实际发生合并的方式。

四个字段中的三个获得 Noul 问题：名称、酿酒厂和风格。酒精度没有获得任何内容，因为比较两个数字是算术运算；如果你想得到它，就在代码中计算它。要在其他类型的数据上使用此方法，请重写 QUESTIONS 和 LEVELS。其他知道啤酒的代码只有两个打印结果的函数，它们命名了字段。

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?",
    ),
}


@json_cache
def score(pair_id: str) -> dict:
    """One request about one candidate pair -> the score plus the three noul answers."""
    pair = BY_ID[pair_id]
    response = client.system_one(
        state={"entity_a": pair["entity_a"], "entity_b": pair["entity_b"]},
        questions=QUESTIONS,
        model=TYPESAFE_MODEL,
    )
    link = response.answers["link_state"]
    return {
        "score": link.score,
        "probabilities": link.probabilities,
        "confidence": link.confidence,
        "properties": {
            k: response.answers[k].noul for k in QUESTIONS if k != "link_state"
        },
        # tokens and requests are the durable units; don't cache a derived cost
        "input_tokens": response.usage.input_tokens or 0,
        "output_tokens": response.usage.output_tokens or 0,
    }


def route(score_value: float) -> str:
    """The whole decision rule: the nearest level names the outcome."""
    return OUTCOME[min(int(score_value + 0.5), len(LEVELS) - 1)]


def show(pair_id: str) -> None:
    pair, result = BY_ID[pair_id], score(pair_id)
    print(
        f"{pair_id}  score {result['score']:.2f}  confidence {result['confidence']:.2f}"
        f"  ->  {route(result['score'])}"
    )
    for side in ("entity_a", "entity_b"):
        e = pair[side]
        print(f"    {e['name'][:44]:<46}{e['brewery'][:30]:<32}{e['style'][:22]}")
    nouls = result["properties"]
    print(
        f"    name {nouls['same_name']:.2f}   brewery {nouls['same_brewery']:.2f}   "
        f"style {nouls['same_style']:.2f}"
    )

四对。c446 是一个产品，c427 是两个产品。其他两对因不同原因落在中间级别：c100 具有相同的名称和酿酒厂，但来源对其风格的措辞不同，而 c428 将啤酒与其果味啤酒花变体配对。

for pair_id in ("c446", "c427", "c100", "c428"):
    show(pair_id)
    print()

c446  score 1.94  confidence 0.92  ->  assert sameAs
    Thomas Hooker Old Marley Barleywine           Thomas Hooker Brewing Company   American Barleywine
    Thomas Hooker Old Marley Barleywine           Thomas Hooker Brewing Company   Barley Wine
    name 0.97   brewery 0.99   style 0.81

c427  score 0.03  confidence 0.95  ->  leave unlinked
    Frost Quake Bourbon Barrel Aged Barley Wine   Wellington County Brewery       American Barleywine
    Lompoc Bourbon Barrel Aged Proletariat Red A  Lompoc Brewing                  Amber Ale
    name 0.02   brewery 0.09   style 0.08

c100  score 1.30  confidence 0.27  ->  curator queue
    Belle Gueule Rousse                           Brasseurs R.J.                  American Amber / Red A
    Belle Gueule Rousse                           Brasseurs RJ                    Amber Lager/Vienna
    name 0.95   brewery 0.94   style 0.35

c428  score 1.10  confidence 0.77  ->  curator queue
    Ambleside Amber Ale                           Bridge Brewing Company          American Amber / Red A
    Bridge Ambleside Amber Ale - Pomegranate & G  Bridge Brewing Company          Amber Ale
    name 0.63   brewery 0.98   style 0.74

路由每个候选配对

# 450 candidate pairs, one request each; a small pool keeps a live run to a few minutes.
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"])

SURFACE, INK, INK2, MUTED = "#fcfcfb", "#0b0b0b", "#52514e", "#898781"
GRID, AXIS, BLUE, ORANGE = "#e1e0d9", "#c3c2b7", "#2a78d6", "#eb6834"

BINS, TOP = 20, len(LEVELS) - 1
counts = [0] * BINS
for s in scores:
    counts[min(int(s / TOP * BINS), BINS - 1)] += 1
centers = [(i + 0.5) / BINS * TOP for i in range(BINS)]
queued = [c if route(x) == "curator queue" else 0 for c, x in zip(counts, centers)]
settled = [c if route(x) != "curator queue" else 0 for c, x in zip(counts, centers)]

fig, ax = plt.subplots(figsize=(7.2, 3.6), facecolor=SURFACE)
ax.set_facecolor(SURFACE)
for side in ("top", "right"):
    ax.spines[side].set_visible(False)
for side in ("left", "bottom"):
    ax.spines[side].set_color(AXIS)
ax.tick_params(colors=MUTED, labelcolor=INK2, labelsize=9)
ax.set_axisbelow(True)
ax.grid(axis="y", color=GRID, linewidth=0.8)
ax.bar(
    centers, settled, width=TOP / BINS * 0.9, color=BLUE, label="settled automatically"
)
ax.bar(
    centers, queued, width=TOP / BINS * 0.9, color=ORANGE, label="sent to the curator"
)
for edge in (0.5, 1.5):
    ax.axvline(edge, color=INK2, linewidth=1, linestyle="--")
ax.set_xticks([0, 0.5, 1, 1.5, 2])
ax.set_xticklabels(["0\ndifferent", "0.5", "1\nrelated", "1.5", "2\nsame"])
ax.set_xlabel("score for the pair", color=INK2, fontsize=9)
ax.set_ylabel("candidate pairs", color=INK2, fontsize=9)
ax.set_title(
    f"{len(PAIRS)} candidate pairs, scored once each",
    loc="left",
    color=INK,
    fontsize=11,
)
ax.legend(frameon=False, labelcolor=INK2, fontsize=9)
display(fig)
plt.close(fig)

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%})")

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

route() 更改答案的两个分数值是切割点。大多数配对已确定：360 个得分低于较低的切割点，40 个高于较高的切割点，剩下 50 个由策展人决定。

在这组数据中，评分并未整齐地落在整数上。大多数接近 0.25。两种毫无共同点的啤酒可能仍然共享风格名称，且它们的酿酒厂名称可能看起来很相似，因此模型将部分概率分配给中间级别，而非零。决定一对是否匹配的是它落在哪一侧的切分点。它距离级别的远近并不计入其中。

两个切割点的拥挤程度不同。九个配对位于 1.5 的较高切割点附近 0.1 的范围内，这是决定哪些内容合并到图谱中的切割点。四十七个配对位于 0.5 的较低切割点附近，这仅决定策展人是否看到该配对。这两个数字都不是你可以调整的参数。两者都取决于你对级别的措辞，而中间级别的措辞决定了哪些配对移交策展人，哪些配对保持未链接状态。

Open it in the playground

下方的 playground 链接打开了 c428，其得分为 1.10，并被送交策展人。它将 Ambleside Amber Ale 与 Bridge Ambleside Amber Ale - Pomegranate & Galena Hops 配对：同一酿酒厂，相同的酒精度。所有四个问题都随附于此。

playground_link = make_playground_link(
    {"entity_a": BY_ID["c428"]["entity_a"], "entity_b": BY_ID["c428"]["entity_b"]},
    QUESTIONS,
    models=[TYPESAFE_MODEL],
)
display(
    Markdown(
        f"🔗 [Open this pair + questions in the TypeSafe playground]({playground_link})"
    )
)

在 TypeSafe playground 中打开此配对及问题 →