第 25 章

第 25 章：层级分类

本文整理自 Datawhale 开源项目 datawhalechina/jev-cookbook（CC BY-NC-SA 4.0），源文件：main/04_实战指南/16_层级分类.ipynb。

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

笔记本结构

章节	内容	实验
0. 准备	安装、客户端、连通性、离线回退	—
1. 迷你商品树	三级中文数码分类树（约 10 个叶子）	打印树结构
2. 贪心下行	每层 Choice 选子节点，沿 argmax 走到叶子	3 条商品描述
3. 束搜索 K=2	保留 top-2 路径，path_score = 几何平均边概率	同 3 条对比

每个主题按固定节奏展开：原理 → 理论根基 → 定义数据 → 定义问题 → 调用 → 解读结果，每个单元格只做一件事，可直接顺着跑完（约 15–25 次 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 本章离线示例数据

仅在 Key 无效（401）时使用。数值按“清晰手机 / 笔记本 / 耳机”三类商品拟制，保证贪心与束搜索代码路径都能跑通。

# 离线：每层节点名 → 子选项概率分布（与 TREE 结构对齐）
# 键为“父路径字符串”（根用 ""）；值为 {child_key: prob}
HIER_OFFLINE = {
    # 商品 0：旗舰手机 → 数码 > 手机 > 旗舰机
    0: {
        "": {"digital": 0.82, "appliance": 0.12, "fashion": 0.06},
        "digital": {"phone": 0.78, "computer": 0.15, "audio": 0.07},
        "digital/phone": {"flagship": 0.71, "midrange": 0.22, "budget": 0.07},
    },
    # 商品 1：轻薄本 → 数码 > 电脑 > 笔记本
    1: {
        "": {"digital": 0.80, "appliance": 0.14, "fashion": 0.06},
        "digital": {"phone": 0.12, "computer": 0.75, "audio": 0.13},
        "digital/computer": {"laptop": 0.80, "desktop": 0.14, "tablet": 0.06},
    },
    # 商品 2：降噪耳机 → 数码 > 音频 > 耳机（束搜索时第二路径可能偏向手机配件感）
    2: {
        "": {"digital": 0.76, "appliance": 0.16, "fashion": 0.08},
        "digital": {"phone": 0.28, "computer": 0.18, "audio": 0.54},
        "digital/audio": {"headphones": 0.72, "speaker": 0.20, "mic": 0.08},
        "digital/phone": {"flagship": 0.35, "midrange": 0.40, "budget": 0.25},
    },
}

📖 理论速览（精简）

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

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

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

1. 迷你中文商品分类树

官方实战指南在 CPC / Shopify / MeSH 等巨型树上做层级分类。本笔记大幅简化：一棵三级中文数码零售树（约 10 个叶子），只保留算法骨架——

贪心下行：从根开始，对当前节点的子节点发一次 Choice，取 argmax，直到叶子；
束搜索（K=2）：每层保留概率最高的 2 条路径；路径分用长度归一化的几何平均 product(probs) ** (1 / decisions)，避免深浅叶子不公平。

出处：Hierarchical Classification · 中文镜像

1.1 原理

每个内部节点的子节点构成一次 Choice 的选项；
叶子没有子节点，到达叶子即分类完成；
问题 ID 不发给模型——语义写在 instructions / criteria 里；
代码掌控制权：遍历顺序、剪枝、最终取哪条路径都在你这边。

1.2 📖 理论根基

层级把一次“从几百个叶子里直接选”拆成多次“在少量兄弟里选”，好处是：

点	说明
可观测	能看出错分常发生在哪一层
可测试	改树结构后可单独测某节点的准确率
可校准	边概率可聚合为路径分，用于束搜索排序

束搜索的 path_score 做了长度归一化，否则浅叶子会系统性吃亏或占便宜。

1.3 定义分类树与展示函数

TREE = {
    "digital": {
        "_label": "数码电子",
        "phone": {
            "_label": "手机",
            "flagship": {"_label": "旗舰机"},
            "midrange": {"_label": "中端机"},
            "budget": {"_label": "入门机"},
        },
        "computer": {
            "_label": "电脑",
            "laptop": {"_label": "笔记本"},
            "desktop": {"_label": "台式机"},
            "tablet": {"_label": "平板"},
        },
        "audio": {
            "_label": "音频",
            "headphones": {"_label": "耳机"},
            "speaker": {"_label": "音箱"},
            "mic": {"_label": "麦克风"},
        },
    },
    "appliance": {
        "_label": "家电",
        "kitchen": {
            "_label": "厨电",
            "blender": {"_label": "搅拌机"},
            "rice_cooker": {"_label": "电饭煲"},
        },
        "cleaning": {
            "_label": "清洁",
            "vacuum": {"_label": "吸尘器"},
        },
    },
    "fashion": {
        "_label": "服饰",
        "wearable": {
            "_label": "可穿戴",
            "watch": {"_label": "手表"},
            "band": {"_label": "手环"},
        },
    },
}


def children_of(node):
    """返回 (key, child_subtree) 列表，跳过元数据键。"""
    return [(k, v) for k, v in node.items() if not k.startswith("_") and isinstance(v, dict)]


def is_leaf(node):
    return len(children_of(node)) == 0


def label_of(node, key):
    return node.get("_label", key)


def print_tree(node=None, indent=0, key="ROOT"):
    node = TREE if node is None else node
    if node is TREE:
        print("ROOT")
        for k, child in children_of(TREE):
            print_tree(child, 1, k)
        return
    prefix = "  " * indent
    mark = "🍃" if is_leaf(node) else "📁"
    print(f"{prefix}{mark} {key}（{label_of(node, key)}）")
    for k, child in children_of(node):
        print_tree(child, indent + 1, k)


print_tree()
n_leaves = 0


def count_leaves(node):
    global n_leaves
    kids = children_of(node)
    if not kids:
        n_leaves += 1
        return
    for _, c in kids:
        count_leaves(c)


count_leaves(TREE)
print(f"\n叶子数: {n_leaves}")

1.4 定义待分类的商品描述

PRODUCTS = [
    "全新旗舰智能手机，6.7 寸 OLED，徕卡三摄，支持卫星通信，适合重度摄影用户。",
    "14 寸轻薄商务笔记本，锐龙 7，16GB 内存，续航约 18 小时，重量 1.2kg。",
    "头戴式主动降噪耳机，40mm 动圈，续航 30 小时，支持多设备切换。",
]

for i, p in enumerate(PRODUCTS):
    print(f"[{i}] {p}")

2. 贪心下行（Greedy Walk）

每到一个内部节点：对该节点的子节点发一次 Choice，选概率最高的子节点，再继续。直到叶子。路径上的每条边概率都会记录下来，便于和束搜索对比。

2.1 定义：从节点构造 Choice 问题

def choice_at(node, path_keys):
    """对当前节点的直接子节点构造 Choice。"""
    kids = children_of(node)
    criteria = {k: label_of(child, k) for k, child in kids}
    depth = len(path_keys)
    instructions = (
        f"根据商品描述，判断它最属于哪一类（当前层级深度 {depth}）。"
        "只依据描述中的产品形态与用途，忽略营销夸张用语。"
    )
    return Choice(instructions=instructions, criteria=criteria)


def resolve_node(path_keys):
    """按 key 列表从 TREE 走到节点。"""
    node = TREE
    for k in path_keys:
        node = node[k]
    return node


def path_str(path_keys):
    return "/".join(path_keys)

2.2 定义贪心分类函数

def greedy_classify(text, product_idx, verbose=True):
    path = []
    edge_probs = []
    node = TREE
    while not is_leaf(node):
        kids = children_of(node)
        q = {"pick": choice_at(node, path)}
        # 离线表按“当前路径”取分布
        parent_key = path_str(path)
        dist = HIER_OFFLINE[product_idx].get(parent_key)
        if dist is None:
            # 兜底：均匀
            dist = {k: 1.0 / len(kids) for k, _ in kids}
        offline = {
            "pick": _FakeAnswer(
                "choice",
                choice=max(dist, key=dist.get),
                confidence=max(dist.values()),
                probabilities=dist,
            )
        }
        resp = ts.call(text, q, offline_answers=offline)
        ans = resp.choices["pick"]
        chosen = ans.choice
        prob = float(ans.probabilities.get(chosen, 0.0))
        edge_probs.append(prob)
        path.append(chosen)
        node = node[chosen]
        if verbose:
            labels = " > ".join(
                label_of(resolve_node(path[: i + 1]), path[i]) for i in range(len(path))
            )
            print(f"   层{len(path)}: {chosen}（{label_of(node, chosen)}）  p={prob:.2f}  conf={ans.confidence:.2f}")
            print(f"        路径: {labels}")
    return path, edge_probs

2.3 对 3 条商品跑贪心分类

print("=== 贪心下行 ===\n")
GREEDY_RESULTS = []
for i, text in enumerate(PRODUCTS):
    print(f"商品[{i}] {text[:36]}…")
    path, probs = greedy_classify(text, i)
    leaf_label = label_of(resolve_node(path), path[-1])
    geo = 1.0
    for p in probs:
        geo *= p
    geo = geo ** (1 / len(probs)) if probs else 0.0
    GREEDY_RESULTS.append((path, probs, geo))
    print(f"   → 叶子: {path[-1]}（{leaf_label}）  path_score={geo:.3f}\n")

观察要点

每层只在兄弟节点间做一次窄判断，比一次从全部叶子里选更稳；
path_score（几何平均）可用来比较不同深度的路径；
若某一层概率很分散，贪心可能锁死错误分支——这正是束搜索要缓解的。

3. 简化束搜索（Beam K=2）

官方做法是并行评估 K 条路径。本笔记用串行简化版：每层对存活路径各自发 Choice，只保留 path_score 最高的 K=2 条。评分公式：

path_score = product(edge_probabilities) ** (1 / decisions)

3.1 定义束搜索

def path_score(edge_probs):
    if not edge_probs:
        return 0.0
    prod = 1.0
    for p in edge_probs:
        prod *= max(p, 1e-9)
    return prod ** (1 / len(edge_probs))


def beam_classify(text, product_idx, k=2, verbose=True):
    # 每项: (path_keys, edge_probs)
    beam = [([], [])]
    while True:
        # 若所有路径都到叶子，结束
        if all(is_leaf(resolve_node(p)) for p, _ in beam):
            break
        candidates = []
        for path, probs in beam:
            node = resolve_node(path)
            if is_leaf(node):
                candidates.append((path, probs))
                continue
            kids = children_of(node)
            q = {"pick": choice_at(node, path)}
            parent_key = path_str(path)
            dist = HIER_OFFLINE[product_idx].get(parent_key)
            if dist is None:
                dist = {ck: 1.0 / len(kids) for ck, _ in kids}
            offline = {
                "pick": _FakeAnswer(
                    "choice",
                    choice=max(dist, key=dist.get),
                    confidence=max(dist.values()),
                    probabilities=dict(dist),
                )
            }
            resp = ts.call(text, q, offline_answers=offline)
            ans = resp.choices["pick"]
            # 取全部子选项概率，扩展候选
            for ck, _ in kids:
                p = float(ans.probabilities.get(ck, 0.0))
                candidates.append((path + [ck], probs + [p]))
        # 按 path_score 排序，保留 top-k
        candidates.sort(key=lambda x: path_score(x[1]), reverse=True)
        beam = candidates[:k]
        if verbose:
            print("   束状态:")
            for path, probs in beam:
                labels = " > ".join(
                    label_of(resolve_node(path[: i + 1]), path[i]) for i in range(len(path))
                )
                print(f"      [{path_score(probs):.3f}] {labels or '(根)'}")
    best = max(beam, key=lambda x: path_score(x[1]))
    return best

3.2 对同一批商品跑束搜索并对比贪心

print("=== 束搜索 K=2 ===\n")
for i, text in enumerate(PRODUCTS):
    print(f"商品[{i}] {text[:36]}…")
    path, probs = beam_classify(text, i, k=2)
    leaf_label = label_of(resolve_node(path), path[-1])
    g_path, g_probs, g_score = GREEDY_RESULTS[i]
    b_score = path_score(probs)
    same = path == g_path
    print(f"   束搜索 → {path[-1]}（{leaf_label}）  score={b_score:.3f}")
    print(f"   贪心   → {g_path[-1]}（{label_of(resolve_node(g_path), g_path[-1])}）  score={g_score:.3f}")
    print(f"   路径一致: {same}\n")

观察要点

K=2 时，若第一层就几乎确定（如“明显是数码”），束与贪心结果通常一致；
当中间层概率接近时，束可能保留另一条分支，最终叶子可能不同；
生产环境可把 K 条路径的 Choice 并行放进同一次 system_one 请求（官方做法）。

小结

方法	行为	适用
贪心下行	每层 argmax，一条路走到黑	层级浅、节点可分性强
束搜索 K=2	保留 top-K，按几何平均边概率排序	中间层易混淆、需要召回

知识补充

逐层 Choice 的代价结构：错误在高层会污染整棵子树，所以高层要么用更保守的阈值，要么用束搜索保留多条路径——本篇两种都演示了。
束宽 vs 成本：束搜索 = 路径数 × 每层一次调用（同调用内并行）；层数深时先把每层候选压到 2-3 个再搜。
进阶阅读：置信度门控决定"何时停止搜索直接接受"，见 ../02_核心概念/04_置信度.ipynb。

第 25 章：层级分类

第 25 章：层级分类

笔记本结构

运行要求

0. 准备

0.1 安装所需的库

0.2 导入库并创建客户端

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

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

0.5 连通性测试

0.6 本章离线示例数据

📖 理论速览（精简）

1. 迷你中文商品分类树

1.1 原理

1.2 📖 理论根基

1.3 定义分类树与展示函数

1.4 定义待分类的商品描述

2. 贪心下行（Greedy Walk）

2.1 定义：从节点构造 Choice 问题

2.2 定义贪心分类函数

2.3 对 3 条商品跑贪心分类

3. 简化束搜索（Beam K=2）

3.1 定义束搜索

3.2 对同一批商品跑束搜索并对比贪心

小结

延伸阅读

知识补充

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

第 25 章：层级分类

第 25 章：层级分类

笔记本结构

运行要求

0. 准备

0.1 安装所需的库

0.2 导入库并创建客户端

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

0.4 调用助手 ts.call()

0.5 连通性测试

0.6 本章离线示例数据

📖 理论速览（精简）

1. 迷你中文商品分类树

1.1 原理

1.2 📖 理论根基

1.3 定义分类树与展示函数

1.4 定义待分类的商品描述

2. 贪心下行（Greedy Walk）

2.1 定义：从节点构造 Choice 问题

2.2 定义贪心分类函数

2.3 对 3 条商品跑贪心分类

3. 简化束搜索（Beam K=2）

3.1 定义束搜索

3.2 对同一批商品跑束搜索并对比贪心

小结

延伸阅读

知识补充

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

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