Hướng dẫn lập trình với Jev từ TypeSafe AI: Ra quyết định kiểu dữ liệu, độ tin cậy và truy vấn song song

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