Modern Python by Version

while (batch := queue.get_batch(64)):
    process(batch)

if (n := len(tokens)) > MAX_TOKENS:
    raise ValueError(f"prompt has {n} tokens (max {MAX_TOKENS})")

clean = [y for x in raw if (y := normalize(x))]

def cosine_similarity(a, b, /, *, eps: float = 1e-8) -> float: ...

logger.debug(f"{batch_size=} {lr=:.1e}")      # "batch_size=32 lr=3.0e-04"

from functools import cached_property

class Corpus:
    def __init__(self, docs: list[str]):
        self.docs = docs

    @cached_property
    def vocabulary(self) -> set[str]:
        return {tok for doc in self.docs for tok in doc.split()}

from collections.abc import Iterable, Iterator, Mapping, Sequence

def batch(items: Sequence[str], size: int) -> Iterator[list[str]]: ...
def merge(a: Mapping[str, float], b: Mapping[str, float]) -> dict[str, float]: ...

params = DEFAULT_GEN_PARAMS | {"temperature": 0.2}

deployment = name.removeprefix("azure/")
stem = filename.removesuffix(".jsonl")

from zoneinfo import ZoneInfo
from datetime import datetime
local = datetime.now(ZoneInfo("Europe/Berlin"))

def handle(event: dict) -> None:
    match event:
        case {"type": "content_block_delta", "delta": {"text": str(text)}}:
            sink.write(text)
        case {"type": "message_stop"}:
            sink.close()
        case {"type": "error", "error": {"message": msg}}:
            raise ProviderError(msg)
        case _:
            pass

from dataclasses import dataclass

@dataclass
class ToolCall:
    name: str
    args: dict

match call:
    case ToolCall(name="search", args={"query": str(q)}):
        return search(q)
    case ToolCall(name="calculator", args={"expr": str(e)}):
        return calculate(e)
    case ToolCall(name=other):
        raise UnknownTool(other)

def to_list(x: str | list[str]) -> list[str]:
    return [x] if isinstance(x, str) else x
isinstance(value, int | float)

with (
    open("prompts.jsonl", encoding="utf-8") as src,
    open("completions.jsonl", "w", encoding="utf-8") as dst,
):
    ...

from itertools import pairwise
deltas = [b - a for a, b in pairwise(timestamps)]

@dataclass(slots=True, kw_only=True, frozen=True)
class SamplingParams:
    temperature: float = 0.0
    top_p: float = 1.0
    max_tokens: int = 512

async def retrieve_all(queries: list[str]) -> list[list[Doc]]:
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(retrieve(q)) for q in queries]
    return [t.result() for t in tasks]

async with asyncio.timeout(15):
    reply = await llm.chat.completions.create(model=model, messages=messages)

try:
    results = await retrieve_all(queries)
except* TimeoutError as eg:
    logger.warning("%d retrievals timed out", len(eg.exceptions))
    raise
except* ValueError as eg:
    for err in eg.exceptions:
        logger.error("bad query: %s", err)
    raise

try:
    record = Example.model_validate(row)
except ValidationError as err:
    err.add_note(f"row {i} in {path}")
    raise

import tomllib
with open("pyproject.toml", "rb") as f:     # binary mode is required
    config = tomllib.load(f)

from enum import StrEnum

class Role(StrEnum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"

{"role": Role.USER, "content": "hi"}   # serialises as "user"

from typing import Self

class PromptBuilder:
    def __init__(self) -> None:
        self.parts: list[str] = []

    def add(self, text: str) -> Self:
        self.parts.append(text)
        return self

def first[T](items: list[T]) -> T:
    return items[0]

class Cache[K, V]:
    def __init__(self) -> None:
        self._data: dict[K, V] = {}

type Embedding = list[float]
type Retriever = Callable[[str, int], list[Doc]]

from itertools import batched
for batch in batched(ids, 100):            # tuples of up to 100
    index.upsert(list(batch))

f"{', '.join(f'{k}={v}' for k, v in params.items())}"
f"{payload["model"]}"                      # same quotes inside: 3.12+ only

from typing import override

class OpenAIEmbedder(BaseEmbedder):
    @override
    def embed(self, texts: list[str]) -> list[list[float]]: ...

for dirpath, dirnames, filenames in Path("corpus").walk():
    dirnames[:] = [d for d in dirnames if not d.startswith(".")]