Strategy Pattern with a Perception Layer

The strategy pattern agent performs well, but it's not perfect in avoiding thermal runaway. One good way to address that is to add a perception layer.

are special skills that process and interpret sensor data before passing it to the rest of the agent. To improve the strategy pattern's performance on temperature control, you can add a perception layer that uses machine learning to predict thermal runaway. This ML model is trained to interpret the sensor data and check for conditions that might indicate an elevated risk of thermal runaway, and then pass that information to the selector along with the rest of the sensor data.

Think of the perception layer as an additional set of senses that helps the agent predict when something might go wrong, like a teacher monitoring the class for early signs of trouble.

Let's get started configuring this agent!

1. Add the Perceptor Skill to Your Project

This agent has a perceptor skill called thermal_runaway_predictor. To publish it to your project you will need to open up your favorite code editor and terminal. In your terminal, navigate to the perceptors folder and use this command with the Composabl CLI.

composabl perceptor publish thermal_runaway_predictor

Return to the agent builder studio and refresh the page. The skill will appear in the skills menu on the left side of your page.

Explore the Code Files

All skills, perceptors, and selectors have a minimum of two files in them. A Python file contains the code that the agent will use, and a config file. Perceptors have some more files to load in ML models and other python packages.

File Structure

Thermal Runaway Perceptor

pyproject.toml

See the code

[project]
name = "Thermal Runaway Predictor - ML 1.2.2"
version = "0.1.0"
description = "ML thermal runaway predictor"
authors = [{ name = "John Doe", email = "[e-mail]" }]
dependencies = [
    "composabl-core",
    "scikit-learn==1.2.2"
]

[composabl]
type = "perceptor"
entrypoint = "thermal_runaway_predictor.perceptor:ThermalRunawayPredict"

# Include additional data files
[tool.setuptools.packages.find]
where = ["thermal_runaway_predictor"]

[tool.setuptools.package-data]
"*" = ["*.json", "*.pkl"]

thermal_runaway_predictor.py

See the code

from composabl_core import PerceptorImpl

#######
import os
import pickle

# Determine the directory where the current script is located
path = os.path.dirname(os.path.realpath(__file__))

class ThermalRunawayPredict(PerceptorImpl):
    def __init__(self, *args, **kwargs):
        """
        Initialize the ThermalRunawayPredict perceptor with default values and load the machine learning model.
        
        Args:
            *args: Variable length argument list.
            **kwargs: Arbitrary keyword arguments.
        """
        # Initialize the prediction output variable
        self.y = 0
        
        # Initialize a flag to indicate thermal runaway status
        self.thermal_run = 0
        
        # Load the pre-trained machine learning model from a pickle file
        # The model is expected to be located in the 'ml_models' directory relative to the script's path
        model_path = os.path.join(path, "ml_models", "ml_predict_temperature_122.pkl")
        try:
            with open(model_path, 'rb') as model_file:
                self.ml_model = pickle.load(model_file)
        except FileNotFoundError:
            print(f"Machine learning model not found at {model_path}. Please ensure the model file exists.")
            self.ml_model = None
        except Exception as e:
            print(f"An error occurred while loading the ML model: {e}")
            self.ml_model = None
        
        # Initialize a list to store historical ML predictions if needed
        self.ML_list = []
        
        # Initialize the last recorded 'Tc' value to compute its change (ΔTc)
        self.last_Tc = 0

    async def compute(self, obs_spec, obs):
        """
        Compute the thermal runaway prediction based on current sensor observations.
        
        Args:
            obs_spec: Observation specification (not used in this implementation).
            obs: Current sensor observations. Can be a list or a dictionary.
        
        Returns:
            dict: A dictionary containing the thermal runaway prediction.
                  Example: {"thermal_runaway_predict": 1}
        """
        # Ensure that 'obs' is a dictionary. If not, convert it using predefined sensor keys.
        if not isinstance(obs, dict):
            # Define the expected sensor keys
            obs_keys = ['T', 'Tc', 'Ca', 'Cref', 'Tref', 'Conc_Error', 'Eps_Yield', 'Cb_Prod']
            # Convert the list to a dictionary by zipping it with the sensor keys
            obs = dict(zip(obs_keys, obs))
            print("Converted 'obs' to dictionary format using predefined sensor keys.")
        
        # Calculate the change in 'Tc' (ΔTc) since the last observation
        if self.last_Tc == 0:
            # If this is the first observation, assume an initial ΔTc of 5
            self.ΔTc = 5
        else:
            # Compute ΔTc as the difference between current 'Tc' and the last recorded 'Tc'
            try:
                current_Tc = float(obs['Tc'])
                self.ΔTc = current_Tc - self.last_Tc
            except (KeyError, ValueError, TypeError) as e:
                # Handle cases where 'Tc' is missing or cannot be converted to float
                print(f"Error accessing or converting 'Tc': {e}")
                self.ΔTc = 0  # Default to 0 if there's an error
        
        # Initialize the prediction output
        y = 0
        
        # Check if the current temperature 'T' exceeds or equals 340
        try:
            current_T = float(obs['T'])
        except (KeyError, ValueError, TypeError) as e:
            print(f"Error accessing or converting 'T': {e}")
            current_T = 0  # Default to 0 if there's an error
        
        if current_T >= 340:
            # Prepare the feature vector for the ML model
            try:
                Ca = float(obs['Ca'])
                Cref = float(obs['Cref'])
            except (KeyError, ValueError, TypeError) as e:
                print(f"Error accessing or converting 'Ca' or 'Cref': {e}")
                Ca = 0
                Cref = 0
            
            # Feature vector: [Ca, T, Tc, ΔTc]
            X = [[Ca, current_T, self.ΔTc]]
            
            # If the ML model was loaded successfully, make a prediction
            if self.ml_model:
                try:
                    # Predict the probability of thermal runaway
                    y_proba = self.ml_model.predict_proba(X)
                    
                    # Get the predicted class label (e.g., 0 or 1)
                    y = self.ml_model.predict(X)[0]
                    
                    # Optionally, use the probability to adjust prediction confidence
                    # For example, set y=1 only if the probability of class 1 is >= 0.3
                    if y_proba[0][1] >= 0.3:
                        y = 1
                    else:
                        y = 0
                except Exception as e:
                    print(f"Error during ML model prediction: {e}")
                    y = 0
            else:
                print("ML model is not loaded. Cannot make predictions.")
                y = 0
        
        # Update the last recorded 'Tc' with the current value for the next computation
        try:
            self.last_Tc = float(obs['Tc'])
        except (KeyError, ValueError, TypeError) as e:
            print(f"Error accessing or converting 'Tc' for updating last_Tc: {e}")
            self.last_Tc = self.last_Tc  # Keep the previous value if there's an error
        
        # Optionally, store the prediction in ML_list for historical tracking
        self.ML_list.append(y)
        
        # Update the prediction output variable
        self.y = y
        
        # Return the prediction as a dictionary
        return {"thermal_runaway_predict": y}

    def filtered_sensor_space(self, obs):
        """
        Define which sensors are relevant for this perceptor.
        
        Args:
            obs: Current sensor observations (not used in this implementation).
        
        Returns:
            list: Names of the sensors to be used.
        """
        # Specify the sensors that this perceptor will use
        return ['T', 'Tc', 'Ca', 'Cref', 'Tref', 'Conc_Error', 'Eps_Yield', 'Cb_Prod']

2. Copy the Strategy Pattern Agent, name it Strategy Pattern with Perceptor, and add the Perceptor Skill to your Strategy Pattern Agent

Drag the Perceptor thermal_runaway_predictor that you can now see on the left-hand side of your project onto the perception layer.

3. Run Your Training Session

Now, we are ready to train your agent and see the results. We suggest you run 50 training cycles.

4. View Results

When the training has been completed, you can view your results in the training sessions tab in the UI. This will show you information on how well the agent is learning.

The agent training results will be a little bit different from the strategy pattern alone. That's because the thermal runaway predictor is making a difference in how the agent performs.

Analyzing the Strategy Pattern Agent’s Performance with Perception

Conversion rate: 92% Thermal runaway risk: Very low

We tested this fully trained agent and plotted the results.

Adding perception improves agent temperature control performance.

The red lines on the graph show where the perceptors helped the agent make adjustments to avoid thermal runaway. This agent gets the same yield as the strategy pattern agent, but the improved temperature control has reduced the thermal runaway incidents from low to 0.