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.
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 .
