Exercises with Hints and Solutions

Role: data analyst
Task: summarise sales trends
Tone: friendly
You are a data analyst. Summarise sales trends. Use a friendly tone.

role = input("Role: ").strip()
task = input("Task: ").strip()
tone = input("Tone: ").strip()

prompt = f"You are a {role}. {task[:1].upper() + task[1:]}. Use a {tone} tone."
print("---")
print(prompt)

cost_per_call(1200, 300, in_price=0.0005, out_price=0.0015) -> 0.00105
one call: $0.0010
10,000 calls: $10.50

def cost_per_call(tokens_in, tokens_out, in_price, out_price):
    return tokens_in / 1000 * in_price + tokens_out / 1000 * out_price


one = cost_per_call(1200, 300, in_price=0.0005, out_price=0.0015)
print(f"one call: ${one:.4f}")                 # $0.0010 (0.00105 rounded for display)
print(f"10,000 calls: ${one * 10_000:,.2f}")   # $10.50

word_counts("The cat sat. The cat ran!")   # {'the': 2, 'cat': 2, 'sat': 1, 'ran': 1}

import re


def word_counts(text):
    counts = {}
    for word in re.findall(r"[a-z']+", text.lower()):
        counts[word] = counts.get(word, 0) + 1
    return counts


counts = word_counts("The cat sat. The cat ran!")
print(counts)
top3 = sorted(counts.items(), key=lambda kv: kv[1], reverse=True)[:3]
print(top3)                                   # [('the', 2), ('cat', 2), ('sat', 1)]

history = [
    {"role": "system", "content": "Be brief."},
    {"role": "user", "content": "What is RAG?"},
    {"role": "assistant", "content": "Retrieval-augmented generation."},
    {"role": "user", "content": "Give an example."},
]
# 1. What is RAG?
# 2. Give an example.

history = [
    {"role": "system", "content": "Be brief."},
    {"role": "user", "content": "What is RAG?"},
    {"role": "assistant", "content": "Retrieval-augmented generation."},
    {"role": "user", "content": "Give an example."},
]

user_texts = [m["content"] for m in history if m["role"] == "user"]
for i, text in enumerate(user_texts, start=1):
    print(f"{i}. {text}")

dedupe(["a", "b", "a", "c", "b"])   # ['a', 'b', 'c']

def dedupe(items):
    seen = set()
    out = []
    for item in items:
        if item not in seen:
            seen.add(item)
            out.append(item)
    return out


print(dedupe(["a", "b", "a", "c", "b"]))     # ['a', 'b', 'c']
print(list(dict.fromkeys(["a", "b", "a"])))  # one-liner: dicts keep insertion order (3.7+)

batches(list(range(7)), 3)   # [[0, 1, 2], [3, 4, 5], [6]]

def batches(items, n):
    if n < 1:
        raise ValueError("n must be at least 1")
    return [items[start:start + n] for start in range(0, len(items), n)]


print(batches(list(range(7)), 3))   # [[0, 1, 2], [3, 4, 5], [6]]
print(batches([], 3))               # []

chunk("w0 w1 w2 w3 w4 w5 w6 w7", size=4, overlap=1)
# ['w0 w1 w2 w3', 'w3 w4 w5 w6', 'w6 w7']

def chunk(text, size=200, overlap=20):
    if not 0 <= overlap < size:
        raise ValueError("need 0 <= overlap < size")
    words = text.split()
    step = size - overlap
    chunks = []
    for start in range(0, len(words), step):
        chunks.append(" ".join(words[start:start + size]))
        if start + size >= len(words):      # this chunk reached the end: stop
            break
    return chunks


print(chunk("w0 w1 w2 w3 w4 w5 w6 w7", size=4, overlap=1))  # ['w0 w1 w2 w3', 'w3 w4 w5 w6', 'w6 w7']
print(chunk("w0 w1 w2 w3", size=4, overlap=1))              # ['w0 w1 w2 w3']  (no trailing 'w3' chunk)
print(chunk("", size=4, overlap=1))                         # []

render("Answer {question} using {context}", question="Q?")
# ValueError: missing values for: context

import string


def placeholders(template):
    return {field for _, field, _, _ in string.Formatter().parse(template) if field}


def render(template, **values):
    missing = sorted(placeholders(template) - values.keys())
    if missing:
        raise ValueError(f"missing values for: {', '.join(missing)}")
    return template.format(**values)


print(render("Answer {question} using {context}", question="What is RAG?", context="notes", extra=1))
try:
    render("Answer {question} using {context}", question="Q?")
except ValueError as e:
    print(e)                                    # missing values for: context

cosine([1, 0], [1, 0])     # 1.0
cosine([1, 0], [0, 1])     # 0.0
cosine([1, 2], [2, 4])     # 1.0 (same direction)

import math


def cosine(a, b):
    if len(a) != len(b):
        raise ValueError(f"length mismatch: {len(a)} vs {len(b)}")
    dot = sum(x * y for x, y in zip(a, b))
    norm_a = math.sqrt(sum(x * x for x in a))
    norm_b = math.sqrt(sum(y * y for y in b))
    if norm_a == 0 or norm_b == 0:
        return 0.0
    return dot / (norm_a * norm_b)


print(cosine([1, 0], [1, 0]), cosine([1, 0], [0, 1]), round(cosine([1, 2], [2, 4]), 6))
print(math.isclose(cosine([0.1, 0.2, 0.3], [0.1, 0.2, 0.3]), 1.0))   # compare floats with isclose

import heapq
import math


def cosine(a, b):
    dot = sum(x * y for x, y in zip(a, b))
    na, nb = math.sqrt(sum(x * x for x in a)), math.sqrt(sum(y * y for y in b))
    return dot / (na * nb) if na and nb else 0.0


def top_k(query, docs, k=2):
    scored = [(doc_id, cosine(query, vec)) for doc_id, vec in docs.items()]
    return heapq.nlargest(k, scored, key=lambda pair: pair[1])


docs = {"cats": [0.9, 0.1, 0.0], "dogs": [0.8, 0.3, 0.1], "stocks": [0.0, 0.1, 0.9]}
for doc_id, score in top_k([1.0, 0.2, 0.0], docs):
    print(f"{doc_id}: {score:.3f}")

fence = "`" * 3      # a Markdown code fence, as models often add
parse_reply(f'Sure! {fence}json\n{{"sentiment": "positive", "confidence": 0.92}}\n{fence}')
# {'sentiment': 'positive', 'confidence': 0.92}
parse_reply('{"sentiment": "great", "confidence": 2}')
# ValueError: sentiment must be one of ['negative', 'neutral', 'positive'], got 'great'

import json

ALLOWED = {"positive", "negative", "neutral"}


def parse_reply(raw):
    start, end = raw.find("{"), raw.rfind("}")
    if start == -1 or end < start:
        raise ValueError(f"no JSON object found in: {raw[:60]!r}")
    try:
        data = json.loads(raw[start:end + 1])
    except json.JSONDecodeError as e:
        raise ValueError(f"invalid JSON ({e.msg} at char {e.pos})") from e
    if not isinstance(data, dict):
        raise ValueError("expected a JSON object")
    sentiment = data.get("sentiment")
    if sentiment not in ALLOWED:
        raise ValueError(f"sentiment must be one of {sorted(ALLOWED)}, got {sentiment!r}")
    confidence = data.get("confidence")
    if isinstance(confidence, bool) or not isinstance(confidence, (int, float)) or not 0 <= confidence <= 1:
        raise ValueError(f"confidence must be a number between 0 and 1, got {confidence!r}")
    return {"sentiment": sentiment, "confidence": float(confidence)}


fence = "`" * 3
print(parse_reply(f'Sure! {fence}json\n{{"sentiment": "positive", "confidence": 0.92}}\n{fence}'))
for bad in ['{"sentiment": "great", "confidence": 2}', "I cannot answer that.", '{"sentiment": "positive",}']:
    try:
        parse_reply(bad)
    except ValueError as e:
        print("rejected:", e)

from dataclasses import dataclass, field

ROLES = {"user", "assistant", "tool"}


@dataclass
class Conversation:
    system: str
    messages: list = field(default_factory=list)

    def add(self, role, content):
        if role not in ROLES:
            raise ValueError(f"unknown role {role!r}; expected one of {sorted(ROLES)}")
        self.messages.append({"role": role, "content": content})

    def last(self, n):
        return self.messages[-n:] if n > 0 else []

    def to_payload(self):
        return [{"role": "system", "content": self.system}, *self.messages]


chat = Conversation("You are concise.")
chat.add("user", "Hi")
chat.add("assistant", "Hello!")
other = Conversation("Different bot")
print(len(chat.messages), len(other.messages))     # 2 0  (separate lists)
print(chat.to_payload())

import time


def stream(text, piece=4, delay=0.02):
    for i in range(0, len(text), piece):
        time.sleep(delay)              # pretend network latency
        yield text[i:i + piece]


parts = []
for delta in stream("Embeddings turn text into vectors."):
    print(delta, end="", flush=True)
    parts.append(delta)
print()
full = "".join(parts)
print(len(full))

import functools
import time


def retry(times=3, exceptions=(Exception,), base_delay=0.1):
    def decorator(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(times):
                try:
                    return fn(*args, **kwargs)
                except exceptions as e:
                    if attempt == times - 1:
                        raise
                    wait = base_delay * 2 ** attempt
                    print(f"{fn.__name__} failed ({e}); retry {attempt + 1} in {wait:.2f}s")
                    time.sleep(wait)
        return wrapper
    return decorator


calls = {"n": 0}


@retry(times=3, exceptions=(TimeoutError,), base_delay=0.01)
def flaky():
    calls["n"] += 1
    if calls["n"] < 3:
        raise TimeoutError("model timed out")
    return "ok"


print(flaky(), calls["n"])        # ok 3

import asyncio
import random
import time


async def fake_model(prompt):
    await asyncio.sleep(random.uniform(0.1, 0.3))
    return prompt[::-1]


async def run_all(prompts, limit=3):
    sem = asyncio.Semaphore(limit)

    async def one(p):
        async with sem:
            return await fake_model(p)

    return await asyncio.gather(*(one(p) for p in prompts))


start = time.perf_counter()
results = asyncio.run(run_all([f"prompt {i}" for i in range(10)]))
print(results[:2], f"{time.perf_counter() - start:.2f}s")

import math
import re
from collections import Counter

notes = {
    "python.md": "Lists are ordered and mutable. Tuples are ordered and immutable.",
    "rag.md": "RAG retrieves relevant chunks and adds them to the prompt so the model can cite them.",
    "venv.md": "A virtual environment keeps each project's packages separate.",
}


def tokenize(text):
    return re.findall(r"[a-z]+", text.lower())


def chunk(text, size=8, overlap=2):
    words, step, out = text.split(), size - overlap, []
    for start in range(0, len(words), step):
        out.append(" ".join(words[start:start + size]))
        if start + size >= len(words):
            break
    return out


chunks = [(name, c) for name, text in notes.items() for c in chunk(text)]
vocab = sorted({w for _, c in chunks for w in tokenize(c)})


def embed(text):
    counts = Counter(tokenize(text))
    return [counts.get(w, 0) for w in vocab]


def cosine(a, b):
    dot = sum(x * y for x, y in zip(a, b))
    na, nb = math.sqrt(sum(x * x for x in a)), math.sqrt(sum(y * y for y in b))
    return dot / (na * nb) if na and nb else 0.0


index = [(name, c, embed(c)) for name, c in chunks]
question = "How does RAG add chunks to the prompt?"
q = embed(question)
best = sorted(index, key=lambda item: cosine(q, item[2]), reverse=True)[:2]
context = "\n".join(f"[{name}] {c}" for name, c, _ in best)
prompt = f"Answer using only the context. Cite the file.\n\nContext:\n{context}\n\nQuestion: {question}"
print(prompt)

a = [1, 2, 3]
b = a
b.append(4)
print(a)

def add(item, bucket=[]):
    bucket.append(item)
    return bucket

print(add(1), add(2))

nums = [10, 20, 30, 40]
print(nums[1:3], nums[-1], nums[:-1], len(nums[1:]))

print(7 / 2, 7 // 2, -7 // 2, 7 % 3, 0.1 + 0.2 == 0.3)

count = 0
def bump():
    count = count + 1
bump()

items = ["a", "b", "c", "d"]
for x in items:
    if x in ("b", "c"):
        items.remove(x)
print(items)

gen = (n * n for n in range(3))
print(list(gen), list(gen))