Idioms Catalog

# Before
for i in range(len(chunks)):
    embed(chunks[i])
# After
for chunk in chunks:
    embed(chunk)

# Before
i = 0
for doc in docs:
    print(i, doc.title)
    i += 1
# After
for i, doc in enumerate(docs):
    print(i, doc.title)
for rank, hit in enumerate(hits, start=1):   # 1-based ranks
    print(f"{rank}. {hit.doc_id}")

# Before
for i in range(len(questions)):
    score(questions[i], answers[i])
# After
for question, answer in zip(questions, answers):
    score(question, answer)
# 3.10+: raise if lengths differ instead of silently truncating
for question, answer in zip(questions, answers, strict=True):
    score(question, answer)

# Before
texts = []
for doc in docs:
    if doc.text.strip():
        texts.append(doc.text.strip())
# After
texts = [doc.text.strip() for doc in docs if doc.text.strip()]
# 3.8+: avoid calling strip twice
texts = [t for doc in docs if (t := doc.text.strip())]

by_id = {doc.id: doc for doc in docs}
vocab = {tok for text in corpus for tok in text.split()}
inverted = {v: k for k, v in label2id.items()}

total_tokens = sum(len(enc.encode(t)) for t in texts)
longest = max(docs, key=lambda d: len(d.text))

# Before
def has_pii(text):
    for pattern in PII_PATTERNS:
        if pattern.search(text):
            return True
    return False
# After
def has_pii(text):
    return any(p.search(text) for p in PII_PATTERNS)

all_valid = all(0.0 <= s <= 1.0 for s in scores)

# Before
for key in config.keys():
    print(key, config[key])
# After
for key, value in config.items():
    print(key, value)
if "temperature" in config: ...      # not: in config.keys()

for msg in reversed(history): ...
for doc in sorted(docs, key=lambda d: d.score, reverse=True): ...

for model in candidates:
    if model in available:
        chosen = model
        break
else:
    raise LookupError("no candidate model is available")

tool = next((t for t in tools if t.name == call.name), None)
if tool is None:
    raise KeyError(call.name)

from itertools import batched          # 3.12+
for batch in batched(texts, 64):
    vectors.extend(model.encode(list(batch)))

# 3.9–3.11 equivalent
def batched(seq, n):
    for i in range(0, len(seq), n):
        yield seq[i:i + n]

from itertools import pairwise         # 3.10+
gaps = [b.start - a.end for a, b in pairwise(spans)]
# overlapping chunk windows
def windows(tokens, size, stride):
    return [tokens[i:i + size] for i in range(0, max(len(tokens) - size, 0) + 1, stride)]

from itertools import chain
all_chunks = list(chain.from_iterable(doc.chunks for doc in docs))
# or
all_chunks = [c for doc in docs for c in doc.chunks]

# Before
pair = best_match(query)
score = pair[0]
doc = pair[1]
# After
score, doc = best_match(query)

system, *turns = messages
head, *_, tail = log_lines

for _ in range(n_retries): ...
name, _, ext = filename.rpartition(".")

for (q, a), score in zip(pairs, scores): ...
for i, (input_ids, labels) in enumerate(loader): ...

# Before
m = re.search(r"```json\s*(.*?)```", reply, re.S)
if m:
    payload = json.loads(m.group(1))
# After
if m := re.search(r"```json\s*(.*?)```", reply, re.S):
    payload = json.loads(m.group(1))

# Read-loop
while chunk := stream.read(8192):
    sink.write(chunk)

device = "cuda" if torch.cuda.is_available() else "cpu"

if response is None: ...
if usage is not None: ...

if not docs: return []           # not: if len(docs) == 0
if history: ...                   # not: if len(history) > 0

if stream: ...                    # not: if stream == True
if not cached: ...                # not: if cached == False

if token not in vocab: ...        # not: if not token in vocab

if isinstance(value, (int, float)): ...
# 3.10+
if isinstance(value, int | float): ...

if 0.0 <= top_p <= 1.0: ...

if finish_reason in {"length", "content_filter"}: ...

import math
math.isclose(similarity, 1.0, rel_tol=1e-6)
# numpy
np.allclose(a, b, atol=1e-6)

# Before
"%s used %d tokens" % (model, n)
"{} used {} tokens".format(model, n)
# After
f"{model} used {n:,} tokens"

f"{latency_ms:.1f} ms"
f"{ratio:.1%}"                    # 0.873 -> 87.3%
f"{n:>8,}"                        # right-aligned with thousands separator
f"{now:%Y-%m-%d %H:%M}"
f"{name!r}"                       # repr
f"{tokens = }"                    # 3.8+: "tokens = 412" (debugging)

context = "\n\n".join(f"[{i}] {c.text}" for i, c in enumerate(chunks, 1))

import textwrap
SYSTEM = textwrap.dedent("""\
    You are a support assistant.
    Answer only from the provided context.
""")
RAG_PROMPT = "Context:\n{context}\n\nQuestion: {question}\nAnswer:"
prompt = RAG_PROMPT.format(context=context, question=question)

model = name.removeprefix("models/")       # not: name[len("models/"):] if name.startswith(...)

if path.endswith((".md", ".mdx", ".txt")): ...

key, _, value = line.partition("=")
words = text.split()                       # splits on any whitespace, drops empties

CODE_BLOCK = re.compile(r"```(\w+)?\n(.*?)```", re.S)

temperature = params.get("temperature", 0.7)

# Before
groups = {}
for chunk in chunks:
    if chunk.doc_id not in groups:
        groups[chunk.doc_id] = []
    groups[chunk.doc_id].append(chunk)
# After
from collections import defaultdict
groups = defaultdict(list)
for chunk in chunks:
    groups[chunk.doc_id].append(chunk)

from collections import Counter
label_counts = Counter(example["label"] for example in dataset)
label_counts.most_common(3)

params = DEFAULTS | user_params            # new dict, right side wins
params |= {"stream": True}                 # in place

client.chat.completions.create(model=model, messages=messages, **gen_kwargs)

nonnull = {k: v for k, v in payload.items() if v is not None}

missing = required_fields - payload.keys()
shared = set(a_tokens) & set(b_tokens)
jaccard = len(shared) / len(set(a_tokens) | set(b_tokens))

unique_ids = list(dict.fromkeys(doc_ids))

def retrieve(query: str, filters: dict | None = None) -> list[Doc]:   # | in annotation: 3.10+ or __future__
    filters = {} if filters is None else filters
    ...

def generate(prompt: str, *, temperature: float = 0.0, max_tokens: int = 512) -> str: ...
generate("hi", temperature=0.2)           # generate("hi", 0.2) is a TypeError: intent is explicit

def cosine(a, b, /): ...

# Before
score = lambda d: d.similarity * d.boost
# After
def score(d: Doc) -> float:
    return d.similarity * d.boost

def parse_tool_call(msg):
    if msg.get("type") != "tool_use":
        return None
    if "name" not in msg:
        raise ValueError("tool_use without name")
    return ToolCall(msg["name"], msg.get("input", {}))

def read_jsonl(path: Path):
    with path.open(encoding="utf-8") as f:
        for line in f:
            if line.strip():
                yield json.loads(line)

from functools import partial
embed_query = partial(embed, model="text-embedding-3-small", dimensions=512)

try:
    data = json.loads(reply)
except json.JSONDecodeError as err:
    raise InvalidModelOutput(reply[:200]) from err

try:
    cfg = CONFIGS[name]
except KeyError:
    raise UnknownModel(name) from None     # hide the KeyError deliberately

from contextlib import suppress
with suppress(FileNotFoundError):
    tmp_path.unlink()
# 3.8+: Path.unlink(missing_ok=True) is even shorter
tmp_path.unlink(missing_ok=True)

try:
    result = tool.run(args)
except ToolError:
    logger.exception("tool %s failed", tool.name)
    raise

try:
    resp = client.post(url, json=body, timeout=30)
except httpx.TimeoutException:
    metrics.incr("timeout")
    raise
else:
    metrics.incr("ok")          # only when no exception
finally:
    span.end()

try:
    async with asyncio.TaskGroup() as tg:
        for url in urls:
            tg.create_task(fetch(url))
except* httpx.TimeoutException as eg:
    logger.warning("%d fetches timed out", len(eg.exceptions))

from dataclasses import dataclass, field

@dataclass(frozen=True)
class GenerationConfig:
    model: str
    temperature: float = 0.0
    stop: tuple[str, ...] = ()
    extra: dict = field(default_factory=dict)

class Usage:
    def __init__(self, prompt: int, completion: int):
        self.prompt, self.completion = prompt, completion

    @property
    def total(self) -> int:
        return self.prompt + self.completion

from enum import Enum
class Role(str, Enum):           # 3.11+: class Role(StrEnum)
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"

match block:
    case {"type": "text", "text": str(text)}:
        out.append(text)
    case {"type": "tool_use", "name": name, "input": dict(args)}:
        calls.append(ToolCall(name, args))
    case _:
        raise ValueError(f"unknown block {block!r}")

from typing import Protocol
class Embedder(Protocol):
    def embed(self, texts: list[str]) -> list[list[float]]: ...

with open(path, encoding="utf-8") as f:
    config = json.load(f)

from pathlib import Path
root = Path(__file__).resolve().parent
data = root / "data" / "eval.jsonl"
for md in (root / "docs").rglob("*.md"):
    ingest(md.read_text(encoding="utf-8"))
data.parent.mkdir(parents=True, exist_ok=True)

import os, tempfile
def write_atomic(path: Path, text: str) -> None:
    fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp")
    with os.fdopen(fd, "w", encoding="utf-8") as f:
        f.write(text)
    os.replace(tmp, path)        # atomic on POSIX and Windows

with tempfile.TemporaryDirectory() as tmp:
    ckpt = Path(tmp) / "model.pt"

def main() -> int:
    ...
    return 0

if __name__ == "__main__":
    raise SystemExit(main())