LogoLogo
  • Welcome to Composabl
  • Get Started
  • Reference
    • CLI Reference
    • SDK Reference
    • Glossary
    • Sample Use Cases
  • Tutorials
    • Industrial Mixer
      • Get Started
      • Deep Reinforcement Learning
      • Strategy Pattern
      • Strategy Pattern with a Perception Layer
      • Plan-Execute Pattern
  • Establish a Simulation Environment
    • Simulation Overview
    • Connect a Simulator to Composabl
    • Composabl Simulation API
  • Build Multi-Agent Systems
    • Anatomy of a Multi-Agent System
    • Create a Use Case
    • Set Goals, Constraints, and Success Criteria
    • Create Skill Agents
      • Create Skill Agents
      • Create Skill Agents with Rewards Using the SDK
      • Configure Programmed Algorithms as Skill Agents
      • Configure API Connections to Third-Party Software as Skill Agents
    • Orchestrate Skill Agents
    • Configure Scenarios
    • Add a Perception Layer
      • Create a New Perceptor
      • Configure an ML Model as a Perceptor
      • Configure an LLM Model as a Perceptor
    • Publish Skill Agent Components to the UI
  • Train Agents
    • Submit a Training Job through the UI
    • Analyze Agent System Behavior
      • View Training Session Information
      • Analyze Data in Detail with the Historian
  • Evaluate Performance
    • Set KPI and ROI
    • Analyze Data
  • Deploy Agents
    • Access a Trained Agent System
    • Deploy an Agent System in a Container
    • Deploy an Agent System as an API
    • Connect Runtime Container to Your Operation
    • Connecting to Agent System Runtime and Plotting Results of Agent System Operations
  • clusters
    • Creating a Cluster
      • Manual
      • Automated
      • Azure
    • Connecting a Cluster
  • Troubleshooting
    • Resolving Certificate Issues for Installing the Composabl SDK on WSL
Powered by GitBook
On this page
  • 1. Add the Perceptor Skill Agent to Your Use Case
  • Explore the Code Files
  • Thermal Runaway Perceptor
  • 2. Copy the Strategy Pattern Agent System, name it Strategy Pattern with Perceptor, and add the Perceptor Skill Agent to your Strategy Pattern Agent System
  • 3. Run Your Training Session
  • 4. View Results
  • Analyzing the Strategy Pattern Agent System’s Performance with Perception
Export as PDF
  1. Tutorials
  2. Industrial Mixer

Strategy Pattern with a Perception Layer

PreviousStrategy PatternNextPlan-Execute Pattern

Last updated 1 month ago

The 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 skill agents that process and interpret sensor data before passing it to the rest of the agent system. To improve the strategy pattern's performance on temperature control, you can add a perception layer that uses machine learning to predict thermal runaway.

While skill agents can be built within Composabl's no-code studio or created externally and then imported, perceptors are always built outside of Composabl and then published to the platform to use in agents.

In this case, the perceptor is a pre-built ML model saved as a pickle file. 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.

In this tutorial, you'll use Composabl's to to publish the perceptor to your use case so that you can add it to your agent configuration.

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

Let's get started configuring this agent system!

1. Add the Perceptor Skill Agent to Your Use Case

composabl perceptor publish thermal_runaway_predictor

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

Explore the Code Files

All skill agents, perceptors, and selectors have a minimum of two files in them. A Python file contains the code that the agent system 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 = "john.doe@composabl.com" }]
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 System, name it Strategy Pattern with Perceptor, and add the Perceptor Skill Agent to your Strategy Pattern Agent System

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

3. Run Your Training Session

We are ready to train your agent system and see the results. Select the cluster you want to use and the number of training cycles. We suggest you run 150 training cycles. You will see the skill agents training one at a time, and you assign the number of cycles you want each skill agent to use. It will automatically assign an equal number of training sessions for each skill agent, but in some agent system designs, some skill agents might require more training than others.

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 system is learning.

The agent system 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 system performs.

Analyzing the Strategy Pattern Agent System’s Performance with Perception

Conversion rate: 92% Thermal runaway risk: Very low

We tested this fully trained agent system and plotted the results.

Adding perception improves agent system temperature control performance.

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

This agent system has a perceptor skill agent called thermal_runaway_predictor. To publish it to your use case, 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
strategy pattern agent system
data science workflow