forked from EC-KitY/EC-KitY
-
Notifications
You must be signed in to change notification settings - Fork 0
/
novelty_search.py
124 lines (109 loc) · 5.02 KB
/
novelty_search.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
import numpy as np
from overrides import override
from eckity.evaluators.individual_evaluator import IndividualEvaluator
from eckity.evaluators.simple_population_evaluator import SimplePopulationEvaluator
from eckity.creators.ga_creators.simple_vector_creator import GAVectorCreator
from eckity.fitness.simple_fitness import SimpleFitness
from eckity.genetic_encodings.ga.bit_string_vector import BitStringVector
from eckity.algorithms.simple_evolution import SimpleEvolution
from eckity.breeders.simple_breeder import SimpleBreeder
from eckity.subpopulation import Subpopulation
from eckity.genetic_operators.selections.tournament_selection import TournamentSelection
class NoveltySearchCreator(GAVectorCreator):
def __init__(self,
operators_sequence,
length=1,
gene_creator=None,
bounds=(0.0, 1.0),
vector_type=BitStringVector,
fitness_type=SimpleFitness,
events=None,
k=15,
max_archive_size=1000):
super().__init__(length=length, gene_creator=gene_creator,
bounds=bounds, vector_type=vector_type,
fitness_type=fitness_type, events=events)
self.k = k
self.max_archive_size = max_archive_size
self.archive = []
self.operators_sequence = operators_sequence
self.individual_evaluator = NoveltySearchEvaluator(
k=self.k,
max_archive_size=self.max_archive_size
)
@override
def create_individuals(self, n_individuals, higher_is_better):
# create individuals with random vectors
# higher_is_better = True because we are maximizing novelty
individuals = super().create_individuals(n_individuals,
higher_is_better=True)
individuals = self.novelty_search(individuals)
# set original higher_is_better
for ind in individuals:
ind.fitness.higher_is_better = higher_is_better
return individuals
def novelty_search(self, individuals):
evo_novelty = SimpleEvolution(
Subpopulation(creators=None,
individuals=individuals,
population_size=100,
evaluator=self.individual_evaluator,
higher_is_better=True,
elitism_rate=0.0,
operators_sequence=self.operators_sequence,
selection_methods=[
# (selection method, selection probability) tuple
(TournamentSelection(tournament_size=4,
higher_is_better=True), 1)
]),
breeder=SimpleBreeder(),
population_evaluator=SimplePopulationEvaluator(executor_method='submit'),
max_generation=200,
executor='process',
max_workers=10
)
evo_novelty.evolve()
individuals = evo_novelty.population.sub_populations[0].individuals
return individuals
class NoveltySearchEvaluator(IndividualEvaluator):
def __init__(self, k=15, max_archive_size=1000):
self.archive = []
self.k = k
self.max_archive_size = max_archive_size
self.individuals = None
def evaluate(self, individual, environment_individuals):
idx = environment_individuals.index(individual)
vectors = [ind.get_vector() for ind in environment_individuals]
novelty_score = self.novelty_metric_genotypic(idx, vectors)
return novelty_score
def insert_to_archive(self, point, avg_nn):
i = 0
while i < len(self.archive):
if avg_nn < self.archive[i][1]:
break
else:
i += 1
self.archive = self.archive[:i] + [(point, avg_nn)] + self.archive[i:]
def add_to_archive(self, point, avg_nn):
# add point to archive if it is greater than minimum
# archive is maintained sorted by avg_nn: [(point1, avg_nn1), (point2, avg_nn2),... ]
if len(self.archive) == 0:
self.archive.append((point, avg_nn))
elif len(self.archive) < self.max_archive_size:
self.insert(point, avg_nn)
else:
# archive is full
if avg_nn > self.archive[0][1]:
# greater than minimum
del self.archive[0]
self.insert(point, avg_nn)
def novelty_metric_genotypic(self, i, vectors):
vec_i = np.array(vectors[i])
dists = [np.linalg.norm(vec_i - np.array(vectors[j]))
for j in range(len(vectors)) if j != i]
dists = [np.linalg.norm(vec_i - np.array(self.archive[j][0]))
for j in range(len(self.archive))]
dists = sorted(dists)
avg_nn = sum(dists[:self.k])/self.k # average distance to nearest neighbors
self.add_to_archive(vectors[i], avg_nn)
return avg_nn