<jupyter_start><jupyter_text>The reach of Logic Tensor Networks**Premise:** we suggest going through the tutorials (`/tutorial` folder) before going through these examples.The objective of this series of examples is to show how the language of Real Logic can be used to specify a number oftasks that involve learning from data and reasoning. Examples of such tasks are classification, regression, clustering,link prediction, etc.The solution of a problem specified in Real Logic is obtained by interpreting such a specification in Logic Tensor Networks.The LTN library implements Real Logic in PyTorch. Every logical operator is grounded using PyTorch primitives such thatLTN implements directly a PyTorch graph. Due to PyTorch built-in optimization, LTN is relatively efficient whileproviding the expressive power of first-order logic.The first tutorial that we will see is based on one of the most simple and intuitive machine learning tasks, the binaryclassification. Binary classificationThe simplest machine learning task is binary classification. Suppose that one wants to learn a binary classifier $A$ fora set of points in $[0, 1]^2$. Suppose that a set of positive and negative training examples is given.LTN uses the following language and grounding:**Domains:**- $points$ (denoting the examples);**Variables:**- $x_+$ for the positive examples;- $x_-$ for the negative examples;- $x$ for all examples;- $D(x) = D(x_+) = D(x_-) = points$, where $D(.)$ is a function which returns the domain of a non-logical symbol given ininput.**Predicates:**- $A(x)$ for the trainable classifier;- $D_{in}(A) = points$, where $D_{in}(.)$ is a function which returns the domain of the input of a function or predicategiven in input.**Axioms:**- $\forall x_+ A(x_+)$: the predicate must be true for positive examples;- $\forall x_- \lnot A(x_-)$: the predicate must be false for negative examples.**Grounding:**- $\mathcal{G}(points)=[0,1]^{2}$;- $\mathcal{G}(x) \in[0,1]^{m \times 2}$: $\mathcal{G}(x)$ is a sequence of $m$ points, that is, $m$ examples;- $\mathcal{G}\left(x_{+}\right)=\langle d \in \mathcal{G}(x) \mid\|d-(0.5,0.5)\|<0.09\rangle$: $\mathcal{G}(x_+)$ are,by definition in this example, the training examples with Euclidean distance to the center $(0.5, 0.5)$ smaller thanthe threshold of 0.09;- $\mathcal{G}\left(x_{+}\right)=\langle d \in \mathcal{G}(x) \mid\|d-(0.5,0.5)\|>=0.09\rangle$: $\mathcal{G}(x_+)$ are,by definition in this example, the training examples with Euclidean distance to the center $(0.5, 0.5)$ larger or equalto the threshold of 0.09;- $\mathcal{G}(A \mid \theta): x \mapsto \sigma\left(\operatorname{MLP}_{\theta}(x)\right)$, where $MLP$is a Multi-Layer Perceptron with a single output neuron, whose parameters $\theta$ have to be learned. DatasetNow, let's define our toy dataset with points in $[0, 1]^2$. We randomly generate 100 points, and we give them some labelsaccording to the *grounding* defined above. So, points near the center will be classified as positives, while points distantfrom the center will be classified as negatives.<jupyter_code>import torch
import matplotlib.pyplot as plt
nr_samples = 100
dataset = torch.rand((nr_samples, 2))
labels_dataset = torch.sum(torch.square(dataset - torch.tensor([.5, .5])), dim=1) < .09<jupyter_output><empty_output><jupyter_text>Let's plot our dataset and see that points around the center have a positive label, while points far from the center havenegative labels.<jupyter_code>plt.figure(figsize=(4,4))
plt.scatter(dataset[labels_dataset][:,0],dataset[labels_dataset][:,1],label='A')
plt.scatter(dataset[torch.logical_not(labels_dataset)][:,0],dataset[torch.logical_not(labels_dataset)][:,1],label='~A')
plt.title("Ground truth")
plt.legend()
plt.show()<jupyter_output><empty_output><jupyter_text>LTN settingIn order to define our knowledge base (axioms), we need to define predicate $A$, the connectives, quantifiers, and the`SatAgg` operator.For the connective and quantifiers, we use the stable product configuration (seen in the tutorials). For predicate $A$, we use a simple $MLP$implemented as a `torch.nn.Module`.`SatAgg` is the SAT aggregator, an operator which aggregates the truth values of all the closedformulas included in the knowledge base. The `SatAgg`, by default, is implemented using the `pMeanError` aggregator.<jupyter_code>import ltn
# we define predicate A
class ModelA(torch.nn.Module):
    def __init__(self):
        super(ModelA, self).__init__()
        self.sigmoid = torch.nn.Sigmoid()
        self.layer1 = torch.nn.Linear(2, 16)
        self.layer2 = torch.nn.Linear(16, 16)
        self.layer3 = torch.nn.Linear(16, 1)
        self.elu = torch.nn.ELU()

    def forward(self, x):
        x = self.elu(self.layer1(x))
        x = self.elu(self.layer2(x))
        return self.sigmoid(self.layer3(x))


A = ltn.Predicate(ModelA())

# we define the connectives, quantifiers, and the SatAgg
Not = ltn.Connective(ltn.fuzzy_ops.NotStandard())
Forall = ltn.Quantifier(ltn.fuzzy_ops.AggregPMeanError(p=2), quantifier="f")
SatAgg = ltn.fuzzy_ops.SatAgg()<jupyter_output><empty_output><jupyter_text>UtilsNow, we need to define some utility classes and functions.We define a standard PyTorch data loader, which takes as input the dataset and returns a generator of batches of data.In particular, we need a data loader instance for training data and one for testing data. 50 examples are selectedfor training, 50 examples are selected for testing.Then, we define functions to evaluate the model performances. The model is evaluated on the test set using the following metrics:- the satisfaction level of the knowledge base: measure the ability of LTN to satisfy the knowledge;- the classification accuracy: measure the quality of the predictions.<jupyter_code>from sklearn.metrics import accuracy_score
import numpy as np

# this is a standard PyTorch DataLoader to load the dataset for the training and testing of the model
class DataLoader(object):
    def __init__(self,
                 data,
                 labels,
                 batch_size=1,
                 shuffle=True):
        self.data = data
        self.labels = labels
        self.batch_size = batch_size
        self.shuffle = shuffle

    def __len__(self):
        return int(np.ceil(self.data.shape[0] / self.batch_size))

    def __iter__(self):
        n = self.data.shape[0]
        idxlist = list(range(n))
        if self.shuffle:
            np.random.shuffle(idxlist)

        for _, start_idx in enumerate(range(0, n, self.batch_size)):
            end_idx = min(start_idx + self.batch_size, n)
            data = self.data[idxlist[start_idx:end_idx]]
            labels = self.labels[idxlist[start_idx:end_idx]]

            yield data, labels


# define metrics for evaluation of the model

# it computes the overall satisfaction level on the knowledge base using the given data loader (train or test)
def compute_sat_level(loader):
    mean_sat = 0
    for data, labels in loader:
        x_A = ltn.Variable("x_A", data[torch.nonzero(labels)])  # positive examples
        x_not_A = ltn.Variable("x_not_A",
                               data[torch.nonzero(torch.logical_not(labels))])  # negative examples
        mean_sat += SatAgg(
            Forall(x_A, A(x_A)),
            Forall(x_not_A, Not(A(x_not_A)))
        )
    mean_sat /= len(loader)
    return mean_sat

# it computes the overall accuracy of the predictions of the trained model using the given data loader
# (train or test)
def compute_accuracy(loader):
    mean_accuracy = 0.0
    for data, labels in loader:
        predictions = A.model(data).detach().numpy()
        predictions = np.where(predictions > 0.5, 1., 0.).flatten()
        mean_accuracy += accuracy_score(labels, predictions)

    return mean_accuracy / len(loader)

# create train and test loader, 50 points each
# batch size is 64, meaning there is only one batch for epoch
train_loader = DataLoader(dataset[:50], labels_dataset[:50], 64, True)
test_loader = DataLoader(dataset[50:], labels_dataset[50:], 64, False)<jupyter_output><empty_output><jupyter_text>LearningLet us define $D$ the data set of all examples. The objective function with $\mathcal{K}=\left\{\forall x_{+} A\left(x_{+}\right), \forall x_{-} \neg A\left(x_{-}\right)\right\}$is given by $\operatorname{SatAgg}_{\phi \in \mathcal{K}} \mathcal{G}_{\boldsymbol{\theta}, x \leftarrow \boldsymbol{D}}(\phi)$.In practice, the optimizer uses the following loss function:$\boldsymbol{L}=\left(1-\underset{\phi \in \mathcal{K}}{\operatorname{SatAgg}} \mathcal{G}_{\boldsymbol{\theta}, x \leftarrow \boldsymbol{B}}(\phi)\right)$where $B$ is a mini batch sampled from $D$.The notation $\mathcal{G}_{x \leftarrow \boldsymbol{D}}(\phi(x))$ means that the variable $x$ is grounded with the data $D$ whengrounding $\phi(x)$.By minimizing the loss, we will try to maximize the `SatAgg` operator applied to the knowledge base. Maximizing the `SatAgg` operatormeans maximizing the satisfaction level of each formula included in the knowledge base.Notice that the objective and the loss function depend on the following hyper-parameters:- the choice of fuzzy logic operator semantics used to approximate each connective and quantifier;- the choice of hyper-parameters underlying the operators, such as the value of the exponent $p$ in any generalized mean;- the choice of formula aggregator function `SatAgg`.In the following, we learn our LTN in the binary classification task using the satisfaction of the knowledge base asan objective. In other words, we want to learn the parameters $\theta$ of unary predicate $A$ in such a way the twoaxioms in the knowledge base are maximally satisfied. We train our model for 1000 epochs and use the `Adam` optimizer.The following figure shows the LTN computational graph for this specific task.![Computational graph](/examples/images/binary-classification.png)<jupyter_code>optimizer = torch.optim.Adam(A.parameters(), lr=0.001)

# training of the predicate A using a loss containing the satisfaction level of the knowledge base
# the objective it to maximize the satisfaction level of the knowledge base
for epoch in range(1000):
    train_loss = 0.0
    for batch_idx, (data, labels) in enumerate(train_loader):
        optimizer.zero_grad()
        # we ground the variables with current batch data
        x_A = ltn.Variable("x_A", data[torch.nonzero(labels)]) # positive examples
        x_not_A = ltn.Variable("x_not_A", data[torch.nonzero(torch.logical_not(labels))]) # negative examples
        sat_agg = SatAgg(
            Forall(x_A, A(x_A)),
            Forall(x_not_A, Not(A(x_not_A)))
        )
        loss = 1. - sat_agg
        loss.backward()
        optimizer.step()
        train_loss += loss.item()
    train_loss = train_loss / len(train_loader)

    # we print metrics every 20 epochs of training
    if epoch % 20 == 0:
        print(" epoch %d | loss %.4f | Train Sat %.3f | Test Sat %.3f | Train Acc %.3f | Test Acc %.3f"
              %(epoch, train_loss, compute_sat_level(train_loader), compute_sat_level(test_loader),
                    compute_accuracy(train_loader), compute_accuracy(test_loader)))<jupyter_output>epoch 0 | loss 0.5040 | Train Sat 0.497 | Test Sat 0.496 | Train Acc 0.680 | Test Acc 0.760
 epoch 20 | loss 0.4983 | Train Sat 0.502 | Test Sat 0.502 | Train Acc 0.340 | Test Acc 0.280
 epoch 40 | loss 0.4954 | Train Sat 0.505 | Test Sat 0.505 | Train Acc 0.640 | Test Acc 0.580
 epoch 60 | loss 0.4919 | Train Sat 0.508 | Test Sat 0.509 | Train Acc 0.700 | Test Acc 0.660
 epoch 80 | loss 0.4875 | Train Sat 0.513 | Test Sat 0.514 | Train Acc 0.680 | Test Acc 0.660
 epoch 100 | loss 0.4821 | Train Sat 0.518 | Test Sat 0.520 | Train Acc 0.680 | Test Acc 0.660
 epoch 120 | loss 0.4760 | Train Sat 0.524 | Test Sat 0.526 | Train Acc 0.680 | Test Acc 0.700
 epoch 140 | loss 0.4699 | Train Sat 0.530 | Test Sat 0.532 | Train Acc 0.680 | Test Acc 0.700
 epoch 160 | loss 0.4647 | Train Sat 0.536 | Test Sat 0.537 | Train Acc 0.680 | Test Acc 0.700
 epoch 180 | loss 0.4606 | Train Sat 0.540 | Test Sat 0.541 | Train Acc 0.660 | Test Acc 0.700
 epoch 200 | loss 0.4575 | Train Sat 0.543 | Test Sat 0.[...]<jupyter_text>Notice that variables $x_+$ and $x_-$ are grounded batch by batch with new data arriving from the data loader. This is exactly whatwe mean with $\mathcal{G}_{x \leftarrow \boldsymbol{B}}(\phi(x))$, where $B$ is a mini-batch sampled by the data loader.Notice also that `SatAgg` takes as input the two axioms and returns one truth value which can be interpreted as the satisfactionlevel of the knowledge base.Note that after 800 epochs the test accuracy is around 1. This shows the power of LTN in learningthe binary classification task only using the satisfaction of a knowledge base as an objective.In the following, we query predicate $A$ on training data and test data. The plot shows the extent of generalization.<jupyter_code>nr_samples_train = 50

fig = plt.figure(figsize=(9, 11))

plt.subplots_adjust(wspace=0.2,hspace=0.3)
ax = plt.subplot2grid((3,8),(0,2),colspan=4)
ax.set_title("ground truth")
ax.scatter(dataset[labels_dataset][:,0],dataset[labels_dataset][:,1],label='A')
ax.scatter(dataset[torch.logical_not(labels_dataset)][:,0],dataset[torch.logical_not(labels_dataset)][:,1],label='~A')
ax.legend()

# Training data
x = ltn.Variable("x",dataset[:nr_samples_train])
fig.add_subplot(3, 2, 3)
result=A(x)
plt.title("A(x) - training data")
plt.scatter(dataset[:nr_samples_train,0],dataset[:nr_samples_train,1],c=result.value.detach().numpy().squeeze())
plt.colorbar()

fig.add_subplot(3, 2, 4)
result=Not(A(x))
plt.title("~A(x) - training data")
plt.scatter(dataset[:nr_samples_train,0],dataset[:nr_samples_train,1],c=result.value.detach().numpy().squeeze())
plt.colorbar()

# Test data
x = ltn.Variable("x",dataset[nr_samples_train:])
fig.add_subplot(3, 2, 5)

result=A(x)
plt.title("A(x) - test data")
plt.scatter(dataset[nr_samples_train:,0],dataset[nr_samples_train:,1],c=result.value.detach().numpy().squeeze())
plt.colorbar()

fig.add_subplot(3, 2, 6)
result=Not(A(x))
plt.title("~A(x) - test data")
plt.scatter(dataset[nr_samples_train:,0],dataset[nr_samples_train:,1],c=result.value.detach().numpy().squeeze())
plt.colorbar()
plt.savefig("ex_binary_testing.pdf")
plt.show()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11450914676721129, 'fraction_numerical': 0.02946515849741656, 'mean_word_length': 4.2350146198830405, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 26, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '337490', 'license': 'MIT', 'n_tokens_mistral': 4687, 'n_tokens_neox': 4265, 'n_words': 1551, 'path': 'examples/1-binary_classification.ipynb', 'repo_name': 'bmxitalia/LTNtorch', 'size': '14322'}