Generative Py Torch

import collections
import random
import re

import numpy as np
import torch

# course-edition deterministic NLP utilities v2

COURSE_NEWS_CLASSES = ["World", "Sports", "Business", "Sci/Tech"]

# This compact corpus was written for this edition. It keeps the notebooks
# deterministic, network-free, and quick enough to run on a learner's CPU.
COURSE_NEWS_TRAIN = [
    (1, "Paris hosted a regional summit where diplomats discussed water security, peaceful trade routes, and a shared emergency plan for neighboring communities."),
    (1, "The elected council approved a cross-border health agreement after delegates reviewed hospital capacity, medicine access, and transparent public reporting."),
    (1, "A coastal city welcomed observers for a public vote, while local groups published clear guidance about polling places and accessible transportation."),
    (1, "The king and queen met a woman who leads the relief agency and a man who coordinates volunteers, then the group announced a neutral humanitarian corridor."),
    (1, "Ministers signed a climate adaptation pledge that funds drought monitoring, resilient farms, and open scientific exchange across several countries."),
    (1, "Community mediators opened a week of talks focused on civilian safety, reliable food deliveries, and practical steps toward a lasting ceasefire."),
    (2, "The basketball team completed a patient comeback in the final quarter, using quick passes, strong defense, and a balanced scoring plan to win the tournament."),
    (2, "A young runner broke the course record after months of careful training, recovery sessions, nutrition planning, and support from her local athletics club."),
    (2, "The football coach praised disciplined play after the squad protected an early lead and created chances through accurate movement on both wings."),
    (2, "Fans filled the arena for a close volleyball final in which both teams served aggressively, defended long rallies, and respected every referee decision."),
    (2, "The tennis champion returned after injury and won a demanding match by varying pace, placing serves carefully, and staying calm during two tie breaks."),
    (2, "Organizers added inclusive swimming events to the city games so more athletes can compete with safe facilities, trained officials, and fair timing equipment."),
    (3, "Microsoft announced a small-business program that combines cloud credits, security workshops, and practical accounting support for new local companies."),
    (3, "Global funds moved cautiously after the central bank held interest rates steady and asked lenders to publish clearer information about consumer borrowing costs."),
    (3, "A neighborhood market expanded its delivery service, hired twelve workers, and invested revenue in reusable packaging supplied by another local business."),
    (3, "The manufacturer reported stable quarterly sales while noting that shipping delays and higher material prices could reduce margins later in the year."),
    (3, "Two payment companies agreed to share fraud signals through a privacy-preserving system designed to protect customers without exposing purchase details."),
    (3, "A cooperative offered farmers transparent contracts and faster invoices, helping members plan equipment repairs before the next growing season begins."),
    (4, "Researchers trained a compact neural network to identify damaged solar panels, then published the evaluation data and documented where the model still fails."),
    (4, "A university technology lab released an open sensor design that measures classroom air quality and stores only anonymous readings for public analysis."),
    (4, "Engineers improved a battery recycling process by recovering more useful minerals at lower temperatures, reducing energy use during the pilot study."),
    (4, "The space telescope captured a detailed spectrum from a distant planet, giving scientists new evidence about clouds and molecules in its atmosphere."),
    (4, "A software team tested an accessibility assistant with keyboard users and screen-reader experts before changing the interface and publishing the results."),
    (4, "Students built a small robot that maps indoor obstacles with inexpensive sensors, explains each route choice, and says hello when a test run begins."),
]

COURSE_NEWS_TEST = [
    (1, "Regional delegates published a joint disaster response schedule after reviewing evacuation routes and communication gaps."),
    (1, "Election monitors confirmed the final count and recommended clearer access rules for voters who need assistance."),
    (2, "The basketball captain scored late, but credited the victory to defense, passing, and careful preparation across the whole season."),
    (2, "A cycling club opened a safe youth race with trained marshals, marked turns, and free equipment checks."),
    (3, "Technology shares rose while retail funds remained cautious after companies issued mixed forecasts for the next quarter."),
    (3, "The family business secured a modest loan to replace old equipment and expand its apprenticeship program."),
    (4, "A research team released a smaller language model with documented energy measurements and a public evaluation set."),
    (4, "Scientists used open satellite data to improve flood warnings and shared the software with local emergency teams."),
]

def course_tokenize(text):
    return re.findall(r"[a-z0-9]+(?:'[a-z0-9]+)?", str(text).lower())

def course_ngrams_iterator(tokens, ngrams=1):
    tokens = list(tokens)
    for token in tokens:
        yield token
    for size in range(2, max(1, ngrams) + 1):
        for start in range(0, len(tokens) - size + 1):
            yield " ".join(tokens[start:start + size])

class CourseVocabulary:
    def __init__(self, counter, min_freq=1, max_tokens=None):
        ordered = sorted(
            (token for token, count in counter.items() if count >= min_freq),
            key=lambda token: (-counter[token], token),
        )
        if max_tokens is not None:
            ordered = ordered[:max(0, max_tokens - 1)]
        self.itos = ["<unk>"] + [token for token in ordered if token != "<unk>"]
        self.stoi = {token: index for index, token in enumerate(self.itos)}

    def __len__(self):
        return len(self.itos)

    def __getitem__(self, token):
        return self.stoi.get(token, 0)

    def get_stoi(self):
        return dict(self.stoi)

    def get_itos(self):
        return list(self.itos)

def build_course_vocab(dataset, ngrams=1, min_freq=1, max_tokens=None, tokenizer=course_tokenize):
    counter = collections.Counter()
    for _label, text in dataset:
        counter.update(course_ngrams_iterator(tokenizer(text), ngrams=ngrams))
    return CourseVocabulary(counter, min_freq=min_freq, max_tokens=max_tokens)

def load_course_fixture(ngrams=1, min_freq=1, vocab_size=None, lines_cnt=None):
    global vocab, tokenizer
    tokenizer = course_tokenize
    train_dataset = list(COURSE_NEWS_TRAIN)
    test_dataset = list(COURSE_NEWS_TEST)
    vocabulary_rows = train_dataset if lines_cnt is None else train_dataset[:lines_cnt]
    vocab = build_course_vocab(
        vocabulary_rows,
        ngrams=ngrams,
        min_freq=min_freq,
        max_tokens=vocab_size,
        tokenizer=tokenizer,
    )
    return train_dataset, test_dataset, list(COURSE_NEWS_CLASSES), vocab

def encode(text, voc=None, unk=0, tokenizer=course_tokenize):
    selected_vocab = vocab if voc is None else voc
    stoi = selected_vocab.get_stoi()
    return [stoi.get(token, unk) for token in tokenizer(text)]

def train_epoch(net, dataloader, lr=0.01, optimizer=None, loss_fn=None, epoch_size=None, report_freq=200):
    optimizer = optimizer or torch.optim.Adam(net.parameters(), lr=lr)
    loss_fn = (loss_fn or torch.nn.CrossEntropyLoss()).to(device)
    net.train()
    total_loss, accuracy, count, batch_index = 0.0, 0, 0, 0
    for labels, features in dataloader:
        optimizer.zero_grad()
        features, labels = features.to(device), labels.to(device)
        output = net(features)
        loss = loss_fn(output, labels)
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
        accuracy += (output.argmax(1) == labels).sum().item()
        count += len(labels)
        batch_index += 1
        if batch_index % report_freq == 0:
            print(f"{count}: acc={accuracy / count:.4f}")
        if epoch_size and count > epoch_size:
            break
    return total_loss / max(count, 1), accuracy / max(count, 1)

def padify(batch, voc=None, tokenizer=course_tokenize):
    vectors = [encode(item[1], voc=voc, tokenizer=tokenizer) for item in batch]
    max_length = max(map(len, vectors))
    return (
        torch.LongTensor([item[0] - 1 for item in batch]),
        torch.stack([
            torch.nn.functional.pad(torch.tensor(vector), (0, max_length - len(vector)), value=0)
            for vector in vectors
        ]),
    )

def offsetify(batch, voc=None):
    vectors = [torch.tensor(encode(item[1], voc=voc)) for item in batch]
    offsets = torch.tensor(([0] + [len(vector) for vector in vectors])[:-1]).cumsum(dim=0)
    return torch.LongTensor([item[0] - 1 for item in batch]), torch.cat(vectors), offsets

def train_epoch_emb(net, dataloader, lr=0.01, optimizer=None, loss_fn=None, epoch_size=None, report_freq=200, use_pack_sequence=False):
    optimizer = optimizer or torch.optim.Adam(net.parameters(), lr=lr)
    loss_fn = (loss_fn or torch.nn.CrossEntropyLoss()).to(device)
    net.train()
    total_loss, accuracy, count, batch_index = 0.0, 0, 0, 0
    for labels, text, offsets in dataloader:
        optimizer.zero_grad()
        labels, text = labels.to(device), text.to(device)
        offsets = offsets.to("cpu" if use_pack_sequence else device)
        output = net(text, offsets)
        loss = loss_fn(output, labels)
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
        accuracy += (output.argmax(1) == labels).sum().item()
        count += len(labels)
        batch_index += 1
        if batch_index % report_freq == 0:
            print(f"{count}: acc={accuracy / count:.4f}")
        if epoch_size and count > epoch_size:
            break
    return total_loss / max(count, 1), accuracy / max(count, 1)

random.seed(42)
np.random.seed(42)
torch.manual_seed(42)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
tokenizer = course_tokenize
vocab = None

train_dataset, test_dataset, classes, vocab = load_course_fixture()

def char_tokenizer(text):
    return list(text)

counter = collections.Counter()
for _label, line in train_dataset:
    counter.update(char_tokenizer(line))
vocab = CourseVocabulary(counter, min_freq=1)

vocab_size = len(vocab)
print(f"Vocabulary size = {vocab_size}")
print(f"Encoding of 'a' is {vocab['a']}")
print(f"Character with code 13 is {vocab.itos[13]}")

def enc(x):
    return torch.LongTensor(encode(x,voc=vocab,tokenizer=char_tokenizer))

enc(train_dataset[0][1])

nchars = 100

def get_batch(s,nchars=nchars):
    ins = torch.zeros(len(s)-nchars,nchars,dtype=torch.long,device=device)
    outs = torch.zeros(len(s)-nchars,nchars,dtype=torch.long,device=device)
    for i in range(len(s)-nchars):
        ins[i] = enc(s[i:i+nchars])
        outs[i] = enc(s[i+1:i+nchars+1])
    return ins,outs

get_batch(train_dataset[0][1])

class LSTMGenerator(torch.nn.Module):
    def __init__(self, vocab_size, hidden_dim):
        super().__init__()
        self.rnn = torch.nn.LSTM(vocab_size,hidden_dim,batch_first=True)
        self.fc = torch.nn.Linear(hidden_dim, vocab_size)

    def forward(self, x, s=None):
        x = torch.nn.functional.one_hot(x,vocab_size).to(torch.float32)
        x,s = self.rnn(x,s)
        return self.fc(x),s

def generate(net, size=100, start="today "):
    chars = list(start)
    out, state = net(enc(chars).view(1, -1).to(device))
    for _ in range(size):
        next_token = torch.argmax(out[0, -1]).reshape(1, 1)
        chars.append(vocab.itos[int(next_token.item())])
        out, state = net(next_token, state)
    return "".join(chars)

net = LSTMGenerator(vocab_size,64).to(device)

samples_to_train = 10000
optimizer = torch.optim.Adam(net.parameters(),0.01)
loss_fn = torch.nn.CrossEntropyLoss()
net.train()
for i,x in enumerate(train_dataset):
    # x[0] is class label, x[1] is text
    if len(x[1])-nchars<10:
        continue
    samples_to_train-=1
    if not samples_to_train: break
    text_in, text_out = get_batch(x[1])
    optimizer.zero_grad()
    out,s = net(text_in)
    loss = torch.nn.functional.cross_entropy(out.view(-1,vocab_size),text_out.flatten()) #cross_entropy(out,labels)
    loss.backward()
    optimizer.step()
    if i%1000==0:
        print(f"Current loss = {loss.item()}")
        print(generate(net))

def generate_soft(net, size=100, start="today ", temperature=1.0):
    if temperature <= 0:
        raise ValueError("temperature must be greater than zero")
    chars = list(start)
    out, state = net(enc(chars).view(1, -1).to(device))
    for _ in range(size):
        probabilities = torch.softmax(out[0, -1] / temperature, dim=0)
        next_token = torch.multinomial(probabilities, 1).reshape(1, 1)
        chars.append(vocab.itos[int(next_token.item())])
        out, state = net(next_token, state)
    return "".join(chars)

for temperature in [0.3, 0.8, 1.0, 1.3, 1.8]:
    sample = generate_soft(net, size=300, start="Today ", temperature=temperature)
    print(f"--- Temperature = {temperature}\n{sample}\n")