added torch model integration

Author: dyumanaditya

.github/workflows/python-package-version-test.yml

@@ -28,6 +28,7 @@ jobs:
       run: |
         python -m pip install --upgrade pip
         python -m pip install flake8 pytest
+        pip install torch==2.6.0
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
     - name: Lint with flake8
       run: |

pyreason/pyreason.py

@@ -25,6 +25,15 @@
 import pyreason.scripts.numba_wrapper.numba_types.fact_edge_type as fact_edge
 import pyreason.scripts.numba_wrapper.numba_types.interval_type as interval
 from pyreason.scripts.utils.reorder_clauses import reorder_clauses
+try:
+    import torch
+except ImportError:
+    LogicIntegratedClassifier = None
+    ModelInterfaceOptions = None
+    print('torch is not installed, model integration is disabled')
+else:
+    from pyreason.scripts.learning.classification.classifier import LogicIntegratedClassifier
+    from pyreason.scripts.learning.utils.model_interface import ModelInterfaceOptions
 
 
 # USER VARIABLES

pyreason/scripts/facts/fact.py

@@ -26,3 +26,11 @@ def __init__(self, fact_text: str, name: str = None, start_time: int = 0, end_ti
         self.component = component
         self.bound = bound
         self.type = fact_type
+
+    def __str__(self):
+        s = f'{self.pred}({self.component}) : {self.bound}'
+        if self.static:
+            s += ' | static'
+        else:
+            s += f' | start: {self.start_time} -> end: {self.end_time}'
+        return s

pyreason/scripts/learning/__init__.py

@@ -0,0 +1,89 @@
+from typing import List
+
+import torch.nn
+import torch.nn.functional as F
+
+from pyreason.scripts.facts.fact import Fact
+from pyreason.scripts.learning.utils.model_interface import ModelInterfaceOptions
+
+
+class LogicIntegratedClassifier(torch.nn.Module):
+    """
+    Class to integrate a PyTorch model with PyReason. The output of the model is returned to the
+    user in the form of PyReason facts. The user can then add these facts to the logic program and reason using them.
+    """
+    def __init__(self, model, class_names: List[str], model_name: str = 'classifier', interface_modes: ModelInterfaceOptions = None):
+        """
+        :param model:
+        :param class_names:
+        """
+        super(LogicIntegratedClassifier, self).__init__()
+        self.model = model
+        self.class_names = class_names
+        self.model_name = model_name
+        self.interface_modes = interface_modes
+
+    def get_class_facts(self, t1: int, t2: int) -> List[Fact]:
+        """
+        Return PyReason facts to create nodes for each class. Each class node will have bounds `[1,1]` with the
+         predicate corresponding to the model name.
+        :param t1: Start time for the facts
+        :param t2: End time for the facts
+        :return: List of PyReason facts
+        """
+        facts = []
+        for c in self.class_names:
+            fact = Fact(f'{self.model_name}({c})', name=f'{self.model_name}-{c}-fact', start_time=t1, end_time=t2)
+            facts.append(fact)
+        return facts
+
+    def forward(self, x, t1: int, t2: int):
+        """
+        Forward pass of the model
+        :param x: Input tensor
+        :param t1: Start time for the facts
+        :param t2: End time for the facts
+        :return: Output tensor
+        """
+        output = self.model(x)
+
+        # Convert logits to probabilities assuming a multi-class classification.
+        probabilities = F.softmax(output, dim=1).squeeze()
+        opts = self.integration_options
+
+        # Prepare threshold tensor.
+        threshold = torch.tensor(opts.threshold, dtype=probabilities.dtype, device=probabilities.device)
+        condition = probabilities > threshold
+
+        if opts.snap_value is not None:
+            snap_value = torch.tensor(opts.snap_value, dtype=probabilities.dtype, device=probabilities.device)
+            # For values that pass the threshold:
+            lower_val = snap_value if opts.set_lower_bound else torch.tensor(0.0, dtype=probabilities.dtype,
+                                                                             device=probabilities.device)
+            upper_val = snap_value if opts.set_upper_bound else torch.tensor(1.0, dtype=probabilities.dtype,
+                                                                             device=probabilities.device)
+        else:
+            # If no snap_value is provided, keep original probabilities for those passing threshold.
+            lower_val = probabilities
+            upper_val = probabilities
+
+        # For probabilities that pass the threshold, apply the above; else, bounds are fixed to [0,1].
+        lower_bounds = torch.where(condition, lower_val, torch.zeros_like(probabilities))
+        upper_bounds = torch.where(condition, upper_val, torch.ones_like(probabilities))
+
+        # Convert bounds to Python floats for fact creation.
+        bounds_list = []
+        for i in range(len(self.class_names)):
+            lower = lower_bounds[i].item()
+            upper = upper_bounds[i].item()
+            bounds_list.append([lower, upper])
+
+        # Define time bounds for the facts.
+        facts = []
+        for class_name, bounds in zip(self.class_names, bounds_list):
+            lower, upper = bounds
+            fact_str = f'{self.model_name}({class_name}) : [{lower:.3f}, {upper:.3f}])'
+            fact = Fact(fact_str, name=f'{self.model_name}-{class_name}-fact', start_time=t1, end_time=t2)
+            facts.append(fact)
+        return output, facts
+

pyreason/scripts/learning/classification/__init__.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+from typing import Optional
+
+
+@dataclass
+class ModelInterfaceOptions:
+    threshold: float = 0.5
+    set_lower_bound: bool = True
+    set_upper_bound: bool = True
+    snap_value: Optional[float] = 1.0

pyreason/scripts/learning/classification/classifier.py

@@ -0,0 +1,53 @@
+# Test cases for classifier integration with pyreason
+import pyreason as pr
+import torch
+import torch.nn as nn
+
+
+def test_classifier_integration():
+    # Reset PyReason
+    pr.reset()
+    pr.reset_rules()
+    pr.reset_settings()
+
+    # Create a dummy PyTorch model: input size 10, output 3 classes.
+    model = nn.Linear(10, 3)
+
+    # Define class names for the output classes.
+    class_names = ["cat", "dog", "rabbit"]
+
+    # Create integration options.
+    # Only probabilities exceeding 0.6 will be considered.
+    # For those, if set_lower_bound is True, lower bound becomes 0.95; if set_upper_bound is False, upper bound is forced to 1.
+    dummy_options = pr.ModelInterfaceOptions(
+        threshold=0.6,
+        set_lower_bound=True,
+        set_upper_bound=False,
+        snap_value=0.95
+    )
+
+    # Create an instance of LogicIntegratedClassifier.
+    logic_classifier = pr.LogicIntegratedClassifier(model, class_names, model_name="classifier",
+                                                    interface_modes=dummy_options)
+
+    # Create a dummy input tensor with 10 features.
+    input_tensor = torch.rand(1, 10)
+
+    # Set time bounds for the facts.
+    t1 = 0
+    t2 = 0
+
+    # Run the forward pass to get the model output and the corresponding PyReason facts.
+    output, facts = logic_classifier(input_tensor, t1, t2)
+
+    # Assert that the output is a tensor.
+    assert isinstance(output, torch.Tensor), "The model output should be a torch.Tensor"
+    # Assert that we have one fact per class.
+    assert len(facts) == len(class_names), "Expected one fact per class"
+
+    # Print results for visual inspection.
+    print("Model output (logits):")
+    print(output)
+    print("\nGenerated PyReason Facts:")
+    for fact in facts:
+        print(fact)