feat/agent-urgency-logic

README.md

@@ -317,6 +317,50 @@ The other is by defining the agent programmatically, with statements like these:
 
 You can also combine both approaches, using the JSON file as a base and then adding or modifying details programmatically.
 
+#### Deciding When to Act: Urgency Score
+
+To provide more nuanced control over agent behavior and manage simulation costs (as each agent action can involve an LLM call), TinyTroupe includes an "urgency score" mechanism. This allows an agent to assess how important it is to act upon a given observation or stimulus.
+
+The core of this logic resides in the `TinyPerson` class with the `_calculate_interaction_urgency(observation: str)` method. When called with an `observation` (a string describing the current situation or stimulus), this method:
+
+1.  Retrieves relevant memories for the agent.
+2.  Constructs a prompt including the agent's persona, current mental state, these memories, and the observation.
+3.  Queries the LLM to return an integer "urgency score" (typically between 0 and 100), indicating how critical it is for the agent to respond.
+
+This score is then compared against a threshold. The library provides `URGENCY_TO_ACT_THRESHOLD` (defaulting to 75) in the `tinytroupe.control` module.
+
+To facilitate using this logic in your simulation scripts, `tinytroupe.control` offers a helper function: `handle_agent_action_with_urgency(agent: TinyPerson, observation: str)`.
+
+Here's a conceptual example of how you might use it in a simulation loop:
+
+```python
+from tinytroupe.agent import TinyPerson
+from tinytroupe.control import handle_agent_action_with_urgency, URGENCY_TO_ACT_THRESHOLD
+# Assume 'my_agent' is an initialized TinyPerson instance
+
+observation = "A new critical task has just been assigned to you."
+
+# The handle_agent_action_with_urgency function will internally call
+# my_agent._calculate_interaction_urgency(observation)
+# and then my_agent.act() if the score >= URGENCY_TO_ACT_THRESHOLD.
+handle_agent_action_with_urgency(my_agent, observation)
+
+# If you want to use a custom threshold:
+# (Note: handle_agent_action_with_urgency currently uses the global URGENCY_TO_ACT_THRESHOLD)
+# For more direct control, you could replicate its logic:
+#
+# urgency = my_agent._calculate_interaction_urgency(observation)
+# print(f"Agent {my_agent.name}: Urgency = {urgency} for observation '{observation}'")
+# CUSTOM_THRESHOLD = 80
+# if urgency >= CUSTOM_THRESHOLD:
+# print(f"Agent {my_agent.name} acts.")
+# my_agent.act()
+# else:
+# print(f"Agent {my_agent.name} does not act.")
+```
+
+This mechanism helps ensure that agents only engage in detailed action processing (and thus incur LLM costs) when the situation warrants it, leading to more efficient and focused simulations.
+
 #### Fragments
 
 `TinyPerson`s can also be further enriched via **fragments**, which are sub-specifications that can be added to the main specification. This is useful to reuse common parts across different agents. For example, the following fragment can be used to specify love of travel ([examples/fragments/travel_enthusiast.agent.fragment.json](./examples/fragments/travel_enthusiast.agent.fragment.json)):

tinytroupe/agent/tiny_person.py

@@ -884,6 +884,47 @@ def retrieve_relevant_memories_for_current_context(self, top_k=7) -> list:
 
         return self.retrieve_relevant_memories(target, top_k=top_k)
 
+    def _calculate_interaction_urgency(self, observation: str) -> int:
+        """
+        Calculates the urgency for the agent to act based on an observation.
+        """
+        relevant_memories_list = self.retrieve_relevant_memories(observation, top_k=5)
+        relevant_memories_str = "\n".join([f"- {memory}" for memory in relevant_memories_list])
+
+        prompt_content = f"""
+        Persona:
+        {json.dumps(self._persona, indent=2)}
+
+        Mental State:
+        {json.dumps(self._mental_state, indent=2)}
+
+        Relevant Memories:
+        {relevant_memories_str}
+
+        Observation:
+        {observation}
+
+        Instruction:
+        Based on the persona, current mental state, relevant memories, and the current observation, provide an integer score from 0 to 100 indicating the urgency for the agent to act. 0 means no urgency, 100 means maximum urgency. Only return the integer score.
+        """
+        try:
+            response_message = openai_utils.client().send_message(
+                messages=[{"role": "user", "content": prompt_content}],
+                # No specific response model, expect plain text.
+            )
+            # Assuming response_message is a dict like {'role': 'assistant', 'content': '75'}
+            urgency_score_str = response_message['content'].strip()
+            urgency_score = int(urgency_score_str)
+            if 0 <= urgency_score <= 100:
+                return urgency_score
+            else:
+                # Score out of range, default to low
+                logger.warning(f"Urgency score {urgency_score} out of range (0-100) for observation: {observation}. Defaulting to 0.")
+                return 0
+        except (ValueError, KeyError, TypeError) as e:
+            # Error in parsing or unexpected response, default to low
+            logger.warning(f"Error parsing urgency score for observation '{observation}': {e}. Defaulting to 0.")
+            return 0
 
     ###########################################################
     # Inspection conveniences

tinytroupe/control.py

@@ -7,10 +7,14 @@
 
 import tinytroupe
 import tinytroupe.utils as utils
+from tinytroupe.agent.tiny_person import TinyPerson # Added import
+from typing import TYPE_CHECKING # Added import
 
 import logging
 logger = logging.getLogger("tinytroupe")
 
+URGENCY_TO_ACT_THRESHOLD = 75
+
 class Simulation:
 
     STATUS_STOPPED = "stopped"
@@ -641,4 +645,23 @@ def cache_misses(id="default"):
     """
     return _simulation(id).cache_misses
 
-reset() # initialize the control state
+reset() # initialize the control state
+
+
+if TYPE_CHECKING:
+    from tinytroupe.agent.tiny_person import TinyPerson
+
+def handle_agent_action_with_urgency(agent: 'TinyPerson', observation: str):
+    # URGENCY_TO_ACT_THRESHOLD is defined in this module
+
+    # Ensure agent has the _calculate_interaction_urgency method.
+    # This function assumes it's being called with a fully capable TinyPerson instance.
+    urgency = agent._calculate_interaction_urgency(observation)
+
+    print(f"Agent {agent.name}: Urgency = {urgency} for observation '{observation}'")
+
+    if urgency >= URGENCY_TO_ACT_THRESHOLD:
+        print(f"Agent {agent.name} acts.")
+        agent.act()  # Call the existing act method on the agent
+    else:
+        print(f"Agent {agent.name} does not act.")