Посібник із програмування TypeSafe AI Jev: типізовані рішення, калібрована впевненість і спекулятивне розгалуження з моделлю System One

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

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()

@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()

def confidence_from(probabilities):
    """TypeSafe's published statistic: (count x peak - 1) / (count - 1)."""
    p = list(probabilities.values())
    return (len(p) * max(p) - 1) / (len(p) - 1)
 
 
@section("3. Confidence is a statistic of the distribution, and you can recompute it")
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()

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. Speculative fan-out: ten questions in one call versus ten calls")
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()

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",
             "there's a 89.99 charge from a gym i never joined",
             "shut everything down, i'm done with this bank",
             "what's the weather like in lisbon"]
    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()

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()

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)                # every argument asked speculatively, one call
        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()

import concurrent.futures
from typesafe_sdk import (AsyncTypeSafeClient, ChoiceAnswer, NoulAnswer, RetryPolicy, ScoreAnswer,
                          SystemOneResponse, TypeSafeAPIError, TypeSafeError)
 
 
class TicketDecision(SystemOneResponse):
    """Declare the answers you expect and read them as attributes, validated by 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):                    # attribute access, no dict lookups, no parsing
        print(f"  {text[:58]:<58s} {r.department.choice:<10s} {r.frustration.score:7.2f} {r.refund_requested.noul:7.2f}")
 
    print("\n  errors are typed too:")
    try:
        client.system_one("anything", {})
    except TypeSafeError as e:
        print(f"    empty questions, caught before any request : {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"    unknown model, rejected by the 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()

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  whole tutorial: {LEDGER['calls']} calls, {LEDGER['input_tokens']:,} input tokens, "
      f"{LEDGER['output_tokens']:,} output tokens (free)  ->  about ${cost:.5f}")
print("""
Where to go next
 - Patterns: docs.typesafe.ai/patterns  (fan-out, confidence routing, composite scoring, intent routing)
 - Cookbooks: re-ranking, RAG passage filtering, citation checks, LLM guardrails, hierarchical classification
 - Known rough edges of jev-1.13: docs.typesafe.ai/model-jaggedness/jev-1.13  (literal reading, arithmetic,
   counting, date comparison, large irrelevant state)
 - Compare against an LLM on the same questions: github.com/typesafe-ai/system-one-adapter-python
 - Pin a version for production: TypeSafeClient(model="jev-1.13.0"); response.model reports what answered
""")