4. 设计决策收件箱

来源：Jev

预计阅读时间：45～60 分钟
本篇目标：完成 Decision Inbox 第一版。程序从本地文件读取留言，用 Jev 判断类别、紧急程度和行动请求，再把结构化结果写回文件。

前面三篇解决了“Jev 是什么”和“怎样提出一个可判断的问题”。从这一篇开始，我们把零散代码变成一个真正可运行的小项目。

Decision Inbox 的场景很简单：你的备忘录、个人网站或活动报名页里积累了一批留言，人工逐条浏览很慢。我们希望先判断每条留言的主题、紧急程度，以及对方是否明确要求采取行动，然后生成一份待处理清单。

它不会自动回复、删消息或替你做决定。第一版只生成建议结果。这样既能看清 Jev 在系统中的位置，也不会因为一次错误判断产生外部后果。

一、先划清 Jev 与普通代码的边界

一个常见误区是：既然用了 AI，就把整个流程都交给 AI。实际工程中，下面这些事情更适合普通代码：

• 读取和校验 JSON；
• 检查消息 ID 是否重复；
• 决定输入、输出文件路径；
• 保存判断结果；
• 记录错误；
• 根据明确规则排序或路由。

Jev 只负责普通 if/else 很难稳定处理的语义判断：

• 这条自然语言留言主要属于哪一类？
• 它表达的情况有多紧急？
• 它是否明确要求收件人采取行动？

可以把整个项目理解成下面这条数据流：

messages.json
      ↓ 普通代码读取并校验
单条留言的结构化 state
      ↓ Jev 完成三项语义判断
带概率和置信度的结构化结果
      ↓ 普通代码整理并保存
output/decisions.json

模型不是应用本身，而是应用中的一个判断部件。文件路径、失败策略和实际动作仍掌握在代码手中。

二、准备输入数据

进入上一篇使用的项目目录：

macOS 或 Linux：

cd ~/jev-course/decision-inbox

Windows PowerShell：

Set-Location "$HOME\jev-course\decision-inbox"

新建 messages.json，写入下面四条虚构留言：

[
  {
    "id": "msg-001",
    "source": "personal_site",
    "text": "我订阅了两次同一个服务，今天还能帮我核对一下扣款吗？"
  },
  {
    "id": "msg-002",
    "source": "notes",
    "text": "刚才上传照片时应用连续闪退三次，我用的是最新版。"
  },
  {
    "id": "msg-003",
    "source": "event_form",
    "text": "周六的分享很有意思，谢谢整理这些资料，不用回复。"
  },
  {
    "id": "msg-004",
    "source": "personal_site",
    "text": "明天中午前需要确认场地，否则预约会自动取消，请尽快告诉我结果。"
  }
]

这里使用中文，是因为教程面向中国用户；但要记住，官方 Models 文档 提示英语目前表现最好，其他语言需要用自己的内容验证。我们的练习目标不是证明它一定判断正确，而是搭好一个可以用真实中文数据评估的流程。第 7 篇会补上系统化评估。

示例数据不要换成真实身份证号、合同、病历或公司机密。判断服务需要接收 state 才能工作，练习阶段没有必要上传敏感信息。

三、设计输入 state

Jev 的 state 可以是一段字符串，也可以是结构化 JSON。这个项目选择对象形式：

state = {
    "message": {
        "id": "msg-001",
        "source": "personal_site",
        "text": "……",
    }
}

对象形式的优点是字段关系明确。以后即使增加 received_at 或 sender_type，问题说明仍可准确引用 message.text，不用要求模型从一大段拼接文本中猜测各部分含义。

不过，不要把“结构化”误解成“把数据库整行原样发送”。只传本次判断真正需要的信息。比如消息分类并不需要用户手机号，就不应把手机号放进 state。

产物一致性前置模块

先把配套 artifact_io.py 复制到项目根目录。新运行开始会使旧的三类生成产物失效；网络失败后不要继续路由。模块以临时文件原子替换，保存 run_id、输入摘要和成功状态；下游必须验证元数据链，不能只检查文件存在。

四、写出 Decision Inbox 第一版

在同一目录中新建 decision_inbox.py：

import json
from artifact_io import begin_run, save_artifact
from pathlib import Path

from typesafe_sdk import Choice, Noul, NoulCriteria, Score, TypeSafeClient


BASE_DIR = Path(__file__).resolve().parent
INPUT_FILE = BASE_DIR / "messages.json"
OUTPUT_DIR = BASE_DIR / "output"
OUTPUT_FILE = OUTPUT_DIR / "decisions.json"

CATEGORIES = {
    "billing": "Charges, invoices, payments, refunds, or subscriptions.",
    "technical": "Software or hardware is broken or unavailable.",
    "scheduling": "Dates, times, appointments, reservations, or travel plans.",
    "feedback": "An opinion, compliment, or experience without another main request.",
    "other": "None of the other options clearly fits.",
}

QUESTIONS = {
    "category": Choice(
        instructions="What is the main topic of `message.text`?",
        criteria=CATEGORIES,
    ),
    "urgency": Score(
        instructions="How urgent is the situation described in `message.text`?",
        criteria=[
            "No time pressure or active harm is expressed.",
            "A prompt response would help, but there is no strict deadline or active harm.",
            "A strict deadline, ongoing loss, or immediate harm is described.",
        ],
    ),
    "action_requested": Noul(
        instructions=(
            "Does `message.text` explicitly ask the recipient to take an action?"
        ),
        criteria=NoulCriteria(
            true="The sender asks the recipient to reply, check, send, change, schedule, or complete something.",
            false="The sender only shares information, emotion, or status and requests no action.",
        ),
    ),
}


def load_messages(path: Path) -> list[dict]:
    with path.open("r", encoding="utf-8") as file:
        data = json.load(file)

    if not isinstance(data, list):
        raise ValueError("messages.json 顶层必须是数组")

    seen_ids = set()
    for index, item in enumerate(data, start=1):
        if not isinstance(item, dict):
            raise ValueError(f"第 {index} 条消息必须是对象")

        missing = {"id", "source", "text"} - item.keys()
        if missing:
            raise ValueError(f"第 {index} 条消息缺少字段：{sorted(missing)}")

        if not all(isinstance(item[key], str) for key in ("id", "source", "text")):
            raise ValueError(f"第 {index} 条消息的 id、source、text 必须是字符串")

        if not item["text"].strip():
            raise ValueError(f"第 {index} 条消息正文为空")

        if item["id"] in seen_ids:
            raise ValueError(f"消息 ID 重复：{item['id']}")
        seen_ids.add(item["id"])

    return data


def evaluate_message(client: TypeSafeClient, message: dict) -> dict:
    response = client.system_one(
        state={"message": message},
        questions=QUESTIONS,
    )

    category = response.answers["category"]
    urgency = response.answers["urgency"]
    action_requested = response.answers["action_requested"]

    if category.choice not in CATEGORIES:
        raise ValueError(f"未知分类：{category.choice}")
    if not 0 <= urgency.score <= 2:
        raise ValueError(f"紧急分数越界：{urgency.score}")
    if not 0 <= action_requested.noul <= 1:
        raise ValueError(f"行动请求概率越界：{action_requested.noul}")

    return {
        "id": message["id"],
        "source": message["source"],
        "text": message["text"],
        "judgment": {
            "category": category.choice,
            "category_probabilities": dict(category.probabilities),
            "category_confidence": category.confidence,
            "urgency_score": urgency.score,
            "urgency_confidence": urgency.confidence,
            "action_requested_probability": action_requested.noul,
        },
    }


def main() -> None:
    input_hash = begin_run(BASE_DIR)
    messages = load_messages(INPUT_FILE)
    decisions = []

    with TypeSafeClient() as client:
        for position, message in enumerate(messages, start=1):
            print(f"[{position}/{len(messages)}] 正在判断 {message['id']}...")
            decisions.append(evaluate_message(client, message))

    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    save_artifact(OUTPUT_FILE, decisions, BASE_DIR / "messages.json", input_hash)

    print(f"已生成 {len(decisions)} 条结果：{OUTPUT_FILE}")


if __name__ == "__main__":
    main()

这段代码看起来比上一篇长，但职责其实很清楚：

• load_messages() 负责检查输入，不调用模型；
• evaluate_message() 只处理一条消息，并保留判断依据相关的数值；
• main() 负责串起读取、循环调用和文件写入。

我们没有只保存 category.choice。如果丢掉概率和置信度，下一步就无法判断某个标签是“明显胜出”，还是仅仅比第二名高一点点。

五、先检查语法和输入，再调用 API

真实 API 调用可能计费，也可能受速率限制。先做不联网的检查。

macOS 或 Linux：

python -m py_compile decision_inbox.py
python -m json.tool messages.json

Windows PowerShell：

.\.venv\Scripts\python.exe -m py_compile decision_inbox.py
.\.venv\Scripts\python.exe -m json.tool messages.json

第一条命令没有输出，通常表示 Python 语法正确；第二条会重新格式化打印 JSON。如果它指出具体行列，先检查漏掉的逗号、引号或方括号。

然后确认当前终端仍然有 API Key：

macOS 或 Linux：

python -c "import os; print('key loaded:', bool(os.getenv('TYPESAFE_API_KEY')))"

Windows PowerShell：

.\.venv\Scripts\python.exe -c "import os; print('key loaded:', bool(os.getenv('TYPESAFE_API_KEY')))"

如果显示 False，按照第 2 篇重新设置环境变量。关闭终端以后，临时环境变量不会自动保留。

六、运行项目并检查产物

macOS 或 Linux：

python decision_inbox.py

Windows PowerShell：

.\.venv\Scripts\python.exe decision_inbox.py

终端应该依次显示四条消息的进度，最后给出 decisions.json 的绝对路径：

[1/4] 正在判断 msg-001...
[2/4] 正在判断 msg-002...
[3/4] 正在判断 msg-003...
[4/4] 正在判断 msg-004...
已生成 4 条结果：.../output/decisions.json

文件中的具体小数会变化，结构应当类似：

[
  {
    "id": "msg-001",
    "source": "personal_site",
    "text": "我订阅了两次同一个服务，今天还能帮我核对一下扣款吗？",
    "judgment": {
      "category": "billing",
      "category_probabilities": {
        "billing": 0.0,
        "technical": 0.0,
        "scheduling": 0.0,
        "feedback": 0.0,
        "other": 0.0
      },
      "category_confidence": 0.0,
      "urgency_score": 0.0,
      "urgency_confidence": 0.0,
      "action_requested_probability": 0.0
    }
  }
]

这里的 0.0 只是结构占位，不是预期答案。不要手工修改自己的实际结果去匹配示例。

用一条独立命令验证输出文件，而不是只凭肉眼看终端。

macOS 或 Linux：

python -c "import json; d=json.load(open('output/decisions.json', encoding='utf-8')); assert len(d)==4; assert all('judgment' in x for x in d); print('output verified')"

Windows PowerShell：

.\.venv\Scripts\python.exe -c "import json; d=json.load(open('output/decisions.json', encoding='utf-8')); assert len(d)==4; assert all('judgment' in x for x in d); print('output verified')"

看到下面这行，说明四条结果都已落盘，而且每条都包含判断对象：

output verified

这项验证只证明程序产出了结构正确的数据，不证明语义判断正确。后者需要人工标注样例和评估集，两者不能混为一谈。

七、给项目补一份 .gitignore

即使现在还没使用 Git，也建议在项目目录创建 .gitignore：

.venv/
__pycache__/
*.pyc
.env
output/

这能避免虚拟环境、临时产物和可能保存密钥的 .env 文件进入仓库。不过要注意，.gitignore 不是保险箱：一旦 Key 已经提交过，再把文件加入忽略列表也不会清除历史记录，应立即撤销旧 Key 并重新创建。

本教程暂时不要求安装 python-dotenv。使用终端环境变量，能少引入一个“到底从哪里读取密钥”的排查点。

八、第一版故意没有做什么

现在的程序已经能工作，但离“无人值守的自动系统”还有距离：

• 任意一条 API 调用失败，整个程序就会停止；
• 多条判断虽然放在同一次请求中，却还没有组合成最终处理建议；
• 没有定义何时相信模型、何时交给人工；
• 没有用自己的标注数据衡量中文效果；
• 没有日志、重试、缓存和成本控制。

这些不是遗漏，而是接下来几篇的学习路径。先保留一份简单、能运行、容易看懂的基线，再逐步完善，比一开始堆满异常处理和框架更容易定位问题。

目前尤其不要根据 urgency_score 自动发消息，也不要根据 category 自动删除或移动真实文件。判断结果只写到本地 JSON，错误的代价很低。

九、常见问题

1. 提示找不到 messages.json

脚本使用 __file__ 定位同目录文件，正常情况下从其他目录运行也能找到。如果仍然报错，确认文件名是完整的 messages.json，而不是系统隐藏扩展名后的 messages.json.txt。

2. 中文结果和文章预期不同

先确认输出结构正确，再观察概率与置信度。不要为了让四条示例“看起来全对”而不断微调描述。当前样例太少，无法证明一次改动真的改善了系统。把有争议的消息记下来，留到第 7 篇放进评估集。

3. dict(category.probabilities) 报错

SDK 版本更新后，概率字段的具体容器可能变化。先直接打印 category.probabilities，再参考当前 Python SDK 文档 的序列化方式。不要用字符串截取模型输出。

4. 运行到一半遇到网络或速率错误

第一版是故意设计成失败即停止，避免悄悄生成不完整结果。记住失败的消息 ID，确认网络、配额和 Key 后重新运行。后面接入真实应用时，再加入逐条保存、有限重试和幂等处理。

十、完成本篇的验收

现在检查项目目录，至少应有：

decision-inbox/
├── .gitignore
├── .venv/
├── decision_inbox.py
├── first_call.py
├── messages.json
├── output/
│   └── decisions.json
└── primitives_lab.py

并确认：

• messages.json 能通过 JSON 格式检查；
• decision_inbox.py 能通过语法检查并完成四条调用；
• output/decisions.json 包含四条结果；
• 每条结果都保留分类概率、分类置信度、紧急分数与行动请求概率；
• 你能说明哪些工作属于普通代码，哪些判断交给 Jev；
• 程序没有根据判断结果执行不可逆动作。

下一篇会继续完善这个项目：不再只是并排保存三个答案，而是把多个判断组合成真正的处理建议，并让代码清楚表达“先判断什么、再根据结果做什么”。