Multiple Agents

[BastianMannerow]

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.

[jakdot]

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