Ръководство за програмиране на TypeSafe AI Jev: типизирани решения, калибрирана увереност и спекулативно разклоняване със System One модел

В този урок работим с Jev, първия модел System One на TypeSafe AI, който изобщо не генерира текст: подаваме му част от програмното състояние и набор от типизирани въпроси, а той връща избори, оценки и вероятности „да/не“, върху които кодът ни може директно да разклонява логиката си. Инсталираме официалния 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 list price; output tokens are free
 
 
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, фиксиран към версията, с която е написан този notebook, и зареждаме API ключа от средата, от раздела Secrets в Colab или чрез скрито подканване, така че той никога да не се появява в notebook-а. TypeSafeClient сам прочита TYPESAFE_API_KEY и по подразбиране използва псевдонима jev-latest; списъкът с моделите показва кои имена и фиксирани версии може да използва ключът. Малкият помощник ask обвива system_one, така че всяко извикване в останалата част на notebook-а да бъде измервано, а използването на токени да се записва в регистър, който обобщаваме в края.

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 има две части: състояние, което може да бъде произволен текст, JSON обект или масив, описващ ситуацията, и речник с именувани въпроси. Choice избира един етикет от зададените критерии и връща вероятност за всеки етикет; Score поставя състоянието върху подредена скала и връща претегленото според вероятностите ниво, така че резултатът може да попадне между две нива; Noul връща една вероятност, че дадено твърдение е вярно. Имената на въпросите се задават от нас и никога не достигат до модела, затова инструкциите съдържат цялото значение и могат да посочват вложени полета чрез пътища в обратни кавички. И четирите въпроса се оценяват в една заявка, паралелно и независимо един от друг, а отговорът показва фиксираната версия на модела, който е отговорил, както и таксуваните токени.

@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 обектът добавя поръчката с двете осчетоводени такси и самата политика за възстановяване. Въпросът не се променя, така че всяка разлика във върнатата вероятност се дължи на състоянието, а колоната с токените показва цената на допълнителния контекст. Именуваните полета са препоръчителният според документацията подход, когато контекстът има няколко части, защото инструкциите могат да се обръщат към тях по име.

TypeSafe описва увереността като статистика, изчислена от разпределението, което вече съдържа отговорът: броят на опциите, умножен по най-високата вероятност, минус едно, разделено на броя на опциите минус едно. Преизчисляваме я от вероятностите на Choice и я сравняваме с полето confidence, а също преизчисляваме Score като сумата от всяко ниво, умножено по вероятността му. Прекарването на рязко формулирано и нарочно неясно съобщение през едни и същи два въпроса показва как разпределението и съответно увереността реагират на двусмислието. Noul изобщо няма поле за увереност, тъй като стойността му вече е вероятността за „да“, а стойност близо до 0,5 означава нерешителност, а не умерена степен.

Тъй като въпросите в една заявка не могат да виждат отговорите на другите въпроси, можем предварително да зададем всичко, което би могло да ни потрябва, включително въпроси, важни само за един клон, и след това да прочетем само съответните отговори. Поставяме десет въпроса за доклад след инцидент — два Choice, два Score и шест Noul — в едно извикване, след което задаваме всеки от тях отново в самостоятелно извикване и сравняваме времето, входните токени и отговорите. Състоянието се изпраща веднъж вместо десет пъти, което осигурява както спестяването на време, така и на токени, а колоната за съвпадение проверява директно твърдението за изолация: въпросът трябва да получи същия отговор независимо от това дали е изпратен самостоятелно или заедно с други.

Типизираните отговори имат значение само ако кодът около тях описва каква сигурност е необходима за дадено действие. Класифицираме всяко съобщение според намерението му и маршрутизираме въз основа на две неща: самото намерение и това дали увереността му надхвърля праг, който се увеличава според риска — от 0,5 за проверка на баланс до 0,9 за закриване на сметка. Всичко, класифицирано като „друго“ или с увереност под 0,5, се насочва към човек; намерение, което е разпознато, но не достига своя праг, първо се потвърждава от потребителя. Праговете са обикновени Python стойности, така че толерантността към риска може да се преглежда, версионира и тества като всеки друг код, вместо да бъде скрита в prompt.

Комбинираното оценяване ограничава задачата на модела и прави политиката явна. За всеки кандидат задаваме четири въпроса Score, всеки от които описва конкретни ситуации, а не степени, нормализираме всяка оценка спрямо най-високото ниво и съхраняваме получената таблица. Класирането е обикновена аритметика: един вектор от тегла за старши индивидуален сътрудник и друг за ръководител на екип. Тъй като преценките се съхраняват отделно от теглата, промяната на това, което ценим, незабавно пренарежда кандидатите и не изисква ново извеждане от модела. Всяка позиция в класирането може да бъде проследена обратно до измерението, което я е породило.

Извикването на функции се превръща в набор от въпроси със затворено множество от отговори: един Choice избира инструмента, включително изрична опция „няма“ за неподдържани команди, а по един Choice за всеки аргумент се задава предварително в същата заявка. Кодът прочита само аргументите, принадлежащи на избрания инструмент, отчита най-слабата преценка като увереност на цялото извикване и след това изпълнява обикновена Python функция с валидирани, изброими стойности. Втората половина прилага документирано заобиколно решение: Jev не брои надеждно в рамките на един въпрос, затова задаваме по един Noul за всеки елемент в една заявка и извършваме сумата в кода.

Четири подробности превръщат примерите в услуга за вземане на решения. Наследяването на SystemOneResponse и декларирането на очакваните отговори осигурява достъп чрез атрибути, валидиран от Pydantic, така че типизираното решение остава типизирано до самото приложение, вместо да се превръща в търсене в речник. Тъй като всяка заявка е независима, опашката от билети е опашка от независими решения: AsyncTypeSafeClient с asyncio.gather ги изпраща едновременно, а помощникът run_async позволява същият код да работи както в скрипт, така и в notebook, където вече има стартиран event loop. RetryPolicy ограничава броя на повторните опити, интервала между тях и общия времеви бюджет за всяко извикване. Грешките също са типизирани: празен набор от въпроси се отхвърля, преди да бъде направена заявка, а неизвестно име на модел се връща от API като подклас на TypeSafeAPIError, съдържащ HTTP статуса.

Обобщението отпечатва резултата от един ред, върнат от всеки раздел, след което сумира регистъра, към който е добавяло всяко извикване: броя на заявките, входните и изходните токени и разхода по публикуваната цена за входа, като изходните токени са безплатни.

В заключение използвахме Jev по предназначение: като източник на малки, типизирани преценки, които кодът комбинира, а не като генератор на текст, към който се подават prompt-ове и отговорите му се парсват. Всеки отговор пристигна като етикет, ниво или вероятност с приложено разпределение, което ни позволи да зададем прагове, тегла и правила за маршрутизиране в Python, където могат да бъдат тествани. Групирането на въпроси върху общо състояние намалява както времето, така и броя на токените, защото състоянието се изпраща веднъж за всеки елемент; Noul заменя броене, което не можем да поверим на модела, а Choice със затворено множество превръща команди на естествен език във валидирани извиквания на функции. Производствените компоненти — типизираните модели за отговори, асинхронният клиент, политиките за повторни опити и типизираните грешки — са малки, а регистърът изчислява цената на целия notebook. Остава частта, която нито един SDK не може да свърши вместо нас: да оценим въпросите, критериите и праговете върху собствените си данни, преди да им се доверим за реални действия.

Разгледайте ПЪЛНИЯ КОД тук. Всички заслуги са за изследователя на този проект. Също така можете да ни последвате в Twitter и не забравяйте да се присъедините към нашия ML SubReddit с над 150 хиляди членове и да се абонирате за нашия бюлетин. Чакайте! В Telegram ли сте? Вече можете да се присъедините към нас и в Telegram.

Имате нужда от партньор за популяризиране на вашето GitHub хранилище, страница в Hugging Face, продуктово представяне, уебинар и др.? Свържете се с нас