Walkthrough Tutorial: the Canonical Naming Game
This tutorial presents a PyFCG-powered implementation of the canonical naming game experiment, in which a population of agents converges on a naming convention used to refer to objects in their environment.
[ ]:
# Run this cell if you have not yet installed these packages
! pip install pyfcg
! pip install numpy
! pip install matplotlib
[1]:
import pyfcg as fcg
import random
import numpy as np
fcg.init()
Representating naming game agents
A first step in setting up a language game experiment concerns the creation of a population of agents. We define our agents as instances of a new class NGAgent that subclasses from PyFCG’s fcg.Agent class. The agents are thereby initialised with an empty grammar and inherit a collection of methods for interacting with instances of the fcg.Grammar and fcg.Construction classes.
[2]:
# The NGAgent subclasses from fcg.Agent
class NGAgent(fcg.Agent):
"""
The NGAgent subclasses from fcg.Agent to inherit its linguistic capabilities. It defines
additional functionality to participate in a naming game.
"""
def __init__(self):
"""Defines a number of slots for this experiment"""
self.discourse_role = None
self.scene = None
self.topic = None
self.applied_cxn = None
self.competitor_cxns = None
self.utterance = None
self.communicated_successfully = False
# we call the init method of the fcg.Agent superclass
# to initialise the agent's grammar
super().__init__()
def clear(self):
"""Clear all data. """
self.discourse_role = None
self.scene = None
self.topic = None
self.applied_cxn = None
self.competitor_cxns = None
self.utterance = None
self.communicated_successfully = False
def comprehend(self,utterance):
""" Comprehend utterance, collect meaning, applied_cxn and competitors. """
meanings, applied_cxn_names_per_meaning = self.comprehend_all(utterance)
if meanings == [None]:
return None
else:
self.applied_cxn = self.find_cxn_by_name(applied_cxn_names_per_meaning[0][0])
self.competitor_cxns = []
for cxn_names in applied_cxn_names_per_meaning[1:]:
self.competitor_cxns.append(self.find_cxn_by_name(cxn_names[0]))
return meanings[0]
def formulate(self, meaning):
"""Formulate an utterance given a meaning. Use the highest-scored cxn and collect the competing constructions on the go. """
utterances, applied_cxn_names_per_utterance = super().formulate_all(meaning)
if utterances == [None]:
return None
else:
self.utterance = utterances[0][0]
self.applied_cxn = self.find_cxn_by_name(applied_cxn_names_per_utterance[0][0])
self.competitor_cxns = []
for cxn_names in applied_cxn_names_per_utterance[1:]:
self.competitor_cxns.append(self.find_cxn_by_name(cxn_names[0]))
return self.utterance
def learn(self, form, meaning):
""" Create a new cxn given a form and a meaning. """
new_cxn = fcg.Construction(name=form+'-cxn',
conditional_pole= [["?name-unit",
{"#meaning": [(meaning, "?t")]},
{"#form": [("sequence", '\"' + form + '\"', "?left", "?right")]}]],
attributes= {"object": meaning, "name": form, "score": 0.5})
self.add_cxn(new_cxn)
return new_cxn
def reward(self):
""" Reward through lateral inhibition. """
inc_delta = CONFIGURATION['cxn_positive_reward']
dec_delta = CONFIGURATION['cxn_negative_reward']
if self.communicated_successfully:
# If success, reward the applied cxn and punish the competitors
self.applied_cxn.increase_score(delta=inc_delta)
for competitor in self.competitor_cxns:
competitor.decrease_score(delta=dec_delta)
# Delete cxns that reach a 0 score
if competitor.get_score() <= 0.0:
self.delete_cxn(competitor)
else:
# If failure, speaker punishes applied cxn
if self.discourse_role == 'speaker':
self.applied_cxn.decrease_score(delta=dec_delta)
# Delete cxns that reach a 0 score
if self.applied_cxn.get_score() <= 0.0:
self.delete_cxn(self.applied_cxn)
NGAgent()
[2]:
<Agent: agent (id: agent-23) ~ 0 constructions>
The comprehend, formulate and reward methods, as implemented for the NGAgent in the cell above, illustrate how high-level PyFCG functionality facilitates the implementation of language game experiments. Not only do comprehend and formulate return the highest-scored solution, they also yield all competing solutions as a second return value. Successfully used constructions can then be rewarded positively through calls to their increase_score method and their competitors
can be rewarded negatively through calls to their decrease_score method. Constructions that reach a score of 0 can be deleted from an agent’s grammar using the delete_cxn method.
Representing a naming game experiment
We also define a new experiment classNGExperiment. Upon initialisation, a population is created as a set of NGAgent instances, and a world is created as a set of abstract objects. Two methods are also associated to this class. The run_interaction method (cf. below) initiates a new communicative interaction as an instance of the NGInteraction class, makes the interaction happen, and records its outcome. The run_series method runs a given number of interactions.
[3]:
! pip install alive_progress
import alive_progress
class NGExperiment():
""" The NGExperiment class holds the population and world and defines methods to run (series of) communicative interactions."""
def __init__(self, configuration={}):
"""Upon initialisation, the world and population are created."""
# The configuration passed to the experiment is merged with the default configuration.
global CONFIGURATION
CONFIGURATION = fcg.merge_dicts(CONFIGURATION,configuration)
# World and population are created.
self.world = ['obj-%d' % i for i in range(CONFIGURATION['nr_of_objects'])]
self.population = [NGAgent() for i in range(CONFIGURATION['nr_of_agents'])]
# With a new experiment, we also reset the monitors
global MONITORS
MONITORS = {}
def run_interaction(self):
""" Create a new interaction, make it happen and record the outcome. """
interaction = NGInteraction(self)
interaction.interact()
interaction.record_communicative_success()
interaction.record_lexicon_size()
interaction.record_conventionality()
def run_series(self, nr_of_interactions):
"""Run a series of interactions."""
with alive_progress.alive_bar(nr_of_interactions, force_tty=True) as bar:
for i in range(nr_of_interactions):
self.run_interaction()
bar()
Requirement already satisfied: alive_progress in /Users/paul/Projects/pyfcg/.venv/lib/python3.12/site-packages (3.2.0)
Requirement already satisfied: about-time==4.2.1 in /Users/paul/Projects/pyfcg/.venv/lib/python3.12/site-packages (from alive_progress) (4.2.1)
Requirement already satisfied: grapheme==0.6.0 in /Users/paul/Projects/pyfcg/.venv/lib/python3.12/site-packages (from alive_progress) (0.6.0)
Representing a communicative interaction
The interact method of the NGInteraction class defines the script according to which each communicative interaction takes place. A randomly selected agent, the speaker, formulates an utterance to draw the attention of another randomly selected agent, the hearer, to a randomly selected object in the environment, the topic. If there exists no construction in the speaker’s grammar that associates a name with the topic object, the speaker calls its learn method to invent such a
construction. The hearer then calls its comprehend method to retrieve the topic object in the environment, and its learn method in case it could not understand. The agents achieve communicative success if the hearer could identify the topic object, and both agents will positively or negatively reward their constructions at the end of the interaction, based on its outcome.
[4]:
class NGInteraction():
""" The NGInteraction implements the interaction script of the Naming Game experiment. """
def __init__(self, experiment):
"""Upon initialisation, sample interacting agents and scene."""
self.experiment = experiment
# choose 2 interacting agents at random
self.interacting_agents = random.sample(self.experiment.population, 2)
# Sample a new scene
self.scene = random.sample(self.experiment.world, CONFIGURATION['nr_of_objects'])
# Clear and re-initialise the interacting agents
for agent in self.interacting_agents:
agent.clear()
agent.scene = self.scene
# the first object of the randomly created scene functions as the topic:
agent.topic = self.scene[0]
# Assign discourse roles
self.speaker = self.interacting_agents[0]
self.hearer = self.interacting_agents[1]
self.speaker.discourse_role = 'speaker'
self.hearer.discourse_role = 'hearer'
def interact(self):
"""Defines the interaction script."""
# Run the speaker side (production and possibly invention)
self.speaker.utterance = self.speaker.formulate([[self.speaker.topic, "t"]])
if self.speaker.utterance is None:
self.speaker.learn(fcg.generate_word_form(), self.speaker.topic)
self.speaker.utterance = self.speaker.formulate([[self.speaker.topic, "t"]])
# Pass the utterance to the hearer
self.hearer.utterance = self.speaker.utterance
# Run the hearer side (parsing and possibly adoption)
if self.hearer.comprehend(self.hearer.utterance) is None:
self.hearer.topic = self.scene[0] # pointing
self.hearer.learn(self.hearer.utterance, self.hearer.topic)
else:
self.speaker.communicated_successfully = True
self.hearer.communicated_successfully = True
# Reward the agents, positively and/or negatively
for agent in self.interacting_agents:
agent.reward()
def record_communicative_success(self):
""" Record communicative success of the interaction """
success = all([agent.communicated_successfully for agent in self.interacting_agents])
if success:
notify('communicative_success', 1)
else:
notify('communicative_success', 0)
def record_lexicon_size(self):
""" Record the average lexicon size across the population. """
agents_with_cxns = [agent for agent in self.experiment.population if agent.grammar_size() > 0]
avg_nr_of_cxns = np.mean([agent.grammar_size() for agent in agents_with_cxns])
notify('construction_inventory_size', avg_nr_of_cxns)
def record_conventionality(self):
""" Record the lexicon coherence between speaker and hearer """
speaker = self.speaker
hearer = self.hearer
if not speaker.communicated_successfully:
notify('conventionality', 0)
else:
hearer_form = hearer.formulate([[hearer.topic, "o"]])
if speaker.utterance == hearer_form: ##check!
notify('conventionality', 1)
else:
notify('conventionality', 0)
Monitors and plotting
We create a number of helper functions for monitoring and plotting the results of the experiment.
[5]:
from itertools import islice
def notify(monitor, value):
""" Notify will add 'value' to the 'monitor' key in the global variable MONITORS. """
if monitor in MONITORS:
MONITORS[monitor].append(value)
else:
MONITORS[monitor] = [value]
def window(seq, n=2):
""" Returns a sliding window (of width n) over the data from the sequence. """
it = iter(seq)
result = tuple(islice(it, n))
if len(result) == n:
yield result
for elem in it:
result = result[1:] + (elem,)
yield result
def make_plot_points(monitors, window_size=100):
""" Creates a new dictionary with the averages of the sliding windows, using the same keys as the monitors. """
plot_points = {}
for key in monitors:
plot_points[key] = []
generator = window(monitors[key], n=window_size)
for w in generator:
plot_points[key].append(np.mean(w))
return plot_points
Running a naming game experiment with a population of FCG agents.
An experiment can be run by first creating a new instance of the NGExperiment class and then calling its run_series method, passing the desired number of interactions as an argument. Let us first define a number of default configurations for the experiment:
[6]:
# Default configuration settings #
##################################
# The experiment configuration can be overwritten by passing a configuration dict
# when instantiating an experiment.
CONFIGURATION = {
'nr_of_agents': 10,
'nr_of_objects': 10,
'cxn_positive_reward': 0.1,
'cxn_negative_reward': 0.2
}
# We will store monitoring data for the experiment under a global variable.
MONITORS = {}
[7]:
experiment = NGExperiment({'nr_of_agents': 10, 'nr_of_objects': 5})
experiment.run_series(2500)
|████████████████████████████████████████| 2500/2500 [100%] in 13.8s (181.50/s)
Visualising the results
The results of an experiment can be visualised using any plotting library from Python’s extensive ecosystem, such as matplotlib, seaborn or plotly. We demonstrate here the use of matplotlib to visualise the experiment’s dynamics through graphs, where the degree of communicative success, degree of conventionality and average number of constructions are plotted in function of the number of interactions that have taken place.
[8]:
import matplotlib.pyplot as plt
pp = make_plot_points(MONITORS, window_size=100)
# for each key in plot points (pp), create a plot
fig, axes = plt.subplots(len(pp.keys()), figsize=(5, 7), sharex=True)
for i, key in enumerate(pp.keys()):
ax = axes[i]
ax.plot(list(range(len(pp[key]))), pp[key], label=key)
ax.grid()
ax.legend()
plt.show()
[ ]: