第 11 章：自一致性 · Choice

本文整理自 Datawhale 开源项目 datawhalechina/jev-cookbook（CC BY-NC-SA 4.0），源文件：main/04_实战指南/02_自一致性Choice.ipynb。

本 notebook 把中文镜像站中的 Cookbook 改写成可以逐格运行、修改输入并观察结果的最小实验。 重复 Choice 审核标签并用置信度门控自动动作。

运行方式与 ../03_架构模式/01_架构模式.ipynb 一致：有有效的 TYPESAFE_API_KEY 时调用真实的 TypeSafe API；没有 Key 或返回 401 时使用内置的离线示例答案。后续代码不区分两种模式，便于先学习 控制流，再切换到真实模型观察概率和置信度。

学习提示：先顺序运行全部单元格，再回到“定义 state”或“定义问题”的单元格修改内容，重新运行后面的单元格。 API Key 只从环境变量读取，不能写进 notebook。

0. 准备

0.1 安装依赖

%pip install -q -U typesafe-sdk

0.2 创建客户端

import os
import statistics
import time
from pprint import pprint

from typesafe_sdk import (
    Choice,
    Score,
    Noul,
    TypeSafeClient,
    TypeSafeAuthenticationError,
)

API_KEY = os.environ.get("TYPESAFE_API_KEY", "")
client = TypeSafeClient(api_key=API_KEY, model="jev-latest") if API_KEY else None
print("客户端已创建：模型=jev-latest，Key=", "已配置" if API_KEY else "未配置（将使用离线示例）")

0.3 离线响应与统一调用入口

class _FakeAnswer:
    def __init__(self, type_, **values):
        self.type = type_
        for key, value in values.items():
            setattr(self, key, value)


class _FakeResponse:
    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)


class TS:
    offline = False
    _warned = False

    @classmethod
    def call(cls, state, questions, offline_answers):
        if client is None:
            cls.offline = True
            if not cls._warned:
                cls._warned = True
                print("⚠️ 未设置有效 TYPESAFE_API_KEY，以下输出使用内置离线示例。")
            return _FakeResponse(offline_answers)
        try:
            return client.system_one(state, questions)
        except TypeSafeAuthenticationError:
            cls.offline = True
            if not cls._warned:
                cls._warned = True
                print("⚠️ 未设置有效 TYPESAFE_API_KEY，以下输出使用内置离线示例。")
            return _FakeResponse(offline_answers)


def answer_line(name, answer):
    if answer.type == "noul":
        return f"{name}: noul={answer.noul:.2f}"
    if answer.type == "choice":
        return f"{name}: choice={answer.choice} confidence={answer.confidence:.2f}"
    return f"{name}: score={answer.score:.2f} confidence={answer.confidence:.2f}"


print("模式：", "离线示例" if TS.offline else "真实 API（首次调用后确定）")

0.4 连通性测试

if client is None:
    TS.offline = True
    print("⚠️ API Key 未设置，后续单元格使用离线示例。")
else:
    try:
        ping = client.system_one("你好", {"is_greeting": Noul(instructions="这段文字是在打招呼吗？")})
        print("✅ API 连通正常，后续单元格会使用真实结果。")
    except TypeSafeAuthenticationError:
        TS.offline = True
        print("⚠️ API Key 无效，后续单元格使用离线示例。")

1. 自一致性：Choice

对一条处于临界状态的审核帖子重复运行多个 Choice 问题。先观察标签是否稳定，再要求最高概率达到 0.60 才允许自动动作，否则返回 uncertain。

1.1 定义帖子和审核问题

POST = {
    "author": "用户 1842",
    "text": "有人说这个功能会泄露数据，我没有证据，但大家最好先别用了。",
    "reports": 3,
}

QUESTIONS = {
    "moderation": Choice(
        instructions="这条帖子最适合采取哪种审核动作？",
        criteria={"keep": "保留帖子", "remove": "删除帖子", "escalate": "升级给人工审核"},
    ),
    "queue": Choice(
        instructions="这条帖子应该进入哪个队列？",
        criteria={"general": "普通队列", "threat": "潜在威胁队列", "spam": "垃圾内容队列"},
    ),
    "severity": Choice(
        instructions="这条帖子的风险等级是什么？",
        criteria={"low": "低风险", "medium": "中风险", "high": "高风险"},
    ),
}
print("state/questions 已定义：字段数=", len(POST), "，问题数=", len(QUESTIONS))

1.2 重复调用并保留完整概率

OFFLINE_RUNS = [
    {"moderation": ("escalate", .58), "queue": ("threat", .64), "severity": ("medium", .57)},
    {"moderation": ("escalate", .62), "queue": ("threat", .61), "severity": ("medium", .55)},
    {"moderation": ("escalate", .59), "queue": ("general", .52), "severity": ("medium", .58)},
    {"moderation": ("keep", .51), "queue": ("threat", .60), "severity": ("medium", .56)},
    {"moderation": ("escalate", .60), "queue": ("threat", .63), "severity": ("medium", .59)},
]
runs = []
for row in OFFLINE_RUNS:
    offline = {}
    for name, (choice, confidence) in row.items():
        offline[name] = _FakeAnswer(
            "choice", choice=choice, confidence=confidence,
            probabilities={choice: confidence, "other": 1 - confidence},
        )
    response = TS.call(POST, QUESTIONS, offline)
    runs.append(response.choices)
    print(" | ".join(answer_line(name, answer) for name, answer in response.choices.items()))

1.3 置信度门控自动动作

def choice_decision(answer, threshold=0.60):
    return answer.choice if answer.confidence >= threshold else "uncertain"


for run_number, answers in enumerate(runs, 1):
    actions = {name: choice_decision(answer) for name, answer in answers.items()}
    print(f"第 {run_number} 次：", actions)

观察：Choice 的标签可以跨次变化，尤其是接近的概率分布。把 confidence 作为自动动作的门槛，会把临界答案明确转交人工。

知识补充

• 门控自动化：多次判定一致才自动执行、否则转人工，是生产里最常用的一道闸。confidence 是服务端综合指标（官方未公开公式），别把它当概率用，当"稳定度"用。
• 阈值属于应用：0.60 只是演示值；正式上线要用独立标注数据选阈值，并保留"不确定→人工"的出口（官方所有高危用例都是这么配的）。
• 进阶阅读：三路分流（自动/复核/人工）完整版见 18_基于置信度的分类.ipynb。

小结

这本 notebook 的边界很清楚：TypeSafe 只负责受限、可编程的判断；排序、阈值、分组、重建文本和 函数分派都由 Python 代码完成。修改输入或问题后重新运行，就能观察“模型答案 → 确定性代码”的变化。