TypeSafe AI Jev 编程指南：用 System One 模型实现类型化决策、校准置信度和投机式扩散

人工智能

在本教程中，我们使用 TypeSafe AI 的首个 System One 模型 Jev，它完全不生成文本：我们向它发送一段程序状态和一組类型化问题，它返回选择、评分和是/否概率，我们的代码可以直接基于这些结果进行分支判断。我们安装官方 Python SDK，发出第一次调用——同时使用三种问题原语，并观察状态的结构如何影响模型能够感知的信息。然后我们根据返回的概率重新计算已发布的置信度统计量，测量将十个问题批量放入一次调用相比十次单独调用能带来多少收益，并构建该 API 设计用来支持的模式：基于置信度的路由、带代码权重的复合评分、类型化函数调用，以及以模型真正能做到的方式计数。最后我们给出生产环境形态：Pydantic 响应模型、通过 asyncio 扩散的异步客户端、重试策略、类型化错误，以及一个记录整本笔记本消耗的账本。

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: key stored under the Secrets tab
            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，固定到编写本笔记本时的版本，并从环境变量、Colab 的 Secrets 标签页或隐藏提示中加载 API key，这样它就不会出现在笔记本中。TypeSafeClient 自己读取 TYPESAFE_API_KEY，默认使用 jev-latest 别名；列出模型可以查看该 key 能使用的名称和固定版本。小的 ask 辅助函数包装了 system_one，使笔记本中后续每个调用都被计时，消耗的 token 会进入我们在最后汇总的账本。

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": "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 in `ticket.messages[0].text`",
            criteria=["Calm, just stating facts", "Frustrated but civil", "Very angry, strong language"],
        ),
        "refund_requested": Noul(instructions="The customer is explicitly asking for a refund"),
        "policy_supports": Noul(instructions="The stated `refund_policy` covers this situation"),
    })
 
    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 从我们定义的 criteria 中选择一个标签，并返回每个标签的概率；Score 将状态放在一个有序的评分标准上，返回概率加权的级别，因此它可以落在两个级别之间；Noul 返回一个语句为真的单一概率。问题名称是我们自己定义的，永远不会到达模型，这就是为什么 instructions 携带完整含义，并可以用反引号路径指向嵌套字段。所有四个问题在一次请求中并行评估，彼此隔离，响应报告回答问题的固定模型版本及其计费的 token。

@section("2. State is program state: the same question over a string and over named fields")
def state_shapes():
    question = {"eligible": Noul(
        instructions="The customer is eligible for a refund under the company's written policy",
        criteria={"true": "A policy is present and it covers the customer's situation",
                  "false": "No policy is given, or the policy does not cover the situation"},
    )}
    bare = "I was charged twice for order A-104. Please refund the duplicate."
    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. Confidence 是分布的统计量，你可以重新计算它")
def confidence_math():
    tone = Choice(instructions="What is the tone of the message",
                  criteria={"angry": "Upset or hostile", "calm": "Neutral or polite", "excited": "Enthusiastic or eager"})
    urgency = Score(instructions="How soon this needs attention",
                    criteria=["Can wait", "Needs attention this week", "Needs attention today"])
    messages = {
        "clear    ": "This is the third outage this week and nobody answers. Fix it NOW or I cancel today.",
        "ambiguous": "Well. That was certainly an experience. Let me know when you get a chance.",
    }
    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  A Noul has no confidence field: its value already is the probability of yes,")
    print("  so 0.5 means undecided, not medium.")
    return f"lowest tone confidence {worst:.2f}"


confidence_math()

TypeSafe 将 confidence 文档化为从答案已包含的分布中计算出的统计量：选项数乘以峰值概率，减去一，再除以选项数减一。我们从一个 Choice 的概率中重新计算它，并与 confidence 字段进行比较，同时将 Score 重新计算为每个 level 及其概率的乘积之和。用同一个两条问题分别处理一条措辞直白的消息和一条故意模糊的消息，可以看出分布——进而 confidence——如何响应歧义。A Noul 完全不携带 confidence 字段，因为它的值本身就是 yes 的概率，接近 0.5 的值表示未决定而非中等。

POSTMORTEM = """Incident 2291 - checkout latency, 14 March. At 09:12 UTC the payments gateway began timing out
for roughly 18 percent of checkout requests in the EU region. The on-call engineer was paged at 09:15 and
acknowledged at 09:21. Initial suspicion fell on the new fraud-scoring service deployed the previous evening,
and it was rolled back at 09:40 with no improvement. At 10:05 the database team found that a connection pool
limit had been lowered from 400 to 40 by an automated configuration sync, which had silently overwritten a
manual override. The limit was restored at 10:11 and error rates returned to baseline by 10:19. Customer
impact: 3,420 failed checkouts and an estimated 61,000 USD in delayed revenue; no data was lost and no
customer data was exposed. Customers were not notified during the incident; the status page was updated at
10:30, after recovery. Follow-ups: alert on pool saturation, require review for configuration-sync overrides,
and add the status page update to the first fifteen minutes of the on-call checklist."""

FANOUT = {
    "root_cause": Choice(instructions="What was the root cause of the incident",
                         criteria={"bad_deploy": "A faulty code or service deployment",
                                   "config_change": "An incorrect configuration value",
                                   "capacity": "Organic traffic exceeded provisioned capacity",
                                   "third_party": "A failure at an external vendor",
                                   "unknown": "The text does not establish a cause"}),
    "detected_by": Choice(instructions="How the incident was first detected",
                          criteria={"alerting": "Automated monitoring or paging", "customer": "Customer reports",
                                    "employee": "An employee noticed by chance", "unclear": "Not stated"}),
    "severity": Score(instructions="Severity of customer impact",
                      criteria=["No customer-visible impact", "Minor degradation for a few customers",
                                "A core flow failed for a meaningful share of customers",
                                "Full outage of a core flow for most customers"]),
    "comms_quality": Score(instructions="Quality of customer communication during the incident",
                           criteria=["Customers were informed promptly while it was happening",
                                     "Customers were informed, but late",
                                     "Customers were only informed after recovery, or never"]),
    "data_exposed": Noul(instructions="Customer data was exposed or leaked"),
    "rollback_helped": Noul(instructions="Rolling back the fraud-scoring service resolved the incident"),
    "human_error": Noul(instructions="A person making a manual mistake directly caused the incident"),
    "has_followups": Noul(instructions="The text lists concrete follow-up actions"),
    "revenue_lost": Noul(instructions="Revenue was permanently lost, as opposed to delayed"),
    "eu_only": Noul(instructions="The impact was limited to the EU region"),
}


def value_of(answer):
    for field in ("choice", "score", "noul"):              # a score of 0.0 is a real value, not a miss
        if hasattr(answer, field):
            return getattr(answer, field)


@section("4. 推测式广播：一次调用问十个问题与十次调用")
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.0

因为一个请求中的问题彼此不可见，我们可以提前询问所有可能需要的内容，包括那些仅在某一分支才需要的问题，之后再读取相关答案。我们将关于一份事故复盘报告的十个问题（两个 Choice、两个 Score 和六个 Noul）打包进一次调用，然后分别对每个问题单独调用一次，并比较运行时间、输入 token 和答案。状态只发送一次而不是十次，这就是延迟和 token 节省的来源，而 agreement 列直接验证了隔离声明：一个问题的答案不应因其与其他问题同行而有所不同。

5. 置信度门控路由：门槛随风险级别上升

类型化答案只有在周围代码明确指定了动作所需确定程度时才有意义。我们将每条消息分类到某个意图，然后根据两个因素进行路由：该意图本身，以及其置信度是否超过了随风险级别上升的门槛——查看余额为 0.5，关闭账户为 0.9。任何被分类为 other（其他），或置信度低于 0.5 的消息，都会转给人工处理；识别的意图但未超过其门槛的，会先与用户确认。阈值是普通的 Python 值，因此风险容忍度像其他代码一样被审查、版本化管理，而不是埋藏在提示词中。

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":  "Eight years of Python; maintains an open-source data validation library with 4k stars. "
             "Deployed fraud models at a bank and ran their monitoring. Mentors two juniors. Writes a technical blog.",
    "Bruno": "Engineering manager for three teams (22 people). Wrote Java for a decade, some Python scripting. "
             "Sponsored the company's ML platform but did not build it. Keynoted two industry conferences.",
    "Chen":  "PhD in statistics; trains models in notebooks, no production deployments. Python for analysis. "
             "Teaching assistant for two courses. Several internal reports.",
    "Dara":  "Built and owned the serving infrastructure for a recommender at 40k requests per second in Python "
             "and C++. Led a five-person platform team. Internal design docs only.",
}
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. Composite scoring: atomic judgments from the model, weights from code")
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()

6. 复合评分：模型的原子判断与代码中的权重

复合评分将模型的工作范围限定得很窄，同时让策略保持显式。对于每个候选人，我们提出四个 Score 问题，每个问题描述的是具体情境而非程度，对每个评分按其最高级别进行归一化，然后存储结果表。排名就是简单的算术运算：高级个体贡献者使用一套权重向量，团队负责人使用另一套。由于判断结果与权重是分开存储的，改变我们重视的东西会立即重新排列候选人，而无需重新调用推理。你可以将排名中的每个位置追溯到产生它的维度。

7. 工具调用提取：Schema 驱动的参数解析

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}),
}

当 AI 智能体生成自然语言响应时，结构化参数提取一直是个难题。传统的做法是让模型直接输出 JSON 或其他格式，这种方式容易因格式细微偏差而中断，且无法利用类型系统进行验证。我们采用的方法是将参数解析卸载到分类层：用一个 Choice 字段识别所调用的工具，另一个 Choice 字段选择目标房间，然后根据前两个决策的结果，从代码中动态加载对应的参数 Schema。这样参数验证就变成了一个组合分类问题，而不是一个脆弱的生成问题。

def extract_and_execute(command):
    response, _ = ask(command, CALL_SPEC)
    tool_choice = response.choices["tool"]
    if tool_choice.choice == "none":
        return "No tool matched."
    tool_fn, param_key = TOOLS[tool_choice.choice]
    room_choice = response.choices["room"]
    if room_choice.choice not in ROOMS:
        return f"Room '{room_choice.choice}' not found."
    param_choice = response.choices[param_key]
    return tool_fn(room_choice.choice, param_choice.choice)
 
 
commands = [
    "turn on the lights in the bedroom",
    "make the office warmer please",
    "play some jazz in the kitchen",
    "set thermostat to eco in the living room",
    "dim the lights in the bedroom",
    "make it cooler in the office",
]
for cmd in commands:
    result = extract_and_execute(cmd)
    print(f"  {cmd:<50s} -> {result}")

8. 总结

我们讨论了四种让 AI 智能体决策更可预测的模式：结构化输出保证响应符合预期 Schema；工具选择通过置信度路由；复合评分将模型判断与代码策略分离；以及 Schema 驱动的参数解析。每种模式都遵循同一个核心原则：让模型做它擅长的事——理解上下文、在维度上进行判断——然后由代码处理后续的验证、组合与策略执行。

这类架构不会取代凭感觉编程，但它能有效约束当模型出错时的代价：不是一次全面的错误响应，而是一个可识别、可处理、有清晰升级路径的错误响应。上下文工程正是关于这种分界的艺术——在模型能力与系统护栏之间找到平衡，确保在模型无法可靠完成时，有一个安全网接住它。