Curriculum: From Zero to "Can Build an LLM App"

python3 --version                 # 3.12 or newer recommended
python3 -m venv .venv             # create a virtual environment in this folder
source .venv/bin/activate         # Windows: .venv\Scripts\activate
python -m pip install --upgrade pip
# or, with uv (fast, manages Python versions too):
uv venv && source .venv/bin/activate
uv pip install httpx

# hello_model.py
model = "tiny-llm"
print("Hello from", model)

user_name = input("Your name: ").strip()
question = "What is RAG?"
temperature = 0.2

prompt = f"""You are a helpful tutor.
Student: {user_name}
Question: {question}
Answer clearly (temperature={temperature:.1f})."""
print(prompt)
print(len(prompt), "characters,", len(prompt.split()), "words")

history = [
    {"role": "system", "content": "You are concise."},
    {"role": "user", "content": "Define embedding."},
]
history.append({"role": "assistant", "content": "A vector that represents meaning."})

print(history[-1]["content"])           # last message's text
print([m["role"] for m in history])     # preview of stage 9: ['system', 'user', 'assistant']
roles_seen = {m["role"] for m in history}
print("user" in roles_seen)             # True
last_two = history[-2:]                 # slicing keeps the list type

MAX_TOKENS = 50
messages = ["Hello there", "Explain transformers in detail please", "Thanks!"]

used = 0
for i, text in enumerate(messages):
    cost = len(text.split())          # pretend 1 word = 1 token
    if used + cost > MAX_TOKENS:
        print(f"Message {i} would exceed the budget, stopping.")
        break
    used += cost
    print(f"#{i}: {cost} tokens (total {used})")
else:
    print("All messages fit.")

def build_prompt(question, context="", style="concise"):
    """Return a prompt string for the model."""
    parts = [f"Answer in a {style} style."]
    if context:
        parts.append(f"Context:\n{context}")
    parts.append(f"Question: {question}")
    return "\n\n".join(parts)


def truncate(text, max_words=20):
    words = text.split()
    return " ".join(words[:max_words]) + (" ..." if len(words) > max_words else "")


print(build_prompt("What is a vector store?", context=truncate("A vector store keeps embeddings " * 10)))

my_bot/
  .venv/
  pyproject.toml        # or requirements.txt
  prompts.py            # build_prompt, truncate
  main.py               # from prompts import build_prompt

python -m pip install httpx           # always "python -m pip" so pip matches the python
python -m pip freeze > requirements.txt
# with uv:
uv init my_bot && cd my_bot && uv add httpx && uv run main.py

import json
from pathlib import Path

faq_path = Path("faq.json")
faq_path.write_text(json.dumps([
    {"q": "What is RAG?", "a": "Retrieval-augmented generation."},
    {"q": "What is an embedding?", "a": "A vector representing meaning."},
], indent=2), encoding="utf-8")

with faq_path.open(encoding="utf-8") as f:
    faq = json.load(f)
for item in faq:
    print(f"Q: {item['q']}\nA: {item['a']}\n")

import json

FENCE = "`" * 3          # three backticks, the Markdown code fence models like to add


def parse_model_json(raw):
    """Model replies sometimes wrap JSON in Markdown fences or add chatter around it."""
    text = raw.strip()
    if text.startswith(FENCE):
        text = text.strip("`")
        text = text.removeprefix("json").strip()      # 3.9+
    start, end = text.find("{"), text.rfind("}")
    if start == -1 or end == -1:
        raise ValueError(f"no JSON object in model reply: {raw[:40]!r}")
    try:
        return json.loads(text[start:end + 1])
    except json.JSONDecodeError as e:
        raise ValueError(f"model returned invalid JSON: {e}") from e


reply = f'Sure! {FENCE}json\n{{"sentiment": "positive", "score": 0.9}}\n{FENCE}'
print(parse_model_json(reply))      # {'sentiment': 'positive', 'score': 0.9}

from dataclasses import dataclass, field


@dataclass
class Message:
    role: str
    content: str


@dataclass
class Conversation:
    system: str
    messages: list = field(default_factory=list)   # never a bare [] default

    def add(self, role, content):
        self.messages.append(Message(role, content))

    def as_payload(self):
        return [{"role": "system", "content": self.system}] + [
            {"role": m.role, "content": m.content} for m in self.messages
        ]


chat = Conversation(system="You are concise.")
chat.add("user", "What is a token?")
print(chat.as_payload())

def chunk_words(text, size=50, overlap=10):
    """Yield chunks of `size` words, each overlapping the previous by `overlap` words."""
    if overlap >= size:
        raise ValueError("overlap must be smaller than size")
    words = text.split()
    step = size - overlap
    for start in range(0, len(words), step):
        yield " ".join(words[start:start + size])
        if start + size >= len(words):
            break


doc = " ".join(f"w{i}" for i in range(120))
for n, chunk in enumerate(chunk_words(doc, size=50, overlap=10)):
    print(n, chunk.split()[0], "...", chunk.split()[-1])

import functools
import random
import time


class RateLimitError(Exception):
    pass


def retry(times: int = 3, delay: float = 0.1):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(1, times + 1):
                try:
                    return fn(*args, **kwargs)
                except RateLimitError:
                    if attempt == times:
                        raise
                    time.sleep(delay * 2 ** (attempt - 1))   # exponential backoff
        return wrapper
    return decorator


@retry(times=5)
def flaky_model(prompt: str) -> str:
    if random.random() < 0.5:
        raise RateLimitError("429 Too Many Requests")
    return prompt.upper()


print(flaky_model("hello"))

import asyncio
import time


async def fake_model(prompt: str, delay: float = 0.5) -> str:
    await asyncio.sleep(delay)            # stands in for a network call
    return f"answer to {prompt!r}"


async def main():
    prompts = ["a", "b", "c", "d"]
    limit = asyncio.Semaphore(2)          # at most 2 calls at once

    async def guarded(p):
        async with limit:
            return await fake_model(p)

    start = time.perf_counter()
    answers = await asyncio.gather(*(guarded(p) for p in prompts))
    print(answers, f"{time.perf_counter() - start:.1f}s")   # ~1.0s, not 2.0s


asyncio.run(main())

import numpy as np
import pandas as pd

a = np.array([0.1, 0.3, 0.5])
b = np.array([0.2, 0.1, 0.4])
cosine = a @ b / (np.linalg.norm(a) * np.linalg.norm(b))
print(round(float(cosine), 3))

evals = pd.DataFrame({
    "model": ["small", "small", "large", "large"],
    "question": ["q1", "q2", "q1", "q2"],
    "correct": [True, False, True, True],
})
print(evals.groupby("model")["correct"].mean())    # accuracy per model

import os

import httpx

API_URL = os.environ.get("LLM_API_URL", "https://api.example.com/v1/chat")
API_KEY = os.environ["LLM_API_KEY"]          # export LLM_API_KEY=... in the shell, or a .env file


def ask(question: str, model: str = "small-model") -> str:
    payload = {"model": model, "messages": [{"role": "user", "content": question}], "max_tokens": 200}
    response = httpx.post(API_URL, json=payload, headers={"Authorization": f"Bearer {API_KEY}"}, timeout=30)
    response.raise_for_status()               # turns 4xx/5xx into an exception
    data = response.json()
    return data["choices"][0]["message"]["content"]   # the shape depends on the provider: print(data) first!

import math
import re
from collections import Counter


def embed(text: str) -> Counter:
    """Toy embedding: word counts. Replace with a real embedding model later."""
    return Counter(re.findall(r"[a-z]+", text.lower()))


def cosine(a: Counter, b: Counter) -> float:
    dot = sum(a[w] * b[w] for w in a)
    norm = math.sqrt(sum(v * v for v in a.values())) * math.sqrt(sum(v * v for v in b.values()))
    return dot / norm if norm else 0.0


notes = [
    "Python lists are ordered and mutable.",
    "Embeddings map text to vectors so similar meanings are close.",
    "A virtual environment isolates project dependencies.",
]
index = [(chunk, embed(chunk)) for chunk in notes]


def retrieve(question: str, k: int = 2) -> list:
    q = embed(question)
    ranked = sorted(index, key=lambda pair: cosine(q, pair[1]), reverse=True)
    return [chunk for chunk, _ in ranked[:k]]


question = "How do embeddings represent meaning?"
context = "\n".join(f"- {c}" for c in retrieve(question))
prompt = f"Answer using only this context:\n{context}\n\nQuestion: {question}"
print(prompt)       # send this to ask() from stage 13