第 4 章：状态：让判断连续可追溯

本文整理自 Datawhale 开源项目 datawhalechina/jev-cookbook（CC BY-NC-SA 4.0），源文件：main/02_核心概念/02_状态.ipynb。

针对官方文档对应章节的可运行实验笔记，全部实验使用中文场景与中文提示词。 面向会基础 Python、刚接触 AI Agent 的读者。

学习目标： 使用字符串、对象和数组表达相同事实，再单独研究订单与政策上下文的作用。

官方原文 · 中文参考。本章以中文重述理论、复刻对应场景；扩展实验会单独说明。 所有客户、订单及消息均为教学合成数据。

笔记本结构

章节	内容
0. 准备	安装库、配置客户端、连通性测试与离线示例
1	同事实、不同格式
2	添加订单与政策
3	字段白名单与避免标签泄漏
练习与小结	练习、自查、总结与本次执行记录

实验按原理 → 理论根基 → 定义数据 → 定义问题 → 调用 → 解读结果展开，每个代码单元格只做一件事。

运行要求

• Python ≥ 3.10；本章使用 typesafe-sdk==0.7.0。
• 真实实验需要启动进程的 TYPESAFE_API_KEY 环境变量，密钥不要写进 Notebook。

在本仓库 notebooks/ 目录创建环境并打开本文件：

./setup_env.sh
.venv/bin/python -m pip install -r requirements.txt -c generators/constraints-foundations.txt
.venv/bin/jupyter lab 05_状态.ipynb

产品名、字段名和选项 key 保持英文，state、提示词与解说使用中文。 默认 JEV_RUN_MODE=auto（在线优先）：检测到 TYPESAFE_API_KEY 即调用真实模型；未检测到才回退离线替身，每格输出都标注来源。 正式验收设置 JEV_RUN_MODE=live（无密钥直接报错、不回退）；强制纯离线学习设置 JEV_RUN_MODE=offline。

验证状态：2026-09-25 已用真实 API 在线执行本章全部调用，下方输出即实测结果。 批量执行、离线预览和验收记录见本目录 MAINTENANCE.md。

0. 准备

本节可折叠阅读，但独立运行时不能跳过。客户端、辅助对象和示例数据都在本文件中定义。

0.1 安装依赖

推荐先运行 setup_env.sh。只有当前内核缺少 SDK 时，本格才安装依赖。

import importlib.util
if importlib.util.find_spec("typesafe_sdk") is None:
    %pip install -q typesafe-sdk==0.7.0

观察与理解： 安装包的名字是 typesafe-sdk，Python 导入名是 typesafe_sdk。安装成功不代表 API 已连通。

0.2 导入与配置

默认模型固定版本，便于记录实验条件；可通过环境变量更换。不要从 Notebook 输入密钥。

import os
import json
import time
import math
from datetime import datetime, timezone
from importlib.metadata import version
from typesafe_sdk import (
    Choice, Score, Noul, NoulCriteria, TypeSafeClient,
    TypeSafeAuthenticationError, RetryPolicy,
)

MODEL = os.environ.get("TYPESAFE_DEFAULT_MODEL", "jev-1.13.0")
RUN_MODE = os.environ.get("JEV_RUN_MODE", "auto")
API_KEY = os.environ.get("TYPESAFE_API_KEY", "")
if RUN_MODE not in {"live", "offline", "auto"}:
    raise ValueError("JEV_RUN_MODE 只能是 live、offline 或 auto")
if RUN_MODE == "live" and not API_KEY:
    raise RuntimeError("JEV_RUN_MODE=live 需要真实密钥：请配置 TYPESAFE_API_KEY；仅学习可改用默认 auto（无密钥自动离线）")
client = None if (RUN_MODE == "offline" or not API_KEY) else TypeSafeClient(
    api_key=API_KEY, model=MODEL, timeout=30, retry=RetryPolicy(max_retries=0))
mode_note = ("在线优先：本次会话调用真实模型" if client is not None else
             ("强制离线（JEV_RUN_MODE=offline）" if RUN_MODE == "offline" else
              "未检测到 TYPESAFE_API_KEY → 离线替身；配置密钥后重跑本格即切在线实测"))
print("模式：", RUN_MODE, "｜", mode_note, "｜SDK：", version("typesafe-sdk"), "｜模型配置：", MODEL)

模式： auto ｜ 在线优先：本次会话调用真实模型 ｜SDK： 0.7.1 ｜模型配置： jev-1.13.0

默认在线优先：有密钥即走真实模型，每次调用的来源（live/offline）都记录在 CALL_LOG 与输出中；auto 仅在未配置密钥或 401 时回退离线替身。正式验收设 JEV_RUN_MODE=live（禁用回退、不自动重试，请求数量有界）。

0.3 连通性测试

用一条 Noul 检查真实响应能否返回。网络、限流与输入错误直接抛出，不伪装成不确定判断。

PING = {"source": "offline", "reason": "未发起连通性请求（无 client：未配置密钥或强制离线）"}
if client is not None:
    try:
        ping = client.system_one("你好", {"greeting": Noul(
            instructions="这段文字是否在打招呼？")})
        PING = {"source": "live", "model": ping.model,
                "input_tokens": ping.usage.input_tokens,
                "output_tokens": ping.usage.output_tokens}
    except TypeSafeAuthenticationError:
        if RUN_MODE == "live":
            raise
        client.close()
        client = None
        PING["reason"] = "401 鉴权失败，仅教学模式允许回退"
print(json.dumps(PING, ensure_ascii=False))

{"source": "live", "model": "jev-1.13.0", "input_tokens": 278, "output_tokens": 22}

观察与理解： source=live 表示这一次连通性请求成功；仍要查看后续实验记录，不能用它代替整章验收。

0.4 离线替身

沿用参考模板的 _FakeAnswer 与 _FakeResponse 访问方式。人工数字仅用来检验读取字段和代码分支。

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


class _FakeResponse:
    def __init__(self, answers):
        self.answers = answers
        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.nouls = {k: v for k, v in answers.items() if v.type == "noul"}
        self.model = "人工示例，非模型预测"
        self.usage = _FakeAnswer("usage", input_tokens=0, output_tokens=0)

人工 Score 由概率计算期望，避免模板中的分数与分布不一致。人工 confidence 只是指定的演示字段，不是在复现服务端的计算公式。

定义两种示例答案构造器；Noul 可直接用 _FakeAnswer。所有具体答案集中在下一节。

def fake_choice(probabilities, confidence):
    return _FakeAnswer("choice", choice=max(probabilities, key=probabilities.get),
                       probabilities=probabilities, confidence=confidence)


def fake_score(probabilities, legend, confidence):
    return _FakeAnswer("score", score=sum(k * p for k, p in probabilities.items()),
                       probabilities=probabilities, legend=dict(enumerate(legend)),
                       confidence=confidence)

观察与理解： 例如概率 {0:0.05, 1:0.26, 2:0.69} 对应 1.64。不能把另一个数与该分布配在一起。

0.5 统一调用入口

每次调用记录来源、模型与 token 用量。离线耗时记为 None，不把本地字典访问当成模型速度。

CALL_LOG = []


class TS:
    def call(self, state, questions, offline_answers, label):
        start = time.perf_counter()
        source = "live"
        if client is None:
            response, source = _FakeResponse(offline_answers), "offline"
        else:
            try:
                response = client.system_one(state, questions)
            except TypeSafeAuthenticationError:
                if RUN_MODE != "auto":
                    raise
                response, source = _FakeResponse(offline_answers), "offline"
        validate_response(response, questions)
        CALL_LOG.append({"case": label, "source": source, "model": response.model,
                         "seconds": time.perf_counter() - start if source == "live" else None,
                         "input_tokens": response.usage.input_tokens,
                         "output_tokens": response.usage.output_tokens})
        if source == "offline":
            print("离线替身（未调用真实模型）：", label, "；人工答案，仅演示代码路径")
        return response


ts = TS()

与参考模板相比，这里增加了严格 live 模式和逐次记录。保留 401 教学回退，但超时、429 等错误继续失败，防止验收被回退掩盖。

校验结构与数值契约；只断言接口应满足的性质，不断言真实模型必须预测某个标签。

def validate_response(response, questions):
    if set(response.answers) != set(questions):
        raise ValueError("答案 ID 与问题 ID 不一致")
    for key, question in questions.items():
        answer = response.answers[key]
        if isinstance(question, Noul):
            if not 0 <= answer.noul <= 1:
                raise ValueError("Noul 超出概率范围")
            continue
        probabilities = answer.probabilities
        if not all(math.isfinite(p) and 0 <= p <= 1 for p in probabilities.values()):
            raise ValueError("概率值无效")
        if not math.isclose(sum(probabilities.values()), 1, abs_tol=0.02):
            raise ValueError("概率之和偏离 1")
        if not 0 <= answer.confidence <= 1:
            raise ValueError("confidence 超出范围")
        if isinstance(question, Choice):
            if set(probabilities) != set(question.criteria):
                raise ValueError("Choice 选项集合不一致")
            if answer.choice not in probabilities:
                raise ValueError("Choice 标签不在选项中")
        else:
            expected = sum(int(k) * p for k, p in probabilities.items())
            if not math.isclose(answer.score, expected, abs_tol=0.03):
                raise ValueError("Score 与概率加权期望不一致")

观察与理解： 容差用于服务端数值舍入。结构检查通过只说明响应可读取，不证明语义判断正确。

显示结果时统一列出类型、概率和置信度；Noul 不额外制造 confidence 字段。

def show(response):
    rows = {}
    for key, answer in response.answers.items():
        rows[key] = {name: getattr(answer, name) for name in
                     ("type", "choice", "score", "noul", "confidence", "probabilities", "legend")
                     if hasattr(answer, name)}
    print(json.dumps(rows, ensure_ascii=False, indent=2))

0.6 本章离线示例数据

以下数值全部人工构造，专门测试分支；不来自 Jev，也不能用于估计中文准确率或校准情况。正式 live 运行不会使用这些答案。

FORMAT_OFFLINE = [
    {"refund_requested": _FakeAnswer("noul", noul=0.96)},
    {"refund_requested": _FakeAnswer("noul", noul=0.96)},
    {"refund_requested": _FakeAnswer("noul", noul=0.96)},
]
CONTEXT_OFFLINE = [
    {"support": fake_choice({"supported": 0.1, "unsupported": 0.1, "insufficient": 0.8}, 0.76)},
    {"support": fake_choice({"supported": 0.93, "unsupported": 0.03, "insufficient": 0.04}, 0.89)},
]

📖 理论根基：state 是提供给判断者的材料

state 放内容、事实和相关记录；questions 放评判任务。需要比较的事实放在同一状态里， 例如客户消息、交易记录和退款政策。对象中的具名字段有助于表达各部分关系；简单消息也可以直接用字符串。

state 的数组表达一组材料，不代表多条输入的批量 API。同一请求仍是一个 state，所有问题共享它。 Jev 当前的文本接口不直接接收图片、音频和视频。中文输入的效果需要在本任务上验证。

官方原文 · 中文参考 · 官方原文 · 中文参考

1. 控制实验 A：只改变表达格式

原理：固定订单号与消息，分别用字符串、对象、数组表达。不能在对象版额外加入政策，再把结果改善归因于“对象更好”。 三种表达的内容一致，但序列化后的文本仍有差异；一次小样本比较不能推断普遍优劣。

第一步：固定事实

MESSAGE = "订单 A-104 被扣了两次款，请退还重复扣取的那一笔。"
STATE_VARIANTS = {
    "字符串": "订单号：A-104。客户消息：" + MESSAGE,
    "对象": {"order_id": "A-104", "message": MESSAGE},
    "数组": ["订单号：A-104", "客户消息：" + MESSAGE],
}

第二步：固定问题

这里不用只在对象中存在的路径，保证问题文本也一致。

FORMAT_QUESTIONS = {
    "refund_requested": Noul(instructions="材料中的客户是否明确要求退还款项？"),
}

观察与理解： 固定模型、题目与事实，记录格式变化。概率不同并不自动代表某个结果更准确。

第三步：逐一调用

format_responses = {
    name: ts.call(state, FORMAT_QUESTIONS, FORMAT_OFFLINE[i], "格式对照：" + name)
    for i, (name, state) in enumerate(STATE_VARIANTS.items())
}

第四步：显示实际观测

for name, response in format_responses.items():
    print({"格式": name, "退款请求概率": response.nouls["refund_requested"].noul})

{'格式': '字符串', '退款请求概率': 0.99}
{'格式': '对象', '退款请求概率': 0.98}
{'格式': '数组', '退款请求概率': 0.98}

观察与理解： 离线预览中人为设成相同数值，不能据此声称模型对格式不敏感。真实实验应原样记录全部返回。

2. 控制实验 B：固定对象格式，再补充事实

📖 理论根基

原文展示了包含对话、订单和政策的完整状态。这里沿用重复扣款场景，比较材料不足与材料完整两种情况。 把‘信息不足’作为显式选项，避免让模型必须在支持／不支持之间猜测。

先定义不包含政策和支付证据的对象。

MINIMAL_STATE = {"ticket": {"message": MESSAGE}, "order_id": "A-104"}

再补充原文场景中的支持对话、订单和政策。

ENRICHED_STATE = {
    "ticket": {
        "message": MESSAGE,
        "support_reply": "我们正在核查扣款记录。",
    },
    "order": {
        "id": "A-104",
        "charges": [
            {"amount_usd": 49, "status": "captured"},
            {"amount_usd": 49, "status": "captured"},
        ],
    },
    "refund_policy": "同一订单的重复扣款可以退还重复部分。",
}

观察与理解： 新增的是可供判断的证据，不是‘正确答案为 supported’这种评测标签。

两种状态使用同一个问题。

CONTEXT_QUESTIONS = {
    "support": Choice(
        instructions="仅按提供的扣款证据和退款政策，能否支持客户的退款请求？不要补造缺失事实。",
        criteria={
            "supported": "必要证据与政策已提供，并支持退款",
            "unsupported": "必要证据与政策已提供，但不支持退款",
            "insufficient": "缺少必要证据或政策，无法完成判断",
        },
    ),
}

执行上下文对照。

context_responses = [
    ts.call(state, CONTEXT_QUESTIONS, CONTEXT_OFFLINE[i], label)
    for i, (state, label) in enumerate([
        (MINIMAL_STATE, "上下文：只有客户消息"),
        (ENRICHED_STATE, "上下文：增加交易与政策"),
    ])
]

观察与理解： 这是两个状态各发一次请求。不能把两者的差异直接称为结构化格式带来的提升。

查看选项和不确定性。

for name, response in zip(["材料不足", "材料完整"], context_responses):
    print(name)
    show(response)

材料不足
{
  "support": {
    "type": "choice",
    "choice": "insufficient",
    "confidence": 1.0,
    "probabilities": {
      "supported": 0.0,
      "unsupported": 0.0,
      "insufficient": 1.0
    }
  }
}
材料完整
{
  "support": {
    "type": "choice",
    "choice": "supported",
    "confidence": 0.85,
    "probabilities": {
      "unsupported": 0.0,
      "insufficient": 0.09,
      "supported": 0.91
    }
  }
}

观察与理解： 预期前者更可能选 insufficient，但这只是实验假设。如果真实返回不符合预期，应检查问题边界并记录失败。

3. 字段白名单：模型该看什么

原理：数据库记录常含人工标签、内部备注和无关信息。显式挑选输入字段，让实验的数据边界可检查。 下面是本教程的工程扩展：它补充输入构造方法，不新增模型调用。

构造一条含评测答案的教学记录。

RAW_RECORD = {
    "message": MESSAGE, "order_id": "A-104",
    "expected_label": "supported", "split": "holdout",
    "internal_note": "用于人工验收的记录，不发送模型",
}

白名单函数只返回所需事实。

def make_state(record):
    return {"message": record["message"], "order_id": record["order_id"]}


filtered_state = make_state(RAW_RECORD)

检查真正将发送的字段。

print(json.dumps(filtered_state, ensure_ascii=False, indent=2))
assert "expected_label" not in filtered_state
assert "split" not in filtered_state

{
  "message": "订单 A-104 被扣了两次款，请退还重复扣取的那一笔。",
  "order_id": "A-104"
}

观察与理解： 这是确定性输入检查，适合断言。真实预测是否符合人工标签则应进入评估统计。

练习与自查

为至少五条不同消息设计格式对照表；对每条消息保持事实一致。你还需要哪些记录才能让别人复现？

参考思路：先完成练习再展开

保存完整 state、instructions、criteria、请求模型、实际模型、SDK 版本和运行日期。多样本报告差异；不要只挑最支持结论的一条。

小结

对照	保持不变	改变什么
A	事实、问题、模型	格式
B	对象格式、问题、模型	可用事实

下一章：原语。

离线运行只说明教材代码能执行。正式交付必须实际运行 live，并阅读每条输出；缺失的分支应记为未观察到。

本次执行记录

先关闭连接，再生成记录。下面的 JSON 由实际运行计算，批量执行器会据此检查来源。

if client is not None:
    client.close()

本章拿几句话试了试真实模型，看它给的概率怎么反应——这只说明模型对这类输入的反应方式，不构成准确率评测。特别提醒：如果哪句答得合心意就专门挑出来当考题，再拿这些挑过的句子去算准确率，数字必然虚高。输出里的耗时也只是当时网络的快照，每次都会不一样。

AUDIT = {
    "kind": "jev_execution_audit",
    "executed_at_utc": datetime.now(timezone.utc).isoformat(),
    "sdk": version("typesafe-sdk"), "requested_model": MODEL,
    "mode": RUN_MODE, "ping": PING,
    "real_calls": sum(x["source"] == "live" for x in CALL_LOG),
    "offline_calls": sum(x["source"] == "offline" for x in CALL_LOG),
    "cases": CALL_LOG,
    "coverage": globals().get("COVERAGE", {}),
    "validation_status": "live_executed_requires_review" if (
        PING["source"] == "live" and CALL_LOG
        and all(x["source"] == "live" for x in CALL_LOG)
    ) else "offline_only_not_model_evidence",
}
print(json.dumps(AUDIT, ensure_ascii=False, indent=2))

{
  "kind": "jev_execution_audit",
  "executed_at_utc": "2026-09-25T09:08:20.654825+00:00",
  "sdk": "0.7.1",
  "requested_model": "jev-1.13.0",
  "mode": "auto",
  "ping": {
    "source": "live",
    "model": "jev-1.13.0",
    "input_tokens": 278,
    "output_tokens": 22
  },
  "real_calls": 5,
  "offline_calls": 0,
  "cases": [
    {
      "case": "格式对照：字符串",
      "source": "live",
      "model": "jev-1.13.0",
      "seconds": 0.2988031250424683,
      "input_tokens": 326,
      "output_tokens": 21
    },
    {
      "case": "格式对照：对象",
      "source": "live",
      "model": "jev-1.13.0",
      "seconds": 0.2801559576764703,
      "input_tokens": 329,
      "output_tokens": 21
    },
    {
      "case": "格式对照：数组",
      "source": "live",
      "model": "jev-1.13.0",
      "seconds": 0.4913866249844432,
      "input_tokens": 330,
      "output_tokens": 21
    },
    {
      "case": "上下文：只有客户消息",
      "source": "live",
      "model": "jev-1.13.0",
      "seconds": 0.3016924583353102,
      "input_tokens": 449,
      "output_tokens": 44
    },
    {
      "case": "上下文：增加交易与政策",
      "source": "live",
      "model": "jev-1.13.0",
      "seconds": 0.3462674170732498,
      "input_tokens": 545,
      "output_tokens": 41
    }
  ],
  "coverage": {},
  "validation_status": "live_executed_requires_review"
}

读完输出后，在本仓库 notebooks/MAINTENANCE.md 的验收表中记录日期、真实模型、观察到的分支和偏离预期之处。不要把人工演示数值抄进实测记录。