Add basic_agent_without_plan and use it in a scenario for state_forma…

…tion.

PiperOrigin-RevId: 686277090
Change-Id: I457d83dda871add58a2a913174f6008a5c3f88eb

concordia/factory/agent/basic_agent_without_plan.py

@@ -0,0 +1,201 @@
+# Copyright 2024 DeepMind Technologies Limited.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""A factory implementing the three key questions agent as an entity."""
+
+import datetime
+
+from concordia.agents import entity_agent_with_logging
+from concordia.associative_memory import associative_memory
+from concordia.associative_memory import formative_memories
+from concordia.clocks import game_clock
+from concordia.components import agent as agent_components
+from concordia.language_model import language_model
+from concordia.memory_bank import legacy_associative_memory
+from concordia.utils import measurements as measurements_lib
+
+
+def _get_class_name(object_: object) -> str:
+  return object_.__class__.__name__
+
+
+def build_agent(
+    *,
+    config: formative_memories.AgentConfig,
+    model: language_model.LanguageModel,
+    memory: associative_memory.AssociativeMemory,
+    clock: game_clock.MultiIntervalClock,
+    update_time_interval: datetime.timedelta,
+) -> entity_agent_with_logging.EntityAgentWithLogging:
+  """Build an agent.
+
+  Args:
+    config: The agent config to use.
+    model: The language model to use.
+    memory: The agent's memory object.
+    clock: The clock to use.
+    update_time_interval: Agent calls update every time this interval passes.
+
+  Returns:
+    An agent.
+  """
+  del update_time_interval
+  if not config.extras.get('main_character', False):
+    raise ValueError('This function is meant for a main character '
+                     'but it was called on a supporting character.')
+
+  agent_name = config.name
+
+  raw_memory = legacy_associative_memory.AssociativeMemoryBank(memory)
+
+  measurements = measurements_lib.Measurements()
+  instructions = agent_components.instructions.Instructions(
+      agent_name=agent_name,
+      logging_channel=measurements.get_channel('Instructions').on_next,
+  )
+
+  observation_label = '\nObservation'
+  observation = agent_components.observation.Observation(
+      clock_now=clock.now,
+      timeframe=clock.get_step_size(),
+      pre_act_key=observation_label,
+      logging_channel=measurements.get_channel('Observation').on_next,
+  )
+  observation_summary_label = '\nSummary of recent observations'
+  observation_summary = agent_components.observation.ObservationSummary(
+      model=model,
+      clock_now=clock.now,
+      timeframe_delta_from=datetime.timedelta(hours=24),
+      timeframe_delta_until=datetime.timedelta(hours=0),
+      pre_act_key=observation_summary_label,
+      logging_channel=measurements.get_channel('ObservationSummary').on_next,
+  )
+  time_display = agent_components.report_function.ReportFunction(
+      function=clock.current_time_interval_str,
+      pre_act_key='\nCurrent time',
+      logging_channel=measurements.get_channel('TimeDisplay').on_next,
+  )
+  identity_label = '\nIdentity characteristics'
+  identity_characteristics = (
+      agent_components.question_of_query_associated_memories.IdentityWithoutPreAct(
+          model=model,
+          logging_channel=measurements.get_channel(
+              'IdentityWithoutPreAct'
+          ).on_next,
+          pre_act_key=identity_label,
+      )
+  )
+  self_perception_label = (
+      f'\nQuestion: What kind of person is {agent_name}?\nAnswer')
+  self_perception = agent_components.question_of_recent_memories.SelfPerception(
+      model=model,
+      components={_get_class_name(identity_characteristics): identity_label},
+      pre_act_key=self_perception_label,
+      logging_channel=measurements.get_channel('SelfPerception').on_next,
+  )
+  situation_perception_label = (
+      f'\nQuestion: What kind of situation is {agent_name} in '
+      'right now?\nAnswer')
+  situation_perception = (
+      agent_components.question_of_recent_memories.SituationPerception(
+          model=model,
+          components={
+              _get_class_name(observation): observation_label,
+              _get_class_name(observation_summary): observation_summary_label,
+          },
+          clock_now=clock.now,
+          pre_act_key=situation_perception_label,
+          logging_channel=measurements.get_channel(
+              'SituationPerception'
+          ).on_next,
+      )
+  )
+  person_by_situation_label = (
+      f'\nQuestion: What would a person like {agent_name} do in '
+      'a situation like this?\nAnswer')
+  person_by_situation = (
+      agent_components.question_of_recent_memories.PersonBySituation(
+          model=model,
+          components={
+              _get_class_name(self_perception): self_perception_label,
+              _get_class_name(situation_perception): situation_perception_label,
+          },
+          clock_now=clock.now,
+          pre_act_key=person_by_situation_label,
+          logging_channel=measurements.get_channel('PersonBySituation').on_next,
+      )
+  )
+  relevant_memories_label = '\nRecalled memories and observations'
+  relevant_memories = agent_components.all_similar_memories.AllSimilarMemories(
+      model=model,
+      components={
+          _get_class_name(observation_summary): observation_summary_label,
+          _get_class_name(time_display): 'The current date/time is'},
+      num_memories_to_retrieve=10,
+      pre_act_key=relevant_memories_label,
+      logging_channel=measurements.get_channel('AllSimilarMemories').on_next,
+  )
+
+  plan_components = {}
+  if config.goal:
+    goal_label = '\nGoal'
+    overarching_goal = agent_components.constant.Constant(
+        state=config.goal,
+        pre_act_key=goal_label,
+        logging_channel=measurements.get_channel(goal_label).on_next)
+    plan_components[goal_label] = goal_label
+  else:
+    goal_label = None
+    overarching_goal = None
+
+  entity_components = (
+      # Components that provide pre_act context.
+      instructions,
+      observation,
+      observation_summary,
+      relevant_memories,
+      self_perception,
+      situation_perception,
+      person_by_situation,
+      time_display,
+
+      # Components that do not provide pre_act context.
+      identity_characteristics,
+  )
+  components_of_agent = {_get_class_name(component): component
+                         for component in entity_components}
+  components_of_agent[
+      agent_components.memory_component.DEFAULT_MEMORY_COMPONENT_NAME] = (
+          agent_components.memory_component.MemoryComponent(raw_memory))
+  component_order = list(components_of_agent.keys())
+  if overarching_goal is not None:
+    components_of_agent[goal_label] = overarching_goal
+    # Place goal after the instructions.
+    component_order.insert(1, goal_label)
+
+  act_component = agent_components.concat_act_component.ConcatActComponent(
+      model=model,
+      clock=clock,
+      component_order=component_order,
+      logging_channel=measurements.get_channel('ActComponent').on_next,
+  )
+
+  agent = entity_agent_with_logging.EntityAgentWithLogging(
+      agent_name=agent_name,
+      act_component=act_component,
+      context_components=components_of_agent,
+      component_logging=measurements,
+  )
+
+  return agent

concordia/factory/agent/factories_test.py

@@ -25,6 +25,8 @@
 from concordia.associative_memory import formative_memories
 from concordia.clocks import game_clock
 from concordia.factory.agent import basic_agent
+from concordia.factory.agent import basic_agent_without_plan
+from concordia.factory.agent import observe_recall_prompt_agent
 from concordia.factory.agent import paranoid_agent
 from concordia.factory.agent import rational_agent
 from concordia.factory.agent import synthetic_user
@@ -44,15 +46,17 @@
 
 AGENT_FACTORIES = {
     'basic_agent': basic_agent,
+    'basic_agent_without_plan': basic_agent_without_plan,
+    'observe_recall_prompt_agent': observe_recall_prompt_agent,
+    'paranoid_agent': paranoid_agent,
     'rational_agent': rational_agent,
     'synthetic_user': synthetic_user,
-    'paranoid_agent': paranoid_agent,
 }
 
 
 def _embedder(text: str):
   del text
-  return np.random.rand(16)
+  return np.random.rand(3)
 
 
 class AgentFactoriesTest(parameterized.TestCase):
@@ -64,20 +68,30 @@ class AgentFactoriesTest(parameterized.TestCase):
           main_role=True
       ),
       dict(
-          testcase_name='rational_agent',
-          agent_name='rational_agent',
+          testcase_name='basic_agent_without_plan',
+          agent_name='basic_agent_without_plan',
           main_role=True,
       ),
       dict(
-          testcase_name='synthetic_user',
-          agent_name='synthetic_user',
+          testcase_name='observe_recall_prompt_agent',
+          agent_name='observe_recall_prompt_agent',
           main_role=True,
       ),
       dict(
           testcase_name='paranoid_agent',
           agent_name='paranoid_agent',
           main_role=True,
       ),
+      dict(
+          testcase_name='rational_agent',
+          agent_name='rational_agent',
+          main_role=True,
+      ),
+      dict(
+          testcase_name='synthetic_user',
+          agent_name='synthetic_user',
+          main_role=True,
+      ),
   )
   def test_output_in_right_format(self, agent_name: str, main_role: bool):
     agent_factory = AGENT_FACTORIES[agent_name]

examples/modular/scenario/scenarios.py

@@ -145,6 +145,17 @@ class ScenarioConfig:
         environment='reality_show',
         supporting_agent_module=None,
     ),
+    state_formation=SubstrateConfig(
+        description=(
+            'players are elders in two pre-state agrarian villages which are '
+            'being threatened by a common enemy and have the option of '
+            'working together; the elders must negotiate an agreement with '
+            'the other village and then sell their deal to influential '
+            'stakeholders back home'
+        ),
+        environment='state_formation',
+        supporting_agent_module='basic_agent',
+    ),
 )
 
 SCENARIO_CONFIGS: Mapping[str, ScenarioConfig] = immutabledict.immutabledict(
@@ -303,6 +314,21 @@ class ScenarioConfig:
             'persuasion',
         ),
     ),
+    state_formation_0=ScenarioConfig(
+        description=(
+            'player must negotiate a treaty to enable division of labor in '
+            'common defense and agriculture and sell it to village stakeholders'
+        ),
+        substrate_config=SUBSTRATE_CONFIGS['state_formation'],
+        background_agent_module='basic_agent_without_plan',
+        time_and_place_module='pre_state_villages',
+        focal_is_resident=True,
+        tags=(
+            'negotiation',
+            'persuasion',
+            'division of labor',
+        ),
+    ),
 )