Source code for d2l.tensorflow
Open the notebook in Colab
Open the notebook in Colab
Open the notebook in Colab

DATA_HUB = dict()

import numpy as np
import tensorflow as tf

nn_Module = tf.keras.Model

#################   WARNING   ################
# The below part is generated automatically through:
#    d2lbook build lib
# Don't edit it directly

import collections
import hashlib
import math
import os
import random
import re
import shutil
import sys
import tarfile
import time
import zipfile
from collections import defaultdict
import pandas as pd
import requests
from IPython import display
from matplotlib import pyplot as plt
from matplotlib_inline import backend_inline

d2l = sys.modules[__name__]

import numpy as np
import tensorflow as tf

[docs]def use_svg_display(): """Use the svg format to display a plot in Jupyter. Defined in :numref:`sec_calculus`""" backend_inline.set_matplotlib_formats('svg')
[docs]def set_figsize(figsize=(3.5, 2.5)): """Set the figure size for matplotlib. Defined in :numref:`sec_calculus`""" use_svg_display() d2l.plt.rcParams['figure.figsize'] = figsize
[docs]def set_axes(axes, xlabel, ylabel, xlim, ylim, xscale, yscale, legend): """Set the axes for matplotlib. Defined in :numref:`sec_calculus`""" axes.set_xlabel(xlabel) axes.set_ylabel(ylabel) axes.set_xscale(xscale) axes.set_yscale(yscale) axes.set_xlim(xlim) axes.set_ylim(ylim) if legend: axes.legend(legend) axes.grid()
[docs]def plot(X, Y=None, xlabel=None, ylabel=None, legend=None, xlim=None, ylim=None, xscale='linear', yscale='linear', fmts=('-', 'm--', 'g-.', 'r:'), figsize=(3.5, 2.5), axes=None): """Plot data points. Defined in :numref:`sec_calculus`""" if legend is None: legend = [] set_figsize(figsize) axes = axes if axes else d2l.plt.gca() # Return True if `X` (tensor or list) has 1 axis def has_one_axis(X): return (hasattr(X, "ndim") and X.ndim == 1 or isinstance(X, list) and not hasattr(X[0], "__len__")) if has_one_axis(X): X = [X] if Y is None: X, Y = [[]] * len(X), X elif has_one_axis(Y): Y = [Y] if len(X) != len(Y): X = X * len(Y) axes.cla() for x, y, fmt in zip(X, Y, fmts): if len(x): axes.plot(x, y, fmt) else: axes.plot(y, fmt) set_axes(axes, xlabel, ylabel, xlim, ylim, xscale, yscale, legend)
[docs]class Timer: """Record multiple running times.""" def __init__(self): """Defined in :numref:`subsec_linear_model`""" self.times = [] self.start()
[docs] def start(self): """Start the timer.""" self.tik = time.time()
[docs] def stop(self): """Stop the timer and record the time in a list.""" self.times.append(time.time() - self.tik) return self.times[-1]
[docs] def avg(self): """Return the average time.""" return sum(self.times) / len(self.times)
[docs] def sum(self): """Return the sum of time.""" return sum(self.times)
[docs] def cumsum(self): """Return the accumulated time.""" return np.array(self.times).cumsum().tolist()
[docs]def synthetic_data(w, b, num_examples): """Generate y = Xw + b + noise. Defined in :numref:`sec_linear_scratch`""" X = d2l.zeros((num_examples, w.shape[0])) X += tf.random.normal(shape=X.shape) y = d2l.matmul(X, tf.reshape(w, (-1, 1))) + b y += tf.random.normal(shape=y.shape, stddev=0.01) y = d2l.reshape(y, (-1, 1)) return X, y
[docs]def linreg(X, w, b): """The linear regression model. Defined in :numref:`sec_linear_scratch`""" return d2l.matmul(X, w) + b
[docs]def squared_loss(y_hat, y): """Squared loss. Defined in :numref:`sec_linear_scratch`""" return (y_hat - d2l.reshape(y, y_hat.shape)) ** 2 / 2
[docs]def sgd(params, grads, lr, batch_size): """Minibatch stochastic gradient descent. Defined in :numref:`sec_linear_scratch`""" for param, grad in zip(params, grads): param.assign_sub(lr*grad/batch_size)
[docs]def load_array(data_arrays, batch_size, is_train=True): """Construct a TensorFlow data iterator. Defined in :numref:`sec_linear_concise`""" dataset = if is_train: dataset = dataset.shuffle(buffer_size=1000) dataset = dataset.batch(batch_size) return dataset
[docs]def get_fashion_mnist_labels(labels): """Return text labels for the Fashion-MNIST dataset. Defined in :numref:`sec_fashion_mnist`""" text_labels = ['t-shirt', 'trouser', 'pullover', 'dress', 'coat', 'sandal', 'shirt', 'sneaker', 'bag', 'ankle boot'] return [text_labels[int(i)] for i in labels]
[docs]def show_images(imgs, num_rows, num_cols, titles=None, scale=1.5): """Plot a list of images. Defined in :numref:`sec_fashion_mnist`""" figsize = (num_cols * scale, num_rows * scale) _, axes = d2l.plt.subplots(num_rows, num_cols, figsize=figsize) axes = axes.flatten() for i, (ax, img) in enumerate(zip(axes, imgs)): ax.imshow(d2l.numpy(img)) ax.axes.get_xaxis().set_visible(False) ax.axes.get_yaxis().set_visible(False) if titles: ax.set_title(titles[i]) return axes
[docs]def load_data_fashion_mnist(batch_size, resize=None): """Download the Fashion-MNIST dataset and then load it into memory. Defined in :numref:`sec_fashion_mnist`""" mnist_train, mnist_test = tf.keras.datasets.fashion_mnist.load_data() # Divide all numbers by 255 so that all pixel values are between # 0 and 1, add a batch dimension at the last. And cast label to int32 process = lambda X, y: (tf.expand_dims(X, axis=3) / 255, tf.cast(y, dtype='int32')) resize_fn = lambda X, y: ( tf.image.resize_with_pad(X, resize, resize) if resize else X, y) return (*mnist_train)).batch( batch_size).shuffle(len(mnist_train[0])).map(resize_fn),*mnist_test)).batch( batch_size).map(resize_fn))
[docs]def accuracy(y_hat, y): """Compute the number of correct predictions. Defined in :numref:`sec_softmax_scratch`""" if len(y_hat.shape) > 1 and y_hat.shape[1] > 1: y_hat = d2l.argmax(y_hat, axis=1) cmp = d2l.astype(y_hat, y.dtype) == y return float(d2l.reduce_sum(d2l.astype(cmp, y.dtype)))
[docs]def evaluate_accuracy(net, data_iter): """Compute the accuracy for a model on a dataset. Defined in :numref:`sec_softmax_scratch`""" metric = Accumulator(2) # No. of correct predictions, no. of predictions for X, y in data_iter: metric.add(accuracy(net(X), y), d2l.size(y)) return metric[0] / metric[1]
[docs]class Accumulator: """For accumulating sums over `n` variables.""" def __init__(self, n): """Defined in :numref:`sec_softmax_scratch`""" = [0.0] * n
[docs] def add(self, *args): = [a + float(b) for a, b in zip(, args)]
[docs] def reset(self): = [0.0] * len(
def __getitem__(self, idx): return[idx]
[docs]def train_epoch_ch3(net, train_iter, loss, updater): """The training loop defined in Chapter 3. Defined in :numref:`sec_softmax_scratch`""" # Sum of training loss, sum of training accuracy, no. of examples metric = Accumulator(3) for X, y in train_iter: # Compute gradients and update parameters with tf.GradientTape() as tape: y_hat = net(X) # Keras implementations for loss takes (labels, predictions) # instead of (predictions, labels) that users might implement # in this book, e.g. `cross_entropy` that we implemented above if isinstance(loss, tf.keras.losses.Loss): l = loss(y, y_hat) else: l = loss(y_hat, y) if isinstance(updater, tf.keras.optimizers.Optimizer): params = net.trainable_variables grads = tape.gradient(l, params) updater.apply_gradients(zip(grads, params)) else: updater(X.shape[0], tape.gradient(l, updater.params)) # Keras loss by default returns the average loss in a batch l_sum = l * float(tf.size(y)) if isinstance( loss, tf.keras.losses.Loss) else tf.reduce_sum(l) metric.add(l_sum, accuracy(y_hat, y), tf.size(y)) # Return training loss and training accuracy return metric[0] / metric[2], metric[1] / metric[2]
[docs]class Animator: """For plotting data in animation.""" def __init__(self, xlabel=None, ylabel=None, legend=None, xlim=None, ylim=None, xscale='linear', yscale='linear', fmts=('-', 'm--', 'g-.', 'r:'), nrows=1, ncols=1, figsize=(3.5, 2.5)): """Defined in :numref:`sec_softmax_scratch`""" # Incrementally plot multiple lines if legend is None: legend = [] d2l.use_svg_display() self.fig, self.axes = d2l.plt.subplots(nrows, ncols, figsize=figsize) if nrows * ncols == 1: self.axes = [self.axes, ] # Use a lambda function to capture arguments self.config_axes = lambda: d2l.set_axes( self.axes[0], xlabel, ylabel, xlim, ylim, xscale, yscale, legend) self.X, self.Y, self.fmts = None, None, fmts
[docs] def add(self, x, y): # Add multiple data points into the figure if not hasattr(y, "__len__"): y = [y] n = len(y) if not hasattr(x, "__len__"): x = [x] * n if not self.X: self.X = [[] for _ in range(n)] if not self.Y: self.Y = [[] for _ in range(n)] for i, (a, b) in enumerate(zip(x, y)): if a is not None and b is not None: self.X[i].append(a) self.Y[i].append(b) self.axes[0].cla() for x, y, fmt in zip(self.X, self.Y, self.fmts): self.axes[0].plot(x, y, fmt) self.config_axes() display.display(self.fig) display.clear_output(wait=True)
[docs]def train_ch3(net, train_iter, test_iter, loss, num_epochs, updater): """Train a model (defined in Chapter 3). Defined in :numref:`sec_softmax_scratch`""" animator = Animator(xlabel='epoch', xlim=[1, num_epochs], ylim=[0.3, 0.9], legend=['train loss', 'train acc', 'test acc']) for epoch in range(num_epochs): train_metrics = train_epoch_ch3(net, train_iter, loss, updater) test_acc = evaluate_accuracy(net, test_iter) animator.add(epoch + 1, train_metrics + (test_acc,)) train_loss, train_acc = train_metrics assert train_loss < 0.5, train_loss assert train_acc <= 1 and train_acc > 0.7, train_acc assert test_acc <= 1 and test_acc > 0.7, test_acc
[docs]class Updater(): """For updating parameters using minibatch stochastic gradient descent. Defined in :numref:`sec_softmax_scratch`""" def __init__(self, params, lr): self.params = params = lr def __call__(self, batch_size, grads): d2l.sgd(self.params, grads,, batch_size)
[docs]def predict_ch3(net, test_iter, n=6): """Predict labels (defined in Chapter 3). Defined in :numref:`sec_softmax_scratch`""" for X, y in test_iter: break trues = d2l.get_fashion_mnist_labels(y) preds = d2l.get_fashion_mnist_labels(d2l.argmax(net(X), axis=1)) titles = [true +'\n' + pred for true, pred in zip(trues, preds)] d2l.show_images( d2l.reshape(X[0:n], (n, 28, 28)), 1, n, titles=titles[0:n])
[docs]def evaluate_loss(net, data_iter, loss): """Evaluate the loss of a model on the given dataset. Defined in :numref:`sec_model_selection`""" metric = d2l.Accumulator(2) # Sum of losses, no. of examples for X, y in data_iter: l = loss(net(X), y) metric.add(d2l.reduce_sum(l), d2l.size(l)) return metric[0] / metric[1]
DATA_HUB = dict() DATA_URL = ''
[docs]def download(name, cache_dir=os.path.join('..', 'data')): """Download a file inserted into DATA_HUB, return the local filename. Defined in :numref:`sec_kaggle_house`""" assert name in DATA_HUB, f"{name} does not exist in {DATA_HUB}." url, sha1_hash = DATA_HUB[name] os.makedirs(cache_dir, exist_ok=True) fname = os.path.join(cache_dir, url.split('/')[-1]) if os.path.exists(fname): sha1 = hashlib.sha1() with open(fname, 'rb') as f: while True: data = if not data: break sha1.update(data) if sha1.hexdigest() == sha1_hash: return fname # Hit cache print(f'Downloading {fname} from {url}...') r = requests.get(url, stream=True, verify=True) with open(fname, 'wb') as f: f.write(r.content) return fname
[docs]def download_extract(name, folder=None): """Download and extract a zip/tar file. Defined in :numref:`sec_kaggle_house`""" fname = download(name) base_dir = os.path.dirname(fname) data_dir, ext = os.path.splitext(fname) if ext == '.zip': fp = zipfile.ZipFile(fname, 'r') elif ext in ('.tar', '.gz'): fp =, 'r') else: assert False, 'Only zip/tar files can be extracted.' fp.extractall(base_dir) return os.path.join(base_dir, folder) if folder else data_dir
[docs]def download_all(): """Download all files in the DATA_HUB. Defined in :numref:`sec_kaggle_house`""" for name in DATA_HUB: download(name)
DATA_HUB['kaggle_house_train'] = ( DATA_URL + 'kaggle_house_pred_train.csv', '585e9cc93e70b39160e7921475f9bcd7d31219ce') DATA_HUB['kaggle_house_test'] = ( DATA_URL + 'kaggle_house_pred_test.csv', 'fa19780a7b011d9b009e8bff8e99922a8ee2eb90')
[docs]def try_gpu(i=0): """Return gpu(i) if exists, otherwise return cpu(). Defined in :numref:`sec_use_gpu`""" if len(tf.config.experimental.list_physical_devices('GPU')) >= i + 1: return tf.device(f'/GPU:{i}') return tf.device('/CPU:0')
[docs]def try_all_gpus(): """Return all available GPUs, or [cpu(),] if no GPU exists. Defined in :numref:`sec_use_gpu`""" num_gpus = len(tf.config.experimental.list_physical_devices('GPU')) devices = [tf.device(f'/GPU:{i}') for i in range(num_gpus)] return devices if devices else [tf.device('/CPU:0')]
[docs]def corr2d(X, K): """Compute 2D cross-correlation.""" h, w = K.shape Y = tf.Variable(tf.zeros((X.shape[0] - h + 1, X.shape[1] - w + 1))) for i in range(Y.shape[0]): for j in range(Y.shape[1]): Y[i, j].assign(tf.reduce_sum( X[i: i + h, j: j + w] * K)) return Y
[docs]class TrainCallback(tf.keras.callbacks.Callback): """A callback to visiualize the training progress. Defined in :numref:`sec_lenet`""" def __init__(self, net, train_iter, test_iter, num_epochs, device_name): self.timer = d2l.Timer() self.animator = d2l.Animator( xlabel='epoch', xlim=[1, num_epochs], legend=[ 'train loss', 'train acc', 'test acc']) = net self.train_iter = train_iter self.test_iter = test_iter self.num_epochs = num_epochs self.device_name = device_name
[docs] def on_epoch_begin(self, epoch, logs=None): self.timer.start()
[docs] def on_epoch_end(self, epoch, logs): self.timer.stop() test_acc = self.test_iter, verbose=0, return_dict=True)['accuracy'] metrics = (logs['loss'], logs['accuracy'], test_acc) self.animator.add(epoch + 1, metrics) if epoch == self.num_epochs - 1: batch_size = next(iter(self.train_iter))[0].shape[0] num_examples = batch_size * self.train_iter).numpy() print(f'loss {metrics[0]:.3f}, train acc {metrics[1]:.3f}, ' f'test acc {metrics[2]:.3f}') print(f'{num_examples / self.timer.avg():.1f} examples/sec on ' f'{str(self.device_name)}')
[docs]def train_ch6(net_fn, train_iter, test_iter, num_epochs, lr, device): """Train a model with a GPU (defined in Chapter 6). Defined in :numref:`sec_lenet`""" device_name = device._device_name strategy = tf.distribute.OneDeviceStrategy(device_name) with strategy.scope(): optimizer = tf.keras.optimizers.SGD(learning_rate=lr) loss = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True) net = net_fn() net.compile(optimizer=optimizer, loss=loss, metrics=['accuracy']) callback = TrainCallback(net, train_iter, test_iter, num_epochs, device_name), epochs=num_epochs, verbose=0, callbacks=[callback]) return net
[docs]class Residual(tf.keras.Model): """The Residual block of ResNet.""" def __init__(self, num_channels, use_1x1conv=False, strides=1): super().__init__() self.conv1 = tf.keras.layers.Conv2D( num_channels, padding='same', kernel_size=3, strides=strides) self.conv2 = tf.keras.layers.Conv2D( num_channels, kernel_size=3, padding='same') self.conv3 = None if use_1x1conv: self.conv3 = tf.keras.layers.Conv2D( num_channels, kernel_size=1, strides=strides) self.bn1 = tf.keras.layers.BatchNormalization() self.bn2 = tf.keras.layers.BatchNormalization()
[docs] def call(self, X): Y = tf.keras.activations.relu(self.bn1(self.conv1(X))) Y = self.bn2(self.conv2(Y)) if self.conv3 is not None: X = self.conv3(X) Y += X return tf.keras.activations.relu(Y)
d2l.DATA_HUB['time_machine'] = (d2l.DATA_URL + 'timemachine.txt', '090b5e7e70c295757f55df93cb0a180b9691891a')
[docs]def read_time_machine(): """Load the time machine dataset into a list of text lines. Defined in :numref:`sec_text_preprocessing`""" with open('time_machine'), 'r') as f: lines = f.readlines() return [re.sub('[^A-Za-z]+', ' ', line).strip().lower() for line in lines]
[docs]def tokenize(lines, token='word'): """Split text lines into word or character tokens. Defined in :numref:`sec_text_preprocessing`""" if token == 'word': return [line.split() for line in lines] elif token == 'char': return [list(line) for line in lines] else: print('ERROR: unknown token type: ' + token)
[docs]class Vocab: """Vocabulary for text.""" def __init__(self, tokens=None, min_freq=0, reserved_tokens=None): """Defined in :numref:`sec_text_preprocessing`""" if tokens is None: tokens = [] if reserved_tokens is None: reserved_tokens = [] # Sort according to frequencies counter = count_corpus(tokens) self._token_freqs = sorted(counter.items(), key=lambda x: x[1], reverse=True) # The index for the unknown token is 0 self.idx_to_token = ['<unk>'] + reserved_tokens self.token_to_idx = {token: idx for idx, token in enumerate(self.idx_to_token)} for token, freq in self._token_freqs: if freq < min_freq: break if token not in self.token_to_idx: self.idx_to_token.append(token) self.token_to_idx[token] = len(self.idx_to_token) - 1 def __len__(self): return len(self.idx_to_token) def __getitem__(self, tokens): if not isinstance(tokens, (list, tuple)): return self.token_to_idx.get(tokens, self.unk) return [self.__getitem__(token) for token in tokens]
[docs] def to_tokens(self, indices): if not isinstance(indices, (list, tuple)): return self.idx_to_token[indices] return [self.idx_to_token[index] for index in indices]
@property def unk(self): # Index for the unknown token return 0 @property def token_freqs(self): # Index for the unknown token return self._token_freqs
[docs]def count_corpus(tokens): """Count token frequencies. Defined in :numref:`sec_text_preprocessing`""" # Here `tokens` is a 1D list or 2D list if len(tokens) == 0 or isinstance(tokens[0], list): # Flatten a list of token lists into a list of tokens tokens = [token for line in tokens for token in line] return collections.Counter(tokens)
[docs]def load_corpus_time_machine(max_tokens=-1): """Return token indices and the vocabulary of the time machine dataset. Defined in :numref:`sec_text_preprocessing`""" lines = read_time_machine() tokens = tokenize(lines, 'char') vocab = Vocab(tokens) # Since each text line in the time machine dataset is not necessarily a # sentence or a paragraph, flatten all the text lines into a single list corpus = [vocab[token] for line in tokens for token in line] if max_tokens > 0: corpus = corpus[:max_tokens] return corpus, vocab
[docs]def seq_data_iter_random(corpus, batch_size, num_steps): """Generate a minibatch of subsequences using random sampling. Defined in :numref:`sec_language_model`""" # Start with a random offset (inclusive of `num_steps - 1`) to partition a # sequence corpus = corpus[random.randint(0, num_steps - 1):] # Subtract 1 since we need to account for labels num_subseqs = (len(corpus) - 1) // num_steps # The starting indices for subsequences of length `num_steps` initial_indices = list(range(0, num_subseqs * num_steps, num_steps)) # In random sampling, the subsequences from two adjacent random # minibatches during iteration are not necessarily adjacent on the # original sequence random.shuffle(initial_indices) def data(pos): # Return a sequence of length `num_steps` starting from `pos` return corpus[pos: pos + num_steps] num_batches = num_subseqs // batch_size for i in range(0, batch_size * num_batches, batch_size): # Here, `initial_indices` contains randomized starting indices for # subsequences initial_indices_per_batch = initial_indices[i: i + batch_size] X = [data(j) for j in initial_indices_per_batch] Y = [data(j + 1) for j in initial_indices_per_batch] yield d2l.tensor(X), d2l.tensor(Y)
[docs]def seq_data_iter_sequential(corpus, batch_size, num_steps): """Generate a minibatch of subsequences using sequential partitioning. Defined in :numref:`sec_language_model`""" # Start with a random offset to partition a sequence offset = random.randint(0, num_steps) num_tokens = ((len(corpus) - offset - 1) // batch_size) * batch_size Xs = d2l.tensor(corpus[offset: offset + num_tokens]) Ys = d2l.tensor(corpus[offset + 1: offset + 1 + num_tokens]) Xs = d2l.reshape(Xs, (batch_size, -1)) Ys = d2l.reshape(Ys, (batch_size, -1)) num_batches = Xs.shape[1] // num_steps for i in range(0, num_batches * num_steps, num_steps): X = Xs[:, i: i + num_steps] Y = Ys[:, i: i + num_steps] yield X, Y
[docs]class SeqDataLoader: """An iterator to load sequence data.""" def __init__(self, batch_size, num_steps, use_random_iter, max_tokens): """Defined in :numref:`sec_language_model`""" if use_random_iter: self.data_iter_fn = d2l.seq_data_iter_random else: self.data_iter_fn = d2l.seq_data_iter_sequential self.corpus, self.vocab = d2l.load_corpus_time_machine(max_tokens) self.batch_size, self.num_steps = batch_size, num_steps def __iter__(self): return self.data_iter_fn(self.corpus, self.batch_size, self.num_steps)
[docs]def load_data_time_machine(batch_size, num_steps, use_random_iter=False, max_tokens=10000): """Return the iterator and the vocabulary of the time machine dataset. Defined in :numref:`sec_language_model`""" data_iter = SeqDataLoader( batch_size, num_steps, use_random_iter, max_tokens) return data_iter, data_iter.vocab
[docs]class RNNModelScratch: """A RNN Model implemented from scratch.""" def __init__(self, vocab_size, num_hiddens, init_state, forward_fn, get_params): """Defined in :numref:`sec_rnn_scratch`""" self.vocab_size, self.num_hiddens = vocab_size, num_hiddens self.init_state, self.forward_fn = init_state, forward_fn self.trainable_variables = get_params(vocab_size, num_hiddens) def __call__(self, X, state): X = tf.one_hot(tf.transpose(X), self.vocab_size) X = tf.cast(X, tf.float32) return self.forward_fn(X, state, self.trainable_variables)
[docs] def begin_state(self, batch_size, *args, **kwargs): return self.init_state(batch_size, self.num_hiddens)
[docs]def predict_ch8(prefix, num_preds, net, vocab): """Generate new characters following the `prefix`. Defined in :numref:`sec_rnn_scratch`""" state = net.begin_state(batch_size=1, dtype=tf.float32) outputs = [vocab[prefix[0]]] get_input = lambda: d2l.reshape(d2l.tensor([outputs[-1]]), (1, 1)).numpy() for y in prefix[1:]: # Warm-up period _, state = net(get_input(), state) outputs.append(vocab[y]) for _ in range(num_preds): # Predict `num_preds` steps y, state = net(get_input(), state) outputs.append(int(y.numpy().argmax(axis=1).reshape(1))) return ''.join([vocab.idx_to_token[i] for i in outputs])
[docs]def grad_clipping(grads, theta): """Clip the gradient. Defined in :numref:`sec_rnn_scratch`""" theta = tf.constant(theta, dtype=tf.float32) new_grad = [] for grad in grads: if isinstance(grad, tf.IndexedSlices): new_grad.append(tf.convert_to_tensor(grad)) else: new_grad.append(grad) norm = tf.math.sqrt(sum((tf.reduce_sum(grad ** 2)).numpy() for grad in new_grad)) norm = tf.cast(norm, tf.float32) if tf.greater(norm, theta): for i, grad in enumerate(new_grad): new_grad[i] = grad * theta / norm else: new_grad = new_grad return new_grad
[docs]def train_epoch_ch8(net, train_iter, loss, updater, use_random_iter): """Train a model within one epoch (defined in Chapter 8). Defined in :numref:`sec_rnn_scratch`""" state, timer = None, d2l.Timer() metric = d2l.Accumulator(2) # Sum of training loss, no. of tokens for X, Y in train_iter: if state is None or use_random_iter: # Initialize `state` when either it is the first iteration or # using random sampling state = net.begin_state(batch_size=X.shape[0], dtype=tf.float32) with tf.GradientTape(persistent=True) as g: y_hat, state = net(X, state) y = d2l.reshape(tf.transpose(Y), (-1)) l = loss(y, y_hat) params = net.trainable_variables grads = g.gradient(l, params) grads = grad_clipping(grads, 1) updater.apply_gradients(zip(grads, params)) # Keras loss by default returns the average loss in a batch # l_sum = l * float(d2l.size(y)) if isinstance( # loss, tf.keras.losses.Loss) else tf.reduce_sum(l) metric.add(l * d2l.size(y), d2l.size(y)) return math.exp(metric[0] / metric[1]), metric[1] / timer.stop()
[docs]def train_ch8(net, train_iter, vocab, lr, num_epochs, strategy, use_random_iter=False): """Train a model (defined in Chapter 8). Defined in :numref:`sec_rnn_scratch`""" with strategy.scope(): loss = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True) updater = tf.keras.optimizers.SGD(lr) animator = d2l.Animator(xlabel='epoch', ylabel='perplexity', legend=['train'], xlim=[10, num_epochs]) predict = lambda prefix: predict_ch8(prefix, 50, net, vocab) # Train and predict for epoch in range(num_epochs): ppl, speed = train_epoch_ch8(net, train_iter, loss, updater, use_random_iter) if (epoch + 1) % 10 == 0: print(predict('time traveller')) animator.add(epoch + 1, [ppl]) device = d2l.try_gpu()._device_name print(f'perplexity {ppl:.1f}, {speed:.1f} tokens/sec on {str(device)}') print(predict('time traveller')) print(predict('traveller'))
[docs]class RNNModel(tf.keras.layers.Layer): """Defined in :numref:`sec_rnn-concise`""" def __init__(self, rnn_layer, vocab_size, **kwargs): super(RNNModel, self).__init__(**kwargs) self.rnn = rnn_layer self.vocab_size = vocab_size self.dense = tf.keras.layers.Dense(vocab_size)
[docs] def call(self, inputs, state): X = tf.one_hot(tf.transpose(inputs), self.vocab_size) # Later RNN like `tf.keras.layers.LSTMCell` return more than two values Y, *state = self.rnn(X, state) output = self.dense(tf.reshape(Y, (-1, Y.shape[-1]))) return output, state
[docs] def begin_state(self, *args, **kwargs): return self.rnn.cell.get_initial_state(*args, **kwargs)
d2l.DATA_HUB['fra-eng'] = (d2l.DATA_URL + '', '94646ad1522d915e7b0f9296181140edcf86a4f5')
[docs]def read_data_nmt(): """Load the English-French dataset. Defined in :numref:`sec_machine_translation`""" data_dir = d2l.download_extract('fra-eng') with open(os.path.join(data_dir, 'fra.txt'), 'r') as f: return
[docs]def preprocess_nmt(text): """Preprocess the English-French dataset. Defined in :numref:`sec_machine_translation`""" def no_space(char, prev_char): return char in set(',.!?') and prev_char != ' ' # Replace non-breaking space with space, and convert uppercase letters to # lowercase ones text = text.replace('\u202f', ' ').replace('\xa0', ' ').lower() # Insert space between words and punctuation marks out = [' ' + char if i > 0 and no_space(char, text[i - 1]) else char for i, char in enumerate(text)] return ''.join(out)
[docs]def tokenize_nmt(text, num_examples=None): """Tokenize the English-French dataset. Defined in :numref:`sec_machine_translation`""" source, target = [], [] for i, line in enumerate(text.split('\n')): if num_examples and i > num_examples: break parts = line.split('\t') if len(parts) == 2: source.append(parts[0].split(' ')) target.append(parts[1].split(' ')) return source, target
[docs]def show_list_len_pair_hist(legend, xlabel, ylabel, xlist, ylist): """Plot the histogram for list length pairs. Defined in :numref:`sec_machine_translation`""" d2l.set_figsize() _, _, patches = d2l.plt.hist( [[len(l) for l in xlist], [len(l) for l in ylist]]) d2l.plt.xlabel(xlabel) d2l.plt.ylabel(ylabel) for patch in patches[1].patches: patch.set_hatch('/') d2l.plt.legend(legend)
[docs]def truncate_pad(line, num_steps, padding_token): """Truncate or pad sequences. Defined in :numref:`sec_machine_translation`""" if len(line) > num_steps: return line[:num_steps] # Truncate return line + [padding_token] * (num_steps - len(line)) # Pad
[docs]def build_array_nmt(lines, vocab, num_steps): """Transform text sequences of machine translation into minibatches. Defined in :numref:`subsec_mt_data_loading`""" lines = [vocab[l] for l in lines] lines = [l + [vocab['<eos>']] for l in lines] array = d2l.tensor([truncate_pad( l, num_steps, vocab['<pad>']) for l in lines]) valid_len = d2l.reduce_sum( d2l.astype(array != vocab['<pad>'], d2l.int32), 1) return array, valid_len
[docs]def load_data_nmt(batch_size, num_steps, num_examples=600): """Return the iterator and the vocabularies of the translation dataset. Defined in :numref:`subsec_mt_data_loading`""" text = preprocess_nmt(read_data_nmt()) source, target = tokenize_nmt(text, num_examples) src_vocab = d2l.Vocab(source, min_freq=2, reserved_tokens=['<pad>', '<bos>', '<eos>']) tgt_vocab = d2l.Vocab(target, min_freq=2, reserved_tokens=['<pad>', '<bos>', '<eos>']) src_array, src_valid_len = build_array_nmt(source, src_vocab, num_steps) tgt_array, tgt_valid_len = build_array_nmt(target, tgt_vocab, num_steps) data_arrays = (src_array, src_valid_len, tgt_array, tgt_valid_len) data_iter = d2l.load_array(data_arrays, batch_size) return data_iter, src_vocab, tgt_vocab
[docs]class Encoder(tf.keras.layers.Layer): """The base encoder interface for the encoder-decoder architecture.""" def __init__(self, **kwargs): super(Encoder, self).__init__(**kwargs)
[docs] def call(self, X, *args, **kwargs): raise NotImplementedError
[docs]class Decoder(tf.keras.layers.Layer): """The base decoder interface for the encoder-decoder architecture. Defined in :numref:`sec_encoder-decoder`""" def __init__(self, **kwargs): super(Decoder, self).__init__(**kwargs)
[docs] def init_state(self, enc_outputs, *args): raise NotImplementedError
[docs] def call(self, X, state, **kwargs): raise NotImplementedError
[docs]class EncoderDecoder(tf.keras.Model): """The base class for the encoder-decoder architecture. Defined in :numref:`sec_encoder-decoder`""" def __init__(self, encoder, decoder, **kwargs): super(EncoderDecoder, self).__init__(**kwargs) self.encoder = encoder self.decoder = decoder
[docs] def call(self, enc_X, dec_X, *args, **kwargs): enc_outputs = self.encoder(enc_X, *args, **kwargs) dec_state = self.decoder.init_state(enc_outputs, *args) return self.decoder(dec_X, dec_state, **kwargs)
[docs]class Seq2SeqEncoder(d2l.Encoder): """The RNN encoder for sequence to sequence learning. Defined in :numref:`sec_seq2seq`""" def __init__(self, vocab_size, embed_size, num_hiddens, num_layers, dropout=0, **kwargs): super().__init__(*kwargs) # Embedding layer self.embedding = tf.keras.layers.Embedding(vocab_size, embed_size) self.rnn = tf.keras.layers.RNN(tf.keras.layers.StackedRNNCells( [tf.keras.layers.GRUCell(num_hiddens, dropout=dropout) for _ in range(num_layers)]), return_sequences=True, return_state=True)
[docs] def call(self, X, *args, **kwargs): # The input `X` shape: (`batch_size`, `num_steps`) # The output `X` shape: (`batch_size`, `num_steps`, `embed_size`) X = self.embedding(X) output = self.rnn(X, **kwargs) state = output[1:] return output[0], state
[docs]def sequence_mask(X, valid_len, value=0): """Mask irrelevant entries in sequences. Defined in :numref:`sec_seq2seq_decoder`""" maxlen = X.shape[1] mask = tf.range(start=0, limit=maxlen, dtype=tf.float32)[ None, :] < tf.cast(valid_len[:, None], dtype=tf.float32) if len(X.shape) == 3: return tf.where(tf.expand_dims(mask, axis=-1), X, value) else: return tf.where(mask, X, value)
[docs]class MaskedSoftmaxCELoss(tf.keras.losses.Loss): """The softmax cross-entropy loss with masks. Defined in :numref:`sec_seq2seq_decoder`""" def __init__(self, valid_len): super().__init__(reduction='none') self.valid_len = valid_len # `pred` shape: (`batch_size`, `num_steps`, `vocab_size`) # `label` shape: (`batch_size`, `num_steps`) # `valid_len` shape: (`batch_size`,)
[docs] def call(self, label, pred): weights = tf.ones_like(label, dtype=tf.float32) weights = sequence_mask(weights, self.valid_len) label_one_hot = tf.one_hot(label, depth=pred.shape[-1]) unweighted_loss = tf.keras.losses.CategoricalCrossentropy( from_logits=True, reduction='none')(label_one_hot, pred) weighted_loss = tf.reduce_mean((unweighted_loss*weights), axis=1) return weighted_loss
[docs]def train_seq2seq(net, data_iter, lr, num_epochs, tgt_vocab, device): """Train a model for sequence to sequence. Defined in :numref:`sec_seq2seq_decoder`""" optimizer = tf.keras.optimizers.Adam(learning_rate=lr) animator = d2l.Animator(xlabel="epoch", ylabel="loss", xlim=[10, num_epochs]) for epoch in range(num_epochs): timer = d2l.Timer() metric = d2l.Accumulator(2) # Sum of training loss, no. of tokens for batch in data_iter: X, X_valid_len, Y, Y_valid_len = [x for x in batch] bos = tf.reshape(tf.constant([tgt_vocab['<bos>']] * Y.shape[0]), shape=(-1, 1)) dec_input = tf.concat([bos, Y[:, :-1]], 1) # Teacher forcing with tf.GradientTape() as tape: Y_hat, _ = net(X, dec_input, X_valid_len, training=True) l = MaskedSoftmaxCELoss(Y_valid_len)(Y, Y_hat) gradients = tape.gradient(l, net.trainable_variables) gradients = d2l.grad_clipping(gradients, 1) optimizer.apply_gradients(zip(gradients, net.trainable_variables)) num_tokens = tf.reduce_sum(Y_valid_len).numpy() metric.add(tf.reduce_sum(l), num_tokens) if (epoch + 1) % 10 == 0: animator.add(epoch + 1, (metric[0] / metric[1],)) print(f'loss {metric[0] / metric[1]:.3f}, {metric[1] / timer.stop():.1f} ' f'tokens/sec on {str(device)}')
[docs]def predict_seq2seq(net, src_sentence, src_vocab, tgt_vocab, num_steps, save_attention_weights=False): """Predict for sequence to sequence. Defined in :numref:`sec_seq2seq_training`""" src_tokens = src_vocab[src_sentence.lower().split(' ')] + [ src_vocab['<eos>']] enc_valid_len = tf.constant([len(src_tokens)]) src_tokens = d2l.truncate_pad(src_tokens, num_steps, src_vocab['<pad>']) # Add the batch axis enc_X = tf.expand_dims(src_tokens, axis=0) enc_outputs = net.encoder(enc_X, enc_valid_len, training=False) dec_state = net.decoder.init_state(enc_outputs, enc_valid_len) # Add the batch axis dec_X = tf.expand_dims(tf.constant([tgt_vocab['<bos>']]), axis=0) output_seq, attention_weight_seq = [], [] for _ in range(num_steps): Y, dec_state = net.decoder(dec_X, dec_state, training=False) # We use the token with the highest prediction likelihood as the input # of the decoder at the next time step dec_X = tf.argmax(Y, axis=2) pred = tf.squeeze(dec_X, axis=0) # Save attention weights if save_attention_weights: attention_weight_seq.append(net.decoder.attention_weights) # Once the end-of-sequence token is predicted, the generation of the # output sequence is complete if pred == tgt_vocab['<eos>']: break output_seq.append(pred.numpy()) return ' '.join(tgt_vocab.to_tokens(tf.reshape(output_seq, shape = -1).numpy().tolist())), attention_weight_seq
[docs]def bleu(pred_seq, label_seq, k): """Compute the BLEU. Defined in :numref:`sec_seq2seq_training`""" pred_tokens, label_tokens = pred_seq.split(' '), label_seq.split(' ') len_pred, len_label = len(pred_tokens), len(label_tokens) score = math.exp(min(0, 1 - len_label / len_pred)) for n in range(1, k + 1): num_matches, label_subs = 0, collections.defaultdict(int) for i in range(len_label - n + 1): label_subs[' '.join(label_tokens[i: i + n])] += 1 for i in range(len_pred - n + 1): if label_subs[' '.join(pred_tokens[i: i + n])] > 0: num_matches += 1 label_subs[' '.join(pred_tokens[i: i + n])] -= 1 score *= math.pow(num_matches / (len_pred - n + 1), math.pow(0.5, n)) return score
[docs]def show_heatmaps(matrices, xlabel, ylabel, titles=None, figsize=(2.5, 2.5), cmap='Reds'): """Show heatmaps of matrices. Defined in :numref:`sec_attention-cues`""" d2l.use_svg_display() num_rows, num_cols = matrices.shape[0], matrices.shape[1] fig, axes = d2l.plt.subplots(num_rows, num_cols, figsize=figsize, sharex=True, sharey=True, squeeze=False) for i, (row_axes, row_matrices) in enumerate(zip(axes, matrices)): for j, (ax, matrix) in enumerate(zip(row_axes, row_matrices)): pcm = ax.imshow(d2l.numpy(matrix), cmap=cmap) if i == num_rows - 1: ax.set_xlabel(xlabel) if j == 0: ax.set_ylabel(ylabel) if titles: ax.set_title(titles[j]) fig.colorbar(pcm, ax=axes, shrink=0.6);
[docs]def masked_softmax(X, valid_lens): """Perform softmax operation by masking elements on the last axis. Defined in :numref:`sec_attention-scoring-functions`""" # `X`: 3D tensor, `valid_lens`: 1D or 2D tensor if valid_lens is None: return tf.nn.softmax(X, axis=-1) else: shape = X.shape if len(valid_lens.shape) == 1: valid_lens = tf.repeat(valid_lens, repeats=shape[1]) else: valid_lens = tf.reshape(valid_lens, shape=-1) # On the last axis, replace masked elements with a very large negative # value, whose exponentiation outputs 0 X = d2l.sequence_mask(tf.reshape(X, shape=(-1, shape[-1])), valid_lens, value=-1e6) return tf.nn.softmax(tf.reshape(X, shape=shape), axis=-1)
[docs]class AdditiveAttention(tf.keras.layers.Layer): """Additive attention. Defined in :numref:`sec_attention-scoring-functions`""" def __init__(self, key_size, query_size, num_hiddens, dropout, **kwargs): super().__init__(**kwargs) self.W_k = tf.keras.layers.Dense(num_hiddens, use_bias=False) self.W_q = tf.keras.layers.Dense(num_hiddens, use_bias=False) self.w_v = tf.keras.layers.Dense(1, use_bias=False) self.dropout = tf.keras.layers.Dropout(dropout)
[docs] def call(self, queries, keys, values, valid_lens, **kwargs): queries, keys = self.W_q(queries), self.W_k(keys) # After dimension expansion, shape of `queries`: (`batch_size`, no. of # queries, 1, `num_hiddens`) and shape of `keys`: (`batch_size`, 1, # no. of key-value pairs, `num_hiddens`). Sum them up with # broadcasting features = tf.expand_dims(queries, axis=2) + tf.expand_dims( keys, axis=1) features = tf.nn.tanh(features) # There is only one output of `self.w_v`, so we remove the last # one-dimensional entry from the shape. Shape of `scores`: # (`batch_size`, no. of queries, no. of key-value pairs) scores = tf.squeeze(self.w_v(features), axis=-1) self.attention_weights = masked_softmax(scores, valid_lens) # Shape of `values`: (`batch_size`, no. of key-value pairs, value # dimension) return tf.matmul(self.dropout( self.attention_weights, **kwargs), values)
[docs]class DotProductAttention(tf.keras.layers.Layer): """Scaled dot product attention. Defined in :numref:`subsec_additive-attention`""" def __init__(self, dropout, **kwargs): super().__init__(**kwargs) self.dropout = tf.keras.layers.Dropout(dropout) # Shape of `queries`: (`batch_size`, no. of queries, `d`) # Shape of `keys`: (`batch_size`, no. of key-value pairs, `d`) # Shape of `values`: (`batch_size`, no. of key-value pairs, value # dimension) # Shape of `valid_lens`: (`batch_size`,) or (`batch_size`, no. of queries)
[docs] def call(self, queries, keys, values, valid_lens, **kwargs): d = queries.shape[-1] scores = tf.matmul(queries, keys, transpose_b=True)/tf.math.sqrt( tf.cast(d, dtype=tf.float32)) self.attention_weights = masked_softmax(scores, valid_lens) return tf.matmul(self.dropout(self.attention_weights, **kwargs), values)
[docs]class AttentionDecoder(d2l.Decoder): """The base attention-based decoder interface. Defined in :numref:`sec_seq2seq_attention`""" def __init__(self, **kwargs): super(AttentionDecoder, self).__init__(**kwargs) @property def attention_weights(self): raise NotImplementedError
[docs]class MultiHeadAttention(tf.keras.layers.Layer): """Multi-head attention. Defined in :numref:`sec_multihead-attention`""" def __init__(self, key_size, query_size, value_size, num_hiddens, num_heads, dropout, bias=False, **kwargs): super().__init__(**kwargs) self.num_heads = num_heads self.attention = d2l.DotProductAttention(dropout) self.W_q = tf.keras.layers.Dense(num_hiddens, use_bias=bias) self.W_k = tf.keras.layers.Dense(num_hiddens, use_bias=bias) self.W_v = tf.keras.layers.Dense(num_hiddens, use_bias=bias) self.W_o = tf.keras.layers.Dense(num_hiddens, use_bias=bias)
[docs] def call(self, queries, keys, values, valid_lens, **kwargs): # Shape of `queries`, `keys`, or `values`: # (`batch_size`, no. of queries or key-value pairs, `num_hiddens`) # Shape of `valid_lens`: # (`batch_size`,) or (`batch_size`, no. of queries) # After transposing, shape of output `queries`, `keys`, or `values`: # (`batch_size` * `num_heads`, no. of queries or key-value pairs, # `num_hiddens` / `num_heads`) queries = transpose_qkv(self.W_q(queries), self.num_heads) keys = transpose_qkv(self.W_k(keys), self.num_heads) values = transpose_qkv(self.W_v(values), self.num_heads) if valid_lens is not None: # On axis 0, copy the first item (scalar or vector) for # `num_heads` times, then copy the next item, and so on valid_lens = tf.repeat(valid_lens, repeats=self.num_heads, axis=0) # Shape of `output`: (`batch_size` * `num_heads`, no. of queries, `num_hiddens` / `num_heads`) output = self.attention(queries, keys, values, valid_lens, **kwargs) # Shape of `output_concat`: (`batch_size`, no. of queries, `num_hiddens`) output_concat = transpose_output(output, self.num_heads) return self.W_o(output_concat)
[docs]def transpose_qkv(X, num_heads): """Transposition for parallel computation of multiple attention heads. Defined in :numref:`sec_multihead-attention`""" # Shape of input `X`: # (`batch_size`, no. of queries or key-value pairs, `num_hiddens`). # Shape of output `X`: # (`batch_size`, no. of queries or key-value pairs, `num_heads`, # `num_hiddens` / `num_heads`) X = tf.reshape(X, shape=(X.shape[0], X.shape[1], num_heads, -1)) # Shape of output `X`: # (`batch_size`, `num_heads`, no. of queries or key-value pairs, # `num_hiddens` / `num_heads`) X = tf.transpose(X, perm=(0, 2, 1, 3)) # Shape of `output`: # (`batch_size` * `num_heads`, no. of queries or key-value pairs, # `num_hiddens` / `num_heads`) return tf.reshape(X, shape=(-1, X.shape[2], X.shape[3]))
[docs]def transpose_output(X, num_heads): """Reverse the operation of `transpose_qkv`. Defined in :numref:`sec_multihead-attention`""" X = tf.reshape(X, shape=(-1, num_heads, X.shape[1], X.shape[2])) X = tf.transpose(X, perm=(0, 2, 1, 3)) return tf.reshape(X, shape=(X.shape[0], X.shape[1], -1))
[docs]class PositionalEncoding(tf.keras.layers.Layer): """Positional encoding. Defined in :numref:`sec_self-attention-and-positional-encoding`""" def __init__(self, num_hiddens, dropout, max_len=1000): super().__init__() self.dropout = tf.keras.layers.Dropout(dropout) # Create a long enough `P` self.P = np.zeros((1, max_len, num_hiddens)) X = np.arange(max_len, dtype=np.float32).reshape( -1,1)/np.power(10000, np.arange( 0, num_hiddens, 2, dtype=np.float32) / num_hiddens) self.P[:, :, 0::2] = np.sin(X) self.P[:, :, 1::2] = np.cos(X)
[docs] def call(self, X, **kwargs): X = X + self.P[:, :X.shape[1], :] return self.dropout(X, **kwargs)
[docs]class PositionWiseFFN(tf.keras.layers.Layer): """Positionwise feed-forward network. Defined in :numref:`sec_transformer`""" def __init__(self, ffn_num_hiddens, ffn_num_outputs, **kwargs): super().__init__(*kwargs) self.dense1 = tf.keras.layers.Dense(ffn_num_hiddens) self.relu = tf.keras.layers.ReLU() self.dense2 = tf.keras.layers.Dense(ffn_num_outputs)
[docs] def call(self, X): return self.dense2(self.relu(self.dense1(X)))
[docs]class AddNorm(tf.keras.layers.Layer): """Residual connection followed by layer normalization. Defined in :numref:`sec_transformer`""" def __init__(self, normalized_shape, dropout, **kwargs): super().__init__(**kwargs) self.dropout = tf.keras.layers.Dropout(dropout) self.ln = tf.keras.layers.LayerNormalization(normalized_shape)
[docs] def call(self, X, Y, **kwargs): return self.ln(self.dropout(Y, **kwargs) + X)
[docs]class EncoderBlock(tf.keras.layers.Layer): """Transformer encoder block. Defined in :numref:`sec_transformer`""" def __init__(self, key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_hiddens, num_heads, dropout, bias=False, **kwargs): super().__init__(**kwargs) self.attention = d2l.MultiHeadAttention(key_size, query_size, value_size, num_hiddens, num_heads, dropout, bias) self.addnorm1 = AddNorm(norm_shape, dropout) self.ffn = PositionWiseFFN(ffn_num_hiddens, num_hiddens) self.addnorm2 = AddNorm(norm_shape, dropout)
[docs] def call(self, X, valid_lens, **kwargs): Y = self.addnorm1(X, self.attention(X, X, X, valid_lens, **kwargs), **kwargs) return self.addnorm2(Y, self.ffn(Y), **kwargs)
[docs]class TransformerEncoder(d2l.Encoder): """Transformer encoder. Defined in :numref:`sec_transformer`""" def __init__(self, vocab_size, key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_hiddens, num_heads, num_layers, dropout, bias=False, **kwargs): super().__init__(**kwargs) self.num_hiddens = num_hiddens self.embedding = tf.keras.layers.Embedding(vocab_size, num_hiddens) self.pos_encoding = d2l.PositionalEncoding(num_hiddens, dropout) self.blks = [EncoderBlock( key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_hiddens, num_heads, dropout, bias) for _ in range( num_layers)]
[docs] def call(self, X, valid_lens, **kwargs): # Since positional encoding values are between -1 and 1, the embedding # values are multiplied by the square root of the embedding dimension # to rescale before they are summed up X = self.pos_encoding(self.embedding(X) * tf.math.sqrt( tf.cast(self.num_hiddens, dtype=tf.float32)), **kwargs) self.attention_weights = [None] * len(self.blks) for i, blk in enumerate(self.blks): X = blk(X, valid_lens, **kwargs) self.attention_weights[ i] = blk.attention.attention.attention_weights return X
[docs]def annotate(text, xy, xytext): d2l.plt.gca().annotate(text, xy=xy, xytext=xytext, arrowprops=dict(arrowstyle='->'))
[docs]def train_2d(trainer, steps=20, f_grad=None): """Optimize a 2D objective function with a customized trainer. Defined in :numref:`subsec_gd-learningrate`""" # `s1` and `s2` are internal state variables that will be used later x1, x2, s1, s2 = -5, -2, 0, 0 results = [(x1, x2)] for i in range(steps): if f_grad: x1, x2, s1, s2 = trainer(x1, x2, s1, s2, f_grad) else: x1, x2, s1, s2 = trainer(x1, x2, s1, s2) results.append((x1, x2)) print(f'epoch {i + 1}, x1: {float(x1):f}, x2: {float(x2):f}') return results
[docs]def show_trace_2d(f, results): """Show the trace of 2D variables during optimization. Defined in :numref:`subsec_gd-learningrate`""" d2l.set_figsize() d2l.plt.plot(*zip(*results), '-o', color='#ff7f0e') x1, x2 = d2l.meshgrid(d2l.arange(-5.5, 1.0, 0.1), d2l.arange(-3.0, 1.0, 0.1)) d2l.plt.contour(x1, x2, f(x1, x2), colors='#1f77b4') d2l.plt.xlabel('x1') d2l.plt.ylabel('x2')
d2l.DATA_HUB['airfoil'] = (d2l.DATA_URL + 'airfoil_self_noise.dat', '76e5be1548fd8222e5074cf0faae75edff8cf93f')
[docs]def get_data_ch11(batch_size=10, n=1500): """Defined in :numref:`sec_minibatches`""" data = np.genfromtxt('airfoil'), dtype=np.float32, delimiter='\t') data = (data - data.mean(axis=0)) / data.std(axis=0) data_iter = d2l.load_array((data[:n, :-1], data[:n, -1]), batch_size, is_train=True) return data_iter, data.shape[1]-1
[docs]def train_ch11(trainer_fn, states, hyperparams, data_iter, feature_dim, num_epochs=2): """Defined in :numref:`sec_minibatches`""" # Initialization w = tf.Variable(tf.random.normal(shape=(feature_dim, 1), mean=0, stddev=0.01),trainable=True) b = tf.Variable(tf.zeros(1), trainable=True) # Train net, loss = lambda X: d2l.linreg(X, w, b), d2l.squared_loss animator = d2l.Animator(xlabel='epoch', ylabel='loss', xlim=[0, num_epochs], ylim=[0.22, 0.35]) n, timer = 0, d2l.Timer() for _ in range(num_epochs): for X, y in data_iter: with tf.GradientTape() as g: l = tf.math.reduce_mean(loss(net(X), y)) dw, db = g.gradient(l, [w, b]) trainer_fn([w, b], [dw, db], states, hyperparams) n += X.shape[0] if n % 200 == 0: timer.stop() p = n/X.shape[0] q = p/ r = (d2l.evaluate_loss(net, data_iter, loss),) animator.add(q, r) timer.start() print(f'loss: {animator.Y[0][-1]:.3f}, {timer.avg():.3f} sec/epoch') return timer.cumsum(), animator.Y[0]
[docs]def train_concise_ch11(trainer_fn, hyperparams, data_iter, num_epochs=2): """Defined in :numref:`sec_minibatches`""" # Initialization net = tf.keras.Sequential() net.add(tf.keras.layers.Dense(1, kernel_initializer=tf.random_normal_initializer(stddev=0.01))) optimizer = trainer_fn(**hyperparams) loss = tf.keras.losses.MeanSquaredError() animator = d2l.Animator(xlabel='epoch', ylabel='loss', xlim=[0, num_epochs], ylim=[0.22, 0.35]) n, timer = 0, d2l.Timer() for _ in range(num_epochs): for X, y in data_iter: with tf.GradientTape() as g: out = net(X) l = loss(y, out) params = net.trainable_variables grads = g.gradient(l, params) optimizer.apply_gradients(zip(grads, params)) n += X.shape[0] if n % 200 == 0: timer.stop() p = n/X.shape[0] q = p/ # `MeanSquaredError` computes squared error without the 1/2 # factor r = (d2l.evaluate_loss(net, data_iter, loss) / 2,) animator.add(q, r) timer.start() print(f'loss: {animator.Y[0][-1]:.3f}, {timer.avg():.3f} sec/epoch')
[docs]class Benchmark: """For measuring running time.""" def __init__(self, description='Done'): """Defined in :numref:`sec_hybridize`""" self.description = description def __enter__(self): self.timer = d2l.Timer() return self def __exit__(self, *args): print(f'{self.description}: {self.timer.stop():.4f} sec')
[docs]def box_corner_to_center(boxes): """Convert from (upper-left, lower-right) to (center, width, height). Defined in :numref:`sec_bbox`""" x1, y1, x2, y2 = boxes[:, 0], boxes[:, 1], boxes[:, 2], boxes[:, 3] cx = (x1 + x2) / 2 cy = (y1 + y2) / 2 w = x2 - x1 h = y2 - y1 boxes = d2l.stack((cx, cy, w, h), axis=-1) return boxes
[docs]def box_center_to_corner(boxes): """Convert from (center, width, height) to (upper-left, lower-right). Defined in :numref:`sec_bbox`""" cx, cy, w, h = boxes[:, 0], boxes[:, 1], boxes[:, 2], boxes[:, 3] x1 = cx - 0.5 * w y1 = cy - 0.5 * h x2 = cx + 0.5 * w y2 = cy + 0.5 * h boxes = d2l.stack((x1, y1, x2, y2), axis=-1) return boxes
[docs]def bbox_to_rect(bbox, color): """Convert bounding box to matplotlib format. Defined in :numref:`sec_bbox`""" # Convert the bounding box (upper-left x, upper-left y, lower-right x, # lower-right y) format to the matplotlib format: ((upper-left x, # upper-left y), width, height) return d2l.plt.Rectangle( xy=(bbox[0], bbox[1]), width=bbox[2]-bbox[0], height=bbox[3]-bbox[1], fill=False, edgecolor=color, linewidth=2)
[docs]def update_D(X, Z, net_D, net_G, loss, optimizer_D): """Update discriminator. Defined in :numref:`sec_basic_gan`""" batch_size = X.shape[0] ones = tf.ones((batch_size,)) # Labels corresponding to real data zeros = tf.zeros((batch_size,)) # Labels corresponding to fake data # Do not need to compute gradient for `net_G`, so it's outside GradientTape fake_X = net_G(Z) with tf.GradientTape() as tape: real_Y = net_D(X) fake_Y = net_D(fake_X) # We multiply the loss by batch_size to match PyTorch's BCEWithLogitsLoss loss_D = (loss(ones, tf.squeeze(real_Y)) + loss( zeros, tf.squeeze(fake_Y))) * batch_size / 2 grads_D = tape.gradient(loss_D, net_D.trainable_variables) optimizer_D.apply_gradients(zip(grads_D, net_D.trainable_variables)) return loss_D
[docs]def update_G(Z, net_D, net_G, loss, optimizer_G): """Update generator. Defined in :numref:`sec_basic_gan`""" batch_size = Z.shape[0] ones = tf.ones((batch_size,)) with tf.GradientTape() as tape: # We could reuse `fake_X` from `update_D` to save computation fake_X = net_G(Z) # Recomputing `fake_Y` is needed since `net_D` is changed fake_Y = net_D(fake_X) # We multiply the loss by batch_size to match PyTorch's BCEWithLogits loss loss_G = loss(ones, tf.squeeze(fake_Y)) * batch_size grads_G = tape.gradient(loss_G, net_G.trainable_variables) optimizer_G.apply_gradients(zip(grads_G, net_G.trainable_variables)) return loss_G
d2l.DATA_HUB['pokemon'] = (d2l.DATA_URL + '', 'c065c0e2593b8b161a2d7873e42418bf6a21106c')# Alias defined in config.ini size = lambda a: tf.size(a).numpy() reshape = tf.reshape ones = tf.ones zeros = tf.zeros meshgrid = tf.meshgrid sin = tf.sin sinh = tf.sinh cos = tf.cos cosh = tf.cosh tanh = tf.tanh linspace = tf.linspace exp = tf.exp normal = tf.random.normal rand = tf.random.uniform matmul = tf.matmul reduce_sum = tf.reduce_sum argmax = tf.argmax tensor = tf.constant arange = tf.range astype = tf.cast int32 = tf.int32 float32 = tf.float32 transpose = tf.transpose concat = tf.concat stack = tf.stack abs = tf.abs eye = tf.eye numpy = lambda x, *args, **kwargs: x.numpy(*args, **kwargs)