Module 1 · LLM Fundamentals
LLM Fundamentals
Module 0 gave you shared/llm_client.py — one function that calls any provider and logs what it cost. This module is about understanding what's actually happening inside that call well enough to predict the cost and the latency before you make it, not just read them off afterward: what a token is, what temperature and streaming actually change, and the two completely different kinds of caching that cut cost without you noticing.
Same assumptions as Module 0: everything runs from the ai-beyond-fundamentals/ repo root with the venv activated. The model names below (claude-haiku-5-5, gpt-5.4-mini, gemini-flash-lite-latest) are the same "check before you paste" caveat from Module 0 — they will go stale, so verify before trusting them.
| Step | Approx. time |
|---|---|
| Lab 1 · Token intuition | about 15 to 30 min |
| Lab 2 · Context builds up | about 22 to 44 min |
| Lab 3 · Temperature sweep | about 3 to 6 min |
| Lab 4 · Streaming vs. blocking | about 2 to 4 min |
| Lab 5 · Prompt caching | about 4 to 8 min |
| Lab 6 · Semantic cache | about 6 to 12 min |
| Project Part 1 · Terminal chat | about 4 to 8 min |
| Project Part 2 · Web chat (Advanced) | about 8 to 16 min |
| Module total | about 1 to 2 hours |
The low end is a timed first run through the module with Module 0 already done. The high end doubles it for a first-time learner who reads the code as they go. Reading the concepts above is not counted, and neither is time spent on a bug you hit.
A virtual environment only stays active in the terminal window where you activated it. Close the window, reboot, or open a new one, and you are back on your system Python, which doesn't have the course's packages. Each time you sit down to work, do this before running any code below.
1. Go to the repo root and activate the venv.
cd ~/path/to/ai-beyond-fundamentals
source .venv/bin/activatecd path\to\ai-beyond-fundamentals
.venv\Scripts\Activate.ps1Your prompt should now start with (.venv). If PowerShell refuses to run the script, allow it for this window only with Set-ExecutionPolicy -Scope Process RemoteSigned and try again.
2. Confirm you are using the venv's Python.
which pythonGet-Command pythonThe path should end in .venv/bin/python (or .venv\Scripts\python.exe on Windows). Anything else means the venv isn't active, and you will see ModuleNotFoundError when you run the code.
3. Check the things that change between sessions.
- If you created or rotated an API key since your last session, use a new terminal window.
load_dotenv()does not override keys already set in a window's environment, so an old window keeps the old key and you get a401. - If you are using the local model, check Ollama is running with
ollama list. A connection error means start the Ollama app, or runollama servein its own terminal. - VS Code users: if the Python interpreter is set to the
.venv(Command Palette, "Python: Select Interpreter"), new integrated terminals activate it for you. You can still confirm with step 2.
Concepts
Messages, roles, and system prompts — and why the shapes differ
Every provider's chat API takes a list of turns, each tagged with a role: user for what you send, assistant for what the model said, and a system prompt that sets standing behavior ("answer only from the provided text," "respond in JSON") before the conversation starts.
Where that system prompt goes is the one thing that genuinely differs per provider, and it's exactly the kind of difference a shared client has to hide. (Module 0's call() sends no system prompt yet, only the user's message; the chat client you build in this module's project adds one.) Anthropic takes system as its own top-level parameter, separate from messages. OpenAI (and Ollama, through the same OpenAI-compatible SDK) treats it as just another message with role: "system". Gemini takes a top-level systemInstruction (like Anthropic's system, not a message), and its own role for the model's turns is "model", not "assistant", which trips people up the first time they build multi-turn history by hand.
The table below puts the shapes side by side. Each link goes to that provider's own API reference, where you can see the complete request payload with every optional field.
| Provider | System prompt goes in | Role names | Full payload example |
|---|---|---|---|
| Claude (Anthropic) | Top-level system parameter | user, assistant | Create a Message · API overview |
| OpenAI | A message with role: "system" (newer models also accept "developer") | system, user, assistant | Chat Completions: create · API overview |
| Gemini (Google) | Top-level systemInstruction field; turns go in contents as parts | user, model | models.generateContent · API overview |
| Ollama (local) | Same as OpenAI when you use its OpenAI-compatible endpoint | system, user, assistant | OpenAI compatibility · Native chat API |
Tokens, context windows, and cost
A token is the unit a model actually reads and writes — usually a few characters or a word-piece, not a whole word. Different providers use different tokenizers, so the same sentence costs a different number of tokens depending on who's counting; Lab 1 makes that concrete instead of abstract.
The context window is the most tokens a single call can hold — input and output combined, for most providers. Go over it and the call fails outright; it doesn't quietly truncate for you. Cost is just arithmetic on top of the token count, and it's the exact formula already sitting in Module 0's call(): input_tokens / 1_000_000 * price_in + output_tokens / 1_000_000 * price_out. Output tokens are usually the expensive half of that sum — providers price them several times higher than input, because generating a token costs more compute than reading one.
Sampling settings: temperature, top-p, max tokens, stop sequences
Temperature controls how much the model is allowed to deviate from its single most-likely next token — near 0 it's close to deterministic, higher values let lower-probability tokens win sometimes. top-p is a second, related knob (nucleus sampling: only consider tokens whose combined probability covers the top p) — most people pick one of the two to tune and leave the other at its default rather than fighting both at once.
What range is allowed depends on the provider, and on the model. We tested the models this course uses:
| Provider and model | Temperature |
|---|---|
OpenAI gpt-5.4-mini | 0 to 2. A value above 2 returns a 400 error. |
Gemini gemini-flash-lite-latest | 0 to 2. A value above 2 returns a 400 error. |
| Ollama (local model) | Accepted every value we tried, up to 5, with no error. Don't read that as meaningful: very high values just produce noise. |
Anthropic claude-haiku-5-5 | Not accepted at all. The API returns a 400 (temperature: Extra inputs are not permitted), and so do top_p and top_k. Earlier Claude models accepted 0 to 1, so check the docs for the model you use. |
That is why Lab 3 sweeps temperature on OpenAI and not on Claude. It is also a reason to read a model's API reference before copying settings from an older tutorial.
max_tokens is a hard ceiling on the reply's length, not a target — the model can stop earlier on its own, but never later. It doesn't affect input cost at all, only caps how much output you could be charged for. Stop sequences end generation early the instant a given string appears, useful for formats where you know the reply should end at a specific marker.
Streaming: time to first token vs. total latency
A blocking call waits for the entire reply before returning anything. A streaming call returns the first chunk as soon as the model has generated it, then keeps delivering chunks as they're produced. Time to first token (TTFT) is how long the first chunk takes; total latency is how long the whole reply takes either way. Streaming barely changes total latency — the model still has to generate every token — but it changes what the user experiences: they see something moving almost immediately instead of staring at a blank screen for the full duration.
The object you iterate to get those chunks is different per SDK: Anthropic gives you a context manager with a .text_stream you loop over; OpenAI (and Ollama) give you an iterator of chunk objects where the text lives at chunk.choices[0].delta.content; Gemini gives you an iterator where it's just chunk.text.
Caching: two different mechanisms, easy to conflate
Prompt caching is the provider reusing its own internal computation (the "KV cache") for a prefix of tokens it has already processed, so a second call that repeats the same long system prompt or document doesn't re-pay for reprocessing it. It requires an identical prefix — one character different and it's a cache miss. Anthropic requires you to mark the cacheable block explicitly with cache_control; OpenAI does it automatically for any prompt over roughly 1,024 tokens, no code change required.
Semantic caching is a completely different, application-level idea: you embed the question itself and compare it by similarity to questions you've answered before, returning the stored answer if it's close enough — even if the wording doesn't match at all. It catches paraphrases that prompt caching can't touch, because prompt caching never looks at whether two questions mean the same thing, only whether the bytes match.
Embeddings and similarity — how a computer compares meaning
A chat model reads text and writes text. An embedding model does something different: it reads text and returns a long list of numbers, called a vector, that stands for what the text means. Texts with similar meaning get vectors that point in similar directions, even if they share almost no words. "Which city is the capital of France?" and "What is the capital of France?" end up close together; "How do I bake a chocolate cake?" ends up far away. The model text-embedding-3-small that Lab 6 uses returns 1,536 numbers for every piece of text you give it.
To compare two vectors you use cosine similarity: the cosine of the angle between them. Identical direction scores 1, and unrelated texts score near 0. In Lab 6 your scores run from 0.109 (a chocolate cake against a question about France) to 0.972 (two ways of asking about spider legs). The score is a measure of closeness, not a probability. A score of 0.8 does not mean "80% likely to be the same question", which is why the threshold needs testing, not guessing.
Three things about embedding models that are easy to miss:
- They are separate models from the chat models. They have their own endpoint and name, they return numbers instead of text, and they bill only for input tokens because there is nothing generated. Providers offer their own: see the OpenAI, Gemini and Ollama embeddings guides.
- Vectors from different embedding models can't be compared. Each model has its own space with its own number of dimensions. If you switch models, you have to re-embed everything you stored.
- Thresholds don't carry over. The same pair of sentences scores differently under a different model, so a cut-off you tuned for one model has to be re-tuned for the next. Lab 6 finds one for
text-embedding-3-smallonly.
You will use the same idea again in Module 3: retrieval-augmented generation embeds your documents once, embeds each question, and fetches the nearest chunks by cosine similarity.
Picking a model: quality, speed, cost — pick two
There's no universally "best" model, only a triangle you're trading off inside: bigger and smarter models cost more and respond slower; smaller "lite" or "mini" variants are cheaper and faster but weaker on hard tasks. The practical move (and the subject of Module 7's router lab) is routing easy requests to a cheap model and reserving the expensive one for requests that actually need it.
That tradeoff isn't hypothetical here: both Module 0 and this module use gemini-flash-lite-latest rather than gemini-flash-latest, specifically because the lite variant answered reliably and fast under real testing (under a second for a short reply), while the larger flash models were overloaded and timed out with 503 and 504 errors. Cheaper and faster isn't always a downgrade — sometimes it's just the better fit for what you're doing.
Hands-on Labs
- Install
tiktoken— it's OpenAI's own tokenizer library, the only way to count OpenAI-style tokens without making a real API call.pip install tiktoken pip freeze > requirements.txt - Create
chatbot/token_intuition.py(code chatbot/token_intuition.py, ornano). Before running anything, write your own guess for each sample as a comment — the point of this lab is checking your intuition, not just reading numbers off a script.# Guess first! Write your estimate for each sample below, then run this # file and see how close you were. # English: ? Code: ? JSON: ? Emoji: ? Japanese: ? import os from dotenv import load_dotenv load_dotenv() SAMPLES = { "English": "The quick brown fox jumps over the lazy dog.", "Code": "def fib(n):\n return n if n < 2 else fib(n-1) + fib(n-2)", "JSON": '{"name": "Ada", "age": 36, "skills": ["math", "engineering"]}', "Emoji": "๐๐ฅ๐๐๐ฏ๐๐๐โจ๐ค", "Japanese": "็ซใฏใใใใใงใใไปๆฅใฏ่ฏใๅคฉๆฐใงใใญใ", } def anthropic_tokens(text): from anthropic import Anthropic client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"]) resp = client.messages.count_tokens( model="claude-haiku-5-5", messages=[{"role": "user", "content": text}], ) return resp.input_tokens def openai_tokens(text): import tiktoken enc = tiktoken.encoding_for_model("gpt-5.4-mini") return len(enc.encode(text)) def gemini_tokens(text): from google import genai client = genai.Client(api_key=os.environ["GEMINI_API_KEY"]) resp = client.models.count_tokens(model="gemini-flash-lite-latest", contents=text) return resp.total_tokens if __name__ == "__main__": for label, text in SAMPLES.items(): a, o, g = anthropic_tokens(text), openai_tokens(text), gemini_tokens(text) print(f"{label:10} anthropic={a:3} openai={o:3} gemini={g:3}") - Run it and compare the three columns, then compare them to your own guesses.
python chatbot/token_intuition.py
anthropic=26 openai=10 gemini=11, Japanese came out anthropic=28 openai=13 gemini=11 — expect your numbers to be close, not identical, since tokenizers change between model versions.A chat API has no memory. Every turn, your code sends the entire conversation again: the system prompt, every earlier question, and every earlier answer. This lab makes that visible and shows why clear, compact prompts are a cost and quality skill, not just a style preference.
- Create
chatbot/context_growth.py(code chatbot/context_growth.py, ornano) and paste in the script. It asks the same four gardening questions twice. The verbose run uses a long system prompt and rambling questions. The compact run asks for the same things in a few words, with a short system prompt that asks for short answers. Each turn appends tomessagesand resends everything, then prints that turn's token usage. Every call is logged throughlog_call, so it also lands inshared/calls.jsonl. Notethinking={"type": "disabled"}on the call: Claude Haiku 5.5 can "think" before it answers, and thinking tokens count againstmax_tokens. With the wordy system prompt it will often spend the entire 400 tokens thinking and return no text at all, sonext(...)raisesStopIteration. Turning thinking off also keeps the token counts about the conversation and not hidden reasoning.import os, sys, time, pathlib from dotenv import load_dotenv load_dotenv() from anthropic import Anthropic sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent / "shared")) from llm_log import log_call client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"]) MODEL = "claude-haiku-5-5" PRICE_IN, PRICE_OUT = 0.10, 0.50 # $ per 1M tokens, from Module 0's models.yaml # The same four questions, asked two ways. The compact version says the same # thing in far fewer words, with a short system prompt that asks for short answers. COMPACT = { "system": "You are a gardening assistant. Answer in at most two sentences.", "turns": [ "I have a sunny 4x8 ft bed. What should I plant in spring?", "Which of those is easiest for a beginner?", "How far apart should I space it?", "And how often do I water it?", ], } VERBOSE = { "system": ( "You are a friendly, thorough, and highly knowledgeable gardening assistant. " "Always greet the user warmly, explain your reasoning step by step, mention any " "relevant caveats, regional differences, seasonal considerations, and common " "mistakes, offer alternative approaches where possible, and finish by inviting " "follow-up questions. Be comprehensive rather than brief." ), "turns": [ "Hi there! So I've been thinking about getting into gardening for a while now, and " "I finally have a bit of free time. I have a raised bed in my backyard, it's about " "4 feet by 8 feet, and it gets a lot of sun, I think maybe six or seven hours on " "a good day. Anyway, I was wondering, what are some things I could plant this spring?", "Thanks, that's really helpful! There's a lot there. Of all of the ones you just " "mentioned, which one do you think would be the easiest for someone who has honestly " "never grown anything before in their life? I'd hate to fail on the first try.", "Great, I think I'll go with that one. Now, I'm not totally sure about the spacing, " "so could you tell me how far apart I should put the plants so they have enough room?", "Okay, that makes sense. One last thing, I'm a little worried about overwatering or " "underwatering, so how often should I actually be watering it, roughly speaking?", ], } def run(label, scenario): messages = [] total_in = total_out = 0 total_cost = 0.0 print(f"--- {label} ---") for turn, user_text in enumerate(scenario["turns"], 1): messages.append({"role": "user", "content": user_text}) start = time.time() resp = client.messages.create( model=MODEL, max_tokens=400, system=scenario["system"], messages=messages, thinking={"type": "disabled"}, # otherwise Haiku can spend the whole token budget thinking and return no text ) elapsed_ms = (time.time() - start) * 1000 reply = next(b.text for b in resp.content if b.type == "text") messages.append({"role": "assistant", "content": reply}) i, o = resp.usage.input_tokens, resp.usage.output_tokens cost = i / 1_000_000 * PRICE_IN + o / 1_000_000 * PRICE_OUT log_call("anthropic", f"{MODEL}-{label}", i, o, elapsed_ms, cost) total_in, total_out, total_cost = total_in + i, total_out + o, total_cost + cost print(f" turn {turn}: input={i:5} output={o:4} cost=${cost:.6f}") print(f" TOTAL : input={total_in:5} output={total_out:4} cost=${total_cost:.6f}\n") return total_in, total_out, total_cost if __name__ == "__main__": v = run("verbose", VERBOSE) c = run("compact", COMPACT) print(f"verbose / compact: input {v[0] / c[0]:.1f}x, output {v[1] / c[1]:.1f}x, cost {v[2] / c[2]:.1f}x") - Run it and read the two tables.
python chatbot/context_growth.py - Before you read our numbers, answer three questions from your own output: (a) In the compact run, the user's turn 4 question is only a few words. Why is the turn 4
inputcount still several times larger than turn 1's? (b) Where did the verbose run's turn 2 jump come from? (c) Which costs more, the system prompt or the conversation history, by turn 4?
input should rise on every turn in both runs, even though your own new question is short each time. That growth is the history: your previous questions and the model's previous answers are sent again, and billed again. In our own run (Claude Haiku 5.5): the compact run's input went 54, 179, 269, 360 across the four turns; the verbose run's went 239, 720, 1,179, 1,637. Each verbose turn added about 460 tokens, because the model's 400-token answer and your next question were both added to the history. The verbose run used about 4.4× the input tokens, 4.7× the output tokens and 4.6× the cost of the compact run, for the same four questions. Two further things to notice: the system prompt is resent and billed on every single turn, and the verbose answers hit the max_tokens=400 ceiling every time, so they were cut off mid-sentence. Your exact numbers will differ, and models don't obey "at most two sentences" perfectly, but the shape holds: cost per turn rises as the conversation grows, so total cost grows faster than the number of turns.
Wordy prompts cost you three times: they are billed when you send them, they are billed again on every later turn, and they invite long answers that are also billed again on every later turn. Long conversations also push toward the context window limit, where the call fails. The fixes are the skills this course keeps coming back to: say what you want in fewer words, ask for short answers when short is enough, and start a fresh conversation (or summarize the old one) instead of dragging a long history along.
- Create
chatbot/temperature_sweep.pyand paste in a function that calls the same promptntimes at a given temperature and collects the answers.import os from dotenv import load_dotenv load_dotenv() from openai import OpenAI client = OpenAI(api_key=os.environ["OPENAI_API_KEY"]) PROMPT = "Name an animal. Reply with just the word." def sweep(temperature, n=10): answers = [] for _ in range(n): resp = client.chat.completions.create( model="gpt-5.4-mini", max_completion_tokens=15, temperature=temperature, messages=[{"role": "user", "content": PROMPT}], ) answers.append(resp.choices[0].message.content.strip().lower()) # lower() so "Elephant" and "elephant" count as one answer return answers if __name__ == "__main__": for temp in [0, 1, 1.8]: answers = sweep(temp) print(f"temp={temp}: {answers} -> {len(set(answers))} distinct") - Run it. Prompt choice matters here more than you'd expect: a narrowly-biased prompt like "name one primary color" stays stuck on whichever answer the model considers overwhelmingly likely (almost always "Red") even at the highest temperature, across all 10 tries — that's a real result if it happens to you, not a bug, it just doesn't make for an interesting lab. "Name an animal" has enough plausible alternatives that temperature actually has room to work.
python chatbot/temperature_sweep.py
temp=0 → 1 distinct (10/10 "elephant"); temp=1 → 3 distinct (seven "elephant", two "dolphin", one "octopus"); temp=1.8 → 5 distinct ("elephant", "dolphin", "otter", "aardvark", "octopus"). Newer OpenAI models reject max_tokens and want max_completion_tokens instead, which is why the call above uses that name. Don't expect temp=0 to be perfectly identical every single run forever; providers don't guarantee strict determinism even at 0. Sampling is random, so your counts will differ from ours, especially at temp=1, where ten samples can easily come back as 1, 2 or 3 distinct answers. The trend across the three temperatures is what matters; raise n to 30 if you want a clearer picture.Notice "elephant" and "dolphin" still lead even at temp=1.8 — that's not noise, it's the real signal. The model's underlying probability distribution for "name an animal" isn't flat across every species; a handful of prototypical answers that show up constantly in training data (elephant, dolphin, lion, tiger) carry most of the probability mass, with thousands of other real animals each holding a tiny sliver of the rest. Temperature rescales that distribution so lower-probability tokens can win sometimes — which is why you see 4 distinct answers instead of 1 — but it doesn't flatten the distribution to uniform. If the two biggest peaks still hold most of the mass after rescaling, they'll still win most draws. "Higher temperature" means "more exposure to how peaked the real distribution already is," not "more randomness" in any general sense.
- Create
chatbot/streaming_vs_blocking.pyand paste in both versions of the same call.import os, time from dotenv import load_dotenv load_dotenv() from anthropic import Anthropic client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"]) PROMPT = "Write a 300-word story about a lighthouse keeper." def call_blocking(show=False): start = time.time() client.messages.create( model="claude-haiku-5-5", max_tokens=500, thinking={"type": "disabled"}, messages=[{"role": "user", "content": PROMPT}], ) elapsed = time.time() - start if show: print(next(b.text for b in resp.content if b.type == "text")) return elapsed def call_streaming(show=False): start = time.time() ttft = None with client.messages.stream( model="claude-haiku-5-5", max_tokens=500, thinking={"type": "disabled"}, messages=[{"role": "user", "content": PROMPT}], ) as stream: for chunk in stream.text_stream: if ttft is None: ttft = time.time() - start if show: print(chunk, end="", flush=True) if show: print() return ttft, time.time() - start if __name__ == "__main__": for i in range(3): blocking_total = call_blocking() ttft, streaming_total = call_streaming() print(f"run {i+1}: blocking={blocking_total:.2f}s | " f"streaming ttft={ttft:.2f}s total={streaming_total:.2f}s") print("\n--- Now watch it. Blocking first: nothing appears until the reply is finished. ---") call_blocking(show=True) print("\n--- Streaming: words appear as they are generated. ---") call_streaming(show=True) - Run it. The three timed runs come first, then the script calls the model once more for you to watch: the blocking version prints nothing for several seconds and then drops the whole story at once, while the streaming version starts writing within about a second. Then look at the gap between
blockingandttftin the numbers — that gap is exactly how much sooner a streaming UI would show the user something.python chatbot/streaming_vs_blocking.py
ttft should be noticeably smaller than blocking's total time, while streaming total should land close to blocking's time — streaming doesn't make the model faster, it just shows you the result sooner. In our own run, across three repeats: blocking averaged ~4.6s total; streaming's first token arrived in 0.6–1.7s (usually under a second). Same total work, but the reader sees words about 4 to 7× sooner. Both calls pass thinking={"type": "disabled"}: with thinking on, Haiku can spend part of its budget on hidden reasoning first, so "first token" would mean the first thinking token, or no text at all.- Grab a real document — any article, README, or chapter, at least a few hundred words across multiple paragraphs — and save it as
chatbot/long_doc.txt. Caching only kicks in above a minimum prompt length (512 tokens for Claude Haiku 5.5 and Sonnet 5.5, per Anthropic's prompt caching docs; shorter prompts are processed without caching and no error comes back), so a single short paragraph won't show anything; a ~700-word piece of ordinary prose clears that comfortably (English runs roughly 2 tokens per word). - Create
chatbot/prompt_caching.py. It asks the same five questions about the document twice — once withcache_controlset, once without — and logs every call through Module 0'slog_callso the comparison lives inshared/calls.jsonl, not just stdout.import os, sys, time, pathlib from dotenv import load_dotenv load_dotenv() sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent / "shared")) from llm_log import log_call DOC = pathlib.Path(__file__).parent.joinpath("long_doc.txt").read_text() QUESTIONS = [ "What is the main topic of this document?", "Summarize it in one sentence.", "What is the second paragraph about?", "Does it mention any numbers? Which ones?", "What tone does it use?", ] def ask(cached): from anthropic import Anthropic client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"]) system = [{"type": "text", "text": DOC}] if cached: system[0]["cache_control"] = {"type": "ephemeral"} label = "claude-haiku-5-5-cached" if cached else "claude-haiku-5-5-uncached" for q in QUESTIONS: start = time.time() resp = client.messages.create( model="claude-haiku-5-5", max_tokens=100, thinking={"type": "disabled"}, system=system, messages=[{"role": "user", "content": q}], ) elapsed_ms = (time.time() - start) * 1000 cost = (resp.usage.input_tokens / 1_000_000 * 0.10 + resp.usage.output_tokens / 1_000_000 * 0.50) log_call("anthropic", label, resp.usage.input_tokens, resp.usage.output_tokens, elapsed_ms, cost) print(f" cache_read={resp.usage.cache_read_input_tokens:5} " f"cache_write={resp.usage.cache_creation_input_tokens:5} {elapsed_ms:.0f}ms") if __name__ == "__main__": print("--- without caching ---") ask(cached=False) print("--- with caching ---") ask(cached=True) - Run it, then pull the two batches out of the log and compare total cost.
python chatbot/prompt_caching.py python -c " import json rows = [json.loads(l) for l in open('shared/calls.jsonl')] for label in ['claude-haiku-5-5-uncached', 'claude-haiku-5-5-cached']: batch = [r for r in rows if r['model'] == label] print(label, 'total cost:', sum(r['cost_usd'] for r in batch)) "
cache_read=0 on every question — no caching ever happens without cache_control, no matter how many times you repeat the document. The cached run should show cache_write on question 1 only, then cache_read equal to roughly the document's token count on questions 2–5. In our own run against a 706-word document: question 1 wrote 1,315 tokens to cache; every question after it read all 1,315 back instead of reprocessing them. The token savings are unambiguous — don't expect the same clarity from elapsed_ms at this document size, though: in our run the cached and uncached latencies actually overlapped (roughly 900–1,600ms either way, apart from the first cached call, which also pays to write the cache), because ordinary network and generation variance was bigger than the time saved on reprocessing ~1,300 tokens. Prompt caching's latency win shows up clearly on much longer documents; at modest sizes, trust the token counts and the cost over the stopwatch — in our run the comparison script printed uncached total cost: 0.0008833 vs. cached total cost: 0.0002408, about 3.7× cheaper for the identical five questions.- Create
chatbot/semantic_cache.py. It embeds a small seed set of question/answer pairs, then for any new question finds the closest seed by cosine similarity (plain Python — no numpy needed for three vectors) and returns its answer if the similarity clears a threshold.import os, math from dotenv import load_dotenv load_dotenv() from openai import OpenAI client = OpenAI(api_key=os.environ["OPENAI_API_KEY"]) def embed(texts): resp = client.embeddings.create(model="text-embedding-3-small", input=texts) return [d.embedding for d in resp.data] def cosine_similarity(a, b): dot = sum(x * y for x, y in zip(a, b)) na = math.sqrt(sum(x * x for x in a)) nb = math.sqrt(sum(x * x for x in b)) return dot / (na * nb) SEED = { "What is the capital of France?": "Paris.", "How many legs does a spider have?": "Eight.", "What's the boiling point of water in Celsius?": "100 degrees.", } seed_questions = list(SEED.keys()) seed_vecs = embed(seed_questions) def lookup(question, threshold): q_vec = embed([question])[0] best_score, best_q = 0, None for sq, sv in zip(seed_questions, seed_vecs): score = cosine_similarity(q_vec, sv) if score > best_score: best_score, best_q = score, sq if best_score >= threshold: return SEED[best_q], best_score, best_q return None, best_score, best_q - Add a test set deliberately mixing three kinds of question — true paraphrases of a seed, questions on the same topic as a seed but asking something different, and fully unrelated questions — then sweep a few thresholds and print every score.
TEST_QUESTIONS = [ # paraphrases -- should hit "Which city is the capital of France?", "What's France's capital city?", "How many legs do spiders have?", "What temperature does water boil at, in Celsius?", # same topic, different question -- should NOT hit "What is the population of France?", "How many eyes does a spider have?", "What's the freezing point of water in Celsius?", # unrelated -- should NOT hit "How do I bake a chocolate cake?", "What's the capital of Japan?", ] if __name__ == "__main__": for threshold in [0.95, 0.85, 0.75, 0.6]: print(f"--- threshold={threshold} ---") hits = 0 for q in TEST_QUESTIONS: answer, score, matched = lookup(q, threshold) status = f"HIT (matched {matched!r})" if answer else "miss" print(f" {score:.3f} {q!r} -> {status}") if answer: hits += 1 print(f" {hits}/{len(TEST_QUESTIONS)} hits\n") - Run it and read every threshold's printout, not just the hit count — the bug you're hunting for is a same-topic question matching a seed it shouldn't.
python chatbot/semantic_cache.py
threshold=0.95 was too strict (1/9 hits — real paraphrases missed); threshold=0.85 landed cleanly (3/9 hits, all genuine paraphrases, zero wrong matches); dropping to threshold=0.75 picked up three more hits: one genuine paraphrase ("What's France's capital city?", 0.814, which 0.85 missed) and two that were the real failure mode — "How many eyes does a spider have?" scored 0.790 against the legs question, and "What's the freezing point of water in Celsius?" scored 0.785 against the boiling-point question, so both would have served the wrong answer. That's the exact thing this lab is designed to surface: somewhere between 0.75 and 0.85, wrong answers start slipping through. Confirm it against your own printout rather than trusting this number — it depends on the exact wording of your seed and test questions.Project
Module 0's call(prompt, model) only handles one-shot, non-streaming calls — a chat app needs two things it doesn't have: a full conversation history instead of a single prompt, and streaming instead of waiting for the whole reply. Part 1 builds that as a terminal app. Part 2 is the same logic behind a minimal local web page, which is also where the one genuinely tricky bit of this project lives.
Part 1 · Terminal REPL
- Extend
shared/llm_client.pywith a streaming, multi-turn sibling tocall(). It takes a full[{"role", "content"}, ...]history instead of one prompt string, and anon_chunkcallback it calls once per piece of text as it streams — so a caller can print each piece the moment it arrives instead of waiting for the full reply.# shared/llm_client.py -- add below the existing call()/_dispatch() def stream_chat(messages, model, on_chunk=None, system=None, max_attempts=5): cfg = CONFIG[model] start = time.time() for attempt in range(max_attempts): try: text, usage = _dispatch_stream(cfg["provider"], model, messages, on_chunk, system) break except RATE_LIMIT_ERRORS: if attempt == max_attempts - 1: raise time.sleep(2 ** attempt + random.random()) elapsed_ms = (time.time() - start) * 1000 cost = (usage["input_tokens"] / 1_000_000 * cfg["price_in"] + usage["output_tokens"] / 1_000_000 * cfg["price_out"]) log_call(cfg["provider"], model, usage["input_tokens"], usage["output_tokens"], elapsed_ms, cost) return text, cost, elapsed_ms - Add
_dispatch_streamto the file.stream_chatabove calls it but it doesn't exist yet, so this is the missing half. They split the work the same waycall()and_dispatch()do in Module 0:stream_chathandles the parts every provider shares (retries, timing, cost, logging), and_dispatch_streamholds the provider-specific code that actually streams the reply. Paste the code below directly understream_chat; it has one branch per provider, the same four as_dispatch. Each branch does three things: it opens a streaming request, adds each piece of text to a runningtextstring while handing the piece toon_chunk(that is what lets a terminal print words as they arrive), and returns the finished text with the token counts. Gemini needs its own role remapped ("assistant"→"model") since it doesn't use OpenAI/Anthropic's role names.def _dispatch_stream(provider, model, messages, on_chunk, system=None): if provider == "anthropic": from anthropic import Anthropic client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"]) extra = {"system": system} if system else {} # Anthropic takes system as its own parameter text = "" with client.messages.stream(model=model, max_tokens=500, messages=messages, thinking={"type": "disabled"}, **extra) as stream: for chunk in stream.text_stream: text += chunk if on_chunk: on_chunk(chunk) final = stream.get_final_message() return text, { "input_tokens": final.usage.input_tokens, "output_tokens": final.usage.output_tokens, } if provider in ("openai", "ollama"): from openai import OpenAI if provider == "ollama": client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama") else: client = OpenAI(api_key=os.environ["OPENAI_API_KEY"]) # OpenAI and Ollama want the system prompt as the first message full = ([{"role": "system", "content": system}] if system else []) + messages resp = client.chat.completions.create( model=model, messages=full, stream=True, stream_options={"include_usage": True}, ) text = "" input_tokens = output_tokens = 0 for chunk in resp: if chunk.choices and chunk.choices[0].delta.content: piece = chunk.choices[0].delta.content text += piece if on_chunk: on_chunk(piece) if chunk.usage: input_tokens = chunk.usage.prompt_tokens output_tokens = chunk.usage.completion_tokens return text, {"input_tokens": input_tokens, "output_tokens": output_tokens} if provider == "gemini": from google import genai from google.genai import types client = genai.Client( api_key=os.environ["GEMINI_API_KEY"], http_options=types.HttpOptions(timeout=20000), ) # Gemini uses "model" instead of "assistant" for its own turns contents = [ {"role": "model" if m["role"] == "assistant" else m["role"], "parts": [{"text": m["content"]}]} for m in messages ] # Gemini takes the system prompt in the request config config = types.GenerateContentConfig(system_instruction=system) if system else None text = "" last_chunk = None for chunk in client.models.generate_content_stream(model=model, contents=contents, config=config): if chunk.text: text += chunk.text if on_chunk: on_chunk(chunk.text) last_chunk = chunk usage = last_chunk.usage_metadata if last_chunk else None return text, { "input_tokens": usage.prompt_token_count if usage else 0, "output_tokens": usage.candidates_token_count if usage else 0, } raise ValueError(f"Unknown provider: {provider}") - Create
chatbot/chat_cli.py: a model picker, a loop that keeps growing the message history each turn, and a cost readout after every reply.import sys, pathlib sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent / "shared")) from llm_client import stream_chat, CONFIG MODELS = list(CONFIG.keys()) SYSTEM_PROMPT = "You are a helpful assistant. Keep answers short." # resent on every turn, so keep it lean def pick_model(): print("Models:") for i, name in enumerate(MODELS, 1): print(f" {i}. {name}") choice = input(f"Pick a model [1-{len(MODELS)}]: ").strip() try: return MODELS[int(choice) - 1] except (ValueError, IndexError): print("Not a valid choice, defaulting to the first model.") return MODELS[0] def main(): model = pick_model() print(f"\nChatting with {model}. Type 'quit' to exit.\n") messages = [] session_cost = 0.0 while True: user_input = input("You: ").strip() if user_input.lower() in ("quit", "exit"): break if not user_input: continue messages.append({"role": "user", "content": user_input}) print(f"{model}: ", end="", flush=True) text, cost, elapsed_ms = stream_chat( messages, model, on_chunk=lambda c: print(c, end="", flush=True), system=SYSTEM_PROMPT, ) print() messages.append({"role": "assistant", "content": text}) session_cost += cost print(f" [turn: ${cost:.6f}, {elapsed_ms:.0f}ms | session total: ${session_cost:.6f}]\n") print(f"\nSession total: ${session_cost:.6f}") if __name__ == "__main__": main() - Run it and have an actual back-and-forth — ask it something, then ask a follow-up that only makes sense if it remembered the first answer.
python chatbot/chat_cli.py
messages is actually accumulating; and the session total only goes up, never resets, across the whole conversation.Part 2 (Advanced) · Minimal local web app
Same chat, same stream_chat, running behind a one-page local Flask app instead of a terminal. The interesting problem here isn't the UI — it's that on_chunk is a plain callback with no way to pause itself and hand control back to Flask, which is exactly what an HTTP response generator needs to do between chunks.
- Install Flask.
pip install flask pip freeze > requirements.txt - Create
chatbot/chat_web.py. The fix for the callback problem: runstream_chaton a background thread that pushes each chunk onto aqueue.Queue, and let Flask's response generator just drain that queue —q.get()blocks until a chunk shows up, which is what turns "a callback fired" into "the generator can yield now."# chatbot/chat_web.py import sys, json, queue, pathlib, threading sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent / "shared")) from llm_client import stream_chat, CONFIG from flask import Flask, request, Response app = Flask(__name__) DONE_MARKER = "\x00DONE\x00" ERROR_MARKER = "\x00ERROR\x00" @app.route("/models") def models(): return {"models": list(CONFIG.keys())} @app.route("/chat", methods=["POST"]) def chat(): body = request.get_json() messages, model = body["messages"], body["model"] q = queue.Queue() def worker(): try: def on_chunk(c): q.put(("chunk", c)) text, cost, elapsed_ms = stream_chat(messages, model, on_chunk=on_chunk) q.put(("done", {"cost": cost, "elapsed_ms": elapsed_ms})) except Exception as e: q.put(("error", str(e))) threading.Thread(target=worker, daemon=True).start() def generate(): while True: kind, payload = q.get() if kind == "chunk": yield payload elif kind == "done": yield DONE_MARKER + json.dumps(payload) return elif kind == "error": yield ERROR_MARKER + payload return return Response(generate(), mimetype="text/plain") - Add the
/route and the page itself — a model<select>, a log, an input box, and JS that reads the streamed response body chunk by chunk, splitting out theDONE_MARKERfooter to update the cost meter. The marker is a null byte sequence that will never appear in normal model output, so a plain string search finds it reliably.# add above the two routes in chat_web.py INDEX_HTML = """<!doctype html> <title>Cost-Meter Chat</title> <select id="model"></select> <div id="log" style="white-space:pre-wrap;border:1px solid #ccc;height:320px;overflow-y:auto;padding:10px"></div> <input id="input"><button id="send">Send</button> <div id="meter"></div> <script> let messages = [], sessionCost = 0; fetch('/models').then(r => r.json()).then(d => { d.models.forEach(m => document.getElementById('model') .appendChild(new Option(m, m))); }); async function send() { const input = document.getElementById('input'); const text = input.value.trim(); if (!text) return; input.value = ''; messages.push({role: 'user', content: text}); const log = document.getElementById('log'); const model = document.getElementById('model').value; log.textContent += 'You: ' + text + '\\n' + model + ': '; const resp = await fetch('/chat', { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({messages, model}), }); const reader = resp.body.getReader(); const decoder = new TextDecoder(); let full = '', buffer = ''; while (true) { const {done, value} = await reader.read(); if (done) break; buffer += decoder.decode(value, {stream: true}); const doneIdx = buffer.indexOf('\\x00DONE\\x00'); if (doneIdx !== -1) { full += buffer.slice(0, doneIdx); log.textContent += buffer.slice(0, doneIdx); const meta = JSON.parse(buffer.slice(doneIdx + 6)); sessionCost += meta.cost; document.getElementById('meter').textContent = `turn: $${meta.cost.toFixed(6)}, ${meta.elapsed_ms.toFixed(0)}ms | session: $${sessionCost.toFixed(6)}`; break; } full += buffer; log.textContent += buffer; buffer = ''; } messages.push({role: 'assistant', content: full}); log.textContent += '\\n'; } document.getElementById('send').onclick = send; document.getElementById('input').addEventListener('keydown', e => { if (e.key === 'Enter') send(); }); </script> """ @app.route("/") def index(): return Response(INDEX_HTML, mimetype="text/html") - Add the launcher as the very last lines of the file, below every route.
app.run()starts the server and doesn't return, so any route written below it is never registered. If you put it earlier, the page loads but the dropdown stays empty (the page asks for/modelsand gets a 404) and/chatfails too.if __name__ == "__main__": app.run(port=5050, debug=True) - Run it and open the page.
python chatbot/chat_web.py # then open http://localhost:5050
curl -N -X POST http://localhost:5050/chat -H "Content-Type: application/json" -d '{"messages":[{"role":"user","content":"hi"}],"model":"gpt-5.4-mini"}' should print text followed immediately by a DONE marker and a JSON cost footer, with no delay between the text ending and the marker appearing. If the whole reply arrives in one burst instead of gradually, the queue isn't actually streaming — check that on_chunk is wired into the worker thread, not called after stream_chat returns.Done when:
- Part 1 (terminal): tokens visibly stream in as they arrive, a multi-turn follow-up gets answered using earlier context, and the session cost total only increases.
- Part 2 (web): the same conversation works in the browser, cost matches what the terminal version would have charged for the same exchange, and the model picker includes the local Ollama model alongside the cloud ones.
- Cost matches the provider's own pricing within 5% — spot-check one turn's
cost_usdinshared/calls.jsonlagainst the provider's pricing page by hand. - A README section (or comment block) compares the same 10 questions across three models — cost, latency, and your read on answer quality, side by side.