在本教程中，我们使用 Jev，TypeSafe AI 的第一个 System One 模型，它完全不生成文本：我们向它发送一段程序状态和一组带类型的问题，它返回选项、分数和是/否概率，我们的代码可以直接据此分支。我们安装官方 Python SDK，进行第一次调用，同时使用全部三种问题原语，并观察状态的形状如何改变模型所能知道的内容。然后我们从返回的概率重新计算已发布的置信度统计量，测量将十个问题批量合并为一次调用相比十次单独调用所获得的收益，并构建 API 所设计的模式：置信度门控路由、权重保留在代码中的复合评分、带类型的函数调用，以及以模型实际能做到的方式进行计数。我们以生产形态收尾：Pydantic 响应模型、用 asyncio 扇出的异步客户端、重试策略、带类型的错误，以及一个为整个 notebook 定价的运行账本。
复制代码已复制使用其他浏览器
import os
import sys
import json
import time
import asyncio
import traceback
import subprocess
from getpass import getpass
RESULTS = {}
LEDGER = {"calls": 0, "input_tokens": 0, "output_tokens": 0}
USD_PER_MILLION_INPUT_TOKENS = 0.042 # Jev 标价；输出 token 免费
def banner(title):
print("\n" + "=" * 78)
print(title)
print("=" * 78)
def section(name):
def wrap(fn):
def run(*a, **kw):
banner(name)
try:
out = fn(*a, **kw)
RESULTS[name] = out if isinstance(out, str) else "ok"
return out
except Exception as e:
RESULTS[name] = f"SKIPPED / FAILED -> {type(e).__name__}: {e}"
print(f"\n[!] {name} did not complete: {type(e).__name__}: {e}")
traceback.print_exc(limit=3)
return None
return run
return wrap
banner("0. Install the SDK, load the API key, list the models")
subprocess.run([sys.executable, "-m", "pip", "install", "-q", "typesafe-sdk==0.7.0"], check=True)
import typesafe_sdk
from typesafe_sdk import Choice, Noul, Score, TypeSafeClient
def load_api_key():
key = os.environ.get("TYPESAFE_API_KEY", "").strip()
if not key:
try:
from google.colab import userdata # Colab：密钥存储在 Secrets 标签页下
key = (userdata.get("TYPESAFE_API_KEY") or "").strip()
except Exception:
key = ""
return key or getpass("TypeSafe API key (console.typesafe.ai/keys): ").strip()
os.environ["TYPESAFE_API_KEY"] = load_api_key()
client = TypeSafeClient() # reads TYPESAFE_API_KEY, defaults to jev-latest
print(f" typesafe-sdk {typesafe_sdk.__version__} | Python {sys.version.split()[0]}")
print(" models available to this key:")
for m in client.models.list().models:
print(f" {m.name:<14s} released {m.release_date} {m.description}")
def ask(state, questions, **kw):
"""One System One call, timed, with its tokens added to the running ledger."""
t0 = time.perf_counter()
response = client.system_one(state, questions, **kw)
ms = (time.perf_counter() - t0) * 1e3
LEDGER["calls"] += 1
LEDGER["input_tokens"] += response.usage.input_tokens or 0
LEDGER["output_tokens"] += response.usage.output_tokens or 0
return response, ms
我们安装 typesafe-sdk，并固定到本 notebook 编写时所对应的版本，然后从环境变量、Colab 的 Secrets 标签页或隐藏提示中加载 API key，这样它就不会出现在 notebook 中。TypeSafeClient 会自行读取 TYPESAFE_API_KEY，并默认使用 jev-latest 别名；列出模型可以显示该 key 能使用哪些名称和固定版本。小小的 ask 辅助函数封装了 system_one，这样 notebook 后续的每次调用都会被计时，其 token 用量也会记入一个账本，我们最后会对其进行汇总。
Copy CodeCopiedUse a different Browser
TICKET = {
"ticket": {
"subject": "Duplicate charge",
"messages": [
{"from": "customer", "text": "I was charged twice for order A-104. This is the second time "
"this year. Please refund the duplicate today."},
{"from": "support", "text": "We are checking the charges."},
],
},
"order": {"id": "A-104", "charges": [{"amount_usd": 49, "status": "captured"},
{"amount_usd": 49, "status": "captured"}]},
"refund_policy": "Duplicate charges are eligible for a full refund within 30 days.",
}
@section("1. Three primitives, one call: Choice, Score, Noul")
def three_primitives():
response, ms = ask(TICKET, {
"department": Choice(
instructions="Which team should handle this ticket",
criteria={"billing": "支付、退款或订阅问题",
"technical": "Bug、故障或集成问题",
"sales": "定价、套餐或账户升级"},
),
"frustration": Score(
instructions="客户在 `ticket.messages[0].text` 中表现出的沮丧程度",
criteria=["平静，只是陈述事实", "沮丧但保持礼貌", "非常愤怒，言辞激烈"],
),
"refund_requested": Noul(instructions="客户明确要求退款"),
"policy_supports": Noul(instructions="所述的 `refund_policy` 涵盖此情况"),
})
dept = response.choices["department"]
print(f" department -> {dept.choice!r} confidence {dept.confidence:.3f}")
print(f" probabilities {({k: round(v, 3) for k, v in dept.probabilities.items()})}")
fr = response.scores["frustration"]
print(f" frustration -> score {fr.score:.3f} on 0..{len(fr.legend) - 1} confidence {fr.confidence:.3f}")
for level, text in fr.legend.items():
print(f" {level}: p={fr.probabilities[level]:.3f} {text}")
print(f" refund_requested -> noul {response.nouls['refund_requested'].noul:.3f}")
print(f" policy_supports -> noul {response.nouls['policy_supports'].noul:.3f}")
print(f"\n answered by {response.model} in {ms:.0f} ms "
f"input tokens {response.usage.input_tokens}, output tokens {response.usage.output_tokens}")
return f"{dept.choice}, frustration {fr.score:.2f}, refund {response.nouls['refund_requested'].noul:.2f}"
three_primitives()
System One 请求包含两部分：state，即任何描述情境的文本、JSON 对象或数组，以及一个命名问题的字典。Choice 从我们定义的标准中选择一个标签，并为每个标签返回一个概率；Score 将 state 置于一个有序评分标准上，并返回概率加权的等级，因此它可以落在两个等级之间；Noul 返回一个表示某陈述为真的单一概率。问题名称是我们自己起的，永远不会传给模型，这就是为什么指令承载了全部含义，并且可以用反引号路径指向嵌套字段。所有四个问题都在一次请求中评估，彼此并行且相互隔离，响应会报告作答的固定模型版本以及计费的 token 数。
复制代码已复制使用其他浏览器
@section("2. State 即程序状态：同一个问题作用于字符串和作用于命名字段")
def state_shapes():
question = {"eligible": Noul(
instructions="根据公司书面政策，该客户符合退款条件",
criteria={"true": "存在一项政策且涵盖该客户的情况",
"false": "未提供政策，或政策不涵盖该情况"},
)}
bare = "订单 A-104 被重复扣费了两次。请退还重复收取的款项。"
as_list = [m["text"] for m in TICKET["ticket"]["messages"]]
shapes = [("string: the message only", bare),
("array : the conversation", as_list),
("object: ticket + order + policy", TICKET)]
print(f" {'state shape':<34s} {'noul':>6s} input tokens ms")
seen = {}
for label, state in shapes:
response, ms = ask(state, question)
seen[label] = response.nouls["eligible"].noul
print(f" {label:<34s} {seen[label]:6.3f} {response.usage.input_tokens:12d} {ms:5.0f}")
print("\n Only the object carries the policy and the two captured charges; the question")
print(" is identical in all three calls, so any movement comes from the state.")
return "noul by state shape: " + ", ".join(f"{v:.2f}" for v in seen.values())
state_shapes()
状态是模型唯一知道的东西，所以我们问一个问题——根据公司书面政策，客户是否有资格获得退款——跨越三种状态形态。裸字符串只包含投诉，别无其他；数组增加了对话；JSON 对象增加了订单及其两笔捕获的费用以及退款政策本身。问题从不改变，因此返回概率中出现的任何差异都可归因于状态，而 token 列显示了额外上下文的成本。当上下文有多个部分时，命名字段是文档推荐的方案，因为指令随后可以按名称引用它们。
复制代码已复制使用其他浏览器
def confidence_from(probabilities):
"""TypeSafe 发布的统计量：(count x peak - 1) / (count - 1)。"""
p = list(probabilities.values())
return (len(p) * max(p) - 1) / (len(p) - 1)
@section("3. 置信度是分布的统计量，你可以重新计算它")
def confidence_math():
tone = Choice(instructions="消息的语气是什么",
criteria={"angry": "愤怒或敌对", "calm": "中立或礼貌", "excited": "热情或急切"})
urgency = Score(instructions="这需要多快得到关注",
criteria=["可以等待", "本周需要关注", "今天需要关注"])
messages = {
"clear ": "这是本周第三次故障，没人回应。现在就修好，否则我今天就取消。",
"ambiguous": "嗯。这确实是一次经历。有空的时候告诉我。",
}
print(f" {'message':<10s} {'choice':<8s} {'API conf':>8s} {'recomputed':>11s} "
f"{'score':>6s} {'sum(level*p)':>13s} {'API conf':>9s}")
worst = 1.0
for label, text in messages.items():
response, _ = ask(text, {"tone": tone, "urgency": urgency})
t, u = response.choices["tone"], response.scores["urgency"]
expected = sum(level * p for level, p in u.probabilities.items())
print(f" {label:<10s} {t.choice:<8s} {t.confidence:8.3f} {confidence_from(t.probabilities):11.3f} "
f"{u.score:6.3f} {expected:13.3f} {u.confidence:9.3f}")
worst = min(worst, t.confidence)
print("\n Noul 没有 confidence 字段：它的值已经是 yes 的概率，")
print(" 所以 0.5 表示未决定，而不是中等。")
return f"lowest tone confidence {worst:.2f}"
confidence_math()
TypeSafe 将 confidence 记录为从答案本身已包含的分布中计算出的统计量：选项数量乘以峰值概率，减一，再除以选项数量减一。我们从 Choice 的概率重新计算它，并与 confidence 字段进行比较，同时将 Score 重新计算为每个级别乘以其概率的总和。用同样两个问题分别跑一条直白的消息和一条刻意含糊的消息，就能看出分布——以及由此决定的 confidence——如何对模糊性作出反应。Noul 完全不携带 confidence 字段，因为它的值本身就是 yes 的概率，而接近 0.5 的值意味着未决定，而非中等。
复制代码已复制请使用其他浏览器
POSTMORTEM = """事件 2291 - 结账延迟，3 月 14 日。09:12 UTC，支付网关开始超时
大约 18% 的欧盟区域结账请求。值班工程师于 09:15 收到寻呼，
并于 09:21 确认。最初的怀疑落在前一晚部署的新欺诈评分服务上，
该服务于 09:40 回滚，但没有改善。10:05，数据库团队发现连接池
上限被自动配置同步从 400 降低到 40，该同步静默覆盖了
手动覆盖设置。该上限于 10:11 恢复，错误率在 10:19 回到基线。客户
影响：3,420 次结账失败，估计 61,000 美元的收入延迟；没有数据丢失，也没有
客户数据泄露。事件期间未通知客户；状态页面于
10:30 恢复后更新。后续行动：对连接池饱和发出告警，要求对配置同步覆盖进行审查，
并将状态页面更新加入值班检查清单的前十五分钟。"""
FANOUT = {
"root_cause": Choice(instructions="事件的根本原因是什么",
criteria={"bad_deploy": "有缺陷的代码或服务部署",
"config_change": "不正确的配置值",
"capacity": "自然流量超出了预置容量",
"third_party": "外部供应商的故障",
"unknown"："文本未说明原因"
"detected_by"：Choice(instructions="事件最初是如何被发现的"
criteria={"alerting"："自动监控或告警","customer"："客户报告"
"employee"："员工偶然发现","unclear"："未说明"})
"severity"：Score(instructions="客户影响的严重程度"
criteria=["没有客户可见的影响","少数客户出现轻微降级"
"核心流程对相当一部分客户失效"
"核心流程对大多数客户完全中断"])
"comms_quality"：Score(instructions="事件期间客户沟通的质量"
criteria=["事件发生时就及时通知了客户"
"通知了客户，但太迟"
"客户仅在恢复后才被通知，或从未被通知"])
"data_exposed"：Noul(instructions="客户数据被暴露或泄露")
"rollback_helped"：Noul(instructions="回滚欺诈评分服务解决了该事件")
"human_error"：Noul(instructions="人为的手动失误直接导致了该事件")
"has_followups"：Noul(instructions="文本列出了具体的后续行动")
"revenue_lost"：Noul(instructions="收入是永久损失，而非延迟")
"eu_only"：Noul(instructions="影响仅限于 EU 地区")
}
def value_of(answer):
for field in ("choice", "score", "noul"): # 0.0 分是真实值，不是缺失
if hasattr(answer, field):
return getattr(answer, field)
@section("4. 推测性扇出：一次调用问十个问题 vs 十次调用")
def fan_out():
batched, batched_ms = ask({"postmortem": POSTMORTEM}, FANOUT)
batched_tokens = batched.usage.input_tokens
seq_ms, seq_tokens, agree = 0.0, 0, 0
print(f" {'question':<16s} {'one call':>10s} {'own call':>10s}")
for name, q in FANOUT.items():
single, ms = ask({"postmortem": POSTMORTEM}, {name: q})
seq_ms, seq_tokens = seq_ms + ms, seq_tokens + single.usage.input_tokens
a, b = value_of(batched.answers[name]), value_of(single.answers[name])
same = a == b if isinstance(a, str) else abs(a - b) < 0.05
agree += same
fmt = (lambda v: f"{v:>10s}") if isinstance(a, str) else (lambda v: f"{v:10.3f}")
print(f" {name:<16s} {fmt(a)} {fmt(b)} {'same' if same else 'differs'}")
print(f"\n one call : {batched_ms:7.0f} ms {batched_tokens:6d} input tokens")
print(f" ten calls: {seq_ms:7.0f} ms {seq_tokens:6d} input tokens")
print(f" -> {seq_ms / batched_ms:.1f}x faster and {seq_tokens / batched_tokens:.1f}x fewer tokens; "
f"{agree}/{len(FANOUT)} answers agree, because questions never see each other")
return f"{seq_ms / batched_ms:.1f}x faster, {seq_tokens / batched_tokens:.1f}x cheaper, {agree}/{len(FANOUT)} agree"
fan_out()
因为一个请求中的问题彼此不可见，我们可以一次性把可能需要的一切都问出来，包括只在某一个分支上才重要的问题，之后再只读取相关的答案。我们把关于一次事故复盘的十个问题——两个 Choice、两个 Score 和六个 Noul——放进一次调用，然后再各自单独调用一次，比较墙钟时间、输入 token 和答案。状态只发送一次而不是十次，延迟和 token 的节省都来自这里，而一致性那一列直接检验了隔离性这一主张：一个问题无论是否与其他问题同行，都应该得到相同的答案。
Copy CodeCopiedUse a different Browser
INTENT = Choice(
instructions="What the user wants the banking assistant to do",
criteria={"check_balance": "See a balance or recent transactions",
"approve_transfer": "Send or approve a transfer of money",
"dispute_charge": "Contest a charge they do not recognise",
"close_account": "Close the account permanently",
"other": "Anything else, or not clear enough to act on"},
)
STAKES = {"check_balance": 0.50, "dispute_charge": 0.70, "approve_transfer": 0.85, "close_account": 0.90}
def route(answer):
if answer.choice == "other" or answer.confidence < 0.50:
return "-> human"
bar = STAKES[answer.choice]
return f"-> run {answer.choice}" if answer.confidence >= bar else f"-> confirm first (needs {bar:.2f})"
@section("5. Confidence-gated routing: the bar rises with the stakes")
def gated_routing():
inbox = ["how much is in my checking account",
"send 2,000 to my landlord like last month",
"i guess maybe move some money around? not sure",
"有一笔 89.99 的扣费来自一家我从没加入过的健身房"
"把一切都关掉，我跟这家银行彻底完了"
"里斯本的天气怎么样"]
print(f" {'message':<50s} {'intent':<17s} {'conf':>5s} decision")
acted = 0
for text in inbox:
response, _ = ask(text, {"intent": INTENT})
a = response.choices["intent"]
decision = route(a)
acted += decision.startswith("-> run")
print(f" {text[:50]:<50s} {a.choice:<17s} {a.confidence:5.2f} {decision}")
print(f"\n thresholds live in code: {STAKES}")
return f"{acted}/{len(inbox)} messages acted on automatically"
gated_routing()
只有当类型化答案周围的代码把某个动作需要多大确定性编码进去时，它们才真正有意义。我们把每条消息分类为一个 intent，并依据两件事来路由：intent 本身，以及它的置信度是否越过一条随风险上升的门槛——从读取余额的 0.5 到关闭账户的 0.9。任何被归类为 other、或低于 0.5 的，都转给人工；一个被识别出但低于其门槛的 intent，会先与用户确认。这些阈值就是普通的 Python 值，因此风险容忍度像其他任何代码一样被审查、版本化和测试，而不是埋在 prompt 里。
Copy CodeCopiedUse a different Browser
DIMENSIONS = {
"python_depth": Score(instructions="Depth of hands-on Python engineering experience", criteria=[
"No Python mentioned", "Scripts or notebooks only", "Ships production Python services",
"Designs Python libraries or frameworks used by others"]),
"ml_systems": Score(instructions="Experience running machine learning systems in production", criteria=[
"None mentioned", "Trained models offline only", "Deployed and monitored models in production",
"Owned large-scale training or serving infrastructure"]),
"leadership": Score(instructions="Evidence of leading people or projects", criteria=[
"None mentioned", "Mentored individuals", "Led a project or a small team",
"Managed several teams or an organisation"]),
"communication": Score(instructions="Evidence of clear written or public communication", criteria=[
"None mentioned", "Internal docs only", "Public posts or talks", "Widely read writing or major conference talks"]),
}
CANDIDATES = {
"Asha"："八年 Python 经验；维护一个拥有 4k stars 的开源数据验证库。"
在一家银行部署过欺诈模型并负责其监控。指导两名初级员工。撰写技术博客。
"Bruno"："三个团队（22 人）的工程经理。写了十年 Java，会一些 Python 脚本。"
赞助了公司的 ML 平台，但并未参与构建。在两次行业会议上做过主题演讲。
"Chen"："统计学博士；在 notebook 中训练模型，没有生产部署。用 Python 做分析。"
担任两门课程的助教。若干内部报告。
"Dara"："用 Python 构建并负责一个每秒 4 万请求的推荐系统服务基础设施
和 C++。带领一个五人平台团队。仅有内部设计文档。
}
WEIGHTS = {"senior IC": {"python_depth": .40, "ml_systems": .40, "leadership": .05, "communication": .15},
"team lead": {"python_depth": .15, "ml_systems": .25, "leadership": .45, "communication": .15}}
@section("6. 综合评分：来自模型的原子判断，来自代码的权重")
def composite_scoring():
table = {}
for name, bio in CANDIDATES.items():
response, _ = ask({"candidate_bio": bio}, DIMENSIONS)
table[name] = {d: response.scores[d].score / (len(q.criteria) - 1) for d, q in DIMENSIONS.items()}
print(f" {'':<7s}" + "".join(f"{d:>15s}" for d in DIMENSIONS) + " (each normalised to 0..1)")
for name, row in table.items():
print(f" {name:<7s}" + "".join(f"{row[d]:15.2f}" for d in DIMENSIONS))
winners = {}
for role, w in WEIGHTS.items():
ranked = sorted(table, key=lambda n: -sum(w[d] * table[n][d] for d in w))
winners[role] = ranked[0]
print(f"\n ranking for {role:<10s}: " +
" > ".join(f"{n} {sum(w[d] * table[n][d] for d in w):.2f}" for n in ranked))
print("\n Two rankings, four model calls: changing the weights re-ran no inference.")
return ", ".join(f"{role}: {who}" for role, who in winners.items())
composite_scoring()
复合评分让模型的任务保持狭窄，让策略保持明确。对于每个候选人，我们提出四个 Score 问题，每个问题描述具体情境而非程度，将每个分数按其最高等级归一化，并存储结果表格。排名随后就是简单的算术：一个权重向量用于高级个人贡献者，另一个用于团队负责人。由于判断与权重分开存储，改变我们看重的因素会立即重新排列候选人，且无需推理。你可以将排名中的每个位置追溯到产生它的维度。
复制代码已复制使用其他浏览器
ROOMS = {"living_room": None, "bedroom": None, "kitchen": None, "office": None}
def set_lights(room, state):
return f"lights in {room} -> {state}"
def set_thermostat(room, mode):
return f"thermostat in {room} -> {mode}"
def play_music(room, genre):
return f"playing {genre} in {room}"
TOOLS = {"set_lights": (set_lights, "state"), "set_thermostat": (set_thermostat, "mode"),
"play_music": (play_music, "genre")}
CALL_SPEC = {
"tool": Choice(instructions="Which smart-home function the command asks for",
criteria={"set_lights": "Turn lights on, off, or dim them",
"set_thermostat": "Make a room warmer, cooler, or set eco mode",
"play_music": "Play music or audio",
"none": "Not a smart-home command this system supports"}),
"room": Choice(instructions="Which room the command refers to", criteria=ROOMS),
"state": Choice(instructions="If this is a lights command: the requested light state",
criteria={"on": None, "off": None, "dim": None}),
"mode": Choice(instructions="If this is a thermostat command: the requested mode",
criteria={"heat": "Warmer", "cool": "Cooler", "eco": "Energy saving"}),
"genre": Choice(instructions="If this is a music command: the requested genre",
criteria={"jazz": None, "classical": None, "rock": None, "ambient": None}),
}
@section("7. Typed function calling, and counting the way Jev can do it")
def function_calling():
commands = ["it's freezing in the office, warm it up", "kill the lights in the bedroom",
"put on something mellow and jazzy in the kitchen", "order me a pizza"]
dispatched = 0
for text in commands:
response, ms = ask(text, CALL_SPEC) # 每个参数都投机性地询问，一次调用
c = response.choices
tool = c["tool"].choice
if tool == "none":
print(f" {text!r:<52s} -> no tool (confidence {c['tool'].confidence:.2f})")
continue
fn, arg = TOOLS[tool]
weakest = min(c["tool"].confidence, c["room"].confidence, c[arg].confidence)
print(f" {text!r:<52s} -> {tool}(room={c['room'].choice!r}, {arg}={c[arg].choice!r}) "
f"weakest judgment {weakest:.2f}, {ms:.0f} ms")
print(f" {'':<52s} {fn(c['room'].choice, c[arg].choice)}")
dispatched += 1
basket = ["mango", "spanner", "kiwi", "router", "plum", "stapler", "fig", "lychee"]
response, _ = ask({"items": basket},
{f"item_{i}": Noul(instructions=f"`items[{i}]` is the name of a fruit") for i in range(len(basket))})
probs = [response.nouls[f"item_{i}"].noul for i in range(len(basket))]
print("\n counting: one Noul per item, summed in code (Jev does not count reliably in one question)")
print(" " + " ".join(f"{item}={p:.2f}" for item, p in zip(basket, probs)))
count = sum(p > 0.5 for p in probs)
print(f" fruits counted: {count} of {len(basket)}")
return f"{dispatched}/{len(commands)} commands dispatched from typed answers; counted {count} fruits"
function_calling()
函数调用变成了一组封闭集问题：一个 Choice 选择工具，其中包含一个显式的 none 选项，用于我们不支持的命令；每个参数对应一个 Choice，在同一次请求中投机性地询问。代码只读取属于所选工具的参数，将最弱的判断报告为整个调用的置信度，然后用经过验证的枚举值执行一个普通的 Python 函数。后半部分应用了一个有文档记录的变通方法：Jev 在单个问题内无法可靠地计数，所以我们在一次请求中为每个条目询问一个 Noul，并在代码中求和。
Copy CodeCopiedUse a different Browser
import concurrent.futures
from typesafe_sdk import (AsyncTypeSafeClient, ChoiceAnswer, NoulAnswer, RetryPolicy, ScoreAnswer,
SystemOneResponse, TypeSafeAPIError, TypeSafeError)
class TicketDecision(SystemOneResponse):
"""声明你期望的答案，并以属性形式读取它们，由 Pydantic 验证。"""
department: ChoiceAnswer
frustration: ScoreAnswer
refund_requested: NoulAnswer
TRIAGE = {
"department": Choice(instructions="Which team should handle this ticket",
criteria={"billing": "Payment, refund or subscription issues",
"technical": "Bugs, outages or integration problems",
"sales": "Pricing, plans or account upgrades"}),
"frustration": Score(instructions="How frustrated the customer appears",
criteria=["Calm, just stating facts", "Frustrated but civil", "Very angry, strong language"]),
"refund_requested": Noul(instructions="The customer is explicitly asking for a refund"),
}
QUEUE = ["My invoice shows two seats but I only have one user.", "The export button does nothing in Safari.",
"Can I get a discount if I pay annually?", "Your API returns 500 on every request since this morning!!",
"I want my money back for last month, the product never worked.", "How do I add a teammate?",
"Webhooks stopped firing after your update.", "Do you offer a plan for nonprofits?",
"Charged after I cancelled. Refund this immediately.", "The dashboard is slow but usable.",
"Is there an on-prem version?", "Login emails never arrive."]
def run_async(coro):
"""Works in a plain script and inside Jupyter/Colab, where an event loop is already running."""
try:
asyncio.get_running_loop()
except RuntimeError:
return asyncio.run(coro)
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
return pool.submit(asyncio.run, coro).result()
async def triage_all(tickets):
retry = RetryPolicy(max_retries=3, backoff_initial=0.5, backoff_max=4.0, timeout=20.0)
async with AsyncTypeSafeClient(retry=retry, timeout=10.0) as aclient:
t0 = time.perf_counter()
results = await asyncio.gather(*(aclient.system_one(t, TRIAGE, response_model=TicketDecision)
for t in tickets))
return results, (time.perf_counter() - t0) * 1e3
@section("8. Production shape: typed response models, async fan-out, retries, errors")
def production():
results, wall_ms = run_async(triage_all(QUEUE))
for r in results:
LEDGER["calls"] += 1
LEDGER["input_tokens"] += r.usage.input_tokens or 0
LEDGER["output_tokens"] += r.usage.output_tokens or 0
print(f" {len(QUEUE)} tickets triaged concurrently in {wall_ms:.0f} ms wall time "
f"({wall_ms / len(QUEUE):.0f} ms per ticket amortised)\n")
print(f" {'ticket':<58s} {'department':<10s} {'frustr.':>7s} {'refund':>7s}")
for text, r in zip(QUEUE, results): # 属性访问，无需字典查找，无需解析
print(f" {text[:58]:<58s} {r.department.choice:<10s} {r.frustration.score:7.2f} {r.refund_requested.noul:7.2f}")
print("\n 错误也是有类型的：")
try:
client.system_one("anything", {})
except TypeSafeError as e:
print(f" 空问题，在任何请求发出前就被捕获 : {type(e).__name__}: {e}")
try:
client.system_one("anything", {"q": Noul(instructions="Is this a test")}, model="jev-does-not-exist",
retry=RetryPolicy(max_retries=0))
except TypeSafeAPIError as e:
print(f" 未知模型，被 API 拒绝 : {type(e).__name__} (HTTP {e.status})")
refunds = sum(r.refund_requested.noul > 0.5 for r in results)
return f"{len(QUEUE)} tickets in {wall_ms:.0f} ms; {refunds} refund requests flagged"
production()
四个细节让这些示例成为一个决策服务。继承 SystemOneResponse 并声明我们期望的答案，就得到了由 Pydantic 验证的属性访问，因此类型化的决策在进入应用程序后始终保持类型化，而不会变成字典查找。由于每个请求都是独立的，一个工单队列就是一个独立决策的队列：AsyncTypeSafeClient 配合 asyncio.gather 并发发送它们，而 run_async 辅助函数让同一份代码既能在脚本中运行，也能在事件循环已在运行的 notebook 中运行。RetryPolicy 限定了每次调用的重试次数、退避和总时间预算。错误也是有类型的：空问题集在任何请求发出前就被拒绝，未知模型名会从 API 返回为携带 HTTP 状态的 TypeSafeAPIError 子类。
Copy CodeCopiedUse a different Browser
banner("SUMMARY")
for name, res in RESULTS.items():
print(f" {name:<86s} {res}")
cost = LEDGER["input_tokens"] / 1e6 * USD_PER_MILLION_INPUT_TOKENS
client.close()
print(f"\n 整个教程：{LEDGER['calls']} calls, {LEDGER['input_tokens']:,} input tokens, "
f"{LEDGER['output_tokens']:,} output tokens (free) -> about ${cost:.5f}")
print("""
接下来该往哪里走
- Patterns：docs.typesafe.ai/patterns（扇出、置信度路由、复合评分、意图路由）
- Cookbooks：重排序、RAG 段落过滤、引用检查、LLM 护栏、层次分类
- jev-1.13 已知的粗糙边缘：docs.typesafe.ai/model-jaggedness/jev-1.13（字面读取、算术、
计数、日期比较、大量无关状态）
- 在相同问题上与 LLM 对比：github.com/typesafe-ai/system-one-adapter-python
- 为生产环境固定版本：TypeSafeClient(model="jev-1.13.0")；response.model 报告实际作答的模型
""")
摘要会打印每个部分返回的单行结果，然后汇总每次调用一直在累积的账本：请求数、输入和输出 token 数，以及按公布的输入价格计算的成本，其中输出 token 免费。
总之，我们按照 Jev 本应被使用的方式来使用它：作为代码可组合的小型、带类型的判断来源，而不是一个需要提示和解析的文本生成器。每个答案都以标签、等级或附带分布的概率形式返回，让我们可以在 Python 中设置阈值、权重和路由规则，并对其进行测试。将问题批量处理在共享状态上，既降低了延迟也减少了 token，因为状态对每个条目只传输一次；Nouls 替代了模型不可信地做出的计数，而封闭集合 Choices 将自然语言命令转化为经过验证的函数调用。生产组件——带类型的响应模型、异步客户端、重试策略和带类型的错误——都很小，而账本为整个 notebook 定价。剩下的部分是任何 SDK 都无法替我们完成的：在信任它们执行真实操作之前，用我们自己的数据评估问题、标准和阈值。
在这里查看完整代码。所有功劳归于该项目的研究者。另外，欢迎在 Twitter 上关注我们，别忘了加入我们 15 万+ 的 ML SubReddit 并订阅我们的 Newsletter。等等！你在 Telegram 上吗？现在你也可以在 Telegram 上加入我们了。
需要与我们合作推广你的 GitHub 仓库、Hugging Face 页面、产品发布或网络研讨会等？联系我们
文章《A Coding Guide to TypeSafe AI Jev: Typed Decisions, Calibrated Confidence, and Speculative Fan-Out with a System One Model》首次发布于 MarkTechPost。