On this page
Ask a chatbot how to make tea, and the recipe fills in a piece at a time. That is not only a screen effect: the model really makes the reply one token at a time. This lesson uses a toy, a tiny program that learned from 17 short recipe lines, 5 of them for tea. Its tokens are whole words and marks like the period. Asked “How do I make tea?”, it writes 30 tokens and then an end mark, a special token that means the reply is finished. That takes 31 passes, each one run of the model. In the first it picks Boil, which had 4 chances out of 5. Real models are far bigger; the lesson says where the toy misleads.
In one minute
AI answers appear word by word because the model makes only one token at a time, and the app shows each token as soon as it arrives. A token is a small piece of text, often a whole word; picture a person writing a reply on a pad one word at a time, rereading everything on the pad before choosing the next word. The model does the same: each run reads the question and the reply so far, and gives chances for the next token. One is picked and added to the text, and the model runs again. Showing each token as it arrives is called streaming the reply.
Explore the numbers
Open the widget. It is the toy from the opening, starting on the tea question. The list shows the chances for the next pass. Press Next token to run that pass: the likeliest token joins the reply, and the count goes up by one. Press Play to stream the rest, or pick another question. Under More settings, a length cap of 10 stops the reply mid-sentence after 10 passes.
How it works
The model makes one token at a time
A language model is a program that learned, from a very large amount of text, which tokens tend to follow which (What is a token? explains them). To answer, it runs a loop. A pass, also called a generation step, is one run of the model: it reads the text so far, the question plus the reply already made, and gives each possible next token a chance. One token is picked, here the likeliest, and added to the text. The next pass starts from the longer text. The loop stops when the model picks the end mark, or when the reply reaches a length limit.
Each pass starts from what the last one made
In the toy, pass 1 starts from “How do I make tea?” It looks at the last three tokens, “make tea ?”, and counts what followed them in the 5 tea lines it read: Boil 4 times, Heat once. So Boil has 4 chances out of 5 (80%) and Heat 1 out of 5 (20%). Boil is added, and pass 2 starts from the text with Boil on the end. Pass 11 is less sure: after “bag in a” the toy read cup 3 times and mug twice, so cup has 60% and mug 40%, and the demo takes the likeliest, cup. A real model also picks from chances, not from a script; Why does AI make things up? covers how adventurous the pick is. The toy is much simpler than a real model. It looks back only three tokens and counts what followed them in the 17 lines it read, while a real model reads all the text so far, up to its limit, and has learned from far more text.
The reply is not written ahead
The model does not write the reply ahead and then reveal it. Inside, the server (the computer running the model) keeps notes about earlier tokens, called the attention cache, so a pass need not redo that work. But no later token exists until its own pass has run. A reply that is still arriving is a reply that is still being made.
Each token costs a pass
One token per pass means the tea reply’s 30 tokens and its end mark need 31 passes, and a longer reply needs more, so the time to finish grows with the length. How long a pass takes depends on the model and the machine, so this lesson gives no speeds. Why a fast AI chip spends most of its time waiting says what makes each pass take time.
Why apps show each token as it arrives
An app could wait for the last pass and then show the whole reply, leaving the reader with a blank screen until then. Many chat apps instead show each token as it arrives. Streaming does not make the model faster: the last token appears at about the same time either way. What changes is the wait before anything shows, which builders call time to first token. Behind the screen, the server sends the reply as a series of small messages over one open connection, and the page adds each message to the text. The widget’s Play button imitates this at a steady pace; a real stream arrives as fast as passes finish.
What this means when you build
- Stream replies that people read. Turn on your provider’s streaming and add each piece to the page as it arrives. Anthropic’s streaming page (read 2026-10-19) says that turning it on sends the reply incrementally as server-sent events, a standard way for a server to push small messages to a page.
- Plan for a stream that breaks. The same page says errors can arrive inside the stream, so part of a reply may already be on screen when one does. Keep what arrived, tell the reader, and offer a retry.
- Do not use half a reply as a whole one. Text that must be complete to be valid, such as JSON (a strict data format), cannot be read until the last piece has arrived.
- Measure two times. Time to first token is the wait before the first piece shows; tokens per second is how fast pieces then arrive. A slow reply can come from either, so record both.
- Set a length cap and show when it was hit. A reply that reaches the cap stops mid-sentence, as the widget shows at a cap of 10.
Optionalabout 12 min
Go deeper
Optional. This part is for readers who write Python. The code is in the code/ folder next to this lesson. python3 show_numbers.py prints every number below, and python3 -B -m pytest -q -p no:cacheprovider code/ runs the tests.
The generation loop
Every system that writes a reply one token at a time runs this loop; only model differs.
text = tokens of the prompt
repeat up to the length cap:
chances = model(text) # one pass: a chance for every possible next token
token = pick(chances) # the likeliest, or a draw by chance
if token is the end mark: stop
add token to text
show token # streaming: the app displays it nowIn the toy, generate is that loop. It returns one record per pass: the text before it, what it looked at, the chances and the token it added. (Docstrings are left out here.)
def generate(model, prompt, cap=30, seed=None):
tokens = tokenize(prompt)
state = seed_state(seed) if seed is not None else None
passes = []
for _ in range(cap):
looked_at, ranked, total = chances(model, tokens)
if not ranked:
break
u = None
if state is not None:
state, u = seed_step(state)
token = pick(ranked, total, u)
passes.append({"before": list(tokens), "looked_at": list(looked_at), "ranked": ranked, "total": total, "token": token})
if token == END:
break
tokens.append(token)
return passesA reply that ends with the end mark has one more pass than it has tokens: 30 tokens and the end mark took 31 passes. A reply stopped by the cap has as many passes as tokens, and it ends mid-sentence.
The toy: count what followed what
The toy has no neural network. train reads the 17 lines of corpus.txt (656 tokens, counting one end mark per line, and 82 different tokens besides the end mark). It pads every line with three <s> marks, so the first word of a line also follows something, and ends it with <end>. For every token it then counts what followed the previous 3, the previous 2, the previous 1 and nothing at all: four tables. The first has 186 different 3-token runs, the second 155, the third 83.
def train(lines):
tables = [{} for _ in range(ORDER + 1)]
for line in lines:
tokens = [START] * ORDER + tokenize(line) + [END]
for i in range(ORDER, len(tokens)):
for k in range(ORDER + 1):
followers = tables[k].setdefault(tuple(tokens[i - k:i]), {})
followers[tokens[i]] = followers.get(tokens[i], 0) + 1
return tableschances looks up the last 3 tokens of the text. If that run never occurred in the corpus it falls back to the last 2, then the last 1, then to all tokens together. The chance of a follower is its count divided by the total, and followers are ranked by count, with ties broken by the token that sorts first, so the demo repeats exactly.
def chances(model, tokens):
text = [START] * ORDER + list(tokens)
for k in range(ORDER, -1, -1):
looked_at = tuple(text[len(text) - k:])
followers = model[k].get(looked_at)
if followers:
ranked = sorted(followers.items(), key=lambda item: (-item[1], item[0]))
return looked_at, ranked, sum(followers.values())
return (), [], 0pick takes the first token of the ranking, or, with a seed, draws by chance: with u between 0 and 1 it computes r = floor(u * total) and walks down the ranking, subtracting each count, until r is smaller than the count it reaches. The seeded draw uses the Park-Miller generator (factor 48271, modulus 2^31 - 1), which needs only whole-number arithmetic, so Python and JavaScript make the same picks.
Three passes by hand
The tea question has 6 tokens: How, do, I, make, tea and ?. The table lists, for the first three passes, what the model looked at (the last three tokens of the text) and what followed them in the corpus.
| pass | looked at | followed by |
|---|---|---|
| 1 | make tea ? | Boil 4, Heat 1 |
| 2 | tea ? Boil | some 4 |
| 3 | ? Boil some | water 4 |
Pass 1: “make tea ?” occurs in the 5 tea lines (the other questions read differently), so the total is 5. Boil follows it 4 times and Heat once, giving 4/5 = 0.8 and 1/5 = 0.2. The likeliest is Boil, so the text grows to 7 tokens. Pass 2: “tea ? Boil” occurs in 4 lines, and each goes on with “some”: 4/4 = 1. The text grows to 8 tokens. Pass 3: “? Boil some” occurs in 4 lines, each followed by “water”: 4/4 = 1. After three passes the text is “How do I make tea? Boil some water”, 9 tokens, and python3 show_numbers.py prints the same kind of line for every pass, up to pass 31.
Why each pass needs a full run of the model
The model’s knowledge sits in weights, billions of numbers kept in memory. To write one token, a standard model’s chip must fetch every weight and use it, and each weight does little work per fetch. Why a fast AI chip spends most of its time waiting works this out, and calls writing a reply token by token decode. The first pass reads the whole prompt together (prefill); every pass after that adds one token. The passes cannot run side by side, because pass 2 reads the token that pass 1 made. Leviathan, Kalman and Matias (2022 preprint, ICML 2023) note that writing K tokens needs K runs of the model, one after another.
The attention cache saves repeated work
Without any saved work, every pass would run the network over the whole text so far: 6 positions, meaning places in the text, for pass 1, 7 for pass 2, 8 for pass 3. That is 6 + 7 + 8 = 21 positions after three passes. A server instead keeps, for every earlier token, notes that later tokens look back at: the attention cache, also called the KV cache. A pass then runs the network over the newest token only (it still reads the cached notes), and the first pass reads the prompt. After three passes that is 6 + 1 + 1 = 8 positions.
positions_read gives the counts: 651 against 36 for the full tea reply, and a test confirms the formula by running the loop on paper. “Positions” counts places in the text, not seconds or bytes: a real server also reads the cached notes with each pass, and they take memory (see Why does a chatbot forget? and Why AI servers waste memory, and how paging fixes it). The cache does not change the tokens, only the work behind them.
More than one token per pass
The plain loop makes exactly one token per pass, and a cache does not change that. Some systems make more. In speculative decoding, a cheaper model’s guesses are checked by the large model, so a pass of the large model can produce several tokens. Leviathan, Kalman and Matias (2022 preprint, ICML 2023) describe it as computing several tokens in parallel without any change to the outputs, and report a 2X-3X acceleration on T5-XXL over the standard T5X implementation, in their own tests. Chen and colleagues (2023) describe speculative sampling, which generates multiple tokens from each transformer call. A later lesson covers the method.
Why apps stream, and what a stream looks like
A reply of a few hundred tokens needs a few hundred passes, so waiting for all of them can leave the screen blank. Provider documentation describes streaming as many small events. The HTML standard defines server-sent events: a server pushes messages over one HTTP response with the content type text/event-stream; a message is one or more data: lines, and a blank line ends it. An invented example in that shape, carrying the toy’s first two tokens:
data: {"token": "Boil"}
data: {"token": " some"}The page adds each message’s token to the screen, which is what the widget’s Play button imitates with a timer. Two numbers describe the result: time to first token, the wait before the first message, and tokens per second, how many arrive each second after that. Both depend on the model, the server and the network, so no figure from this lesson transfers to a real service.
Proof it works
python3 -B -m pytest -q -p no:cacheprovider code/
python3 -B code/embed_widget.py --checkThe tests check these claims. Each pass adds exactly one token, so the passes equal the tokens plus the end mark, and every pass starts from the text the last one left. The chances add up to 1 and are ranked. The first pass on the tea question is Boil 4, Heat 1; a pass looks at the last 3 tokens only, and text before them changes nothing; unseen text falls back to shorter runs. For each of the four questions the greedy reply is the line the corpus repeats most. A cap of N allows exactly N passes. The seeded generator gives the published check value (from state 1, the 10000th state is 399268537), a draw gives each token its share, and 300 seeds per question stay on the question and end. The cache counts match a pass-by-pass count and give 651 and 36.
widget.html runs the same rules in JavaScript, so test_widget.py runs the widget’s own code in Node and compares it with the Python: the tokenizer, the tables, every pass’s chances, picks and reply, on five corpora, ten questions, eleven ways of picking (the likeliest token and ten seeds) and four caps (more than a thousand comparisons). A second test drives the page with a stand-in document and a fake clock through 28 runs, and checks Play, Pause, Next, Reset and the closing messages. embed_widget.py --check fails if the corpus or the questions inside the widget differ from corpus.txt and prompts.py.
Where this is used in practice
The public record is documentation. Anthropic’s streaming page (read 2026-10-19) says that setting "stream": true streams the response incrementally with server-sent events. Each event has a named type and JSON data: a message starts, content blocks arrive as a start, one or more deltas and a stop, and the message ends; text comes in text_delta events. The page also says errors can arrive inside the stream. OpenAI’s streaming guide (read 2026-10-19) lists events such as response.created, response.output_text.delta, response.completed and error. Google’s text generation page for the Gemini API (read 2026-10-19) includes a streaming example. None of the three pages promises a speed, and this lesson gives none.
Skills this lesson builds
- Explain why a reply arrives piece by piece, and what streaming changes (the wait before the first piece) and what it does not (the model’s speed).
- Trace a generation loop by hand: text so far, chances, pick, add, and the two ways it stops (the end mark and the length cap).
- Count passes for a reply, and the positions the network runs over with and without a cache.
- Read a server-sent events stream, add each piece to a page, and handle an error that arrives mid-stream.
- Measure time to first token and tokens per second separately.
Portfolio project: build a small command-line tool that calls any chat model with streaming on, prints each piece as it arrives, and records when each piece arrived. Run it on 10 prompts of different lengths and print a table of time to first token, total time and tokens per second for each. Publish the prompts, the script and the table.
Check yourself
-
A friend says the chatbot already wrote the whole answer and is only revealing it slowly. What do you tell them?
Answer
The model makes one token per pass, from the text so far, and the app shows each token as it arrives. No later token exists until its pass has run, so a reply that is still arriving is still being made.
-
A reply is 120 tokens long, and the model ends it by picking the end mark. How many passes did that take?
Answer
121: one pass for each of the 120 tokens, and one more for the end mark.
-
Streaming does not make the model faster. What does it improve?
Answer
The wait before anything shows. The last token arrives at about the same time either way, but with streaming the reader sees the first piece after the first pass, which is the time to first token.
-
A prompt has 8 tokens and the reply takes 5 passes. How many positions does the network run over without a cache, and with one?
Answer
Without a cache: 8 + 9 + 10 + 11 + 12 = 50. With a cache: 8 for the prompt, then 1 for each of the other 4 passes, so 12.
-
Why can the 31 passes of the tea reply not run at the same time?
Answer
Each pass reads the token the pass before it made, so it cannot start until that token exists. This is why writing K tokens needs K runs of the model, one after another; speculative decoding is a way to get several tokens from one pass of the large model.
Further reading
Gentle
- Streaming messages (Anthropic documentation; read 2026-10-19)
- Streaming API responses (OpenAI documentation; read 2026-10-19)
- Text generation (Google Gemini API documentation, with a streaming example; read 2026-10-19)
- Server-sent events (WHATWG HTML Standard; read 2026-10-19)
Deep
- Fast Inference from Transformers via Speculative Decoding (Leviathan, Kalman and Matias, 2022 preprint, ICML 2023; abstract read 2026-10-19)
- Accelerating Large Language Model Decoding with Speculative Sampling (Chen, Borgeaud, Irving, Lespiau, Sifre and Jumper, 2023; abstract read 2026-10-19)
Optional
Code
Each file runs on its own. Download it, or read it here.
Show code
How do I make tea? Boil some water. Put a tea bag in a cup. Pour the water over the bag. Wait three minutes. Pull the bag out. Enjoy.
How do I make tea? Boil some water. Put a tea bag in a cup. Pour the water over the bag. Wait three minutes. Pull the bag out. Enjoy.
How do I make tea? Boil some water. Put a tea bag in a cup. Pour the water over the bag. Wait three minutes. Pull the bag out. Add some milk. Enjoy.
How do I make tea? Boil some water. Put a tea bag in a mug. Pour the water over the bag. Wait four minutes. Pull the bag out. Enjoy.
How do I make tea? Heat some water. Put a tea bag in a mug. Pour the water over the bag. Wait three minutes. Pull the bag out. Add some milk. Enjoy.
How do I boil an egg? Put the egg in a pan. Cover it with cold water. Bring the water to a boil. Wait ten minutes. Scoop the egg out. Enjoy.
How do I boil an egg? Put the egg in a pan. Cover it with cold water. Bring the water to a boil. Wait ten minutes. Scoop the egg out. Enjoy.
How do I boil an egg? Put the egg in a pan. Cover it with cold water. Bring the water to a boil. Wait six minutes. Scoop the egg out. Enjoy.
How do I boil an egg? Place the egg in a pot. Cover it with cold water. Bring the water to a boil. Wait ten minutes. Scoop the egg out. Enjoy.
How do I make toast? Slice some bread. Put the slice in a toaster. Press the lever down. Wait two minutes. Lift out the toast. Add butter. Enjoy.
How do I make toast? Slice some bread. Put the slice in a toaster. Press the lever down. Wait two minutes. Lift out the toast. Add butter. Enjoy.
How do I make toast? Slice some bread. Put the slice in a toaster. Press the lever down. Wait one minute. Lift out the toast. Add jam. Enjoy.
How do I make toast? Cut some bread. Put the slice in a toaster. Press the lever down. Wait two minutes. Lift out the toast. Add butter. Enjoy.
How do I water a plant? Check the soil with a finger. If the soil is dry, add water. Pour slowly until water drains out. Empty the dish under the pot. Check again next week.
How do I water a plant? Check the soil with a finger. If the soil is dry, add water. Pour slowly until water drains out. Empty the dish under the pot. Check again next week.
How do I water a plant? Check the soil with a finger. If the soil is dry, add water. Pour slowly until water drains out. Empty the dish under the pot. Check again in three days.
How do I water a plant? Check the soil with a finger. If the soil is wet, wait a day. Check again tomorrow.Show code
"""Keep the copy of the corpus and the settings inside ../widget.html equal to the ones here.
python3 embed_widget.py rewrite the two data blocks in widget.html
python3 embed_widget.py --check exit 1 if widget.html differs (test_widget.py runs the same check)
"""
import json
import re
import sys
from pathlib import Path
from prompts import PROMPTS
from toy_gen import CAP, END, ORDER, START, load_corpus
WIDGET = Path(__file__).resolve().parent.parent / "widget.html"
CORPUS_BLOCK = re.compile(r'(<script type="text/plain" id="wbw-corpus">)(.*?)(</script>)', re.S)
CONFIG_BLOCK = re.compile(r'(<script type="application/json" id="wbw-config">)(.*?)(</script>)', re.S)
def config():
"""The settings the page needs besides the corpus: the rules' constants and the questions."""
return {"order": ORDER, "start": START, "end": END, "capMax": CAP, "prompts": PROMPTS}
def embed(html):
"""html with both data blocks rewritten from corpus.txt, toy_gen.py and prompts.py."""
corpus = load_corpus()
if "</script" in corpus.lower() or "<!--" in corpus:
raise ValueError("the corpus would end or confuse the script element")
settings = json.dumps(config(), ensure_ascii=False, separators=(", ", ": ")).replace("<", "\\u003c") # the marks <s> and <end> must not read as tags
html = CORPUS_BLOCK.sub(lambda m: m.group(1) + corpus + m.group(3), html, count=1)
return CONFIG_BLOCK.sub(lambda m: m.group(1) + "\n" + settings + "\n" + m.group(3), html, count=1)
if __name__ == "__main__":
current = WIDGET.read_text(encoding="utf-8")
updated = embed(current)
if "--check" in sys.argv:
sys.exit(0 if updated == current else 1)
WIDGET.write_text(updated, encoding="utf-8")
print("widget.html data blocks now match corpus.txt, toy_gen.py and prompts.py")Show code
"""The four questions the widget offers. Each is the start of at least two lines in corpus.txt."""
PROMPTS = [
{"name": "Tea", "prompt": "How do I make tea?"},
{"name": "Egg", "prompt": "How do I boil an egg?"},
{"name": "Toast", "prompt": "How do I make toast?"},
{"name": "Plant", "prompt": "How do I water a plant?"},
]Show code
"""Print every number the lesson quotes. Run it from this folder: python3 show_numbers.py"""
from prompts import PROMPTS
from toy_gen import END, generate, lines_of, load_corpus, positions_read, reply_of, tokenize, train
lines = lines_of(load_corpus())
model = train(lines)
tea = PROMPTS[0]["prompt"]
def counts(p, n=3):
return ", ".join(f"{t} {c}" for t, c in p["ranked"][:n])
print("== The corpus (corpus.txt)")
print(f" {len(lines)} lines, {sum(len(tokenize(l)) + 1 for l in lines)} tokens counting one end mark per line, "
f"{len({t for l in lines for t in tokenize(l)})} different tokens besides the end mark")
for item in PROMPTS:
print(f" {item['prompt']!r} starts {sum(l.startswith(item['prompt']) for l in lines)} lines")
print(f" the 3-token table has {len(model[3])} contexts, the 2-token table {len(model[2])}, the 1-token table {len(model[1])}")
print("\n== Greedy replies (every pass takes the likeliest token)")
for item in PROMPTS:
passes = generate(model, item["prompt"], 40)
prompt_tokens, written = len(tokenize(item["prompt"])), len(passes) - 1
split = sum(1 for p in passes if p["ranked"][0][1] < p["total"])
print(f" {item['name']}: {prompt_tokens} prompt tokens, {written} reply tokens + end mark = {len(passes)} passes; "
f"{split} passes had more than one possible token")
print(f" {reply_of(passes)}")
print(f" positions read: {positions_read(prompt_tokens, len(passes), False)} without the cache, "
f"{positions_read(prompt_tokens, len(passes), True)} with it")
print("\n== Tea, every pass")
for i, p in enumerate(generate(model, tea, 40), 1):
print(f" {i:2d} looked at {' '.join(p['looked_at']):<22} counts {counts(p)} of {p['total']} adds {p['token']}")
print("\n== Tea, the first three passes by hand")
for i, p in enumerate(generate(model, tea, 3), 1):
shares = ", ".join(f"{t} {c}/{p['total']} = {c / p['total']:.0%}" for t, c in p["ranked"])
print(f" pass {i}: text ends '{' '.join(p['before'][-3:])}', looked at '{' '.join(p['looked_at'])}': {shares}; adds {p['token']}")
print("\n== Cap")
print(f" cap 10 on tea stops after {len(generate(model, tea, 10))} passes; the reply is {reply_of(generate(model, tea, 10))!r}")
print("\n== Seeded picks (tea)")
for seed in (1, 2, 3, 7):
passes = generate(model, tea, 40, seed)
print(f" seed {seed}: {len(passes)} passes: {reply_of(passes)}")
print("\n== Passes with and without the cache, tea, by reply length")
for n in (1, 2, 10, 31):
print(f" {n:2d} passes: {positions_read(6, n, False)} positions without, {positions_read(6, n, True)} with")Show code
"""A toy that writes a reply one token at a time, using nothing but the standard library.
widget.html runs the same rules in JavaScript; test_widget.py checks that the two agree.
The toy reads the text so far (the question and the reply written up to now), looks at its last three tokens,
and asks what followed those three tokens in its corpus. If it never saw them it looks at the last two, then the
last one, then at all tokens together. The share of each follower is its chance. One pass of the loop gives one
chance list and one token, which is added to the text before the next pass.
"""
import re
from math import floor
from pathlib import Path
CORPUS = Path(__file__).resolve().parent / "corpus.txt"
START = "<s>"
END = "<end>"
TOKEN = re.compile(r"[A-Za-z0-9']+|[.,!?]") # ASCII only, so Python and JavaScript split alike
ORDER = 3 # how many earlier tokens one pass looks at
CAP = 40 # the longest reply the widget allows
SEED_MODULUS = 2147483647 # 2**31 - 1, for the seeded pick (Park-Miller)
SEED_FACTOR = 48271
def load_corpus():
return CORPUS.read_text(encoding="utf-8")
def tokenize(text):
"""Words and the marks . , ! ? as separate tokens. Capital letters count: 'Boil' is not 'boil'."""
return TOKEN.findall(text)
def lines_of(text):
"""The non-empty lines of a corpus. Each line is one question followed by its reply."""
return [line for line in text.split("\n") if line.strip()]
def train(lines):
"""Count what followed every run of 0 to ORDER tokens. Each line is padded with START marks and ends with END."""
tables = [{} for _ in range(ORDER + 1)]
for line in lines:
tokens = [START] * ORDER + tokenize(line) + [END]
for i in range(ORDER, len(tokens)):
for k in range(ORDER + 1):
followers = tables[k].setdefault(tuple(tokens[i - k:i]), {})
followers[tokens[i]] = followers.get(tokens[i], 0) + 1
return tables
def chances(model, tokens):
"""What one pass produces: the run of tokens it looked at and every possible next token with its count.
Returns (looked_at, ranked, total). `ranked` is [(token, count)] from most to least likely; a tie goes to the
token that sorts first. The chance of a token is its count divided by the total."""
text = [START] * ORDER + list(tokens)
for k in range(ORDER, -1, -1):
looked_at = tuple(text[len(text) - k:])
followers = model[k].get(looked_at)
if followers:
ranked = sorted(followers.items(), key=lambda item: (-item[1], item[0]))
return looked_at, ranked, sum(followers.values())
return (), [], 0 # an empty corpus
def seed_state(seed):
"""The first state of the seeded pick: a whole number from 1 to 2**31 - 2."""
return seed % (SEED_MODULUS - 1) + 1
def seed_step(state):
"""One move of the seeded generator (Park-Miller). Returns (next state, a number from 0 up to but below 1)."""
state = state * SEED_FACTOR % SEED_MODULUS
return state, (state - 1) / (SEED_MODULUS - 1)
def pick(ranked, total, u=None):
"""The token the pass adds. With u=None the likeliest; with u in [0, 1) a draw in which each token gets
its count out of the total. Only whole numbers are compared, so Python and JavaScript agree."""
if u is None:
return ranked[0][0]
r = floor(u * total)
for token, count in ranked:
if r < count:
return token
r -= count
return ranked[-1][0]
def generate(model, prompt, cap=30, seed=None):
"""Write a reply one pass at a time. Returns one dict per pass: the text before it, what it looked at, its
chances, and the token it added. Stops after the END mark or when the reply reaches `cap` tokens.
With seed=None every pass takes the likeliest token; with a seed it draws by chance."""
tokens = tokenize(prompt)
state = seed_state(seed) if seed is not None else None
passes = []
for _ in range(cap):
looked_at, ranked, total = chances(model, tokens)
if not ranked:
break
u = None
if state is not None:
state, u = seed_step(state)
token = pick(ranked, total, u)
passes.append({"before": list(tokens), "looked_at": list(looked_at), "ranked": ranked, "total": total, "token": token})
if token == END:
break
tokens.append(token)
return passes
def reply_of(passes):
"""The reply as the reader sees it: the tokens without the END mark, with marks . , ! ? attached to the word before."""
out = ""
for p in passes:
if p["token"] == END:
break
out += p["token"] if p["token"] in ".,!?" else (" " if out else "") + p["token"]
return out
def positions_read(prompt_tokens, new_tokens, cached):
"""How many token positions the network runs over to write `new_tokens` tokens after a prompt of `prompt_tokens`.
Without a cache every pass re-reads the whole text so far. With the cache of earlier work (the attention cache)
the first pass reads the prompt and every later pass reads only the newest token."""
if new_tokens <= 0:
return 0
if cached:
return prompt_tokens + (new_tokens - 1)
return sum(prompt_tokens + i for i in range(new_tokens))Show code
import pytest
from prompts import PROMPTS
from toy_gen import (END, ORDER, START, chances, generate, lines_of, load_corpus, pick, positions_read, reply_of,
seed_state, seed_step, tokenize, train)
LINES = lines_of(load_corpus())
MODEL = train(LINES)
TEA = PROMPTS[0]["prompt"]
GREEDY = {
"Tea": "Boil some water. Put a tea bag in a cup. Pour the water over the bag. Wait three minutes. Pull the bag out. Enjoy.",
"Egg": "Put the egg in a pan. Cover it with cold water. Bring the water to a boil. Wait ten minutes. Scoop the egg out. Enjoy.",
"Toast": "Slice some bread. Put the slice in a toaster. Press the lever down. Wait two minutes. Lift out the toast. Add butter. Enjoy.",
"Plant": "Check the soil with a finger. If the soil is dry, add water. Pour slowly until water drains out. Empty the dish under the pot. Check again next week.",
}
def test_every_question_opens_at_least_two_corpus_lines():
"""Guards a widget question the corpus cannot answer: the first pass of each would fall back to guessing from all tokens."""
for item in PROMPTS:
assert sum(line.startswith(item["prompt"]) for line in LINES) >= 2, item["prompt"]
@pytest.mark.parametrize("item", PROMPTS, ids=lambda item: item["name"])
def test_greedy_reply_is_the_most_common_line_for_that_question(item):
"""Guards a pick that is not the likeliest token: the repeatable demo must write the line the corpus repeats most."""
passes = generate(MODEL, item["prompt"], 40)
assert reply_of(passes) == GREEDY[item["name"]]
assert passes[-1]["token"] == END
assert item["prompt"] + " " + GREEDY[item["name"]] in LINES
@pytest.mark.parametrize("item", PROMPTS, ids=lambda item: item["name"])
def test_one_token_per_pass(item):
"""Guards a loop that adds two tokens, or none, in a pass: each pass adds exactly one token to the text, so
the passes equal the reply tokens plus the end mark, and every pass starts from the text the last one left."""
passes = generate(MODEL, item["prompt"], 40)
written = [p["token"] for p in passes if p["token"] != END]
assert len(passes) == len(written) + 1 == len(tokenize(reply_of(passes))) + 1
text = tokenize(item["prompt"])
for p in passes:
assert p["before"] == text
if p["token"] != END:
text = text + [p["token"]]
@pytest.mark.parametrize("item", PROMPTS, ids=lambda item: item["name"])
def test_chances_add_up_to_one_and_are_ranked(item):
for p in generate(MODEL, item["prompt"], 40):
assert sum(count for _, count in p["ranked"]) == p["total"]
assert sum(count / p["total"] for _, count in p["ranked"]) == pytest.approx(1, abs=1e-12)
keys = [(-count, token) for token, count in p["ranked"]]
assert keys == sorted(keys)
def test_first_pass_on_tea_splits_four_to_one():
"""The numbers the lesson opens with: of the 5 lines that start with the tea question, 4 go on with Boil and 1 with Heat."""
looked_at, ranked, total = chances(MODEL, tokenize(TEA))
assert looked_at == ("make", "tea", "?")
assert ranked == [("Boil", 4), ("Heat", 1)] and total == 5
def test_first_three_passes_match_the_hand_worked_table():
"""Guards the lesson's hand-worked table drifting from the code: the three rows, the 5 lines behind the first, and the
text growing from 6 to 9 tokens."""
passes = generate(MODEL, TEA, 3)
assert [(p["looked_at"], p["ranked"], p["total"], p["token"]) for p in passes] == [
(["make", "tea", "?"], [("Boil", 4), ("Heat", 1)], 5, "Boil"),
(["tea", "?", "Boil"], [("some", 4)], 4, "some"),
(["?", "Boil", "some"], [("water", 4)], 4, "water"),
]
assert [len(p["before"]) for p in passes] == [6, 7, 8]
assert sum(line.startswith(TEA) for line in LINES) == 5
assert positions_read(6, 3, False) == 6 + 7 + 8 and positions_read(6, 3, True) == 6 + 1 + 1
def test_a_pass_reads_only_the_last_three_tokens():
"""Guards a model that quietly reads more than three tokens: text before the last three changes nothing."""
tokens = tokenize(TEA) + ["Boil", "some", "water", ".", "Put"]
base = chances(MODEL, tokens)
assert base[0] == ("water", ".", "Put") and len(base[0]) == ORDER
assert chances(MODEL, ["Elephant", "?"] + tokens) == base
assert chances(MODEL, tokens[3:]) == base
def test_unseen_text_falls_back_to_shorter_runs():
"""Guards a crash or an empty list when the last three tokens were never seen: look at two, then one, then all."""
assert chances(MODEL, ["zebra", "make", "tea", "?"])[0] == ("make", "tea", "?")
assert chances(MODEL, ["zebra", "zebra", "tea", "?"])[0] == ("tea", "?") # "zebra tea ?" never occurs, "tea ?" does
assert chances(MODEL, ["zebra", "zebra", "zebra", "water"])[0] == ("water",)
looked_at, ranked, total = chances(MODEL, ["zebra"])
assert looked_at == () and total == sum(count for _, count in ranked) and ranked[0][0] == "."
assert chances(train([]), ["a"]) == ((), [], 0)
def test_start_marks_pad_the_beginning_of_a_line():
"""Guards the first token of a line being counted with no context: two tokens of text still find the lines that begin with them."""
model = train(["Hello there friend."])
assert chances(model, [])[1] == [("Hello", 1)] and chances(model, ["Hello"])[1] == [("there", 1)]
assert START not in [token for k in range(ORDER + 1) for followers in model[k].values() for token in followers]
@pytest.mark.parametrize("item", PROMPTS, ids=lambda item: item["name"])
def test_greedy_runs_are_repeatable_and_stop_at_the_end_mark(item):
assert generate(MODEL, item["prompt"], 40) == generate(MODEL, item["prompt"], 40)
assert len(generate(MODEL, item["prompt"], 40)) < 40
assert len(generate(MODEL, item["prompt"], 1000)) == len(generate(MODEL, item["prompt"], 40))
@pytest.mark.parametrize("cap", [1, 4, 10, 30])
def test_length_cap_stops_the_reply_without_an_end_mark(cap):
"""Guards a cap that is off by one: a cap of N allows N passes, and the reply then stops mid-sentence."""
passes = generate(MODEL, TEA, cap)
assert len(passes) == cap and all(p["token"] != END for p in passes)
def test_seed_generator_matches_the_published_check_value():
"""Park-Miller with factor 48271 (minstd_rand): from state 1, the 10000th state is 399268537. Guards a changed constant,
which would still give plausible numbers but different picks in JavaScript and Python."""
state = 1
for _ in range(10000):
state, u = seed_step(state)
assert state == 399268537
assert seed_state(0) == 1 and seed_state(1) == 2 and seed_state(2147483645) == 2147483646 and seed_state(2147483646) == 1
assert seed_step(seed_state(1)) == (96542, 96541 / 2147483646)
def test_pick_by_chance_gives_each_token_its_share():
"""Guards an off-by-one in the walk: with one draw per unit of the total, each token is picked exactly count times."""
ranked, total = [("a", 3), ("b", 2), ("c", 1)], 6
drawn = [pick(ranked, total, (i + 0.5) / total) for i in range(total)]
assert drawn == ["a", "a", "a", "b", "b", "c"]
assert pick(ranked, total) == "a" and pick(ranked, total, 0.0) == "a" and pick(ranked, total, 0.9999999) == "c"
def test_seeded_runs_repeat_and_differ_between_seeds():
assert generate(MODEL, TEA, 40, seed=7) == generate(MODEL, TEA, 40, seed=7)
replies = {reply_of(generate(MODEL, TEA, 40, seed=s)) for s in range(1, 21)}
assert GREEDY["Tea"] in replies and len(replies) >= 3
assert reply_of(generate(MODEL, TEA, 40, seed=7)) != reply_of(generate(MODEL, TEA, 40, seed=1))
def test_seeded_runs_stay_on_the_question_and_end():
"""Guards a corpus where two questions share three-token runs: a draw could then wander into another question's recipe
(a tea reply that takes out the toast). Every token must come from a line that starts with the same question."""
for item in PROMPTS:
own = {t for line in LINES if line.startswith(item["prompt"]) for t in tokenize(line)} | {END}
for seed in range(1, 301):
passes = generate(MODEL, item["prompt"], 40, seed)
assert passes[-1]["token"] == END and {p["token"] for p in passes} <= own, (item["name"], seed)
def simulate_positions(prompt_tokens, new_tokens, cached):
"""Run the loop on paper: each pass runs the network over the text so far, or only over what the cache lacks."""
done = total = 0
for i in range(new_tokens):
length = prompt_tokens + i
total += length - done if cached else length
done = length
return total
def test_positions_read_matches_counting_pass_by_pass():
"""Guards the closed form behind the cache numbers: run the loop on paper and compare."""
for prompt_tokens in (1, 6, 7):
for new in range(0, 12):
for cached in (False, True):
assert positions_read(prompt_tokens, new, cached) == simulate_positions(prompt_tokens, new, cached)
assert positions_read(6, 1, False) == positions_read(6, 1, True) == 6
assert (positions_read(6, 31, False), positions_read(6, 31, True)) == (651, 36)
def test_reply_text_attaches_marks_to_the_word_before():
assert reply_of([{"token": t} for t in ["Hi", ",", "you", ".", END, "x"]]) == "Hi, you."
assert reply_of([]) == ""Show code
"""Guards for ../widget.html: its data blocks, its size, that its rules give the same passes as toy_gen.py, and what a reader sees."""
import json
import re
import subprocess
from embed_widget import CONFIG_BLOCK, CORPUS_BLOCK, WIDGET, embed
from prompts import PROMPTS
from toy_gen import END, generate, lines_of, load_corpus, pick, reply_of, seed_state, seed_step, tokenize, train
HTML = WIDGET.read_text(encoding="utf-8")
CORPUS = load_corpus()
MODEL = train(lines_of(CORPUS))
# Reads the rules out of the widget between its marker comments, runs them with Node, and compares every pass with the
# ones Python wrote into the fixture: the same text before, the same chances in the same order, the same token.
PARITY = r"""
import fs from 'node:fs';
const [fixture, widget] = process.argv.slice(2);
const fx = JSON.parse(fs.readFileSync(fixture, 'utf8'));
const html = fs.readFileSync(widget, 'utf8');
const cfg = JSON.parse(html.match(/<script type="application\/json" id="wbw-config">([\s\S]*?)<\/script>/)[1]);
const js = html.slice(html.indexOf('/* ---------- the toy ('), html.indexOf('/* ---------- state'));
const api = new Function('cfg', "'use strict';" + js + '; return {tokenize, linesOf, train, chances, seedState, seedStep, pick, generate, textOf};')(cfg);
const bad = [];
let checks = 0;
const same = (a, b) => JSON.stringify(a) === JSON.stringify(b);
fx.tokens.forEach((t) => { checks++; if (!same(api.tokenize(t[0]), t[1])) bad.push(['tokenize', t[0]]); });
fx.seeds.forEach((s) => { checks++; if (api.seedState(s.seed) !== s.state) bad.push(['seedState', s.seed]); });
fx.steps.forEach((s) => { checks++; if (!same(api.seedStep(s.state), { state: s.next, u: s.u })) bad.push(['seedStep', s.state]); });
fx.picks.forEach((p) => { checks++; if (api.pick(p.ranked, p.total, p.u) !== p.token) bad.push(['pick', p.u]); });
fx.corpora.forEach((c) => {
const model = api.train(api.linesOf(c.text));
checks++;
if (!same(model.map((m) => m.size), c.sizes)) bad.push(['table sizes', c.name, model.map((m) => m.size), c.sizes]);
c.runs.forEach((r) => {
const got = api.generate(model, r.prompt, r.cap, r.seed);
checks += 2;
if (got.length !== r.passes.length) { bad.push(['pass count', c.name, r.prompt, r.seed, r.cap]); return; }
if (api.textOf(got.map((p) => p.token)) !== r.reply) bad.push(['reply', c.name, r.prompt, r.seed, r.cap]);
r.passes.forEach((want, i) => {
const p = got[i];
checks += 5;
if (!same(p.before, want.before)) bad.push(['before', c.name, r.prompt, r.seed, i]);
if (!same(p.looked, want.looked)) bad.push(['looked', c.name, r.prompt, r.seed, i]);
if (!same(p.ranked, want.ranked) || p.total !== want.total) bad.push(['chances', c.name, r.prompt, r.seed, i]);
if (p.token !== want.token) bad.push(['token', c.name, r.prompt, r.seed, i, p.token, want.token]);
if (!same(api.chances(model, want.before).ranked, want.ranked)) bad.push(['chances()', c.name, r.prompt, i]);
});
});
});
console.log(JSON.stringify({ checks, bad: bad.slice(0, 5) }));
process.exit(bad.length ? 1 : 0);
"""
# Runs the page script against a small stand-in for the DOM and a fake clock, presses the buttons, and reads back what a reader sees.
PAGE = r"""
import fs from 'node:fs';
const [widget, fixture] = process.argv.slice(2);
const html = fs.readFileSync(widget, 'utf8');
const fx = JSON.parse(fs.readFileSync(fixture, 'utf8'));
const block = (id) => html.match(new RegExp('<script type="[a-z/]+" id="' + id + '">([\\s\\S]*?)</script>'))[1];
const code = html.match(/<script>([\s\S]*?)<\/script>/)[1];
function el(tag) {
const e = { tag, children: [], listeners: {}, className: '', checked: false, hidden: false, disabled: false, value: '', style: {}, attrs: {}, _text: '' };
e.setAttribute = (k, v) => { e.attrs[k] = v; };
e.addEventListener = (type, f) => (e.listeners[type] = e.listeners[type] || []).push(f);
e.appendChild = (c) => (e.children.push(c), c);
Object.defineProperty(e, 'textContent', { get: () => e._text, set: (v) => { e._text = v; e.children = []; } });
return e;
}
const shown = (e) => e._text + e.children.map((c) => ('text' in c ? c.text : shown(c))).join('');
function open() {
const page = { timers: [], root: el('html'), messages: [], parent: { other: true } };
fx.ids.forEach((id) => { page[id] = el('x'); });
page['wbw-config']._text = block('wbw-config');
page['wbw-corpus']._text = block('wbw-corpus');
const doc = { getElementById: (id) => page[id], createElement: el, createTextNode: (text) => ({ text }), documentElement: page.root };
const win = { addEventListener: (type, f) => page.messages.push(f) };
const clock = { set: (f, ms) => { page.timers.push({ f, ms }); return page.timers.length; }, clear: () => { page.timers.length = 0; } };
new Function('document', 'window', 'parent', 'setInterval', 'clearInterval', code)(doc, win, page.parent, clock.set, clock.clear);
return page;
}
const fire = (e, type) => (e.listeners[type || 'click'] || []).forEach((f) => f({ target: e }));
const tick = (page) => page.timers[0].f();
const bad = [];
const want = (what, got, expected) => { if (JSON.stringify(got) !== JSON.stringify(expected)) bad.push([what, got, expected]); };
function look(page, v, what) { // compares the page with one expected view
want(what + ': reply', shown(page['wbw-reply']), v.reply);
want(what + ': count', page['wbw-count']._text, v.count);
want(what + ': next button', page['wbw-next'].disabled, 'closing' in v);
want(what + ': closing hidden', page['wbw-end'].hidden, !('closing' in v));
want(what + ': chances hidden', page['wbw-next-box'].hidden, 'closing' in v);
if ('closing' in v) {
want(what + ': closing', page['wbw-end']._text, v.closing);
want(what + ': closing kind', page['wbw-end'].attrs['data-v'], v.kind);
return;
}
want(what + ': looked', page['wbw-looked']._text, v.looked);
const rows = page['wbw-list'].children;
want(what + ': row count', rows.length, v.rows.length);
rows.forEach((li, i) => {
const r = v.rows[i];
want(what + ': row ' + i + ' token', shown(li.children[0]), r[0] + (i === 0 && !v.chance ? ' next' : ''));
want(what + ': row ' + i + ' share', li.children[2]._text, r[1]);
if (!(Math.abs(parseFloat(li.children[1].style.width) - r[2]) <= 0.051)) bad.push([what + ': row ' + i + ' bar', li.children[1].style.width, r[2]]);
want(what + ': row ' + i + ' top mark', li.attrs['data-top'] === 'true', i === 0 && !v.chance);
});
}
function settle(page, s) { // question, then settings, as a reader would
page['wbw-q'].value = String(s.q); fire(page['wbw-q'], 'change');
page['wbw-chance'].checked = s.chance; page['wbw-seed'].value = String(s.seed); page['wbw-cap'].value = String(s.cap);
['wbw-chance', 'wbw-seed', 'wbw-cap'].forEach((id) => fire(page[id], 'change'));
}
fx.scenarios.forEach((s, n) => {
const page = open();
settle(page, s);
const name = 'scenario ' + n + ' (' + s.q + ', ' + (s.chance ? 'seed ' + s.seed : 'likeliest') + ', cap ' + s.cap + ')';
s.views.forEach((v, k) => {
if (k > 0) {
fire(page['wbw-next']);
const p = s.passes[k - 1];
want(name + ': spoken after press ' + k, page['wbw-live']._text, p.spoken);
if (p.token !== fx.end) {
const marks = page['wbw-reply'].children.filter((c) => c.tag === 'mark');
want(name + ': newest token marked', marks.map(shown), [p.token]);
}
}
look(page, Object.assign({ chance: s.chance }, v), name + ' after ' + k + ' presses');
});
});
// the opening state is the one the lesson shows first
const first = open();
look(first, Object.assign({ chance: false }, fx.scenarios[0].views[0]), 'opening state');
want('options', first['wbw-q'].children.map((o) => [o.value, o._text]), fx.options);
want('nothing spoken yet', first['wbw-live']._text, '');
// Play runs the loop on a clock and speaks only at the end; Pause stops it; Replay starts over
const play = open();
fire(play['wbw-play']);
want('timer started', [play.timers.length, play['wbw-play']._text], [1, 'Pause']);
want('play paces itself', play.timers[0].ms >= 100 && play.timers[0].ms <= 1000, true);
tick(play); tick(play); tick(play);
want('three ticks, three passes', play['wbw-count']._text, 'Passes so far: 3');
want('silent while playing', play['wbw-live']._text, '');
fire(play['wbw-play']);
want('paused', [play.timers.length, play['wbw-play']._text, play['wbw-count']._text], [0, 'Play', 'Passes so far: 3']);
fire(play['wbw-play']);
let guard = 0;
while (play.timers.length && guard++ < 100) tick(play);
look(play, Object.assign({ chance: false }, fx.scenarios[0].views[fx.scenarios[0].views.length - 1]), 'played to the end');
want('play ends', [play.timers.length, play['wbw-play']._text, play['wbw-live']._text], [0, 'Replay', fx.finished]);
fire(play['wbw-play']);
want('play again starts over', [play['wbw-count']._text, play.timers.length, shown(play['wbw-reply'])], ['Passes so far: 0', 1, 'Nothing yet.']);
// Next token stops a running play; Reset empties the reply; a new question starts over
const mix = open();
fire(mix['wbw-play']); tick(mix);
fire(mix['wbw-next']);
want('Next token stops play', [mix.timers.length, mix['wbw-count']._text, mix['wbw-play']._text], [0, 'Passes so far: 2', 'Play']);
fire(mix['wbw-reset']);
look(mix, Object.assign({ chance: false }, fx.scenarios[0].views[0]), 'after Reset');
fire(mix['wbw-next']);
mix['wbw-q'].value = '2'; fire(mix['wbw-q'], 'change');
look(mix, Object.assign({ chance: false }, fx.scenarios.find((s) => s.q === 2 && !s.chance && s.cap === 40).views[0]), 'new question');
// settings are kept inside their range and written back
const odd = open();
[['wbw-cap', '2', '4'], ['wbw-cap', '99', '40'], ['wbw-cap', '', '40'], ['wbw-cap', '12.7', '12'], ['wbw-seed', '-5', '0'], ['wbw-seed', '1e12', '999999999'], ['wbw-seed', '', '1']]
.forEach(([id, typed, kept]) => { odd[id].value = typed; fire(odd[id], 'change'); want(id + ' typed ' + JSON.stringify(typed), odd[id].value, kept); });
// the host page theme: only the parent window may set it, and only to light or dark
const themed = open();
const send = (data, source) => themed.messages.forEach((f) => f({ data, source }));
send({ type: 'spicyshark:theme', theme: 'dark' }, { other: false });
send({ type: 'spicyshark:theme', theme: 'purple' }, themed.parent);
send({ type: 'other', theme: 'dark' }, themed.parent);
send(null, themed.parent);
want('ignored messages', themed.root.attrs, {});
send({ type: 'spicyshark:theme', theme: 'dark' }, themed.parent);
want('theme from the parent', themed.root.attrs, { 'data-theme': 'dark' });
send({ type: 'spicyshark:theme', theme: 'light' }, themed.parent);
want('light again', themed.root.attrs, { 'data-theme': 'light' });
console.log(JSON.stringify({ scenarios: fx.scenarios.length, bad: bad.slice(0, 6) }));
process.exit(bad.length ? 1 : 0);
"""
# Corpora a reader might paste, and questions the corpus never saw: words that are also JavaScript object keys,
# non-ASCII text, an empty corpus. The page itself always uses the lesson corpus.
CORPORA = {
"lesson": CORPUS,
"odd words": "constructor is __proto__ .\nhasOwnProperty is toString !\nThe café is open, isn't it?\nNo. 2026 is 2026.\n\n\n",
"one line": "How do I make tea? Boil some water.\n",
"empty": "",
"one word": "x\n",
}
QUESTIONS = [p["prompt"] for p in PROMPTS] + ["constructor", "", "is __proto__", "zzz qqq zzz", "How do I make", "Café is"]
SEEDS = [None, 0, 1, 2, 3, 7, 12345, 999999999, 2147483645, 2147483646, -5]
CAPS = [1, 4, 10, 40]
TOKEN_CASES = ["How do I make tea?", "", " \t ", "a_b-c", "éè", "end.", "x" * 40, "A.B,C!D?E", "isn't it? 2026!"]
OPTIONS = [[str(i), p["prompt"]] for i, p in enumerate(PROMPTS)]
def plural(n, one, many):
return f"{n} {one if n == 1 else many}"
def passes_of(prompt, cap, seed):
return generate(MODEL, prompt, cap, seed)
def view(passes, k):
"""What the page shows after k presses of Next token: the reply, the count, and then the chances for the next pass or the closing line."""
shown = {"reply": reply_of(passes[:k]) or "Nothing yet.", "count": f"Passes so far: {k}"}
if k < len(passes):
p = passes[k]
looked = f': "{' '.join(p['looked_at'])}"' if p["looked_at"] else ""
shown["looked"] = f"Pass {k + 1} looks at the last {plural(len(p['looked_at']), 'token', 'tokens')}{looked}. Chances for the next token:"
shown["rows"] = [["end mark" if token == END else token, f"{count} of {p['total']} · {int(100 * count / p['total'] + 0.5)}%", 100 * count / p["total"]]
for token, count in p["ranked"]]
elif passes[-1]["token"] == END:
shown["closing"] = f"The model chose the end mark, so the reply is finished: {plural(k - 1, 'token', 'tokens')} and the end mark took {plural(k, 'pass', 'passes')}."
shown["kind"] = "end"
else:
shown["closing"] = f"Stopped at the length cap: {plural(k, 'token', 'tokens')} took {plural(k, 'pass', 'passes')}, and the reply is cut off."
shown["kind"] = "cap"
return shown
def spoken(pass_number, token):
return f"Pass {pass_number}: added " + ("the end mark. The reply is finished." if token == END else f"{token}.")
def scenario(q, chance, seed, cap):
passes = passes_of(PROMPTS[q]["prompt"], cap, seed if chance else None)
return {"q": q, "chance": chance, "seed": seed, "cap": cap, "views": [view(passes, k) for k in range(len(passes) + 1)],
"passes": [{"token": p["token"], "spoken": spoken(i + 1, p["token"])} for i, p in enumerate(passes)]}
def run_node(tmp_path, script, fixture, args):
(tmp_path / "fixture.json").write_text(json.dumps(fixture), encoding="utf-8")
(tmp_path / "check.mjs").write_text(script, encoding="utf-8")
run = subprocess.run(["node", str(tmp_path / "check.mjs"), *args(tmp_path)], capture_output=True, text=True, timeout=120)
assert run.returncode == 0, run.stdout + run.stderr
return json.loads(run.stdout)
def test_widget_data_blocks_match_corpus_and_settings():
"""Guards a corpus, question or constant edit that was not copied into the widget (run embed_widget.py)."""
assert embed(HTML) == HTML
def test_embed_rewrites_a_widget_whose_data_blocks_drifted():
"""Guards embed() skipping a block: a hand-edited corpus line or question must be put back."""
drifted = HTML.replace("Boil some water", "Boil some milk", 1).replace('"Egg"', '"Eggs"', 1)
assert drifted != HTML and embed(drifted) == HTML
def test_every_element_the_script_looks_up_exists_inside_the_root():
"""Guards a widget that never starts: the parity run covers only the rules, so a misspelt id passed there and threw in the browser."""
root = HTML[HTML.index('<div class="wbw-root">'):HTML.index("<script")]
script = HTML[HTML.index("<script>"):]
lookups = set(re.findall(r"\$\('([\w-]+)'\)", script))
handlers = re.search(r"^\s*\['wbw-chance'.*$", script, re.M)
assert handlers, "the handler list that starts with 'wbw-chance' has moved: update this test"
lookups |= set(re.findall(r"'(wbw-[\w-]+)'", handlers.group(0)))
inside = lookups - {"wbw-config", "wbw-corpus"}
assert inside and all(f'id="{i}"' in root for i in inside), sorted(i for i in inside if f'id="{i}"' not in root)
assert 'id="wbw-config"' in HTML and 'id="wbw-corpus"' in HTML
def test_corpus_is_safe_to_embed_in_a_script_element():
assert "\r" not in CORPUS and "</script" not in CORPUS.lower() and "<!--" not in CORPUS
assert "<" not in CONFIG_BLOCK.search(HTML).group(2)
def test_widget_is_small_and_asks_for_nothing_online():
assert len(HTML.splitlines()) < 300 and len(HTML.encode("utf-8")) < 60 * 1024
page = CORPUS_BLOCK.sub("", CONFIG_BLOCK.sub("", HTML))
for banned in ("http://", "https://", "fetch(", "XMLHttpRequest", "WebSocket", "sendBeacon", "import("):
assert banned not in page, banned
assert " src=" not in page and "@import" not in page
def test_widget_has_no_form_element():
"""Guards a dead button: the lesson page loads widget.html in a sandboxed frame, where a form cannot submit."""
assert "<form" not in HTML and 'type="submit"' not in HTML
def test_widget_shows_at_most_five_controls_and_hides_the_rest():
"""Guards the first screen at 390 px: more than five controls push the chances off it. Extra settings stay in a closed details element."""
root = HTML[HTML.index('<div class="wbw-root">'):HTML.index("<script")]
outside = re.sub(r"<details>.*?</details>", "", root, flags=re.S)
assert len(re.findall(r"<(?:select|button|input|textarea)\b", outside)) <= 5
assert "<details><summary>More settings</summary>" in root and "<details open" not in root
def test_widget_keeps_the_site_theme_listener_and_palette():
assert "d.type !== 'spicyshark:theme'" in HTML and "e.source !== parent" in HTML and "parent === window" in HTML
for pair in ("light-dark(#ffffff, #0a0f17)", "light-dark(#ebf0f6, #131b28)", "light-dark(#0a1626, #f4f7fb)",
"light-dark(#44566b, #a3b1c4)", "light-dark(#d2dae4, #1c2635)", "light-dark(#66768a, #7a8a9f)",
"light-dark(#f6f8fb, #05070c)", "--green: light-dark(#00855F, #009E73)", "--orange: light-dark(#b36200, #E69F00)",
":root[data-theme=light]", ":root[data-theme=dark]"):
assert pair in HTML, pair
for banned in ("#0072B2", "#56B4E9", "#D55E00"):
assert banned.lower() not in HTML.lower()
def test_widget_rules_give_the_passes_of_the_python_toy(tmp_path):
"""Guards the widget's JavaScript drifting from toy_gen.py: the lesson prints Python's passes and the reader sees the
widget's, so the two must agree on every pass (text before, chances, token) for odd corpora, unseen questions, every
seed and every cap."""
corpora = []
for name, text in CORPORA.items():
model = train(lines_of(text))
runs = []
for question in QUESTIONS:
for seed in SEEDS:
for cap in CAPS:
passes = generate(model, question, cap, seed)
runs.append({"prompt": question, "seed": seed, "cap": cap, "reply": reply_of(passes),
"passes": [{"before": p["before"], "looked": p["looked_at"], "ranked": [list(r) for r in p["ranked"]],
"total": p["total"], "token": p["token"]} for p in passes]})
corpora.append({"name": name, "text": text, "sizes": [len(table) for table in model], "runs": runs})
seed_values = SEEDS[1:] + [2147483647, 2147483648, 40, 4000000000]
seeds = [{"seed": seed, "state": seed_state(seed)} for seed in seed_values]
steps = []
for state in [seed_state(seed) for seed in seed_values] + [1, 2, 2147483646, 399268537]:
nxt, u = seed_step(state)
steps.append({"state": state, "next": nxt, "u": u})
ranked, total = [("a", 3), ("b", 2), ("c", 1)], 6
picks = [{"ranked": ranked, "total": total, "u": u, "token": pick(ranked, total, u)} for u in (None, 0.0, 0.2, 0.5, 0.51, 0.83, 0.9999999)]
fixture = {"tokens": [[t, tokenize(t)] for t in TOKEN_CASES], "seeds": seeds, "steps": steps, "picks": picks, "corpora": corpora}
result = run_node(tmp_path, PARITY, fixture, lambda d: [str(d / "fixture.json"), str(WIDGET)])
assert result["bad"] == [] and result["checks"] > 1000
assert any(len(r["passes"]) == 31 for r in corpora[0]["runs"]) and any(r["reply"] == "" for r in corpora[3]["runs"])
def test_widget_page_shows_what_the_lesson_names(tmp_path):
"""Guards the part a reader sees: the opening state (Boil 4 of 5, Heat 1 of 5), one more token per press, the pass
count, the marked newest token, the closing line for an end mark and for the cap, Play, Pause, Reset and the settings."""
scenarios = [scenario(0, False, 1, 40)] + [scenario(q, False, 1, 40) for q in (1, 2, 3)]
scenarios += [scenario(0, False, 1, 4), scenario(0, False, 1, 10)]
scenarios += [scenario(q, True, seed, 40) for q in range(4) for seed in (1, 2, 7, 12345, 0)]
scenarios += [scenario(0, True, 3, 6), scenario(3, True, 999999999, 40)]
opening = scenarios[0]["views"][0]
assert opening["rows"][0][:2] == ["Boil", "4 of 5 · 80%"] and opening["rows"][1][:2] == ["Heat", "1 of 5 · 20%"]
assert opening["looked"] == 'Pass 1 looks at the last 3 tokens: "make tea ?". Chances for the next token:' and opening["reply"] == "Nothing yet."
assert scenarios[0]["views"][-1]["closing"].endswith("30 tokens and the end mark took 31 passes.")
assert max(len(v["rows"]) for s in scenarios for v in s["views"] if "rows" in v) >= 2
ids = sorted(set(re.findall(r'id="(wbw-[\w-]+)"', HTML)))
fixture = {"ids": ids, "end": END, "scenarios": scenarios, "options": OPTIONS, "finished": spoken(31, END),
"text": reply_of(passes_of(PROMPTS[0]["prompt"], 40, None))}
result = run_node(tmp_path, PAGE, fixture, lambda d: [str(WIDGET), str(d / "fixture.json")])
assert result == {"scenarios": len(scenarios), "bad": []}