jakdot
commented
2 weeks ago Hi,
Thanks for your interest in pyactr.
pyactr was not built with this functionality in mind. However, I think it is possible to mimic multiple agents interacting in one environment. The following code should work. I am also attaching a trace from running this model with extra explanation on the behavior of this model in some detail. trace.txt
The code simulates two agents, both interacting with the screen on which letters A, B and C are printed one after the other. Agent 1 has to press the key A or C when the letter A or C appears on the screen. If Agent 1 does so, the screen will advance to the next letter. If the letter B appears on the screen, Agent 2 has to press the key B, and the screen advances then to the next letter. If Agent 1 presses A or C when B appears on the screen, nothing happens, and if Agent 2 presses B when A or C are on the screen, nothing happens.
The environments are not completely shared, but the trick to get it to work is to let the environment advance for one agent whenever the other agent made it advance. See the part in the scope of while True at the final part of the code. Most crucial bits are two lines under the "update stimulus" comments. This is a bit of a hack, I had to access some private instance variables to update the simulation from outside of it. At least for this case, it works but I cannot promise it will not break in more complicated cases.
"""
Demo for multiple agents.
This code shows how one can build an environment with multiple agents in which actions of each agent affects the environment for them both.
There are two agents. AG1 should press keys A and C when those letters appear on the screen. AG2 should press B when that letter appears on the screen.
Pressing the right key moves the screen from one letter to the next.
pyactr was not designed with the functionality of supporting multiagents in mind. So the package has to be hacked a bit.
First, sim.run() does not work. You have to go through events manually (see lines at while True how that is done). But that's fine, any serious modeling should do that, instead of using the running function.
Second, in the two lines following the comment "# update stimulus for the other agent..." we have to go into the internal setup of event simulation and manually update the values there. It's not nice and it's likely it will break in more complicated cases. Please, use this code with caution.
Third, there is some inherent asymmetry in the agents. If agent 1 cannot fire any rules whatsoever, the simulation would stop even if agent 2 has some actions ready. This could be avoided with some extra code (e.g., specify that break from the loop only happens after you check both/all agents).
"""
import string
import random
import simpy
import pyactr as actr
stimuli = ['A', 'B', 'C']
text = [{1: {'text': stimuli[0], 'position': (100,100)}}, {2: {'text': stimuli[1], 'position': (100,100)}}, {3: {'text': stimuli[2], 'position': (100,100)}}]
# environments
environ = actr.Environment(focus_position=(100,100))
environproc = environ.environment_process
# chunk types
actr.chunktype("chunk", "value")
actr.chunktype("read", "state")
actr.chunktype("image", "img")
actr.makechunk(nameofchunk="start", typename="chunk", value="start")
actr.makechunk(nameofchunk="start", typename="chunk", value="start")
actr.makechunk(nameofchunk="attend_let", typename="chunk", value="attend_let")
actr.makechunk(nameofchunk="response", typename="chunk", value="response")
actr.makechunk(nameofchunk="done", typename="chunk", value="done")
# rules
encode_letter="""
=g>
isa read
state start
=visual>
isa _visual
value =letter
==>
=g>
isa read
state respond
+g2>
isa image
img =letter"""
respond_toA = """
=g>
isa read
state respond
=g2>
isa image
img A
?manual>
state free
==>
=g>
isa read
state start
=g2>
isa image
img empty
+manual>
isa _manual
cmd 'press_key'
key A"""
respond_toC = """
=g>
isa read
state respond
=g2>
isa image
img C
?manual>
state free
==>
=g>
isa read
state done
=g2>
isa image
img empty
+manual>
isa _manual
cmd 'press_key'
key C"""
dontrespond_toB = """
=g>
isa read
state respond
=g2>
isa image
img B
?manual>
state free
==>
=g>
isa read
state start"""
dontrespond_toA = """
=g>
isa read
state respond
=g2>
isa image
img A
?manual>
state free
==>
=g>
isa read
state start"""
respond_toB = """
=g>
isa read
state respond
=g2>
isa image
img B
?manual>
state free
==>
=g>
isa read
state done
=g2>
isa image
img empty
+manual>
isa _manual
cmd 'press_key'
key B"""
class Agent:
"""
Create agent for environment. Delay specifies delay in creating imaginal buffer (called g2 here).
delay is used for illustration, just to differentiate between agents.
"""
def __init__(self, environment, delay):
self.agent = actr.ACTRModel(environment=environment, motor_prepared=True)
self.agent.goal.add(actr.chunkstring(name="reading", string="""
isa read
state start"""))
self.agent.productionstring(name="encode_letter", string=encode_letter)
g2 = self.agent.set_goal("g2")
g2.delay=delay
# 2 agents with different speed of encoding goals
magent1 = Agent(environ, delay=0.2).agent
magent2 = Agent(environ, delay=0.4).agent
# 2 agents differ in their rules - see the description at the start of this script
magent1.productionstring(name="respond_toA", string=respond_toA)
magent1.productionstring(name="respond_toC", string=respond_toC)
magent1.productionstring(name="dontrespond_toB", string=dontrespond_toB)
magent2.productionstring(name="dontrespond_toA", string=dontrespond_toA)
magent2.productionstring(name="respond_toB", string=respond_toB)
# We start the simulation for both agents
agent1_sim = magent1.simulation(realtime=False, environment_process=environproc, stimuli=text, triggers=stimuli, times=3)
agent2_sim = magent2.simulation(realtime=False, environment_process=environproc, stimuli=text, triggers=stimuli, times=3)
# this is used for internal updates of environment
old_stimulus1, old_stimulus2 = None, None
# The following loop runs the whole simulation
while True:
# store current stimulus
try:
old_stimulus1 = agent1_sim._Simulation__env.stimulus.copy()
except AttributeError:
pass
try:
# do one simulation step
agent1_sim.step()
# print event
print("AGENT1, ", agent1_sim.current_event)
# update stimulus for the other agent if they changed due to the event agent 1 carried
if old_stimulus1 and old_stimulus1 != agent1_sim._Simulation__env.stimulus:
agent2_sim._Simulation__environment_activate.succeed(value=(agent1_sim._Simulation__env.trigger, agent1_sim._Simulation__pr.env_interaction))
# if the schedule is empty, the agent has no rules to do and the action is stopped
except simpy.core.EmptySchedule:
break
# switch to the other agent who should work as long as its internal time is smaller than agent 1 - the timing does not always show well in the trace because of internal time updates in pyactr, which are not always visible to the user
while agent1_sim.show_time() > agent2_sim._Simulation__simulation.peek():
# store stimuli /triggers
try:
old_stimulus2 = agent2_sim._Simulation__env.stimulus.copy()
except AttributeError:
pass
try:
# do one simulation step
agent2_sim.step()
# print event
print("AGENT2, ", agent2_sim.current_event)
# update stimulus for the other agent if they changed due to the event agent 2 carried
if old_stimulus2 and old_stimulus2 != agent2_sim._Simulation__env.stimulus:
agent1_sim._Simulation__environment_activate.succeed(value=(agent2_sim._Simulation__env.trigger, agent2_sim._Simulation__pr.env_interaction))
break
# if the schedule is empty, the agent has no rules to do and the action is stopped
except simpy.core.EmptySchedule:
break
Hey! Thanks a lot for the package. Do you have any idea, how I'm able to let multiple agents interact with each other in the same environment simultaniously? Working on my thesis at the moment.