Ch 1 — Python Fundamentals for AI¶

# Two names, one object
a: list[int] = [1, 2, 3]
b: list[int] = a          # b is an alias, NOT a copy

b.append(4)
print(a)  # [1, 2, 3, 4]  ← a sees the mutation because a and b point to the same list
print(id(a) == id(b))  # True

# Assignment rebinds the name, not the object
b = [99]
print(a)  # [1, 2, 3, 4]  ← a is unaffected; b now points to a new list

x: list[int] = [1, 2, 3]
y: list[int] = [1, 2, 3]

print(x == y)   # True  — same *value*
print(x is y)   # False — different *objects* (different id)

# CPython interns small integers and short strings as an optimisation.
# Never rely on `is` for value equality — use it only to test for None, True, False.
assert (x is None) == False  # correct idiom
assert (x == None) == False  # also correct but triggers linting warnings

# Python integers are arbitrary-precision — no overflow
huge: int = 2 ** 1000          # works without special libraries

# IEEE-754 double precision for float — the usual suspects
print(0.1 + 0.2 == 0.3)        # False
print(abs(0.1 + 0.2 - 0.3) < 1e-9)  # True — always use tolerance comparisons

import math
print(math.isfinite(float("inf")))  # False
print(math.isnan(float("nan")))     # True

text: str = "gradient descent"

# Strings are sequences of Unicode code points
print(len("café"))   # 4 — characters, not bytes
print("café".encode("utf-8"))  # b'caf\xc3\xa9' — 5 bytes (é is 2 bytes in UTF-8)

# f-strings (the modern way — prefer over .format() or %)
loss: float = 0.2413
epoch: int = 42
msg: str = f"Epoch {epoch:03d} | loss={loss:.4f}"
print(msg)  # Epoch 042 | loss=0.2413

# Raw bytes for binary data
weights: bytes = b"\x00\xff\x7f"
print(len(weights))  # 3

from typing import Any

# List — ordered, mutable, allows duplicates
batch: list[dict[str, Any]] = [
    {"input_ids": [101, 7592, 102], "label": 1},
    {"input_ids": [101, 3231, 102], "label": 0},
]

# Tuple — ordered, immutable; great for multi-return values
def split_dataset(
    data: list[Any], ratio: float = 0.8
) -> tuple[list[Any], list[Any]]:
    split = int(len(data) * ratio)
    return data[:split], data[split:]

train, val = split_dataset(batch)

# Set — unordered, unique elements; O(1) membership test
vocab: set[str] = {"[PAD]", "[UNK]", "[CLS]", "[SEP]"}
print("[CLS]" in vocab)  # True — O(1)

# Dict — ordered by insertion (Python 3.7+), O(1) average get/set
token_to_id: dict[str, int] = {"[PAD]": 0, "[UNK]": 1, "[CLS]": 2, "[SEP]": 3}
print(token_to_id.get("[MASK]", -1))  # -1 — safe default

# Basic form: [expression for item in iterable if condition]
logits: list[float] = [2.1, -0.5, 3.8, 1.2, -1.0]

# Softmax numerators (no condition)
import math
exp_logits: list[float] = [math.exp(x) for x in logits]

# Filter: keep only positive logits
positive: list[float] = [x for x in logits if x > 0]

# Nested: flatten a batch of token lists
batch_tokens: list[list[int]] = [[101, 7592], [101, 3231, 102]]
flat: list[int] = [token for seq in batch_tokens for token in seq]
# → [101, 7592, 101, 3231, 102]

# Invert a token→id mapping to id→token
token_to_id: dict[str, int] = {"[PAD]": 0, "[UNK]": 1, "[CLS]": 2}
id_to_token: dict[int, str] = {v: k for k, v in token_to_id.items()}

# Selective update: double the id of every token that starts with "["
adjusted: dict[str, int] = {
    k: (v * 2 if k.startswith("[") else v)
    for k, v in token_to_id.items()
}

sentences: list[str] = ["the cat sat", "the dog ran", "a cat ran"]

# Unique words across all sentences
unique_words: set[str] = {word for sentence in sentences for word in sentence.split()}
print(unique_words)  # {'the', 'cat', 'sat', 'dog', 'ran', 'a'}

import math

logits = [2.1, -0.5, 3.8, 1.2, -1.0]

# sum() consumes the generator without ever materialising the full list
total = sum(math.exp(x) for x in logits)

# Generators can be chained to form a memory-efficient pipeline
def load_lines(path: str):
    with open(path, encoding="utf-8") as fh:
        yield from fh  # each line is produced on demand

def tokenise(lines):
    for line in lines:
        yield line.strip().split()

# No line is read until next() is called somewhere downstream
pipeline = tokenise(load_lines("corpus.txt"))

from typing import Any


def train_epoch(
    model: Any,
    loader: Any,
    lr: float = 1e-3,          # default argument
    clip_grad: float | None = None,
) -> float:
    """Run one training epoch and return the mean loss."""
    total_loss = 0.0
    for batch in loader:
        loss = model.step(batch, lr=lr)
        total_loss += loss
    return total_loss / len(loader)


def log_metrics(*values: float, prefix: str = "train") -> None:
    """Accept any number of metric values; prefix is keyword-only."""
    for i, v in enumerate(values):
        print(f"{prefix}/metric_{i}: {v:.4f}")


def create_optimizer(params: Any, **kwargs: Any) -> dict[str, Any]:
    """Forward arbitrary keyword arguments to the underlying optimizer."""
    defaults = {"lr": 1e-3, "weight_decay": 0.0, "betas": (0.9, 0.999)}
    return {**defaults, **kwargs, "params": params}

def evaluate(
    model: Any,
    loader: Any,
    *,                         # everything after here is keyword-only
    device: str = "cpu",
    amp: bool = False,
) -> dict[str, float]:
    ...
    return {"accuracy": 0.95, "loss": 0.12}


# evaluate(model, loader, "cuda")  ← TypeError: device must be keyword argument
result = evaluate(model, loader, device="cuda", amp=True)  # correct

from __future__ import annotations  # postponed evaluation — allows forward references

from typing import Any, Optional, TypeVar, Union

T = TypeVar("T")  # generic type variable

# Basic annotations
def clip_value(x: float, lo: float, hi: float) -> float:
    return max(lo, min(hi, x))


# Optional[X] is equivalent to X | None (Python 3.10+ syntax)
def load_checkpoint(path: str, device: Optional[str] = None) -> dict[str, Any]:
    ...


# Union (pre-3.10 syntax — kept for library compatibility)
def parse_label(raw: Union[str, int]) -> int:
    return int(raw)


# Collection generics
def pad_sequences(
    sequences: list[list[int]],
    pad_id: int = 0,
    max_len: int | None = None,
) -> list[list[int]]:
    length = max_len or max(len(s) for s in sequences)
    return [s + [pad_id] * (length - len(s)) for s in sequences]


# Generic functions
def first(items: list[T]) -> T:
    if not items:
        raise IndexError("empty list")
    return items[0]

from typing import Protocol, runtime_checkable


@runtime_checkable
class Encoder(Protocol):
    def encode(self, text: str) -> list[int]: ...
    def decode(self, ids: list[int]) -> str: ...


def embed_batch(encoder: Encoder, texts: list[str]) -> list[list[int]]:
    return [encoder.encode(t) for t in texts]


# Any class with matching encode/decode methods satisfies Encoder,
# without inheriting from it — duck typing with static verification.

import functools
import time
from collections.abc import Callable
from typing import ParamSpec, TypeVar

P = ParamSpec("P")  # captures the parameter signature of the wrapped function
R = TypeVar("R")    # return type


def timer(func: Callable[P, R]) -> Callable[P, R]:
    """Measure and print the wall-clock execution time of *func*."""

    @functools.wraps(func)  # preserves __name__, __doc__, etc.
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__qualname__} finished in {elapsed:.4f}s")
        return result

    return wrapper


@timer
def train_step(batch_size: int) -> float:
    time.sleep(0.05)  # simulate work
    return 0.312

loss = train_step(32)  # prints: train_step finished in 0.0502s

import functools
from collections.abc import Callable
from typing import ParamSpec, TypeVar

P = ParamSpec("P")
R = TypeVar("R")


def retry(max_attempts: int = 3, exceptions: tuple[type[Exception], ...] = (Exception,)):
    """Retry *func* up to *max_attempts* times on specified exception types."""

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @functools.wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except exceptions as exc:
                    if attempt == max_attempts:
                        raise
                    print(f"Attempt {attempt} failed ({exc}); retrying...")
            raise RuntimeError("unreachable")  # satisfies type checker

        return wrapper

    return decorator


@retry(max_attempts=5, exceptions=(ConnectionError, TimeoutError))
def fetch_embeddings(url: str) -> list[float]:
    ...

import functools
from collections.abc import Callable
from typing import Any


class RateLimit:
    """Allow at most *calls_per_second* calls to the decorated function."""

    def __init__(self, calls_per_second: float) -> None:
        self._interval = 1.0 / calls_per_second
        self._last_call: float = 0.0

    def __call__(self, func: Callable[..., Any]) -> Callable[..., Any]:
        @functools.wraps(func)
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            import time
            elapsed = time.perf_counter() - self._last_call
            if elapsed < self._interval:
                time.sleep(self._interval - elapsed)
            self._last_call = time.perf_counter()
            return func(*args, **kwargs)

        return wrapper


@RateLimit(calls_per_second=5.0)
def call_llm_api(prompt: str) -> str:
    ...

class ManagedTempDir:
    """Create a temporary directory on enter; delete it on exit."""

    import pathlib
    import shutil
    import tempfile

    def __init__(self, prefix: str = "ai_run_") -> None:
        self._prefix = prefix
        self.path: "pathlib.Path | None" = None

    def __enter__(self) -> "pathlib.Path":
        import pathlib
        import tempfile
        self.path = pathlib.Path(tempfile.mkdtemp(prefix=self._prefix))
        return self.path

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: object | None,
    ) -> bool:
        import shutil
        if self.path is not None:
            shutil.rmtree(self.path, ignore_errors=True)
        return False  # do not suppress exceptions


with ManagedTempDir() as tmp:
    (tmp / "features.npy").write_bytes(b"\x93NUMPY")
    print(tmp.exists())  # True
# Directory and all its contents are deleted here.

import contextlib
import time
from collections.abc import Generator


@contextlib.contextmanager
def timer_block(label: str) -> Generator[None, None, None]:
    """Print elapsed time for the code block."""
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"[{label}] {elapsed:.4f}s")


with timer_block("data loading"):
    import time; time.sleep(0.1)   # simulate work

import contextlib
import pathlib

# Silently ignore FileNotFoundError — equivalent to try/except but more readable
with contextlib.suppress(FileNotFoundError):
    pathlib.Path("stale_cache.pkl").unlink()

from typing import Iterator


class BatchIterator:
    """Yields fixed-size batches from a flat list of samples."""

    def __init__(self, data: list[int], batch_size: int) -> None:
        self._data = data
        self._batch_size = batch_size
        self._pos = 0

    def __iter__(self) -> "BatchIterator":
        return self

    def __next__(self) -> list[int]:
        if self._pos >= len(self._data):
            raise StopIteration
        batch = self._data[self._pos : self._pos + self._batch_size]
        self._pos += self._batch_size
        return batch


for batch in BatchIterator(list(range(10)), batch_size=3):
    print(batch)
# [0, 1, 2]
# [3, 4, 5]
# [6, 7, 8]
# [9]

from collections.abc import Generator
from typing import Any


def infinite_sampler(
    data: list[Any], *, shuffle: bool = True
) -> Generator[Any, None, None]:
    """Yield samples forever, optionally shuffling each epoch."""
    import random
    indices = list(range(len(data)))
    while True:
        if shuffle:
            random.shuffle(indices)
        for i in indices:
            yield data[i]


sampler = infinite_sampler([10, 20, 30, 40])
first_five = [next(sampler) for _ in range(5)]

from collections.abc import Generator


def chain_datasets(*datasets: list[int]) -> Generator[int, None, None]:
    for dataset in datasets:
        yield from dataset  # delegates to each inner iterable


for sample in chain_datasets([1, 2], [3, 4], [5]):
    print(sample, end=" ")
# 1 2 3 4 5

from collections.abc import Generator


def running_mean() -> Generator[float, float, None]:
    """Send each new value; receive the updated mean."""
    total: float = 0.0
    count: int = 0
    while True:
        value: float = yield (total / count if count else 0.0)
        total += value
        count += 1


gen = running_mean()
next(gen)           # prime the generator
print(gen.send(10.0))  # 10.0
print(gen.send(20.0))  # 15.0
print(gen.send(30.0))  # 20.0

from dataclasses import dataclass, field
from typing import Any


@dataclass
class TrainingConfig:
    model_name: str
    learning_rate: float = 1e-4
    batch_size: int = 32
    max_epochs: int = 10
    tags: list[str] = field(default_factory=list)  # safe mutable default

    def lr_at_epoch(self, epoch: int) -> float:
        """Linear warm-up schedule."""
        warmup_steps = 1000
        return self.learning_rate * min(1.0, epoch / warmup_steps)


cfg = TrainingConfig(model_name="bert-base-uncased", batch_size=64)
print(cfg)
# TrainingConfig(model_name='bert-base-uncased', learning_rate=0.0001, ...)

from dataclasses import dataclass


@dataclass(frozen=True)
class TokenizerConfig:
    vocab_size: int
    pad_id: int = 0
    unk_id: int = 1

    def is_special_id(self, token_id: int) -> bool:
        return token_id in (self.pad_id, self.unk_id)


cfg = TokenizerConfig(vocab_size=32_000)
# cfg.vocab_size = 50_000  ← FrozenInstanceError: cannot assign to field 'vocab_size'

# Frozen dataclasses ARE hashable — usable as dict keys and in sets
seen: set[TokenizerConfig] = {cfg}

from dataclasses import dataclass


@dataclass
class SplitRatios:
    train: float
    val: float
    test: float

    def __post_init__(self) -> None:
        total = self.train + self.val + self.test
        if abs(total - 1.0) > 1e-6:
            raise ValueError(f"Split ratios must sum to 1.0; got {total:.6f}")
        for name, ratio in [("train", self.train), ("val", self.val), ("test", self.test)]:
            if not 0.0 < ratio < 1.0:
                raise ValueError(f"{name} ratio must be in (0, 1); got {ratio}")


good = SplitRatios(0.7, 0.15, 0.15)
# bad  = SplitRatios(0.7, 0.2, 0.2)  ← ValueError: Split ratios must sum to 1.0; got 1.1