第 18 章

第 18 章：实体对齐

本文整理自 Datawhale 开源项目 datawhalechina/jev-cookbook（CC BY-NC-SA 4.0），源文件：main/04_实战指南/09_实体对齐.ipynb。

针对官方文档 实战指南 · Knowledge graph entity alignment 的可运行实验笔记，用真实 TypeSafe API（Jev 模型）复刻核心流程并中文化。中文翻译版见 datawhalechina.github.io/jev-cookbook。

笔记本结构

章节	内容	实验
0. 准备	安装库、配置客户端、连通性测试、离线回退	—
📖 理论速览	System One / Score 有序层级 / 扇出 Noul（精简）	—
1. 实体对齐	Score 三档路由 + 三道字段 Noul	4 对中文啤酒实体

每个主题按固定节奏展开：原理 → 理论根基 → 定义数据 → 定义问题 → 调用 → 解读结果，每个单元格只做一件事，可直接顺着跑完（约 4 次 API 调用（每对一次））。

运行要求

Python ≥ 3.10（官方 SDK 要求；macOS 系统自带 python3 是 3.9，装不上 SDK）
一个 TypeSafe API Key（console.typesafe.ai/keys 获取）

推荐：一键创建本地环境（在本 notebooks 目录下）

./setup_env.sh                                  # 创建 .venv：Python 3.12 + 全部依赖
export TYPESAFE_API_KEY=你的key
.venv/bin/jupyter lab <本文件>.ipynb

或者手动创建：python3.12 -m venv .venv && .venv/bin/pip install -r requirements.txt

🔑 API Key 安全提示：本笔记从环境变量 TYPESAFE_API_KEY 读取密钥，不要把 Key 硬编码进笔记本（尤其打算提交到公开仓库时）。

🈶 关于语言：实验全部使用中文 state 与中文提示词。三种原语的选项 key （如 billing、verified）属于代码标识符，保持英文以便代码分支判断；它们的描述文字（criteria 值）均为中文，模型据此理解语义。

0. 准备

0.1 安装所需的库

如果已经用 ./setup_env.sh 创建过环境，本节通常显示“依赖已满足”；在其他环境里首次运行时会自动安装。

%pip install -q -U typesafe-sdk          # 本笔记本必需（要求 Python ≥ 3.10）
# %pip install -q -U jupyterlab         # 如本机还没有 Jupyter，取消注释运行一次
# %pip install -q -U nbformat nbclient  # 仅在需要重新生成/批量执行笔记本时安装

0.2 导入库并创建客户端

Choice / Score / Noul：三种问题原语的构造器（对应官方文档原语一章）；
TypeSafeClient：同步客户端，model="jev-latest" 表示使用官方旗舰模型的最新别名；
Key 从环境变量 TYPESAFE_API_KEY 读取；临时调试也可以直接赋值给 API_KEY（不要提交）。

import os
import time

from typesafe_sdk import (
    Choice,                       # 选择题：从命名选项中选一个，返回 choice + probabilities + confidence
    Score,                        # 打分题：按有序量表打分，返回期望分 + probabilities + confidence
    Noul,                         # 是非题：返回"是"的概率（0~1），本身就是概率所以没有 confidence 字段
    TypeSafeClient,
    TypeSafeAuthenticationError,  # 401 鉴权失败时抛出
)

API_KEY = os.environ.get("TYPESAFE_API_KEY", "")
# API_KEY = "apikey_..."   # ← 仅在临时调试时使用，注意不要提交到公开仓库

# Key 为空时不构造客户端：SDK 在无 Key 时抛 TypeSafeError（非 401 的
# TypeSafeAuthenticationError），不会被下面的回退捕获，会让整本笔记本中断。
client = TypeSafeClient(api_key=API_KEY, model="jev-latest") if API_KEY else None

0.3 离线回退用的两个替身类

真实 API 不可用时，我们需要一个与官方 SDK 响应对象同构的替身，让后续分析代码不用改。官方 SystemOneResponse 的访问方式是：

访问方式	返回
`resp.answers["名称"]`	全部答案（按问题名）
`resp.choices["名称"]`	选择题答案：`.choice` `.confidence` `.probabilities`
`resp.scores["名称"]`	打分题答案：`.score` `.confidence` `.legend` `.probabilities`
`resp.nouls["名称"]`	是非题答案：`.noul`（本身就是概率，无 confidence）
`resp.usage.input_tokens`	本次请求计费的 input token 数

下面的替身类暴露完全相同的属性，仅用于离线模式。

class _FakeAnswer:
    """单个答案的替身：按需挂属性（choice/score/noul/confidence/...）。"""

    def __init__(self, type_, **kw):
        self.type = type_
        for k, v in kw.items():
            setattr(self, k, v)


class _FakeResponse:
    """整个响应的替身：与 SystemOneResponse 同构（answers/nouls/choices/scores/usage）。"""

    def __init__(self, answers):
        self.answers = answers
        self.nouls = {k: v for k, v in answers.items() if v.type == "noul"}
        self.choices = {k: v for k, v in answers.items() if v.type == "choice"}
        self.scores = {k: v for k, v in answers.items() if v.type == "score"}
        self.model = "jev-latest(离线示例)"
        self.usage = _FakeAnswer("usage", input_tokens=0, output_tokens=0)

0.4 调用助手 `ts.call()`

统一入口：优先请求真实 API；只有当鉴权失败（401）时，才回退到各实验预置的离线示例数据，并在第一次回退时给出显著警告。这样拿到无效 Key 也能跑通全流程，而有效 Key 下全程真实。

class TS:
    real_calls = 0     # 成功的真实调用计数
    offline = False    # 一旦回退过就置 True，后续单元格据此跳过真实计时等逻辑
    _warned = False    # 完整警告只打印一次，避免刷屏

    def call(self, state, questions, offline_answers=None):
        if client is None:          # 未配置 Key：直接走离线示例，不触碰任何客户端
            TS.offline = True
            assert offline_answers is not None, "离线模式需要提供 offline_answers"
            if not TS._warned:
                TS._warned = True
                print("⚠️  离线示例模式：未设置 TYPESAFE_API_KEY，以下为内置示例数据而非真实模型结果；"
                      "设置有效的 TYPESAFE_API_KEY 后重跑本笔记本即可得到真实输出。")
            else:
                print("⚠️ （本次为离线示例数据，非真实 API 输出）")
            return _FakeResponse(offline_answers)
        try:
            resp = client.system_one(state, questions)
            TS.real_calls += 1
            return resp
        except TypeSafeAuthenticationError:
            TS.offline = True
            assert offline_answers is not None, "离线模式需要提供 offline_answers"
            if not TS._warned:
                TS._warned = True
                print("⚠️  离线示例模式：API Key 无效(401)，以下输出为内置示例数据而非真实模型结果；"
                      "设置有效的 TYPESAFE_API_KEY 后重跑本笔记本即可得到真实输出。")
            else:
                print("⚠️ （本次为离线示例数据，非真实 API 输出）")
            return _FakeResponse(offline_answers)


ts = TS()

0.5 连通性测试

用一条最简单的是非题试连官方 API：Key 有效则提示通过；若返回 401，后面的实验会自动切换到离线示例模式（见下一节说明），流程照样能走通。

if client is None:
    TS.offline = True
    print("⚠️  未设置 TYPESAFE_API_KEY，以下实验以【离线示例模式】运行：")
    print("    代码路径与真实调用完全一致，仅数据换成本笔记内置的示例值；")
    print("    在有效 Key 下重跑本笔记本即可得到真实模型输出。")
else:
  try:
    ping = client.system_one(
        "你好",
        {"is_greeting": Noul(instructions="这段文字是在打招呼吗？")},
    )
    print("✅ API 连通正常，Key 有效。将进行真实实验。")
  except TypeSafeAuthenticationError:
    TS.offline = True
    print("⚠️  API Key 无效（401）。以下实验将以【离线示例模式】运行：")
    print("    代码路径与真实调用完全一致，仅数据换成本笔记内置的示例值；")
    print("    在有效 Key 下重跑本笔记本即可得到真实模型输出。")

0.6 本章离线示例数据

下面是 4 对实体的预置示例答案，仅在 Key 无效时才会被用到。数值按文档风格拟制，保证后续 route() 与策展人提示路径能走通。

# 实验：4 对啤酒实体的 link_state + 三道字段 Noul
PAIRS_OFFLINE = [
    # 同一产品（应 assert sameAs）
    {
        "link_state": _FakeAnswer(
            "score", score=1.92, confidence=0.91,
            probabilities={0: 0.02, 1: 0.04, 2: 0.94},
            legend={0: "两个不同产品", 1: "相关但不一定相同", 2: "同一产品"},
        ),
        "same_name": _FakeAnswer("noul", noul=0.97),
        "same_brewery": _FakeAnswer("noul", noul=0.95),
        "same_style": _FakeAnswer("noul", noul=0.93),
    },
    # 完全不同（应 leave unlinked）
    {
        "link_state": _FakeAnswer(
            "score", score=0.18, confidence=0.88,
            probabilities={0: 0.85, 1: 0.12, 2: 0.03},
            legend={0: "两个不同产品", 1: "相关但不一定相同", 2: "同一产品"},
        ),
        "same_name": _FakeAnswer("noul", noul=0.08),
        "same_brewery": _FakeAnswer("noul", noul=0.12),
        "same_style": _FakeAnswer("noul", noul=0.15),
    },
    # 风格相关但不是同一商品（应 curator queue；style 偏低）
    {
        "link_state": _FakeAnswer(
            "score", score=1.12, confidence=0.72,
            probabilities={0: 0.18, 1: 0.55, 2: 0.27},
            legend={0: "两个不同产品", 1: "相关但不一定相同", 2: "同一产品"},
        ),
        "same_name": _FakeAnswer("noul", noul=0.81),
        "same_brewery": _FakeAnswer("noul", noul=0.90),
        "same_style": _FakeAnswer("noul", noul=0.22),
    },
    # 变体 / 特别版（应 curator queue）
    {
        "link_state": _FakeAnswer(
            "score", score=1.05, confidence=0.68,
            probabilities={0: 0.15, 1: 0.62, 2: 0.23},
            legend={0: "两个不同产品", 1: "相关但不一定相同", 2: "同一产品"},
        ),
        "same_name": _FakeAnswer("noul", noul=0.74),
        "same_brewery": _FakeAnswer("noul", noul=0.96),
        "same_style": _FakeAnswer("noul", noul=0.88),
    },
]

📖 理论速览（精简）

本笔记本默认你已读过 System One / 原语 / 置信度（中文镜像站有对应页）。此处只提醒实验会反复用到的三点：

请求模型：同一个 state + 一组问题 → 类型化答案；问题彼此独立、按 ID 返回；
三种原语：Choice 选命名选项；Score 返回概率加权期望分（可为小数）；Noul 返回“是”的概率（无 confidence）；
代码掌控制权：阈值、路由、合并答案都在你的代码里；模型只回答狭窄、原子的判断。

💡 问题 ID 不会发给模型——完整语义写在 instructions / criteria 里。

1. 知识图谱实体对齐（Entity alignment）

用一道 TypeSafe Score 判定候选实体对是同一产品 / 交给策展人 / 保持未链接；同一次请求再搭载三道字段 Noul，告诉策展人两边在哪一列上对不上。

为什么用 Score 而不是 Choice / Noul？

三种结果有有序关系：不同 → 相关可疑 → 同一；Score 的层级天然表达这种顺序；
中间档（related）是关键：错误合并的代价高于漏掉一个匹配，因此需要“交给人看”的出口；
阈值不需要拟合——route() 只需把期望分四舍五入到最近层级。

官方原文与中文镜像： entity_alignment · 中文版。

📖 理论根基

防错合并优先：不当合并会把两侧事实与外链一并污染；漏匹配只留下重复项。因此判断需要第三档，而不是硬二分类。
Score 的期望分：score 是概率加权期望值，可能是 1.12 这样的小数；round / int(x+0.5) 取最近层级。
扇出几乎免费：三道字段 Noul 与主 Score 同请求并行；仅当结果落入策展人档时，它们才被代码消费。
代码掌控制权：模型只回答“这对实体作为产品如何相关”；最终动作名写在你的 OUTCOME 表里。

1.1 定义三档结果与路由表

LEVELS 是 Score 的 criteria（中文描述）；OUTCOME 把层级编号映射成英文动作 key，方便代码分支。

LEVELS = [
    "它们描述的是两个不同的产品。",
    "它们描述的是密切相关的产品，可能是同一款，也可能是变体、特别版或名称易混淆的商品。",
    "它们描述的是完全同一款产品。",
]
OUTCOME = {0: "leave unlinked", 1: "curator queue", 2: "assert sameAs"}


def route(score_value: float) -> str:
    """整条决策规则：最近的 Score 层级决定动作。"""
    level = min(int(score_value + 0.5), len(LEVELS) - 1)
    return OUTCOME[level]

1.2 定义 4 对中文啤酒实体

覆盖四种典型情况：同一产品、完全不同、同厂相关但风格不符、变体/特别版。

PAIRS = [
    {
        "id": "same_product",
        "note": "同一款 IPA，两侧字段一致",
        "entity_a": {
            "name": "云岭精酿 · 晨雾 IPA",
            "brewery": "云岭精酿",
            "style": "美式 IPA",
        },
        "entity_b": {
            "name": "晨雾 IPA",
            "brewery": "云岭精酿（Kunling Brewing）",
            "style": "American IPA",
        },
    },
    {
        "id": "different",
        "note": "不同厂、不同酒款",
        "entity_a": {
            "name": "江城世涛",
            "brewery": "江城啤酒厂",
            "style": "帝国世涛",
        },
        "entity_b": {
            "name": "山城小麦",
            "brewery": "山城精酿",
            "style": "德式小麦",
        },
    },
    {
        "id": "style_mismatch",
        "note": "同厂同名系列，但风格描述冲突",
        "entity_a": {
            "name": "南湖淡色艾尔",
            "brewery": "南湖啤酒",
            "style": "英式淡色艾尔",
        },
        "entity_b": {
            "name": "南湖淡色艾尔",
            "brewery": "南湖啤酒",
            "style": "德式黑啤",
        },
    },
    {
        "id": "variant",
        "note": "同一酒款的桶陈特别版 vs 常规版",
        "entity_a": {
            "name": "赤兔世涛",
            "brewery": "赤兔精酿",
            "style": "帝国世涛",
        },
        "entity_b": {
            "name": "赤兔世涛 · 波本桶陈特别版",
            "brewery": "赤兔精酿",
            "style": "桶陈帝国世涛",
        },
    },
]

1.3 定义问题：1 道 Score + 3 道 Noul

问题 ID	类型	作用
`link_state`	Score	主决策：不同 / 相关可疑 / 同一
`same_name`	Noul	策展人提示：名称是否一致
`same_brewery`	Noul	策展人提示：酒厂是否一致
`same_style`	Noul	策展人提示：风格是否一致

QUESTIONS = {
    "link_state": Score(
        instructions="这两段实体描述作为产品如何相关？",
        criteria=LEVELS,
    ),
    "same_name": Noul(
        instructions="两个实体给出的啤酒名称是否相同？",
    ),
    "same_brewery": Noul(
        instructions="两个实体是否来自同一家酒厂？",
    ),
    "same_style": Noul(
        instructions="两个实体描述的啤酒风格是否相同？",
    ),
}

1.4 逐对调用并解读

每对实体一次请求（4 问并行）。打印期望分、路由结果，以及三道字段 Noul——落入 curator queue 时，这些 Noul 就是策展人的排查线索。

for pair, off in zip(PAIRS, PAIRS_OFFLINE):
    state = {"entity_a": pair["entity_a"], "entity_b": pair["entity_b"]}
    resp = ts.call(state, QUESTIONS, offline_answers=off)
    link = resp.answers["link_state"]
    action = route(link.score)
    print(f"【{pair['id']}】{pair['note']}")
    print(f"  A: {pair['entity_a']['name']} / {pair['entity_a']['brewery']} / {pair['entity_a']['style']}")
    print(f"  B: {pair['entity_b']['name']} / {pair['entity_b']['brewery']} / {pair['entity_b']['style']}")
    print(f"  score={link.score:.2f}  confidence={link.confidence:.2f}  →  {action}")
    print(
        f"  same_name={resp.answers['same_name'].noul:.2f}  "
        f"same_brewery={resp.answers['same_brewery'].noul:.2f}  "
        f"same_style={resp.answers['same_style'].noul:.2f}"
    )
    if action == "curator queue":
        hints = []
        if resp.answers["same_name"].noul < 0.5:
            hints.append("名称不一致")
        if resp.answers["same_brewery"].noul < 0.5:
            hints.append("酒厂不一致")
        if resp.answers["same_style"].noul < 0.5:
            hints.append("风格不一致")
        print(f"  策展人提示: {', '.join(hints) if hints else '字段大多一致，请人工确认是否变体'}")
    print()

观察要点

same_product 的期望分靠近 2 → assert sameAs；
different 靠近 0 → leave unlinked；
style_mismatch / variant 落在中间档 → curator queue，并靠 Noul 指出冲突字段或变体嫌疑；
没有阈值要拟合：改动决策只需改 LEVELS / OUTCOME，或调整四舍五入规则。

小结

组件	实验验证的行为
Score 三档	期望分四舍五入 → leave unlinked / curator queue / assert sameAs
字段 Noul	同请求扇出；仅策展人档消费
中文 demo	4 对啤酒覆盖同款、不同、风格冲突、变体

知识补充

组合爆炸提醒：两两 noul 判定是 O(n²)。生产上先用规则/编码器粗筛（同类型、同首字母、向量近邻），只剩少量候选对再做语义精判。
图上的用法：判完"同一实体"由代码合并节点、传播属性——知识图谱构建里 Jev 只负责最难的语义消歧一步。
不确定对：概率居中的候选对进人工队列，不要硬合并（合并错误的修复成本远高于漏合并）。

第 18 章：实体对齐

第 18 章：实体对齐

笔记本结构

运行要求

0. 准备

0.1 安装所需的库

0.2 导入库并创建客户端

0.3 离线回退用的两个替身类

0.4 调用助手 `ts.call()`

0.5 连通性测试

0.6 本章离线示例数据

📖 理论速览（精简）

1. 知识图谱实体对齐（Entity alignment）

📖 理论根基

1.1 定义三档结果与路由表

1.2 定义 4 对中文啤酒实体

1.3 定义问题：1 道 Score + 3 道 Noul

1.4 逐对调用并解读

小结

延伸阅读

知识补充

📑 Jev Cookbook：System One 判断模型实战教程

第 18 章：实体对齐

第 18 章：实体对齐

笔记本结构

运行要求

0. 准备

0.1 安装所需的库

0.2 导入库并创建客户端

0.3 离线回退用的两个替身类

0.4 调用助手 ts.call()

0.5 连通性测试

0.6 本章离线示例数据

📖 理论速览（精简）

1. 知识图谱实体对齐（Entity alignment）

📖 理论根基

1.1 定义三档结果与路由表

1.2 定义 4 对中文啤酒实体

1.3 定义问题：1 道 Score + 3 道 Noul

1.4 逐对调用并解读

小结

延伸阅读

知识补充

📑 Jev Cookbook：System One 判断模型实战教程

0.4 调用助手 `ts.call()`