Module tinytroupe.agent

This module provides the main classes and functions for TinyTroupe's agents.

Agents are the key abstraction used in TinyTroupe. An agent is a simulated person or entity that can interact with other agents and the environment, by receiving stimuli and producing actions. Agents have cognitive states, which are updated as they interact with the environment and other agents. Agents can also store and retrieve information from memory, and can perform actions in the environment. Different from agents whose objective is to provide support for AI-based assistants or other such productivity tools, TinyTroupe agents aim at representing human-like behavior, which includes idiossincracies, emotions, and other human-like traits, that one would not expect from a productivity tool.

The overall underlying design is inspired mainly by Cognitive Psychology, which is why agents have various internal cognitive states, such as attention, emotions, and goals. It is also why agent memory, differently from other LLM-based agent platforms, has subtle internal divisions, notably between episodic and semantic memory. Some behaviorist concepts are also present, such as the explicit and decoupled concepts of "stimulus" and "response" in the listen and act methods, which are key abstractions to understand how agents interact with the environment and other agents.

Expand source code 
"""
This module provides the main classes and functions for TinyTroupe's  agents.

Agents are the key abstraction used in TinyTroupe. An agent is a simulated person or entity that can interact with other agents and the environment, by
receiving stimuli and producing actions. Agents have cognitive states, which are updated as they interact with the environment and other agents. 
Agents can also store and retrieve information from memory, and can perform actions in the environment. Different from agents whose objective is to
provide support for AI-based assistants or other such productivity tools, **TinyTroupe agents aim at representing human-like behavior**, which includes
idiossincracies, emotions, and other human-like traits, that one would not expect from a productivity tool.

The overall underlying design is inspired mainly by Cognitive Psychology, which is why agents have various internal cognitive states, such as attention, emotions, and goals.
It is also why agent memory, differently from other LLM-based agent platforms, has subtle internal divisions, notably between episodic and semantic memory. 
Some behaviorist concepts are also present, such as the explicit and decoupled concepts of "stimulus" and "response" in the `listen` and `act` methods, which are key abstractions
to understand how agents interact with the environment and other agents.
"""

import tinytroupe.utils as utils
from pydantic import BaseModel

import logging
logger = logging.getLogger("tinytroupe")

from tinytroupe import default

###########################################################################
# Types and constants
###########################################################################
from typing import TypeVar, Union
Self = TypeVar("Self", bound="TinyPerson")
AgentOrWorld = Union[Self, "TinyWorld"]


###########################################################################
# Data structures to enforce output format during LLM API call.
###########################################################################
class Action(BaseModel):
    type: str
    content: str
    target: str

class CognitiveState(BaseModel):
    goals: str
    context: list[str]
    attention: str
    emotions: str

class CognitiveActionModel(BaseModel):
    action: Action
    cognitive_state: CognitiveState

class CognitiveActionModelWithReasoning(BaseModel):
    reasoning: str
    action: Action
    cognitive_state: CognitiveState


###########################################################################
# Exposed API
###########################################################################
# from. grounding ... ---> not exposing this, clients should not need to know about detailed grounding mechanisms
from .memory import SemanticMemory, EpisodicMemory, EpisodicConsolidator, ReflectionConsolidator
from .mental_faculty import CustomMentalFaculty, RecallFaculty, FilesAndWebGroundingFaculty, TinyToolUse
from .tiny_person import TinyPerson

__all__ = ["SemanticMemory", "EpisodicMemory", "EpisodicConsolidator", "ReflectionConsolidator",
           "CustomMentalFaculty", "RecallFaculty", "FilesAndWebGroundingFaculty", "TinyToolUse",
           "TinyPerson"]

Sub-modules

tinytroupe.agent.action_generator
tinytroupe.agent.grounding
tinytroupe.agent.memory
tinytroupe.agent.mental_faculty
tinytroupe.agent.tiny_person

Classes

class CustomMentalFaculty (name: str, requires_faculties: list = None, actions_configs: dict = None, constraints: dict = None)

Represents a custom mental faculty of an agent. Custom mental faculties are the cognitive abilities that an agent has and that are defined by the user just by specifying the actions that the faculty can perform or the constraints that the faculty introduces. Constraints might be related to the actions that the faculty can perform or be independent, more general constraints that the agent must follow.

Initializes the custom mental faculty.

Args

name : str
The name of the mental faculty.
requires_faculties : list
A list of mental faculties that this faculty requires to function properly. Format is ["faculty1", "faculty2", …]
actions_configs : dict
A dictionary with the configuration of actions that this faculty can perform. Format is {: {"description": , "function": }}
constraints : dict
A list with the constraints introduced by this faculty. Format is [, , …]
Expand source code 
class CustomMentalFaculty(TinyMentalFaculty):
    """
    Represents a custom mental faculty of an agent. Custom mental faculties are the cognitive abilities that an agent has
    and that are defined by the user just by specifying the actions that the faculty can perform or the constraints that
    the faculty introduces. Constraints might be related to the actions that the faculty can perform or be independent,
    more general constraints that the agent must follow.
    """

    def __init__(self, name: str, requires_faculties: list = None,
                 actions_configs: dict = None, constraints: dict = None):
        """
        Initializes the custom mental faculty.

        Args:
            name (str): The name of the mental faculty.
            requires_faculties (list): A list of mental faculties that this faculty requires to function properly. 
              Format is ["faculty1", "faculty2", ...]
            actions_configs (dict): A dictionary with the configuration of actions that this faculty can perform.
              Format is {<action_name>: {"description": <description>, "function": <function>}}
            constraints (dict): A list with the constraints introduced by this faculty.
              Format is [<constraint1>, <constraint2>, ...]
        """

        super().__init__(name, requires_faculties)

        # {<action_name>: {"description": <description>, "function": <function>}}
        if actions_configs is None:
            self.actions_configs = {}
        else:
            self.actions_configs = actions_configs
        
        # [<constraint1>, <constraint2>, ...]
        if constraints is None:
            self.constraints = {}
        else:
            self.constraints = constraints
    
    def add_action(self, action_name: str, description: str, function: Callable=None):
        self.actions_configs[action_name] = {"description": description, "function": function}

    def add_actions(self, actions: dict):
        for action_name, action_config in actions.items():
            self.add_action(action_name, action_config['description'], action_config['function'])
    
    def add_action_constraint(self, constraint: str):
        self.constraints.append(constraint)
    
    def add_actions_constraints(self, constraints: list):
        for constraint in constraints:
            self.add_action_constraint(constraint)

    def process_action(self, agent, action: dict) -> bool:
        logger.debug(f"Processing action: {action}")

        action_type = action['type']
        if action_type in self.actions_configs:
            action_config = self.actions_configs[action_type]
            action_function = action_config.get("function", None)

            if action_function is not None:
                action_function(agent, action)
            
            # one way or another, the action was processed
            return True 
        
        else:
            return False
    
    def actions_definitions_prompt(self) -> str:
        prompt = ""
        for action_name, action_config in self.actions_configs.items():
            prompt += f"  - {action_name.upper()}: {action_config['description']}\n"
        
        return prompt

    def actions_constraints_prompt(self) -> str:
        prompt = ""
        for constraint in self.constraints:
            prompt += f"  - {constraint}\n"
        
        return prompt

Ancestors

Methods

def add_action(self, action_name: str, description: str, function: Callable = None)
Expand source code 
def add_action(self, action_name: str, description: str, function: Callable=None):
    self.actions_configs[action_name] = {"description": description, "function": function}
def add_action_constraint(self, constraint: str)
Expand source code 
def add_action_constraint(self, constraint: str):
    self.constraints.append(constraint)
def add_actions(self, actions: dict)
Expand source code 
def add_actions(self, actions: dict):
    for action_name, action_config in actions.items():
        self.add_action(action_name, action_config['description'], action_config['function'])
def add_actions_constraints(self, constraints: list)
Expand source code 
def add_actions_constraints(self, constraints: list):
    for constraint in constraints:
        self.add_action_constraint(constraint)

Inherited members

class EpisodicConsolidator

Consolidates episodic memories into a more abstract representation, such as a summary or an abstract fact.

Expand source code 
class EpisodicConsolidator(MemoryProcessor):
    """
    Consolidates episodic memories into a more abstract representation, such as a summary or an abstract fact.
    """

    def process(self, memories: list, timestamp: str=None, context:Union[str, list, dict] = None, persona:Union[str, dict] = None, sequential: bool = True) -> list:
        logger.debug(f"STARTING MEMORY CONSOLIDATION: {len(memories)} memories to consolidate")

        enriched_context = f"CURRENT COGNITIVE CONTEXT OF THE AGENT: {context}" if context else "No specific context provided for consolidation."

        result = self._consolidate(memories, timestamp, enriched_context, persona)
        logger.debug(f"Consolidated {len(memories)} memories into: {result}")
        
        return result

    @utils.llm(enable_json_output_format=True, enable_justification_step=False)
    def _consolidate(self, memories: list, timestamp: str, context:str, persona:str) -> dict:
        """
        Given a list of input episodic memories, this method consolidates them into more organized structured representations, which however preserve all information and important details. 

        For this process, you assume:
          - This consolidation is being carried out by an agent, so the memories are from the agent's perspective. "Actions" refer to behaviors produced by the agent,
            while  "stimulus" refer to events or information from the environment or other agents that the agent has perceived.
                * Thus, in the consoldation you write "I have done X" or "I have perceived Y", not "the agent has done X" or "the agent has perceived Y".
          - The purpose of consolidation is to restructure and organize the most relevant information from the episodic memories, so that any facts learned therein can be used in future reasoning processes.
                * If a `context` is provided, you can use it to guide the consolidation process, making sure that the memories are consolidated in the most useful way under the given context.
                  For example, if the agent is looking for a specific type of information, you can focus the consolidation on that type of information, preserving more details about it
                  than you would otherwise.
                * If a `persona` is provided, you can use it to guide the consolidation process, making sure that the memories are consolidated in a way that is consistent with the persona.
                  For example, if the persona is that of a cat lover, you can focus the consolidation on the agent's experiences with cats, preserving more details about them than you would otherwise.
          - If the memory contians a `content` field, that's where the relevant information is found. Otherwise, consider the whole memory as relevant information.

        The consolidation process follows these rules:
          - Each consolidated memory groups together all similar entries: so actions are grouped together, stimuli go together, facts are grouped together, impressions are grouped together, 
            learned processes are grouped together, and ad-hoc elements go together too. Noise, minor details and irrelevant elements are discarded. 
            In all, you will produce at most the following consolidated entries (you can avoid some if appropriate, but not add more):
              * Actions: all actions are grouped together, giving an account of what the agent has done.
              * Stimuli: all stimuli are grouped together, giving an account of what the agent has perceived.
              * Facts: facts are extracted from the actions and stimuli, and then grouped together in a single entry, consolidating learning of objective facts.
              * Impressions: impressions, feelings, or other subjective experiences are also extracted,  and then grouped together in a single entry, consolidating subjective experiences.
              * Procedural: learned processes (e.g., how to do certain things) are also extracted, formatted in an algorithmic way (i.e., pseudo-code that is self-explanatory), and then grouped together in a 
                single entry, consolidating learned processes.
              * Ad-Hoc: important elements that do not correspond to these options are also grouped together in an ad-hoc single entry, consolidating other types of information.
          - Each consolidated memory is a comprehensive report of the relevant information from the input memories, preserving all details. The consolidation merely reorganizes the information,
            but does not remove any relevant information. The consolidated memories are not summaries, but rather a more organized and structured representation of the information in the input memories.
          

        Each input memory is a dictionary of the form:
            ```
            {
            "role": role, 
            "content": content, 
            "type": "action"/"stimulus"/"feedback"/"reflection", 
            "simulation_timestamp": timestamp
            }
            ``` 

        Each consolidated output memory is a dictionary of the form:
            ```
            {
            "content": content, 
            "type": "consolidated", 
            "simulation_timestamp": timestamp of the consolidation
            }  
            ```


         So the final value outputed **must** be a JSON composed of a list of dictionaries, each representing a consolidated memory, **always** with the following structure:
            ```
            {"consolidation":
                [
                    {
                        "content": content_1, 
                        "type": "consolidated", 
                        "simulation_timestamp": timestamp of the consolidation
                    },
                    {
                        "content": content_2, 
                        "type": "consolidated", 
                        "simulation_timestamp": timestamp of the consolidation
                    },
                    ...
                ]
            }
            ```

        Note:
          - because the output is a JSON, you must use double quotes for the keys and string values.
        ## Example (simplified)

        Here's a simplified example. Suppose the following memory contents are provided as input (simplifying here as just a bullet list of contents):
         - stimulus: "I have seen a cat, walking beautifully in the street"
         - stimulus: "I have seen a dog, barking loudly at a passerby, looking very aggressive"
         - action: "I have petted the cat, run around with him (or her?), saying a thousand times how cute it is, and how much I seem to like cats"
         - action: "I just realized that I like cats more than dogs. For example, look at this one, it is so cute, so civilized, so noble, so elegant, an inspiring animal! I had never noted this before! "
         - stimulus: "The cat is meowing very loudly, it seems to be hungry"
         - stimulus: "Somehow a big capivara has appeared in the room, it is looking at me with curiosity"

        Then, this would be a possible CORRECT output of the consolidation process (again, simplified, showing only contents in bullet list format):
          - consolidated actions: "I have petted the cat, run around with it, and expressed my admiration for cats."
          - consolidated stimuli: "I have seen a beautiful but hungry cat, a loud and agressive-looking dog, and - surprisingly - a capivara"
          - consolidated impressions: "I felt great admiration for the cat, they look like such noble and elegant animals."
          - consolidated facts: "I like cats more than dogs because they are cute and noble creatures."

        These are correct because they focus on the agent's experience. In contrast, this would be an INCORRECT output of the consolidation process:
          - consolidated actions: "the user sent messages about a cat, a dog and a capivara, and about playing with the cat."
          - consolidated facts: "the assistant has received various messages at different times, and has performed actions in response to them."

        These are incorrect because they focus on the agent's cognition and internal implementation mechanisms, not on the agent's experience.

        Args:
            memories (list): The list of memories to consolidate.
            timestamp (str): The timestamp of the consolidation, which will be used in the consolidated memories instead of any original timestamp.
            context (str, optional): Additional context to guide the consolidation process. This can be used to provide specific instructions or constraints for the consolidation.
            persona (str, optional): The persona of the agent, which can be used to guide the consolidation process. This can be used to provide specific instructions or constraints for the consolidation.

        Returns:
            dict: A dictionary with a single key "consolidation", whose value is a list of consolidated memories, each represented as a dictionary with the structure described above.
        """
        # llm annotation will handle the implementation

Ancestors

Inherited members

class EpisodicMemory (fixed_prefix_length: int = 20, lookback_length: int = 100)

Provides episodic memory capabilities to an agent. Cognitively, episodic memory is the ability to remember specific events, or episodes, in the past. This class provides a simple implementation of episodic memory, where the agent can store and retrieve messages from memory.

Subclasses of this class can be used to provide different memory implementations.

Initializes the memory.

Args

fixed_prefix_length : int
The fixed prefix length. Defaults to 20.
lookback_length : int
The lookback length. Defaults to 100.
Expand source code 
class EpisodicMemory(TinyMemory):
    """
    Provides episodic memory capabilities to an agent. Cognitively, episodic memory is the ability to remember specific events,
    or episodes, in the past. This class provides a simple implementation of episodic memory, where the agent can store and retrieve
    messages from memory.
    
    Subclasses of this class can be used to provide different memory implementations.
    """

    MEMORY_BLOCK_OMISSION_INFO = {'role': 'assistant', 'content': "Info: there were other messages here, but they were omitted for brevity.", 'simulation_timestamp': None}

    def __init__(
        self, fixed_prefix_length: int = 20, lookback_length: int = 100
    ) -> None:
        """
        Initializes the memory.

        Args:
            fixed_prefix_length (int): The fixed prefix length. Defaults to 20.
            lookback_length (int): The lookback length. Defaults to 100.
        """
        self.fixed_prefix_length = fixed_prefix_length
        self.lookback_length = lookback_length

        # the definitive memory that records all episodic events
        self.memory = []
        
        # the current episode buffer, which is used to store messages during an episode
        self.episodic_buffer = []


    def commit_episode(self):
        """
        Ends the current episode, storing the episodic buffer in memory.
        """
        self.memory.extend(self.episodic_buffer)
        self.episodic_buffer = []
    
    def get_current_episode(self, item_types:list=None) -> list:
        """
        Returns the current episode buffer, which is used to store messages during an episode.

        Args:
            item_types (list, optional): If provided, only retrieve memories of these types. Defaults to None, which retrieves all types.

        Returns:
            list: The current episode buffer.
        """
        result = copy.copy(self.episodic_buffer)
        result = self.filter_by_item_types(result, item_types) if item_types is not None else result
        return result

    def count(self) -> int:
        """
        Returns the number of values in memory.
        """
        return len(self._memory_with_current_buffer())

    def clear(self, max_prefix_to_clear:int=None, max_suffix_to_clear:int=None):
        """
        Clears the memory, generating a permanent "episodic amnesia". 
        If max_prefix_to_clear is not None, it clears the first n values from memory.
        If max_suffix_to_clear is not None, it clears the last n values from memory. If both are None,
        it clears all values from memory.

        Args:
            max_prefix_to_clear (int): The number of first values to clear.
            max_suffix_to_clear (int): The number of last values to clear.
        """

        # clears all episodic buffer messages
        self.episodic_buffer = []

        # then clears the memory according to the parameters
        if max_prefix_to_clear is not None:
            self.memory = self.memory[max_prefix_to_clear:]

        if max_suffix_to_clear is not None:
            self.memory = self.memory[:-max_suffix_to_clear]

        if max_prefix_to_clear is None and max_suffix_to_clear is None:
            self.memory = []
    
    def _memory_with_current_buffer(self) -> list:
        """
        Returns the current memory, including the episodic buffer.
        This is useful for retrieving the most recent memories, including the current episode.
        """
        return self.memory + self.episodic_buffer
        
    ######################################
    # General memory methods
    ######################################
    def _store(self, value: Any) -> None:
        """
        Stores a value in memory.
        """
        self.episodic_buffer.append(value)

    def retrieve(self, first_n: int, last_n: int, include_omission_info:bool=True, item_type:str=None) -> list:
        """
        Retrieves the first n and/or last n values from memory. If n is None, all values are retrieved.

        Args:
            first_n (int): The number of first values to retrieve.
            last_n (int): The number of last values to retrieve.
            include_omission_info (bool): Whether to include an information message when some values are omitted.
            item_type (str, optional): If provided, only retrieve memories of this type.

        Returns:
            list: The retrieved values.
        
        """

        omisssion_info = [EpisodicMemory.MEMORY_BLOCK_OMISSION_INFO] if include_omission_info else []

        # use the other methods in the class to implement
        if first_n is not None and last_n is not None:
            return self.retrieve_first(first_n, include_omission_info=False, item_type=item_type) + omisssion_info + self.retrieve_last(last_n, include_omission_info=False, item_type=item_type)
        elif first_n is not None:
            return self.retrieve_first(first_n, include_omission_info, item_type=item_type)
        elif last_n is not None:
            return self.retrieve_last(last_n, include_omission_info, item_type=item_type)
        else:
            return self.retrieve_all(item_type=item_type)

    def retrieve_recent(self, include_omission_info:bool=True, item_type:str=None) -> list:
        """
        Retrieves the n most recent values from memory.

        Args:
            include_omission_info (bool): Whether to include an information message when some values are omitted.
            item_type (str, optional): If provided, only retrieve memories of this type.
        """
        omisssion_info = [EpisodicMemory.MEMORY_BLOCK_OMISSION_INFO] if include_omission_info else []
        
        # Filter memories if item_type is provided
        memories = self._memory_with_current_buffer() if item_type is None else self.filter_by_item_type(self._memory_with_current_buffer(), item_type)

        # compute fixed prefix
        fixed_prefix = memories[: self.fixed_prefix_length] + omisssion_info

        # how many lookback values remain?
        remaining_lookback = min(
            len(memories) - len(fixed_prefix) + (1 if include_omission_info else 0), self.lookback_length
        )

        # compute the remaining lookback values and return the concatenation
        if remaining_lookback <= 0:
            return fixed_prefix
        else:
            return fixed_prefix + memories[-remaining_lookback:]

    def retrieve_all(self, item_type:str=None) -> list:
        """
        Retrieves all values from memory.

        Args:
            item_type (str, optional): If provided, only retrieve memories of this type.
        """
        memories = self._memory_with_current_buffer() if item_type is None else self.filter_by_item_type(self._memory_with_current_buffer(), item_type)
        return copy.copy(memories)

    def retrieve_relevant(self, relevance_target: str, top_k:int) -> list:
        """
        Retrieves top-k values from memory that are most relevant to a given target.
        """
        raise NotImplementedError("Subclasses must implement this method.")

    def retrieve_first(self, n: int, include_omission_info:bool=True, item_type:str=None) -> list:
        """
        Retrieves the first n values from memory.

        Args:
            n (int): The number of values to retrieve.
            include_omission_info (bool): Whether to include an information message when some values are omitted.
            item_type (str, optional): If provided, only retrieve memories of this type.
        """
        omisssion_info = [EpisodicMemory.MEMORY_BLOCK_OMISSION_INFO] if include_omission_info else []
        
        memories = self._memory_with_current_buffer() if item_type is None else self.filter_by_item_type(self._memory_with_current_buffer(), item_type)
        return memories[:n] + omisssion_info
    
    def retrieve_last(self, n: int=None, include_omission_info:bool=True, item_type:str=None) -> list:
        """
        Retrieves the last n values from memory.

        Args:
            n (int): The number of values to retrieve, or None to retrieve all values.
            include_omission_info (bool): Whether to include an information message when some values are omitted.
            item_type (str, optional): If provided, only retrieve memories of this type.
        """
        omisssion_info = [EpisodicMemory.MEMORY_BLOCK_OMISSION_INFO] if include_omission_info else []

        memories = self._memory_with_current_buffer() if item_type is None else self.filter_by_item_type(self._memory_with_current_buffer(), item_type)
        memories = memories[-n:] if n is not None else memories
                            
        return omisssion_info + memories

Ancestors

Class variables

var MEMORY_BLOCK_OMISSION_INFO

Methods

def clear(self, max_prefix_to_clear: int = None, max_suffix_to_clear: int = None)

Clears the memory, generating a permanent "episodic amnesia". If max_prefix_to_clear is not None, it clears the first n values from memory. If max_suffix_to_clear is not None, it clears the last n values from memory. If both are None, it clears all values from memory.

Args

max_prefix_to_clear : int
The number of first values to clear.
max_suffix_to_clear : int
The number of last values to clear.
Expand source code 
def clear(self, max_prefix_to_clear:int=None, max_suffix_to_clear:int=None):
    """
    Clears the memory, generating a permanent "episodic amnesia". 
    If max_prefix_to_clear is not None, it clears the first n values from memory.
    If max_suffix_to_clear is not None, it clears the last n values from memory. If both are None,
    it clears all values from memory.

    Args:
        max_prefix_to_clear (int): The number of first values to clear.
        max_suffix_to_clear (int): The number of last values to clear.
    """

    # clears all episodic buffer messages
    self.episodic_buffer = []

    # then clears the memory according to the parameters
    if max_prefix_to_clear is not None:
        self.memory = self.memory[max_prefix_to_clear:]

    if max_suffix_to_clear is not None:
        self.memory = self.memory[:-max_suffix_to_clear]

    if max_prefix_to_clear is None and max_suffix_to_clear is None:
        self.memory = []
def commit_episode(self)

Ends the current episode, storing the episodic buffer in memory.

Expand source code 
def commit_episode(self):
    """
    Ends the current episode, storing the episodic buffer in memory.
    """
    self.memory.extend(self.episodic_buffer)
    self.episodic_buffer = []
def count(self) ‑> int

Returns the number of values in memory.

Expand source code 
def count(self) -> int:
    """
    Returns the number of values in memory.
    """
    return len(self._memory_with_current_buffer())
def get_current_episode(self, item_types: list = None) ‑> list

Returns the current episode buffer, which is used to store messages during an episode.

Args

item_types : list, optional
If provided, only retrieve memories of these types. Defaults to None, which retrieves all types.

Returns

list
The current episode buffer.
Expand source code 
def get_current_episode(self, item_types:list=None) -> list:
    """
    Returns the current episode buffer, which is used to store messages during an episode.

    Args:
        item_types (list, optional): If provided, only retrieve memories of these types. Defaults to None, which retrieves all types.

    Returns:
        list: The current episode buffer.
    """
    result = copy.copy(self.episodic_buffer)
    result = self.filter_by_item_types(result, item_types) if item_types is not None else result
    return result
def retrieve_first(self, n: int, include_omission_info: bool = True, item_type: str = None) ‑> list

Retrieves the first n values from memory.

Args

n : int
The number of values to retrieve.
include_omission_info : bool
Whether to include an information message when some values are omitted.
item_type : str, optional
If provided, only retrieve memories of this type.
Expand source code 
def retrieve_first(self, n: int, include_omission_info:bool=True, item_type:str=None) -> list:
    """
    Retrieves the first n values from memory.

    Args:
        n (int): The number of values to retrieve.
        include_omission_info (bool): Whether to include an information message when some values are omitted.
        item_type (str, optional): If provided, only retrieve memories of this type.
    """
    omisssion_info = [EpisodicMemory.MEMORY_BLOCK_OMISSION_INFO] if include_omission_info else []
    
    memories = self._memory_with_current_buffer() if item_type is None else self.filter_by_item_type(self._memory_with_current_buffer(), item_type)
    return memories[:n] + omisssion_info
def retrieve_last(self, n: int = None, include_omission_info: bool = True, item_type: str = None) ‑> list

Retrieves the last n values from memory.

Args

n : int
The number of values to retrieve, or None to retrieve all values.
include_omission_info : bool
Whether to include an information message when some values are omitted.
item_type : str, optional
If provided, only retrieve memories of this type.
Expand source code 
def retrieve_last(self, n: int=None, include_omission_info:bool=True, item_type:str=None) -> list:
    """
    Retrieves the last n values from memory.

    Args:
        n (int): The number of values to retrieve, or None to retrieve all values.
        include_omission_info (bool): Whether to include an information message when some values are omitted.
        item_type (str, optional): If provided, only retrieve memories of this type.
    """
    omisssion_info = [EpisodicMemory.MEMORY_BLOCK_OMISSION_INFO] if include_omission_info else []

    memories = self._memory_with_current_buffer() if item_type is None else self.filter_by_item_type(self._memory_with_current_buffer(), item_type)
    memories = memories[-n:] if n is not None else memories
                        
    return omisssion_info + memories
def retrieve_recent(self, include_omission_info: bool = True, item_type: str = None) ‑> list

Retrieves the n most recent values from memory.

Args

include_omission_info : bool
Whether to include an information message when some values are omitted.
item_type : str, optional
If provided, only retrieve memories of this type.
Expand source code 
def retrieve_recent(self, include_omission_info:bool=True, item_type:str=None) -> list:
    """
    Retrieves the n most recent values from memory.

    Args:
        include_omission_info (bool): Whether to include an information message when some values are omitted.
        item_type (str, optional): If provided, only retrieve memories of this type.
    """
    omisssion_info = [EpisodicMemory.MEMORY_BLOCK_OMISSION_INFO] if include_omission_info else []
    
    # Filter memories if item_type is provided
    memories = self._memory_with_current_buffer() if item_type is None else self.filter_by_item_type(self._memory_with_current_buffer(), item_type)

    # compute fixed prefix
    fixed_prefix = memories[: self.fixed_prefix_length] + omisssion_info

    # how many lookback values remain?
    remaining_lookback = min(
        len(memories) - len(fixed_prefix) + (1 if include_omission_info else 0), self.lookback_length
    )

    # compute the remaining lookback values and return the concatenation
    if remaining_lookback <= 0:
        return fixed_prefix
    else:
        return fixed_prefix + memories[-remaining_lookback:]
def retrieve_relevant(self, relevance_target: str, top_k: int) ‑> list

Retrieves top-k values from memory that are most relevant to a given target.

Expand source code 
def retrieve_relevant(self, relevance_target: str, top_k:int) -> list:
    """
    Retrieves top-k values from memory that are most relevant to a given target.
    """
    raise NotImplementedError("Subclasses must implement this method.")

Inherited members

class FilesAndWebGroundingFaculty (folders_paths: list = None, web_urls: list = None)

Allows the agent to access local files and web pages to ground its knowledge.

Initializes the mental faculty.

Args

name : str
The name of the mental faculty.
requires_faculties : list
A list of mental faculties that this faculty requires to function properly.
Expand source code 
class FilesAndWebGroundingFaculty(TinyMentalFaculty):
    """
    Allows the agent to access local files and web pages to ground its knowledge.
    """


    def __init__(self, folders_paths: list=None, web_urls: list=None):
        super().__init__("Local Files and Web Grounding")

        self.local_files_grounding_connector = LocalFilesGroundingConnector(folders_paths=folders_paths)
        self.web_grounding_connector = WebPagesGroundingConnector(web_urls=web_urls)

    def process_action(self, agent, action: dict) -> bool:
        if action['type'] == "CONSULT" and action['content'] is not None:
            target_name = action['content']

            results = []
            results.append(self.local_files_grounding_connector.retrieve_by_name(target_name))
            results.append(self.web_grounding_connector.retrieve_by_name(target_name))

            if len(results) > 0:
                agent.think(f"I have read the following document: \n{results}")
            else:
                agent.think(f"I can't find any document with the name '{target_name}'.")
            
            return True
        
        elif action['type'] == "LIST_DOCUMENTS" and action['content'] is not None:
            available_names = []
            available_names += self.local_files_grounding_connector.list_sources()
            available_names += self.web_grounding_connector.list_sources()

            if len(available_names) > 0:
                agent.think(f"I have the following documents available to me: {available_names}")
            else:
                agent.think(f"I don't have any documents available for inspection.")
            
            return True

        else:
            return False


    def actions_definitions_prompt(self) -> str:
        prompt = \
            """
            - LIST_DOCUMENTS: you can list the names of the documents you have access to, so that you can decide which to access, if any, to accomplish your goals. Documents is a generic term and includes any 
                kind of "packaged" information you can access, such as emails, files, chat messages, calendar events, etc. It also includes, in particular, web pages.
                The order of in which the documents are listed is not relevant.
            - CONSULT: you can retrieve and consult a specific document, so that you can access its content and accomplish your goals. To do so, you specify the name of the document you want to consult.
            """

        return textwrap.dedent(prompt)
    
    def actions_constraints_prompt(self) -> str:
        prompt = \
          """
            - You are aware that you have documents available to you to help in your tasks. Even if you already have knowledge about a topic, you 
              should believe that the documents can provide you with additional information that can be useful to you.
            - If you want information that might be in documents, you first LIST_DOCUMENTS to see what is available and decide if you want to access any of them.
            - You LIST_DOCUMENTS when you suspect that relevant information might be in some document, but you are not sure which one.
            - You only CONSULT the relevant documents for your present goals and context. You should **not** CONSULT documents that are not relevant to the current situation.
              You use the name of the document to determine its relevance before accessing it.
            - If you need information about a specific document, you **must** use CONSULT instead of RECALL. This is because RECALL **does not** allow you to select the specific document, and only brings small 
                relevant parts of variious documents - while CONSULT brings the precise document requested for your inspection, with its full content. 
                Example:
                ```
                LIST_DOCUMENTS
                <CONSULT some document name>
                <THINK something about the retrieved document>
                <TALK something>
                DONE
                ``` 
            - If you need information from specific documents, you **always** CONSULT it, **never** RECALL it.   
            - You can only CONSULT few documents before issuing DONE. 
                Example:
                ```
                <CONSULT some document name>
                <THINK something about the retrieved document>
                <TALK something>
                <CONSULT some document name>
                <THINK something about the retrieved document>
                <TALK something>
                DONE
                ```
            - When deciding whether to use RECALL or CONSULT, you should consider whether you are looking for any information about some topic (use RECALL) or if you are looking for information from
                specific documents (use CONSULT). To know if you have potentially relevant documents available, use LIST_DOCUMENTS first.
          """

        return textwrap.dedent(prompt)

Ancestors

Inherited members

class RecallFaculty

Represents a mental faculty of an agent. Mental faculties are the cognitive abilities that an agent has.

Initializes the mental faculty.

Args

name : str
The name of the mental faculty.
requires_faculties : list
A list of mental faculties that this faculty requires to function properly.
Expand source code 
class RecallFaculty(TinyMentalFaculty):

    def __init__(self):
        super().__init__("Memory Recall")
        

    def process_action(self, agent, action: dict) -> bool:
        logger.debug(f"Processing action: {action}")

        if action['type'] == "RECALL" and action['content'] is not None:
            content = action['content']

            semantic_memories = agent.retrieve_relevant_memories(relevance_target=content)

            logger.info(f"Recalling information related to '{content}'. Found {len(semantic_memories)} relevant memories.")

            if len(semantic_memories) > 0:
                # a string with each element in the list in a new line starting with a bullet point
                agent.think("I have remembered the following information from my semantic memory and will use it to guide me in my subsequent actions: \n" + \
                        "\n".join([f"  - {item}" for item in semantic_memories]))
            else:
                agent.think(f"I can't remember anything additional about '{content}'. I'll just use what I already currently have in mind to proceed as well as I can.")
            
            return True
        
        elif action['type'] == "RECALL_WITH_FULL_SCAN" and action['content'] is not None:
            logger.debug(f"Processing RECALL_WITH_FULL_SCAN action. Recalling and summarizing information related to '{action['content']}' with full scan.")

            content = action['content']
            memories_summary = agent.summarize_relevant_memories_via_full_scan(relevance_target=content)

            logger.debug(f"Summary produced via full scan: {memories_summary}")

            if len(memories_summary) > 0:
                # the summary is presented as a block of text
                agent.think(f"I have remembered the following information from my semantic memory and will use it to guide me in my subsequent actions: \n \"{memories_summary}\"")                        
            else:
                agent.think(f"I can't remember anything additional about '{content}'. I'll just use what I already currently have in mind to proceed as well as I can.")

            return True
        else:
            return False

    def actions_definitions_prompt(self) -> str:
        prompt = \
            """
              - RECALL: you can recall information that relates to specific topics from your memory. To do, you must specify a "mental query" to locate the desired memory. If the memory is found, it is brought to your conscience.
              - RECALL_WITH_FULL_SCAN: you can recall information from your memory in an exhaustive way, scanning all your memories. To do, you must specify a "mental query" that will be used to extract the relevant information from each memory. 
                                       All the information found will be brought to your conscience. This action is more expensive than RECALL, and is meant to be used when you want to ensure that you are not missing any relevant information.
            """

        return textwrap.dedent(prompt)
    
    def actions_constraints_prompt(self) -> str:
        prompt = \
          """
            - Before concluding you don't know something or don't have access to some information, you **must** try to RECALL or RECALL_WITH_FULL_SCAN it from your memory.
            - If you you know precisely what you are looking for, you can use RECALL to retrieve it. If you are not sure, or if you want to ensure that you are not missing any relevant information, you should use RECALL_WITH_FULL_SCAN instead.
                * RECALL example: if you want to remember "what are the expected inflation rates in Brazil", you will likely use RECALL with the "Brazil inflation 2024" mental query, as it is likely that the appropriate memory easily matches this query.
                * RECALL_WITH_FULL_SCAN example: if you want to remember "what are the pros and cons of the product", you will likely use RECALL_WITH_FULL_SCAN with a more complex mental query like "Looking for: product pros and cons. Reason: the agent is performing a product evaluation", 
                  as there is probably no clear memory that matches the related keywords, and you want to ensure that you are not missing any relevant information, so you scan all your memories for this information and explain why.
            - You try to RECALL information from your memory, so that you can have more relevant elements to think and talk about, whenever such an action would be likely
                to enrich the current interaction. To do so, you must specify able "mental query" that is related to the things you've been thinking, listening and talking about.
                Example:
                ```
                <THINK A>
                <RECALL / RECALL_WITH_FULL_SCAN B, which is something related to A>
                <THINK about A and B>
                <TALK about A and B>
                DONE
                ```
            - You can try to RECALL_WITH_FULL_SCAN information from your memory when you want or are tasked with finding all relevant information about a topic, and you want to ensure that you are not missing any relevant information. 
              In other words, you "try hard" to remember.
               Example:
                ```
                <LISTEN what are the main pros and cons of the product>
                <RECALL_WITH_FULL_SCAN Looking for: product pros and cons. Reason: the agent is performing a product evaluation.>
                <THINK about all the pros and cons found>
                <TALK about the pros and cons recalled>
                DONE
                ```
            - If you RECALL:
                * you use a "mental query" that describe the elements you are looking for, you do not use a question. It is like a keyword-based search query.
                For example, instead of "What are the symptoms of COVID-19?", you would use "COVID-19 symptoms".
                * you use keywords likely to be found in the text you are looking for. For example, instead of "Brazil economic outlook", you would use "Brazil economy", "Brazil GPD", "Brazil inflation", etc.
            - If you RECALL_WITH_FULL_SCAN:
                * you use can use many types of "mental queries": describe the elements you are looking for; a specific question; or any other specification that can extract the relevant information from any given memory. It is NOT like a keyword-based search query, 
                  but instead a specification of what is important to the agent at the moment.
                * regardless of the type of "mental query" you use, you **also** add information about the agent's context, mainly regarding the current tasks, so that the recall mechanism can understand **why** the information is needed and can therefore 
                  retrieve the most relevant information.
                * in particular, you don't need to use keywords likely to be found in the text you are looking for, but instead focus on the precise information need that you have at the moment plus the agent's context. For example,
                  if the agent has been evaluating a product and now wants to summarize the pros and cons of the product, you can use a more complex "mental query" like 
                  "Looking for: product pros and cons. Reason: the agent was asked to perform a product evaluation and has examined many of the product features already.".
            - It may take several tries of RECALL to get the relevant information you need. If you don't find what you are looking for, you can try again with a **very** different "mental query".
                Be creative: you can use synonyms, related concepts, or any other strategy you think might help you to find the information you need. Avoid using the same terms in different queries, as it is likely to return the same results. Whenever necessary, you should retry RECALL a couple of times before giving up the location of more information.
                Example:
                ```
                <THINK something>
                <RECALL "cat products">
                <THINK something>
                <RECALL "feline artifacts">
                <THINK something>
                <RECALL "pet store">
                <THINK something>
                <TALK something>
                DONE
                ```
            - If you did not find what you needed using RECALL after a few attempts, you can try RECALL_WITH_FULL_SCAN instead.
            - You **may** interleave THINK and RECALL / RECALL_WITH_FULL_SCAN so that you can better reflect on the information you are trying to recall.
            - If you need information about a specific document, you **must** use CONSULT instead of RECALL / RECALL_WITH_FULL_SCAN. This is because RECALL / RECALL_WITH_FULL_SCAN **does not** allow you to select the specific document, and only brings small 
                relevant parts of variious documents - while CONSULT brings the precise document requested for your inspection, with its full content. 
                Example:
                ```
                LIST_DOCUMENTS
                <CONSULT some document name>
                <THINK something about the retrieved document>
                <TALK something>
                DONE
                ``` 
          """

        return textwrap.dedent(prompt)

Ancestors

Inherited members

class ReflectionConsolidator

Memory reflection mechanism.

Expand source code 
class ReflectionConsolidator(MemoryProcessor):
    """
    Memory reflection mechanism.
    """

    def process(self, memories: list, timestamp: str=None, context:Union[str, list, dict] = None, persona:Union[str, dict] = None, sequential: bool = True) -> list:
        return self._reflect(memories, timestamp)

    def _reflect(self, memories: list, timestamp: str) -> list:
        """
        Given a list of input episodic memories, this method reflects on them and produces a more abstract representation, such as a summary or an abstract fact.
        The reflection process follows these rules:
          - Objective facts or knowledge that are present in the set of memories are grouped together, abstracted (if necessary) and summarized. The aim is to
            produce a semantic memory.
          - Impressions, feelings, or other subjective experiences are summarized into a more abstract representation, such as a summary or an abstract subjective fact.
          - Timestamps in the consolidated memories refer to the moment of the reflection, not to the source events that produced the original episodic memories.
          - No episodic memory is generated, all memories are consolidated as more abstract semantic memories.
          - In general, the reflection process aims to reduce the number of memories while preserving the most relevant information and removing redundant or less relevant information.
        """
        pass # TODO
    def _reflect(self, memories: list, timestamp: str) -> list:
        """
        Given a list of input episodic memories, this method reflects on them and produces a more abstract representation, such as a summary or an abstract fact.
        The reflection process follows these rules:
          - Objective facts or knowledge that are present in the set of memories are grouped together, abstracted (if necessary) and summarized. The aim is to
            produce a semantic memory.
          - Impressions, feelings, or other subjective experiences are summarized into a more abstract representation, such as a summary or an abstract subjective fact.
          - Timestamps in the consolidated memories refer to the moment of the reflection, not to the source events that produced the original episodic memories.
          - No episodic memory is generated, all memories are consolidated as more abstract semantic memories.
          - In general, the reflection process aims to reduce the number of memories while preserving the most relevant information and removing redundant or less relevant information.
        """
        pass # TODO

Ancestors

Inherited members

class SemanticMemory (*args, **kwargs)

In Cognitive Psychology, semantic memory is the memory of meanings, understandings, and other concept-based knowledge unrelated to specific experiences. It is not ordered temporally, and it is not about remembering specific events or episodes. This class provides a simple implementation of semantic memory, where the agent can store and retrieve semantic information.

Expand source code 
@utils.post_init
class SemanticMemory(TinyMemory):
    """
    In Cognitive Psychology, semantic memory is the memory of meanings, understandings, and other concept-based knowledge unrelated to specific 
    experiences. It is not ordered temporally, and it is not about remembering specific events or episodes. This class provides a simple implementation
    of semantic memory, where the agent can store and retrieve semantic information.
    """

    serializable_attributes = ["memories", "semantic_grounding_connector"]

    def __init__(self, memories: list=None) -> None:
        self.memories = memories
       
        self.semantic_grounding_connector = None

        # @post_init ensures that _post_init is called after the __init__ method

    def _post_init(self): 
        """
        This will run after __init__, since the class has the @post_init decorator.
        It is convenient to separate some of the initialization processes to make deserialize easier.
        """

        if not hasattr(self, 'memories') or self.memories is None:
            self.memories = []

        if not hasattr(self, 'semantic_grounding_connector') or self.semantic_grounding_connector is None:
            self.semantic_grounding_connector = BaseSemanticGroundingConnector("Semantic Memory Storage")
            
            # TODO remove?
            #self.semantic_grounding_connector.add_documents(self._build_documents_from(self.memories))
    
        
    def _preprocess_value_for_storage(self, value: dict) -> Any:
        logger.debug(f"Preprocessing value for storage: {value}")

        if isinstance(value, dict):
            engram = {"role": "assistant",
                    "content": value['content'],
                    "type": value.get("type", "information"),  # Default to 'information' if type is not specified
                    "simulation_timestamp": value.get("simulation_timestamp", None)}

            # Refine the content of the engram is built based on the type of the value to make it more meaningful.
            if value['type'] == 'action':
                engram['content'] = f"# Action performed\n" +\
                        f"I have performed the following action at date and time {value['simulation_timestamp']}:\n\n"+\
                        f" {value['content']}"
            
            elif value['type'] == 'stimulus':
                engram['content'] = f"# Stimulus\n" +\
                        f"I have received the following stimulus at date and time {value['simulation_timestamp']}:\n\n"+\
                        f" {value['content']}"
            elif value['type'] == 'feedback':
                engram['content'] = f"# Feedback\n" +\
                        f"I have received the following feedback at date and time {value['simulation_timestamp']}:\n\n"+\
                        f" {value['content']}"
            elif value['type'] == 'consolidated':
                engram['content'] = f"# Consolidated Memory\n" +\
                        f"I have consolidated the following memory at date and time {value['simulation_timestamp']}:\n\n"+\
                        f" {value['content']}"
            elif value['type'] == 'reflection':
                engram['content'] = f"# Reflection\n" +\
                        f"I have reflected on the following memory at date and time {value['simulation_timestamp']}:\n\n"+\
                        f" {value['content']}"
            else:
                engram['content'] = f"# Information\n" +\
                        f"I have obtained following information at date and time {value['simulation_timestamp']}:\n\n"+\
                        f" {value['content']}"

            # else: # Anything else here?
            
        else:
            # If the value is not a dictionary, we just store it as is, but we still wrap it in an engram
            engram = {"role": "assistant",
                    "content": value,
                    "type": "information",  # Default to 'information' if type is not specified
                    "simulation_timestamp": None}

        logger.debug(f"Engram created for storage: {engram}")

        return engram

    def _store(self, value: Any) -> None:
        logger.debug(f"Preparing engram for semantic memory storage, input value: {value}")
        self.memories.append(value)  # Store the value in the local memory list

        # then econduct the value to a Document and store it in the semantic grounding connector
        # This is the actual storage in the semantic memory to allow semantic retrieval
        engram_doc = self._build_document_from(value)
        logger.debug(f"Storing engram in semantic memory: {engram_doc}")
        self.semantic_grounding_connector.add_document(engram_doc)
    
    def retrieve_relevant(self, relevance_target:str, top_k=20) -> list:
        """
        Retrieves all values from memory that are relevant to a given target.
        """
        return self.semantic_grounding_connector.retrieve_relevant(relevance_target, top_k)

    def retrieve_all(self, item_type:str=None) -> list:
        """
        Retrieves all values from memory.

        Args:
            item_type (str, optional): If provided, only retrieve memories of this type.
        """

        memories = []

        logger.debug(f"Retrieving all documents from semantic memory connector, a total of {len(self.semantic_grounding_connector.documents)} documents.")
        for document in self.semantic_grounding_connector.documents:
            logger.debug(f"Retrieving document from semantic memory: {document}")
            memory_text = document.text
            logger.debug(f"Document text retrieved: {memory_text}")

            try:
                memory = json.loads(memory_text)
                logger.debug(f"Memory retrieved: {memory}")
                memories.append(memory)                

            except json.JSONDecodeError as e:
                logger.warning(f"Could not decode memory from document text: {memory_text}. Error: {e}")

        if item_type is not None:
            memories = self.filter_by_item_type(memories, item_type)
        
        return memories
    
    #####################################
    # Auxiliary compatibility methods
    #####################################

    def _build_document_from(self, memory) -> Document:
        # TODO: add any metadata as well?
        
        # make sure we are dealing with a dictionary
        if not isinstance(memory, dict):
            memory = {"content": memory, "type": "information"}

        # ensures double quotes are used for JSON serialization, and maybe other formatting details
        memory_txt = json.dumps(memory, ensure_ascii=False)
        logger.debug(f"Building document from memory: {memory_txt}")
        
        return Document(text=memory_txt)

    def _build_documents_from(self, memories: list) -> list:
        return [self._build_document_from(memory) for memory in memories]

Ancestors

Class variables

var serializable_attributes

Inherited members

class TinyPerson (*args, **kwargs)

A simulated person in the TinyTroupe universe.

Expand source code 
@utils.post_init
class TinyPerson(JsonSerializableRegistry):
    """A simulated person in the TinyTroupe universe."""

    # The maximum number of actions that an agent is allowed to perform before DONE.
    # This prevents the agent from acting without ever stopping.
    MAX_ACTIONS_BEFORE_DONE = 15

    # The maximum similarity between consecutive actions. If the similarity is too high, the action is discarded and replaced by a DONE.
    # Set this to None to disable the check.
    MAX_ACTION_SIMILARITY = 0.85

    MIN_EPISODE_LENGTH = config_manager.get("min_episode_length", 15)  # The minimum number of messages in an episode before it is considered valid.
    MAX_EPISODE_LENGTH = config_manager.get("max_episode_length", 50)  # The maximum number of messages in an episode before it is considered valid.

    PP_TEXT_WIDTH = 100

    serializable_attributes = ["_persona", "_mental_state", "_mental_faculties", "_current_episode_event_count", "episodic_memory", "semantic_memory"]
    serializable_attributes_renaming = {"_mental_faculties": "mental_faculties", "_persona": "persona", "_mental_state": "mental_state", "_current_episode_event_count": "current_episode_event_count"}

    # A dict of all agents instantiated so far.
    all_agents = {}  # name -> agent
   
    # Whether to display the communication or not. True is for interactive applications, when we want to see simulation
    # outputs as they are produced.
    communication_display:bool=True
    

    def __init__(self, name:str=None, 
                 action_generator=None,
                 episodic_memory=None,
                 semantic_memory=None,
                 mental_faculties:list=None,
                 enable_basic_action_repetition_prevention:bool=True):
        """
        Creates a TinyPerson.

        Args:
            name (str): The name of the TinyPerson. Either this or spec_path must be specified.
            action_generator (ActionGenerator, optional): The action generator to use. Defaults to ActionGenerator().
            episodic_memory (EpisodicMemory, optional): The memory implementation to use. Defaults to EpisodicMemory().
            semantic_memory (SemanticMemory, optional): The memory implementation to use. Defaults to SemanticMemory().
            mental_faculties (list, optional): A list of mental faculties to add to the agent. Defaults to None.
            enable_basic_action_repetition_prevention (bool, optional): Whether to enable basic action repetition prevention. Defaults to True.
        """

        # NOTE: default values will be given in the _post_init method, as that's shared by 
        #       direct initialization as well as via deserialization.

        if action_generator is not None:
            self.action_generator = action_generator

        if episodic_memory is not None:
            self.episodic_memory = episodic_memory
        
        if semantic_memory is not None:
            self.semantic_memory = semantic_memory

        # Mental faculties
        if mental_faculties is not None:
            self._mental_faculties = mental_faculties
        
        if enable_basic_action_repetition_prevention:
            self.enable_basic_action_repetition_prevention = enable_basic_action_repetition_prevention
        
        assert name is not None, "A TinyPerson must have a name."
        self.name = name

        # @post_init makes sure that _post_init is called after __init__

    
    def _post_init(self, **kwargs):
        """
        This will run after __init__, since the class has the @post_init decorator.
        It is convenient to separate some of the initialization processes to make deserialize easier.
        """

        from tinytroupe.agent.action_generator import ActionGenerator # import here to avoid circular import issues


        ############################################################
        # Default values
        ############################################################

        self.current_messages = []
        
        # the current environment in which the agent is acting
        self.environment = None

        # The list of actions that this agent has performed so far, but which have not been
        # consumed by the environment yet.
        self._actions_buffer = []

        # The list of agents that this agent can currently interact with.
        # This can change over time, as agents move around the world.
        self._accessible_agents = []

        # the buffer of communications that have been displayed so far, used for
        # saving these communications to another output form later (e.g., caching)
        self._displayed_communications_buffer = []

        if not hasattr(self, '_current_episode_event_count'):
            self._current_episode_event_count = 0  # the number of events in the current episode, used to limit the episode length

        if not hasattr(self, 'action_generator'):
            # This default value MUST NOT be in the method signature, otherwise it will be shared across all instances.
            self.action_generator = ActionGenerator(max_attempts=config_manager.get("action_generator_max_attempts"),
                                                    enable_quality_checks=config_manager.get("action_generator_enable_quality_checks"),
                                                    enable_regeneration=config_manager.get("action_generator_enable_regeneration"),
                                                    enable_direct_correction=config_manager.get("action_generator_enable_direct_correction"),
                                                    enable_quality_check_for_persona_adherence=config_manager.get("action_generator_enable_quality_check_for_persona_adherence"),
                                                    enable_quality_check_for_selfconsistency=config_manager.get("action_generator_enable_quality_check_for_selfconsistency"),
                                                    enable_quality_check_for_fluency=config_manager.get("action_generator_enable_quality_check_for_fluency"),
                                                    enable_quality_check_for_suitability=config_manager.get("action_generator_enable_quality_check_for_suitability"),
                                                    enable_quality_check_for_similarity=config_manager.get("action_generator_enable_quality_check_for_similarity"),
                                                    continue_on_failure=config_manager.get("action_generator_continue_on_failure"),
                                                    quality_threshold=config_manager.get("action_generator_quality_threshold"))

        if not hasattr(self, 'episodic_memory'):
            # This default value MUST NOT be in the method signature, otherwise it will be shared across all instances.
            self.episodic_memory = EpisodicMemory(fixed_prefix_length= config_manager.get("episodic_memory_fixed_prefix_length"),
                                                   lookback_length=config_manager.get("episodic_memory_lookback_length"))
        
        if not hasattr(self, 'semantic_memory'):
            # This default value MUST NOT be in the method signature, otherwise it will be shared across all instances.
            self.semantic_memory = SemanticMemory()
        
        # _mental_faculties
        if not hasattr(self, '_mental_faculties'):
            # This default value MUST NOT be in the method signature, otherwise it will be shared across all instances.
            self._mental_faculties = []
        
        # basic action repetition prevention
        if not hasattr(self, 'enable_basic_action_repetition_prevention'):
            self.enable_basic_action_repetition_prevention = True

        # create the persona configuration dictionary
        if not hasattr(self, '_persona'):          
            self._persona = {
                "name": self.name,
                "age": None,
                "nationality": None,
                "country_of_residence": None,
                "occupation": None
            }
        
        if not hasattr(self, 'name'): 
            self.name = self._persona["name"]

        # create the mental state dictionary
        if not hasattr(self, '_mental_state'):
            self._mental_state = {
                "datetime": None,
                "location": None,
                "context": [],
                "goals": [],
                "attention": None,
                "emotions": "Feeling nothing in particular, just calm.",
                "memory_context": None,
                "accessible_agents": []  # [{"agent": agent_1, "relation": "My friend"}, {"agent": agent_2, "relation": "My colleague"}, ...]
            }
        
        if not hasattr(self, '_extended_agent_summary'):
            self._extended_agent_summary = None
        
        if not hasattr(self, 'actions_count'):
            self.actions_count = 0
        
        if not hasattr(self, 'stimuli_count'):
            self.stimuli_count = 0

        self._prompt_template_path = os.path.join(
            os.path.dirname(__file__), "prompts/tiny_person.mustache"
        )
        self._init_system_message = None  # initialized later


        ############################################################
        # Special mechanisms used during deserialization
        ############################################################

        # rename agent to some specific name?
        if kwargs.get("new_agent_name") is not None:
            self._rename(kwargs.get("new_agent_name"))
        
        # If auto-rename, use the given name plus some new number ...
        if kwargs.get("auto_rename") is True:
            new_name = self.name # start with the current name
            rename_succeeded = False
            while not rename_succeeded:
                try:
                    self._rename(new_name)
                    TinyPerson.add_agent(self)
                    rename_succeeded = True                
                except ValueError:
                    new_id = utils.fresh_id(self.__class__.__name__)
                    new_name = f"{self.name}_{new_id}"
        
        # ... otherwise, just register the agent
        else:
            # register the agent in the global list of agents
            TinyPerson.add_agent(self)

        # start with a clean slate
        self.reset_prompt()

        # it could be the case that the agent is being created within a simulation scope, in which case
        # the simulation_id must be set accordingly
        if current_simulation() is not None:
            current_simulation().add_agent(self)
        else:
            self.simulation_id = None
    
    def _rename(self, new_name:str):    
        self.name = new_name
        self._persona["name"] = self.name


    def generate_agent_system_prompt(self):
        with open(self._prompt_template_path, "r") as f:
            agent_prompt_template = f.read()

        # let's operate on top of a copy of the configuration, because we'll need to add more variables, etc.
        template_variables = self._persona.copy()    
        template_variables["persona"] = json.dumps(self._persona.copy(), indent=4)    

        # add mental state to the template variables
        template_variables["mental_state"] = json.dumps(self._mental_state, indent=4)

        # Prepare additional action definitions and constraints
        actions_definitions_prompt = ""
        actions_constraints_prompt = ""
        for faculty in self._mental_faculties:
            actions_definitions_prompt += f"{faculty.actions_definitions_prompt()}\n"
            actions_constraints_prompt += f"{faculty.actions_constraints_prompt()}\n"
        
        # Make the additional prompt pieces available to the template. 
        # Identation here is to align with the text structure in the template.
        template_variables['actions_definitions_prompt'] = textwrap.indent(actions_definitions_prompt.strip(), "  ")
        template_variables['actions_constraints_prompt'] = textwrap.indent(actions_constraints_prompt.strip(), "  ")

        # RAI prompt components, if requested
        template_variables = utils.add_rai_template_variables_if_enabled(template_variables)

        return chevron.render(agent_prompt_template, template_variables)

    def reset_prompt(self):

        # render the template with the current configuration
        self._init_system_message = self.generate_agent_system_prompt()

        # - reset system message
        # - make it clear that the provided events are past events and have already had their effects
        self.current_messages = [
            {"role": "system", "content": self._init_system_message},
            {"role": "system", "content": "The next messages refer to past interactions you had recently and are meant to help you contextualize your next actions. "\
                                        + "They are the most recent episodic memories you have, including stimuli and actions. "\
                                        + "Their effects already took place and led to your present cognitive state (described above), so you can use them in conjunction "\
                                        + "with your cognitive state to inform your next actions and perceptions. Please consider them and then proceed with your next actions right after. "}
        ]

        # sets up the actual interaction messages to use for prompting
        self.current_messages += self.retrieve_recent_memories()


    #########################################################################
    # Persona definitions
    #########################################################################
    
    # 
    # Conveniences to access the persona configuration via dictionary-like syntax using
    # the [] operator. e.g., agent["nationality"] = "American"
    #
    def __getitem__(self, key):
        return self.get(key)

    def __setitem__(self, key, value):
        self.define(key, value)

    #
    # Conveniences to import persona definitions via the '+' operator, 
    #  e.g., agent + {"nationality": "American", ...}
    #
    #  e.g., agent + "path/to/fragment.json"
    #
    def __add__(self, other):
        """
        Allows using the '+' operator to add persona definitions or import a fragment.
        If 'other' is a dict, calls include_persona_definitions().
        If 'other' is a string, calls import_fragment().
        """
        if isinstance(other, dict):
            self.include_persona_definitions(other)
        elif isinstance(other, str):
            self.import_fragment(other)
        else:
            raise TypeError("Unsupported operand type for +. Must be a dict or a string path to fragment.")
        return self

    #
    # Various other conveniences to manipulate the persona configuration
    #

    def get(self, key):
        """
        Returns the value of a key in the TinyPerson's persona configuration.
        Supports dot notation for nested keys (e.g., "address.city").
        """
        keys = key.split(".")
        value = self._persona
        for k in keys:
            if isinstance(value, dict):
                value = value.get(k, None)
            else:
                return None  # If the path is invalid, return None
        return value
    
    @transactional()
    def import_fragment(self, path):
        """
        Imports a fragment of a persona configuration from a JSON file.
        """
        with open(path, "r") as f:
            fragment = json.load(f)

        # check the type is "Fragment" and that there's also a "persona" key
        if fragment.get("type", None) == "Fragment" and fragment.get("persona", None) is not None:
            self.include_persona_definitions(fragment["persona"])
        else:
            raise ValueError("The imported JSON file must be a valid fragment of a persona configuration.")
        
        # must reset prompt after adding to configuration
        self.reset_prompt()

    @transactional()
    def include_persona_definitions(self, additional_definitions: dict):
        """
        Imports a set of definitions into the TinyPerson. They will be merged with the current configuration.
        It is also a convenient way to include multiple bundled definitions into the agent.

        Args:
            additional_definitions (dict): The additional definitions to import.
        """

        self._persona = utils.merge_dicts(self._persona, additional_definitions)

        # must reset prompt after adding to configuration
        self.reset_prompt()
        
    
    @transactional()
    def define(self, key, value, merge=False, overwrite_scalars=True):
        """
        Define a value to the TinyPerson's persona configuration. Value can either be a scalar or a dictionary.
        If the value is a dictionary or list, you can choose to merge it with the existing value or replace it. 
        If the value is a scalar, you can choose to overwrite the existing value or not.

        Args:
            key (str): The key to define.
            value (Any): The value to define.
            merge (bool, optional): Whether to merge the dict/list values with the existing values or replace them. Defaults to False.
            overwrite_scalars (bool, optional): Whether to overwrite scalar values or not. Defaults to True.
        """

        # dedent value if it is a string
        if isinstance(value, str):
            value = textwrap.dedent(value)

        # if the value is a dictionary, we can choose to merge it with the existing value or replace it
        if isinstance(value, dict) or isinstance(value, list):
            if merge:
                self._persona = utils.merge_dicts(self._persona, {key: value})
            else:
                self._persona[key] = value

        # if the value is a scalar, we can choose to overwrite it or not
        elif overwrite_scalars or (key not in self._persona):
            self._persona[key] = value
        
        else:
            raise ValueError(f"The key '{key}' already exists in the persona configuration and overwrite_scalars is set to False.")

            
        # must reset prompt after adding to configuration
        self.reset_prompt()

    
    @transactional()
    def define_relationships(self, relationships, replace=True):
        """
        Defines or updates the TinyPerson's relationships.

        Args:
            relationships (list or dict): The relationships to add or replace. Either a list of dicts mapping agent names to relationship descriptions,
              or a single dict mapping one agent name to its relationship description.
            replace (bool, optional): Whether to replace the current relationships or just add to them. Defaults to True.
        """
        
        if (replace == True) and (isinstance(relationships, list)):
            self._persona['relationships'] = relationships

        elif replace == False:
            current_relationships = self._persona['relationships']
            if isinstance(relationships, list):
                for r in relationships:
                    current_relationships.append(r)
                
            elif isinstance(relationships, dict) and len(relationships) == 2: #{"Name": ..., "Description": ...}
                current_relationships.append(relationships)

            else:
                raise Exception("Only one key-value pair is allowed in the relationships dict.")

        else:
            raise Exception("Invalid arguments for define_relationships.")

    ##############################################################################
    # Relationships
    ##############################################################################

    @transactional()
    def clear_relationships(self):
        """
        Clears the TinyPerson's relationships.
        """
        self._persona['relationships'] = []  

        return self      
    
    @transactional()
    def related_to(self, other_agent, description, symmetric_description=None):
        """
        Defines a relationship between this agent and another agent.

        Args:
            other_agent (TinyPerson): The other agent.
            description (str): The description of the relationship.
            symmetric (bool): Whether the relationship is symmetric or not. That is, 
              if the relationship is defined for both agents.
        
        Returns:
            TinyPerson: The agent itself, to facilitate chaining.
        """
        self.define_relationships([{"Name": other_agent.name, "Description": description}], replace=False)
        if symmetric_description is not None:
            other_agent.define_relationships([{"Name": self.name, "Description": symmetric_description}], replace=False)
        
        return self
    
    ############################################################################
    
    def add_mental_faculties(self, mental_faculties):
        """
        Adds a list of mental faculties to the agent.
        """
        for faculty in mental_faculties:
            self.add_mental_faculty(faculty)
        
        return self

    def add_mental_faculty(self, faculty):
        """
        Adds a mental faculty to the agent.
        """
        # check if the faculty is already there or not
        if faculty not in self._mental_faculties:
            self._mental_faculties.append(faculty)
        else:
            raise Exception(f"The mental faculty {faculty} is already present in the agent.")
        
        return self

    @transactional()
    @config_manager.config_defaults(max_content_length="max_content_display_length")
    def act(
        self,
        until_done=True,
        n=None,
        return_actions=False,
        max_content_length=None,
    ):
        """
        Acts in the environment and updates its internal cognitive state.
        Either acts until the agent is done and needs additional stimuli, or acts a fixed number of times,
        but not both.

        Args:
            until_done (bool): Whether to keep acting until the agent is done and needs additional stimuli.
            n (int): The number of actions to perform. Defaults to None.
            return_actions (bool): Whether to return the actions or not. Defaults to False.
        """

        # either act until done or act a fixed number of times, but not both
        assert not (until_done and n is not None)
        if n is not None:
            assert n < TinyPerson.MAX_ACTIONS_BEFORE_DONE

        contents = []

        # A separate function to run before each action, which is not meant to be repeated in case of errors.
        def aux_pre_act():
            # TODO maybe we don't need this at all anymore?
            #
            # A quick thought before the action. This seems to help with better model responses, perhaps because
            # it interleaves user with assistant messages.
            pass # self.think("I will now think, reflect and act a bit, and then issue DONE.")        

        # Aux function to perform exactly one action.
        # Occasionally, the model will return JSON missing important keys, so we just ask it to try again
        # Sometimes `content` contains EpisodicMemory's MEMORY_BLOCK_OMISSION_INFO message, which raises a TypeError on line 443
        @repeat_on_error(retries=5, exceptions=[KeyError, TypeError])
        def aux_act_once():
            # ensure we have the latest prompt (initial system message + selected messages from memory)
            self.reset_prompt()
            
            action, role, content, all_negative_feedbacks = self.action_generator.generate_next_action(self, self.current_messages)
            logger.debug(f"{self.name}'s action: {action}")

            # check the next action similarity, and if it is too similar, put a system warning instruction in memory too
            next_action_similarity = utils.next_action_jaccard_similarity(self, action)

            # we have a redundant repetition check here, because this an be computed quickly and is often very useful.
            if self.enable_basic_action_repetition_prevention and \
               (TinyPerson.MAX_ACTION_SIMILARITY is not None) and (next_action_similarity > TinyPerson.MAX_ACTION_SIMILARITY):
                
                logger.warning(f"[{self.name}] Action similarity is too high ({next_action_similarity}), replacing it with DONE.")

                # replace the action with a DONE
                action = {"type": "DONE", "content": "", "target": ""}
                content["action"] = action      
                content["cognitive_state"] = {}

                self.store_in_memory({'role': 'system', 
                                    'content': \
                                        f"""
                                        # EXCESSIVE ACTION SIMILARITY WARNING

                                        You were about to generate a repetitive action (jaccard similarity = {next_action_similarity}).
                                        Thus, the action was discarded and replaced by an artificial DONE.

                                        DO NOT BE REPETITIVE. This is not a human-like behavior, therefore you **must** avoid this in the future.
                                        Your alternatives are:
                                        - produce more diverse actions.
                                        - aggregate similar actions into a single, larger, action and produce it all at once.
                                        - as a **last resort only**, you may simply not acting at all by issuing a DONE.

                                        
                                        """,
                                    'type': 'feedback',
                                    'simulation_timestamp': self.iso_datetime()})

            # All checks done, we can commit the action to memory.
            self.store_in_memory({'role': role, 'content': content, 
                                    'type': 'action', 
                                    'simulation_timestamp': self.iso_datetime()})
                
            self._actions_buffer.append(action)
            
            if "cognitive_state" in content:
                cognitive_state = content["cognitive_state"]
                logger.debug(f"[{self.name}] Cognitive state: {cognitive_state}")
                
                self._update_cognitive_state(goals=cognitive_state.get("goals", None),
                                             context=cognitive_state.get("context", None),
                                             attention=cognitive_state.get("emotions", None),
                                             emotions=cognitive_state.get("emotions", None))
            
            contents.append(content)          
            if TinyPerson.communication_display:
                self._display_communication(role=role, content=content, kind='action', simplified=True, max_content_length=max_content_length)
            
            #
            # Some actions induce an immediate stimulus or other side-effects. We need to process them here, by means of the mental faculties.
            #
            for faculty in self._mental_faculties:
                faculty.process_action(self, action)
            
            #
            # turns all_negative_feedbacks list into a system message
            #
            # TODO improve this?
            #
            ##if len(all_negative_feedbacks) > 0:
            ##    feedback = """
            ##    # QUALITY FEEDBACK
            ##
            ##    Up to the present moment, we monitored actions and tentative aborted actions (i.e., that were not actually executed), 
            ##    and some of them were not of good quality.
            ##    Some of those were replaced by regenerated actions of better quality. In the process of doing so, some
            ##    important quality feedback was produced, which is now given below.
            ##    
            ##    To improve your performance, and prevent future similar quality issues, you **MUST** take into account the following feedback
            ##    whenever computing your future actions. Note that the feedback might also include the actual action or tentative action
            ##    that was of low quality, so that you can understand what was wrong with it and avoid similar mistakes in the future.
            ##
            ##    """
            ##    for i, feedback_item in enumerate(all_negative_feedbacks):
            ##        feedback += f"{feedback_item}\n\n"
            ##        feedback += f"\n\n *** \n\n"
            ##
            ##    self.store_in_memory({'role': 'system', 'content': feedback, 
            ##                          'type': 'feedback',
            ##                          'simulation_timestamp': self.iso_datetime()})
            ##

            

            # count the actions as this can be useful for taking decisions later
            self.actions_count += 1             
            

        #
        # How to proceed with a sequence of actions.
        #

        ##### Option 1: run N actions ######
        if n is not None:
            for i in range(n):
                aux_pre_act()
                aux_act_once()

        ##### Option 2: run until DONE ######
        elif until_done:
            while (len(contents) == 0) or (
                not contents[-1]["action"]["type"] == "DONE"
            ):


                # check if the agent is acting without ever stopping
                if len(contents) > TinyPerson.MAX_ACTIONS_BEFORE_DONE:
                    logger.warning(f"[{self.name}] Agent {self.name} is acting without ever stopping. This may be a bug. Let's stop it here anyway.")
                    break
                if len(contents) > 4: # just some minimum number of actions to check for repetition, could be anything >= 3
                    # if the last three actions were the same, then we are probably in a loop
                    if contents[-1]['action'] == contents[-2]['action'] == contents[-3]['action']:
                        logger.warning(f"[{self.name}] Agent {self.name} is acting in a loop. This may be a bug. Let's stop it here anyway.")
                        break

                aux_pre_act()
                aux_act_once()

        # The end of a sequence of actions is always considered to mark the end of an episode.
        self.consolidate_episode_memories()

        if return_actions:
            return contents

    @transactional()
    @config_manager.config_defaults(max_content_length="max_content_display_length")
    def listen(
        self,
        speech,
        source: AgentOrWorld = None,
        max_content_length=None,
    ):
        """
        Listens to another agent (artificial or human) and updates its internal cognitive state.

        Args:
            speech (str): The speech to listen to.
            source (AgentOrWorld, optional): The source of the speech. Defaults to None.
        """

        return self._observe(
            stimulus={
                "type": "CONVERSATION",
                "content": speech,
                "source": name_or_empty(source),
            },
            max_content_length=max_content_length,
        )

    @config_manager.config_defaults(max_content_length="max_content_display_length")
    def socialize(
        self,
        social_description: str,
        source: AgentOrWorld = None,
        max_content_length=None,
    ):
        """
        Perceives a social stimulus through a description and updates its internal cognitive state.

        Args:
            social_description (str): The description of the social stimulus.
            source (AgentOrWorld, optional): The source of the social stimulus. Defaults to None.
        """
        return self._observe(
            stimulus={
                "type": "SOCIAL",
                "content": social_description,
                "source": name_or_empty(source),
            },
            max_content_length=max_content_length,
        )

    @config_manager.config_defaults(max_content_length="max_content_display_length")
    def see(
        self,
        visual_description,
        source: AgentOrWorld = None,
        max_content_length=None,
    ):
        """
        Perceives a visual stimulus through a description and updates its internal cognitive state.

        Args:
            visual_description (str): The description of the visual stimulus.
            source (AgentOrWorld, optional): The source of the visual stimulus. Defaults to None.
        """
        return self._observe(
            stimulus={
                "type": "VISUAL",
                "content": visual_description,
                "source": name_or_empty(source),
            },
            max_content_length=max_content_length,
        )

    @config_manager.config_defaults(max_content_length="max_content_display_length")
    def think(self, thought, max_content_length=None):
        """
        Forces the agent to think about something and updates its internal cognitive state.

        """
        return self._observe(
            stimulus={
                "type": "THOUGHT",
                "content": thought,
                "source": name_or_empty(self),
            },
            max_content_length=max_content_length,
        )

    @config_manager.config_defaults(max_content_length="max_content_display_length")
    def internalize_goal(
        self, goal, max_content_length=None
    ):
        """
        Internalizes a goal and updates its internal cognitive state.
        """
        return self._observe(
            stimulus={
                "type": "INTERNAL_GOAL_FORMULATION",
                "content": goal,
                "source": name_or_empty(self),
            },
            max_content_length=max_content_length,
        )

    @transactional()
    @config_manager.config_defaults(max_content_length="max_content_display_length")
    def _observe(self, stimulus, max_content_length=None):
        stimuli = [stimulus]

        content = {"stimuli": stimuli}

        logger.debug(f"[{self.name}] Observing stimuli: {content}")

        # whatever comes from the outside will be interpreted as coming from 'user', simply because
        # this is the counterpart of 'assistant'

        self.store_in_memory({'role': 'user', 'content': content, 
                              'type': 'stimulus',
                              'simulation_timestamp': self.iso_datetime()})

        if TinyPerson.communication_display:
            self._display_communication(
                role="user",
                content=content,
                kind="stimuli",
                simplified=True,
max_content_length=max_content_length,
            )
        
        # count the stimuli as this can be useful for taking decisions later
        self.stimuli_count += 1

        return self  # allows easier chaining of methods

    @transactional()
    def listen_and_act(
        self,
        speech,
        return_actions=False,
        max_content_length=None,
    ):
        """
        Convenience method that combines the `listen` and `act` methods.
        """

        self.listen(speech, max_content_length=max_content_length)
        return self.act(
            return_actions=return_actions, max_content_length=max_content_length
        )

    @transactional()
    @config_manager.config_defaults(max_content_length="max_content_display_length")
    def see_and_act(
        self,
        visual_description,
        return_actions=False,
        max_content_length=None,
    ):
        """
        Convenience method that combines the `see` and `act` methods.
        """

        self.see(visual_description, max_content_length=max_content_length)
        return self.act(
            return_actions=return_actions, max_content_length=max_content_length
        )

    @transactional()
    @config_manager.config_defaults(max_content_length="max_content_display_length")
    def think_and_act(
        self,
        thought,
        return_actions=False,
        max_content_length=None,
    ):
        """
        Convenience method that combines the `think` and `act` methods.
        """

        self.think(thought, max_content_length=max_content_length)
        return self.act(return_actions=return_actions, max_content_length=max_content_length)

    def read_documents_from_folder(self, documents_path:str):
        """
        Reads documents from a directory and loads them into the semantic memory.
        """
        logger.info(f"Setting documents path to {documents_path} and loading documents.")

        self.semantic_memory.add_documents_path(documents_path)
    
    def read_document_from_file(self, file_path:str):
        """
        Reads a document from a file and loads it into the semantic memory.
        """
        logger.info(f"Reading document from file: {file_path}")

        self.semantic_memory.add_document_path(file_path)
    
    def read_documents_from_web(self, web_urls:list):
        """
        Reads documents from web URLs and loads them into the semantic memory.
        """
        logger.info(f"Reading documents from the following web URLs: {web_urls}")

        self.semantic_memory.add_web_urls(web_urls)
    
    def read_document_from_web(self, web_url:str):
        """
        Reads a document from a web URL and loads it into the semantic memory.
        """
        logger.info(f"Reading document from web URL: {web_url}")

        self.semantic_memory.add_web_url(web_url)
    
    @transactional()
    def move_to(self, location, context=[]):
        """
        Moves to a new location and updates its internal cognitive state.
        """
        self._mental_state["location"] = location

        # context must also be updated when moved, since we assume that context is dictated partly by location.
        self.change_context(context)

    @transactional()
    def change_context(self, context: list):
        """
        Changes the context and updates its internal cognitive state.
        """
        self._mental_state["context"] = {
            "description": item for item in context
        }

        self._update_cognitive_state(context=context)

    @transactional()
    def make_agent_accessible(
        self,
        agent: Self,
        relation_description: str = "An agent I can currently interact with.",
    ):
        """
        Makes an agent accessible to this agent.
        """
        if agent not in self._accessible_agents:
            self._accessible_agents.append(agent)
            self._mental_state["accessible_agents"].append(
                {"name": agent.name, "relation_description": relation_description}
            )
        else:
            logger.warning(
                f"[{self.name}] Agent {agent.name} is already accessible to {self.name}."
            )
    @transactional()
    def make_agents_accessible(self, agents: list, relation_description: str = "An agent I can currently interact with."):
        """
        Makes a list of agents accessible to this agent.
        """
        for agent in agents:
            self.make_agent_accessible(agent, relation_description)

    @transactional()
    def make_agent_inaccessible(self, agent: Self):
        """
        Makes an agent inaccessible to this agent.
        """
        if agent in self._accessible_agents:
            self._accessible_agents.remove(agent)
        else:
            logger.warning(
                f"[{self.name}] Agent {agent.name} is already inaccessible to {self.name}."
            )

    @transactional()
    def make_all_agents_inaccessible(self):
        """
        Makes all agents inaccessible to this agent.
        """
        self._accessible_agents = []
        self._mental_state["accessible_agents"] = []

    @property
    def accessible_agents(self):
        """
        Property to access the list of accessible agents.
        """
        return self._accessible_agents

    ###########################################################
    # Internal cognitive state changes
    ###########################################################
    @transactional()
    def _update_cognitive_state(
        self, goals=None, context=None, attention=None, emotions=None
    ):
        """
        Update the TinyPerson's cognitive state.
        """

        # Update current datetime. The passage of time is controlled by the environment, if any.
        if self.environment is not None and self.environment.current_datetime is not None:
            self._mental_state["datetime"] = utils.pretty_datetime(self.environment.current_datetime)

        # update current goals
        if goals is not None:
            self._mental_state["goals"] = goals

        # update current context
        if context is not None:
            self._mental_state["context"] = context

        # update current attention
        if attention is not None:
            self._mental_state["attention"] = attention

        # update current emotions
        if emotions is not None:
            self._mental_state["emotions"] = emotions
        
        # update relevant memories for the current situation. These are memories that come to mind "spontaneously" when the agent is in a given context,
        # so avoiding the need to actively trying to remember them.
        current_memory_context = self.retrieve_relevant_memories_for_current_context()
        self._mental_state["memory_context"] = current_memory_context

        self.reset_prompt()
        

    ###########################################################
    # Memory management
    ###########################################################

    def store_in_memory(self, value: Any) -> list:
        self.episodic_memory.store(value)
        
        self._current_episode_event_count += 1
        logger.debug(f"[{self.name}] Current episode event count: {self._current_episode_event_count}.")

        if self._current_episode_event_count >= self.MAX_EPISODE_LENGTH:
            # commit the current episode to memory, if it is long enough
            logger.warning(f"[{self.name}] Episode length exceeded {self.MAX_EPISODE_LENGTH} events. Committing episode to memory. Please check whether this was expected or not.")
            self.consolidate_episode_memories()
    
    def consolidate_episode_memories(self):
        """
        Applies all memory consolidation or transformation processes appropriate to the conclusion of one simulation episode.
        """
        # a minimum length of the episode is required to consolidate it, to avoid excessive fragments in the semantic memory
        if self._current_episode_event_count > self.MIN_EPISODE_LENGTH:
            logger.debug(f"[{self.name}] ***** Consolidating current episode memories into semantic memory *****")
        
            # Consolidate latest episodic memories into semantic memory
            if config_manager.get("enable_memory_consolidation"):
                
                
                    episodic_consolidator = EpisodicConsolidator()
                    episode = self.episodic_memory.get_current_episode(item_types=["action", "stimulus"],)
                    logger.debug(f"[{self.name}] Current episode: {episode}")
                    consolidated_memories = episodic_consolidator.process(episode, timestamp=self._mental_state["datetime"], context=self._mental_state, persona=self.minibio())["consolidation"]
                    if consolidated_memories is not None:
                        logger.info(f"[{self.name}] Consolidating current {len(episode)} episodic events as consolidated semantic memories.")
                        logger.debug(f"[{self.name}] Consolidated memories: {consolidated_memories}")
                        self.semantic_memory.store_all(consolidated_memories)
                    else:
                        logger.debug(f"[{self.name}] No memories to consolidate from the current episode.")
                

            else:
                logger.warning(f"[{self.name}] Memory consolidation is disabled. Not consolidating current episode memories into semantic memory.")

            # commit the current episode to episodic memory
            self.episodic_memory.commit_episode()
            self._current_episode_event_count = 0
            logger.debug(f"[{self.name}] Current episode event count reset to 0 after consolidation.")

            # TODO reflections, optimizations, etc.

    def optimize_memory(self):
        pass #TODO

    def clear_episodic_memory(self, max_prefix_to_clear=None, max_suffix_to_clear=None):
        """
        Clears the episodic memory, causing a permanent "episodic amnesia". Note that this does not
        change other memories, such as semantic memory.  
        """
        self.episodic_memory.clear(max_prefix_to_clear=max_prefix_to_clear, max_suffix_to_clear=max_suffix_to_clear)

    def retrieve_memories(self, first_n: int, last_n: int, include_omission_info:bool=True, max_content_length:int=None) -> list:
        episodes = self.episodic_memory.retrieve(first_n=first_n, last_n=last_n, include_omission_info=include_omission_info)

        if max_content_length is not None:
            episodes = utils.truncate_actions_or_stimuli(episodes, max_content_length)

        return episodes


    def retrieve_recent_memories(self, max_content_length:int=None) -> list:
        episodes = self.episodic_memory.retrieve_recent()

        if max_content_length is not None:
            episodes = utils.truncate_actions_or_stimuli(episodes, max_content_length)

        return episodes

    def retrieve_relevant_memories(self, relevance_target:str, top_k=20) -> list:
        relevant = self.semantic_memory.retrieve_relevant(relevance_target, top_k=top_k)

        return relevant

    def retrieve_relevant_memories_for_current_context(self, top_k=7) -> list:
        # current context is composed of th recent memories, plus context, goals, attention, and emotions
        context = self._mental_state["context"]
        goals = self._mental_state["goals"]
        attention = self._mental_state["attention"]
        emotions = self._mental_state["emotions"]
        recent_memories = "\n".join([f"  - {m['content']}"  for m in self.retrieve_memories(first_n=10, last_n=20, max_content_length=500)])

        # put everything together in a nice markdown string to fetch relevant memories
        target = f"""
        Current Context: {context}
        Current Goals: {goals}
        Current Attention: {attention}
        Current Emotions: {emotions}
        Selected Episodic Memories (from oldest to newest):
        {recent_memories}
        """

        logger.debug(f"Retrieving relevant memories for contextual target: {target}")

        return self.retrieve_relevant_memories(target, top_k=top_k)

    def summarize_relevant_memories_via_full_scan(self, relevance_target:str, item_type: str = None) -> str:
        """
        Summarizes relevant memories for a given target by scanning the entire semantic memory.
        
        Args:
            relevance_target (str): The target to retrieve relevant memories for.
            item_type (str, optional): The type of items to summarize. Defaults to None.
            max_summary_length (int, optional): The maximum length of the summary. Defaults to 1000.
        
        Returns:
            str: The summary of relevant memories.
        """
        return self.semantic_memory.summarize_relevant_via_full_scan(relevance_target, item_type=item_type)

    ###########################################################
    # Inspection conveniences
    ###########################################################

    def last_remembered_action(self, ignore_done:bool=True):
        """
        Returns the last remembered action.

        Args:
            ignore_done (bool): Whether to ignore the "DONE" action or not. Defaults to True.
        """
        action = None 
        
        memory_items_list = self.episodic_memory.retrieve_last(include_omission_info=False, item_type="action")

        if len(memory_items_list) > 0:
            # iterate from last to first while the action type is not "DONE"
            for candidate_item in memory_items_list[::-1]:
                if candidate_item["content"]["action"]["type"] != "DONE":
                    action = candidate_item["content"]["action"]
                    break
                else:
                    if ignore_done:
                        continue
                    else:
                        action = candidate_item["content"]["action"]
                        break

        return action 


    ###########################################################
    # Communication display and action execution 
    ###########################################################

    def _display_communication(
        self,
        role,
        content,
        kind,
        simplified=True,
        max_content_length=default["max_content_display_length"],
    ):
        """
        Displays the current communication and stores it in a buffer for later use.
        """
        # CONCURRENT PROTECTION, as we'll access shared display buffers
        with concurrent_agent_action_lock:
            if kind == "stimuli":
                rendering = self._pretty_stimuli(
                    role=role,
                    content=content,
                    simplified=simplified,
                    max_content_length=max_content_length,
                )
                source = content["stimuli"][0].get("source", None)
                target = self.name
                
            elif kind == "action":
                rendering = self._pretty_action(
                    role=role,
                    content=content,
                    simplified=simplified,
                    max_content_length=max_content_length,
                )
                source = self.name
                target = content["action"].get("target", None)

            else:
                raise ValueError(f"Unknown communication kind: {kind}")

            # if the agent has no parent environment, then it is a free agent and we can display the communication.
            # otherwise, the environment will display the communication instead. This is important to make sure that
            # the communication is displayed in the correct order, since environments control the flow of their underlying
            # agents.
            if self.environment is None:
                self._push_and_display_latest_communication({"kind": kind, "rendering":rendering, "content": content, "source":source, "target": target})
            else:
                self.environment._push_and_display_latest_communication({"kind": kind, "rendering":rendering, "content": content, "source":source, "target": target})

    def _push_and_display_latest_communication(self, communication):
        """
        Pushes the latest communications to the agent's buffer.
        """
        self._displayed_communications_buffer.append(communication)
        print(communication["rendering"])

    def pop_and_display_latest_communications(self):
        """
        Pops the latest communications and displays them.
        """
        communications = self._displayed_communications_buffer
        self._displayed_communications_buffer = []

        for communication in communications:
            print(communication["rendering"])

        return communications

    def clear_communications_buffer(self):
        """
        Cleans the communications buffer.
        """
        self._displayed_communications_buffer = []

    @transactional()
    def pop_latest_actions(self) -> list:
        """
        Returns the latest actions performed by this agent. Typically used
        by an environment to consume the actions and provide the appropriate
        environmental semantics to them (i.e., effects on other agents).
        """
        actions = self._actions_buffer
        self._actions_buffer = []
        return actions

    @transactional()
    def pop_actions_and_get_contents_for(
        self, action_type: str, only_last_action: bool = True
    ) -> list:
        """
        Returns the contents of actions of a given type performed by this agent.
        Typically used to perform inspections and tests.

        Args:
            action_type (str): The type of action to look for.
            only_last_action (bool, optional): Whether to only return the contents of the last action. Defaults to False.
        """
        actions = self.pop_latest_actions()
        # Filter the actions by type
        actions = [action for action in actions if action["type"] == action_type]

        # If interested only in the last action, return the latest one
        if only_last_action:
            return actions[-1].get("content", "")

        # Otherwise, return all contents from the filtered actions
        return "\n".join([action.get("content", "") for action in actions])

    #############################################################################################
    # Formatting conveniences
    #
    # For rich colors,
    #    see: https://rich.readthedocs.io/en/latest/appendix/colors.html#appendix-colors
    #############################################################################################

    def __repr__(self):
        return f"TinyPerson(name='{self.name}')"

    @transactional()
    def minibio(self, extended=True, requirements=None):
        """
        Returns a mini-biography of the TinyPerson.

        Args:
            extended (bool): Whether to include extended information or not.
            requirements (str): Additional requirements for the biography (e.g., focus on a specific aspect relevant for the scenario).

        Returns:
            str: The mini-biography.
        """

        # if occupation is a dict and has a "title" key, use that as the occupation 
        if isinstance(self._persona['occupation'], dict) and 'title' in self._persona['occupation']:
            occupation = self._persona['occupation']['title']
        else:
            occupation = self._persona['occupation']

        base_biography = f"{self.name} is a {self._persona['age']} year old {occupation}, {self._persona['nationality']}, currently living in {self._persona['residence']}."

        if self._extended_agent_summary is None and extended:
            logger.debug(f"Generating extended agent summary for {self.name}.")
            self._extended_agent_summary = LLMChat(
                                                system_prompt=f"""
                                                You are given a short biography of an agent, as well as a detailed specification of his or her other characteristics
                                                You must then produce a short paragraph (3 or 4 sentences) that **complements** the short biography, adding details about
                                                personality, interests, opinions, skills, etc. Do not repeat the information already given in the short biography.
                                                repeating the information already given. The paragraph should be coherent, consistent and comprehensive. All information
                                                must be grounded on the specification, **do not** create anything new.

                                                {"Additional constraints: "+ requirements if requirements is not None else ""}
                                                """, 

                                                user_prompt=f"""
                                                **Short biography:** {base_biography}

                                                **Detailed specification:** {self._persona}
                                                """).call()

        if extended:
            biography = f"{base_biography} {self._extended_agent_summary}"
        else:
            biography = base_biography

        return biography

    def pp_current_interactions(
        self,
        simplified=True,
        skip_system=True,
        max_content_length=default["max_content_display_length"],
        first_n=None, 
        last_n=None, 
        include_omission_info:bool=True
    ):
        """
        Pretty prints the current messages.
        """
        print(
            self.pretty_current_interactions(
                simplified=simplified,
                skip_system=skip_system,
                max_content_length=max_content_length,
                first_n=first_n,
                last_n=last_n,
                include_omission_info=include_omission_info
            )
        )

    def pp_last_interactions(
        self,
        n=3,
        simplified=True,
        skip_system=True,
        max_content_length=default["max_content_display_length"],
        include_omission_info:bool=True
    ):
        """
        Pretty prints the last n messages. Useful to examine the conclusion of an experiment.
        """
        print(
            self.pretty_current_interactions(
                simplified=simplified,
                skip_system=skip_system,
                max_content_length=max_content_length,
                first_n=None,
                last_n=n,
                include_omission_info=include_omission_info
            )
        )

    def pretty_current_interactions(self, simplified=True, skip_system=True, max_content_length=default["max_content_display_length"], first_n=None, last_n=None, include_omission_info:bool=True):
      """
      Returns a pretty, readable, string with the current messages.
      """
      lines = [f"**** BEGIN SIMULATION TRAJECTORY FOR {self.name} ****"]
      last_step = 0
      for i, message in enumerate(self.episodic_memory.retrieve(first_n=first_n, last_n=last_n, include_omission_info=include_omission_info)):
        try:
            if not (skip_system and message['role'] == 'system'):
                msg_simplified_type = ""
                msg_simplified_content = ""
                msg_simplified_actor = ""

                last_step = i
                lines.append(f"Agent simulation trajectory event #{i}:")
                lines.append(self._pretty_timestamp(message['role'], message['simulation_timestamp']))

                if message["role"] == "system":
                    msg_simplified_actor = "SYSTEM"
                    msg_simplified_type = message["role"]
                    msg_simplified_content = message["content"]

                    lines.append(
                        f"[dim] {msg_simplified_type}: {msg_simplified_content}[/]"
                    )

                elif message["role"] == "user":
                    lines.append(
                        self._pretty_stimuli(
                            role=message["role"],
                            content=message["content"],
                            simplified=simplified,
                            max_content_length=max_content_length,
                        )
                    )

                elif message["role"] == "assistant":
                    lines.append(
                        self._pretty_action(
                            role=message["role"],
                            content=message["content"],
                            simplified=simplified,
                            max_content_length=max_content_length,
                        )
                    )
                else:
                    lines.append(f"{message['role']}: {message['content']}")
        except:
            # print(f"ERROR: {message}")
            continue

      lines.append(f"The last agent simulation trajectory event number was {last_step}, thus the current number of the NEXT POTENTIAL TRAJECTORY EVENT is {last_step + 1}.")
      lines.append(f"**** END SIMULATION TRAJECTORY FOR {self.name} ****\n\n")
      return "\n".join(lines)

    def _pretty_stimuli(
        self,
        role,
        content,
        simplified=True,
        max_content_length=default["max_content_display_length"],
    ) -> list:
        """
        Pretty prints stimuli.
        """

        lines = []
        msg_simplified_actor = "USER"
        for stimus in content["stimuli"]:
            if simplified:
                if stimus["source"] != "":
                    msg_simplified_actor = stimus["source"]

                else:
                    msg_simplified_actor = "USER"

                msg_simplified_type = stimus["type"]
                msg_simplified_content = utils.break_text_at_length(
                    stimus["content"], max_length=max_content_length
                )

                indent = " " * len(msg_simplified_actor) + "      > "
                msg_simplified_content = textwrap.fill(
                    msg_simplified_content,
                    width=TinyPerson.PP_TEXT_WIDTH,
                    initial_indent=indent,
                    subsequent_indent=indent,
                )

                #
                # Using rich for formatting. Let's make things as readable as possible!
                #

                rich_style = utils.RichTextStyle.get_style_for("stimulus", msg_simplified_type)
                lines.append(
                    f"[{rich_style}][underline]{msg_simplified_actor}[/] --> [{rich_style}][underline]{self.name}[/]: [{msg_simplified_type}] \n{msg_simplified_content}[/]"
                )
            else:
                lines.append(f"{role}: {content}")

        return "\n".join(lines)

    def _pretty_action(
        self,
        role,
        content,
        simplified=True,
        max_content_length=default["max_content_display_length"],
    ) -> str:
        """
        Pretty prints an action.
        """
        if simplified:
            msg_simplified_actor = self.name
            msg_simplified_type = content["action"]["type"]
            msg_simplified_content = utils.break_text_at_length(
                content["action"].get("content", ""), max_length=max_content_length
            )

            indent = " " * len(msg_simplified_actor) + "      > "
            msg_simplified_content = textwrap.fill(
                msg_simplified_content,
                width=TinyPerson.PP_TEXT_WIDTH,
                initial_indent=indent,
                subsequent_indent=indent,
            )

            #
            # Using rich for formatting. Let's make things as readable as possible!
            #
            rich_style = utils.RichTextStyle.get_style_for("action", msg_simplified_type)
            return f"[{rich_style}][underline]{msg_simplified_actor}[/] acts: [{msg_simplified_type}] \n{msg_simplified_content}[/]"
        
        else:
            return f"{role}: {content}"
    
    def _pretty_timestamp(
        self,
        role,
        timestamp,
    ) -> str:
        """
        Pretty prints a timestamp.
        """
        return f">>>>>>>>> Date and time of events: {timestamp}"

    def iso_datetime(self) -> str:
        """
        Returns the current datetime of the environment, if any.

        Returns:
            datetime: The current datetime of the environment in ISO forat.
        """
        if self.environment is not None and self.environment.current_datetime is not None:
            return self.environment.current_datetime.isoformat()
        else:
            return None

    ###########################################################
    # IO
    ###########################################################

    def save_specification(self, path, include_mental_faculties=True, include_memory=False, include_mental_state=False):
        """
        Saves the current configuration to a JSON file.
        """
        
        suppress_attributes = []

        # should we include the mental faculties?
        if not include_mental_faculties:
            suppress_attributes.append("_mental_faculties")

        # should we include the memory?
        if not include_memory:
            suppress_attributes.append("episodic_memory")
            suppress_attributes.append("semantic_memory")

        # should we include the mental state?
        if not include_mental_state:
            suppress_attributes.append("_mental_state")
        

        self.to_json(suppress=suppress_attributes, file_path=path,
                     serialization_type_field_name="type")

    
    @staticmethod
    def load_specification(path_or_dict, suppress_mental_faculties=False, suppress_memory=False, suppress_mental_state=False, 
                           auto_rename_agent=False, new_agent_name=None):
        """
        Loads a JSON agent specification.

        Args:
            path_or_dict (str or dict): The path to the JSON file or the dictionary itself.
            suppress_mental_faculties (bool, optional): Whether to suppress loading the mental faculties. Defaults to False.
            suppress_memory (bool, optional): Whether to suppress loading the memory. Defaults to False.
            suppress_memory (bool, optional): Whether to suppress loading the memory. Defaults to False.
            suppress_mental_state (bool, optional): Whether to suppress loading the mental state. Defaults to False.
            auto_rename_agent (bool, optional): Whether to auto rename the agent. Defaults to False.
            new_agent_name (str, optional): The new name for the agent. Defaults to None.
        """

        suppress_attributes = []

        # should we suppress the mental faculties?
        if suppress_mental_faculties:
            suppress_attributes.append("_mental_faculties")

        # should we suppress the memory?
        if suppress_memory:
            suppress_attributes.append("episodic_memory")
            suppress_attributes.append("semantic_memory")
        
        # should we suppress the mental state?
        if suppress_mental_state:
            suppress_attributes.append("_mental_state")

        return TinyPerson.from_json(json_dict_or_path=path_or_dict, suppress=suppress_attributes, 
                                    serialization_type_field_name="type",
                                    post_init_params={"auto_rename_agent": auto_rename_agent, "new_agent_name": new_agent_name})
    @staticmethod
    def load_specifications_from_folder(folder_path:str, file_suffix=".agent.json", suppress_mental_faculties=False, 
                                        suppress_memory=False, suppress_mental_state=False, auto_rename_agent=False, 
                                        new_agent_name=None) -> list:
        """     
        Loads all JSON agent specifications from a folder.

        Args:
            folder_path (str): The path to the folder containing the JSON files.
            file_suffix (str, optional): The suffix of the JSON files. Defaults to ".agent.json".
            suppress_mental_faculties (bool, optional): Whether to suppress loading the mental faculties. Defaults to False.
            suppress_memory (bool, optional): Whether to suppress loading the memory. Defaults to False.
            suppress_mental_state (bool, optional): Whether to suppress loading the mental state. Defaults to False.
            auto_rename_agent (bool, optional): Whether to auto rename the agent. Defaults to False.
            new_agent_name (str, optional): The new name for the agent. Defaults to None.
        """

        agents = []
        for file in os.listdir(folder_path):
            if file.endswith(file_suffix):
                file_path = os.path.join(folder_path, file)
                agent = TinyPerson.load_specification(file_path, suppress_mental_faculties=suppress_mental_faculties,
                                                      suppress_memory=suppress_memory, suppress_mental_state=suppress_mental_state,
                                                      auto_rename_agent=auto_rename_agent, new_agent_name=new_agent_name)
                agents.append(agent)

        return agents
        


    def encode_complete_state(self) -> dict:
        """
        Encodes the complete state of the TinyPerson, including the current messages, accessible agents, etc.
        This is meant for serialization and caching purposes, not for exporting the state to the user.
        """
        to_copy = copy.copy(self.__dict__)

        # delete the logger and other attributes that cannot be serialized
        del to_copy["environment"]
        del to_copy["_mental_faculties"]
        del to_copy["action_generator"]

        to_copy["_accessible_agents"] = [agent.name for agent in self._accessible_agents]
        to_copy['episodic_memory'] = self.episodic_memory.to_json()
        to_copy['semantic_memory'] = self.semantic_memory.to_json()
        to_copy["_mental_faculties"] = [faculty.to_json() for faculty in self._mental_faculties]

        state = copy.deepcopy(to_copy)

        return state

    def decode_complete_state(self, state: dict) -> Self:
        """
        Loads the complete state of the TinyPerson, including the current messages,
        and produces a new TinyPerson instance.
        """
        state = copy.deepcopy(state)
        
        self._accessible_agents = [TinyPerson.get_agent_by_name(name) for name in state["_accessible_agents"]]
        self.episodic_memory = EpisodicMemory.from_json(state['episodic_memory'])
        self.semantic_memory = SemanticMemory.from_json(state['semantic_memory'])
        
        for i, faculty in enumerate(self._mental_faculties):
            faculty = faculty.from_json(state['_mental_faculties'][i])

        # delete fields already present in the state
        del state["_accessible_agents"]
        del state['episodic_memory']
        del state['semantic_memory']
        del state['_mental_faculties']

        # restore other fields
        self.__dict__.update(state)


        return self
    
    def create_new_agent_from_current_spec(self, new_name:str) -> Self:
        """
        Creates a new agent from the current agent's specification. 

        Args:
            new_name (str): The name of the new agent. Agent names must be unique in the simulation, 
              this is why we need to provide a new name.
        """
        new_agent = TinyPerson(name=new_name, spec_path=None)
        
        new_persona = copy.deepcopy(self._persona)
        new_persona['name'] = new_name

        new_agent._persona = new_persona

        return new_agent
        

    @staticmethod
    def add_agent(agent):
        """
        Adds an agent to the global list of agents. Agent names must be unique,
        so this method will raise an exception if the name is already in use.
        """
        if agent.name in TinyPerson.all_agents:
            raise ValueError(f"Agent name {agent.name} is already in use.")
        else:
            TinyPerson.all_agents[agent.name] = agent

    @staticmethod
    def has_agent(agent_name: str):
        """
        Checks if an agent is already registered.
        """
        return agent_name in TinyPerson.all_agents

    @staticmethod
    def set_simulation_for_free_agents(simulation):
        """
        Sets the simulation if it is None. This allows free agents to be captured by specific simulation scopes
        if desired.
        """
        for agent in TinyPerson.all_agents.values():
            if agent.simulation_id is None:
                simulation.add_agent(agent)

    @staticmethod
    def get_agent_by_name(name):
        """
        Gets an agent by name.
        """
        if name in TinyPerson.all_agents:
            return TinyPerson.all_agents[name]
        else:
            return None
    
    @staticmethod
    def all_agents_names():
        """
        Returns the names of all agents.
        """
        return list(TinyPerson.all_agents.keys())

    @staticmethod
    def clear_agents():
        """
        Clears the global list of agents.
        """
        TinyPerson.all_agents = {}

Ancestors

Class variables

var MAX_ACTIONS_BEFORE_DONE
var MAX_ACTION_SIMILARITY
var MAX_EPISODE_LENGTH
var MIN_EPISODE_LENGTH
var PP_TEXT_WIDTH
var all_agents
var communication_display : bool
var serializable_attributes
var serializable_attributes_renaming

Static methods

def add_agent(agent)

Adds an agent to the global list of agents. Agent names must be unique, so this method will raise an exception if the name is already in use.

Expand source code 
@staticmethod
def add_agent(agent):
    """
    Adds an agent to the global list of agents. Agent names must be unique,
    so this method will raise an exception if the name is already in use.
    """
    if agent.name in TinyPerson.all_agents:
        raise ValueError(f"Agent name {agent.name} is already in use.")
    else:
        TinyPerson.all_agents[agent.name] = agent
def all_agents_names()

Returns the names of all agents.

Expand source code 
@staticmethod
def all_agents_names():
    """
    Returns the names of all agents.
    """
    return list(TinyPerson.all_agents.keys())
def clear_agents()

Clears the global list of agents.

Expand source code 
@staticmethod
def clear_agents():
    """
    Clears the global list of agents.
    """
    TinyPerson.all_agents = {}
def get_agent_by_name(name)

Gets an agent by name.

Expand source code 
@staticmethod
def get_agent_by_name(name):
    """
    Gets an agent by name.
    """
    if name in TinyPerson.all_agents:
        return TinyPerson.all_agents[name]
    else:
        return None
def has_agent(agent_name: str)

Checks if an agent is already registered.

Expand source code 
@staticmethod
def has_agent(agent_name: str):
    """
    Checks if an agent is already registered.
    """
    return agent_name in TinyPerson.all_agents
def load_specification(path_or_dict, suppress_mental_faculties=False, suppress_memory=False, suppress_mental_state=False, auto_rename_agent=False, new_agent_name=None)

Loads a JSON agent specification.

Args

path_or_dict : str or dict
The path to the JSON file or the dictionary itself.
suppress_mental_faculties : bool, optional
Whether to suppress loading the mental faculties. Defaults to False.
suppress_memory : bool, optional
Whether to suppress loading the memory. Defaults to False.
suppress_memory : bool, optional
Whether to suppress loading the memory. Defaults to False.
suppress_mental_state : bool, optional
Whether to suppress loading the mental state. Defaults to False.
auto_rename_agent : bool, optional
Whether to auto rename the agent. Defaults to False.
new_agent_name : str, optional
The new name for the agent. Defaults to None.
Expand source code 
@staticmethod
def load_specification(path_or_dict, suppress_mental_faculties=False, suppress_memory=False, suppress_mental_state=False, 
                       auto_rename_agent=False, new_agent_name=None):
    """
    Loads a JSON agent specification.

    Args:
        path_or_dict (str or dict): The path to the JSON file or the dictionary itself.
        suppress_mental_faculties (bool, optional): Whether to suppress loading the mental faculties. Defaults to False.
        suppress_memory (bool, optional): Whether to suppress loading the memory. Defaults to False.
        suppress_memory (bool, optional): Whether to suppress loading the memory. Defaults to False.
        suppress_mental_state (bool, optional): Whether to suppress loading the mental state. Defaults to False.
        auto_rename_agent (bool, optional): Whether to auto rename the agent. Defaults to False.
        new_agent_name (str, optional): The new name for the agent. Defaults to None.
    """

    suppress_attributes = []

    # should we suppress the mental faculties?
    if suppress_mental_faculties:
        suppress_attributes.append("_mental_faculties")

    # should we suppress the memory?
    if suppress_memory:
        suppress_attributes.append("episodic_memory")
        suppress_attributes.append("semantic_memory")
    
    # should we suppress the mental state?
    if suppress_mental_state:
        suppress_attributes.append("_mental_state")

    return TinyPerson.from_json(json_dict_or_path=path_or_dict, suppress=suppress_attributes, 
                                serialization_type_field_name="type",
                                post_init_params={"auto_rename_agent": auto_rename_agent, "new_agent_name": new_agent_name})
def load_specifications_from_folder(folder_path: str, file_suffix='.agent.json', suppress_mental_faculties=False, suppress_memory=False, suppress_mental_state=False, auto_rename_agent=False, new_agent_name=None) ‑> list

Loads all JSON agent specifications from a folder.

Args

folder_path : str
The path to the folder containing the JSON files.
file_suffix : str, optional
The suffix of the JSON files. Defaults to ".agent.json".
suppress_mental_faculties : bool, optional
Whether to suppress loading the mental faculties. Defaults to False.
suppress_memory : bool, optional
Whether to suppress loading the memory. Defaults to False.
suppress_mental_state : bool, optional
Whether to suppress loading the mental state. Defaults to False.
auto_rename_agent : bool, optional
Whether to auto rename the agent. Defaults to False.
new_agent_name : str, optional
The new name for the agent. Defaults to None.
Expand source code 
@staticmethod
def load_specifications_from_folder(folder_path:str, file_suffix=".agent.json", suppress_mental_faculties=False, 
                                    suppress_memory=False, suppress_mental_state=False, auto_rename_agent=False, 
                                    new_agent_name=None) -> list:
    """     
    Loads all JSON agent specifications from a folder.

    Args:
        folder_path (str): The path to the folder containing the JSON files.
        file_suffix (str, optional): The suffix of the JSON files. Defaults to ".agent.json".
        suppress_mental_faculties (bool, optional): Whether to suppress loading the mental faculties. Defaults to False.
        suppress_memory (bool, optional): Whether to suppress loading the memory. Defaults to False.
        suppress_mental_state (bool, optional): Whether to suppress loading the mental state. Defaults to False.
        auto_rename_agent (bool, optional): Whether to auto rename the agent. Defaults to False.
        new_agent_name (str, optional): The new name for the agent. Defaults to None.
    """

    agents = []
    for file in os.listdir(folder_path):
        if file.endswith(file_suffix):
            file_path = os.path.join(folder_path, file)
            agent = TinyPerson.load_specification(file_path, suppress_mental_faculties=suppress_mental_faculties,
                                                  suppress_memory=suppress_memory, suppress_mental_state=suppress_mental_state,
                                                  auto_rename_agent=auto_rename_agent, new_agent_name=new_agent_name)
            agents.append(agent)

    return agents
def set_simulation_for_free_agents(simulation)

Sets the simulation if it is None. This allows free agents to be captured by specific simulation scopes if desired.

Expand source code 
@staticmethod
def set_simulation_for_free_agents(simulation):
    """
    Sets the simulation if it is None. This allows free agents to be captured by specific simulation scopes
    if desired.
    """
    for agent in TinyPerson.all_agents.values():
        if agent.simulation_id is None:
            simulation.add_agent(agent)

Instance variables

var accessible_agents

Property to access the list of accessible agents.

Expand source code 
@property
def accessible_agents(self):
    """
    Property to access the list of accessible agents.
    """
    return self._accessible_agents

Methods

def act(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def add_mental_faculties(self, mental_faculties)

Adds a list of mental faculties to the agent.

Expand source code 
def add_mental_faculties(self, mental_faculties):
    """
    Adds a list of mental faculties to the agent.
    """
    for faculty in mental_faculties:
        self.add_mental_faculty(faculty)
    
    return self
def add_mental_faculty(self, faculty)

Adds a mental faculty to the agent.

Expand source code 
def add_mental_faculty(self, faculty):
    """
    Adds a mental faculty to the agent.
    """
    # check if the faculty is already there or not
    if faculty not in self._mental_faculties:
        self._mental_faculties.append(faculty)
    else:
        raise Exception(f"The mental faculty {faculty} is already present in the agent.")
    
    return self
def change_context(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def clear_communications_buffer(self)

Cleans the communications buffer.

Expand source code 
def clear_communications_buffer(self):
    """
    Cleans the communications buffer.
    """
    self._displayed_communications_buffer = []
def clear_episodic_memory(self, max_prefix_to_clear=None, max_suffix_to_clear=None)

Clears the episodic memory, causing a permanent "episodic amnesia". Note that this does not change other memories, such as semantic memory.

Expand source code 
def clear_episodic_memory(self, max_prefix_to_clear=None, max_suffix_to_clear=None):
    """
    Clears the episodic memory, causing a permanent "episodic amnesia". Note that this does not
    change other memories, such as semantic memory.  
    """
    self.episodic_memory.clear(max_prefix_to_clear=max_prefix_to_clear, max_suffix_to_clear=max_suffix_to_clear)
def clear_relationships(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def consolidate_episode_memories(self)

Applies all memory consolidation or transformation processes appropriate to the conclusion of one simulation episode.

Expand source code 
def consolidate_episode_memories(self):
    """
    Applies all memory consolidation or transformation processes appropriate to the conclusion of one simulation episode.
    """
    # a minimum length of the episode is required to consolidate it, to avoid excessive fragments in the semantic memory
    if self._current_episode_event_count > self.MIN_EPISODE_LENGTH:
        logger.debug(f"[{self.name}] ***** Consolidating current episode memories into semantic memory *****")
    
        # Consolidate latest episodic memories into semantic memory
        if config_manager.get("enable_memory_consolidation"):
            
            
                episodic_consolidator = EpisodicConsolidator()
                episode = self.episodic_memory.get_current_episode(item_types=["action", "stimulus"],)
                logger.debug(f"[{self.name}] Current episode: {episode}")
                consolidated_memories = episodic_consolidator.process(episode, timestamp=self._mental_state["datetime"], context=self._mental_state, persona=self.minibio())["consolidation"]
                if consolidated_memories is not None:
                    logger.info(f"[{self.name}] Consolidating current {len(episode)} episodic events as consolidated semantic memories.")
                    logger.debug(f"[{self.name}] Consolidated memories: {consolidated_memories}")
                    self.semantic_memory.store_all(consolidated_memories)
                else:
                    logger.debug(f"[{self.name}] No memories to consolidate from the current episode.")
            

        else:
            logger.warning(f"[{self.name}] Memory consolidation is disabled. Not consolidating current episode memories into semantic memory.")

        # commit the current episode to episodic memory
        self.episodic_memory.commit_episode()
        self._current_episode_event_count = 0
        logger.debug(f"[{self.name}] Current episode event count reset to 0 after consolidation.")

        # TODO reflections, optimizations, etc.
def create_new_agent_from_current_spec(self, new_name: str) ‑> ~Self

Creates a new agent from the current agent's specification.

Args

new_name : str
The name of the new agent. Agent names must be unique in the simulation, this is why we need to provide a new name.
Expand source code 
def create_new_agent_from_current_spec(self, new_name:str) -> Self:
    """
    Creates a new agent from the current agent's specification. 

    Args:
        new_name (str): The name of the new agent. Agent names must be unique in the simulation, 
          this is why we need to provide a new name.
    """
    new_agent = TinyPerson(name=new_name, spec_path=None)
    
    new_persona = copy.deepcopy(self._persona)
    new_persona['name'] = new_name

    new_agent._persona = new_persona

    return new_agent
def decode_complete_state(self, state: dict) ‑> ~Self

Loads the complete state of the TinyPerson, including the current messages, and produces a new TinyPerson instance.

Expand source code 
def decode_complete_state(self, state: dict) -> Self:
    """
    Loads the complete state of the TinyPerson, including the current messages,
    and produces a new TinyPerson instance.
    """
    state = copy.deepcopy(state)
    
    self._accessible_agents = [TinyPerson.get_agent_by_name(name) for name in state["_accessible_agents"]]
    self.episodic_memory = EpisodicMemory.from_json(state['episodic_memory'])
    self.semantic_memory = SemanticMemory.from_json(state['semantic_memory'])
    
    for i, faculty in enumerate(self._mental_faculties):
        faculty = faculty.from_json(state['_mental_faculties'][i])

    # delete fields already present in the state
    del state["_accessible_agents"]
    del state['episodic_memory']
    del state['semantic_memory']
    del state['_mental_faculties']

    # restore other fields
    self.__dict__.update(state)


    return self
def define(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def define_relationships(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def encode_complete_state(self) ‑> dict

Encodes the complete state of the TinyPerson, including the current messages, accessible agents, etc. This is meant for serialization and caching purposes, not for exporting the state to the user.

Expand source code 
def encode_complete_state(self) -> dict:
    """
    Encodes the complete state of the TinyPerson, including the current messages, accessible agents, etc.
    This is meant for serialization and caching purposes, not for exporting the state to the user.
    """
    to_copy = copy.copy(self.__dict__)

    # delete the logger and other attributes that cannot be serialized
    del to_copy["environment"]
    del to_copy["_mental_faculties"]
    del to_copy["action_generator"]

    to_copy["_accessible_agents"] = [agent.name for agent in self._accessible_agents]
    to_copy['episodic_memory'] = self.episodic_memory.to_json()
    to_copy['semantic_memory'] = self.semantic_memory.to_json()
    to_copy["_mental_faculties"] = [faculty.to_json() for faculty in self._mental_faculties]

    state = copy.deepcopy(to_copy)

    return state
def generate_agent_system_prompt(self)
Expand source code 
def generate_agent_system_prompt(self):
    with open(self._prompt_template_path, "r") as f:
        agent_prompt_template = f.read()

    # let's operate on top of a copy of the configuration, because we'll need to add more variables, etc.
    template_variables = self._persona.copy()    
    template_variables["persona"] = json.dumps(self._persona.copy(), indent=4)    

    # add mental state to the template variables
    template_variables["mental_state"] = json.dumps(self._mental_state, indent=4)

    # Prepare additional action definitions and constraints
    actions_definitions_prompt = ""
    actions_constraints_prompt = ""
    for faculty in self._mental_faculties:
        actions_definitions_prompt += f"{faculty.actions_definitions_prompt()}\n"
        actions_constraints_prompt += f"{faculty.actions_constraints_prompt()}\n"
    
    # Make the additional prompt pieces available to the template. 
    # Identation here is to align with the text structure in the template.
    template_variables['actions_definitions_prompt'] = textwrap.indent(actions_definitions_prompt.strip(), "  ")
    template_variables['actions_constraints_prompt'] = textwrap.indent(actions_constraints_prompt.strip(), "  ")

    # RAI prompt components, if requested
    template_variables = utils.add_rai_template_variables_if_enabled(template_variables)

    return chevron.render(agent_prompt_template, template_variables)
def get(self, key)

Returns the value of a key in the TinyPerson's persona configuration. Supports dot notation for nested keys (e.g., "address.city").

Expand source code 
def get(self, key):
    """
    Returns the value of a key in the TinyPerson's persona configuration.
    Supports dot notation for nested keys (e.g., "address.city").
    """
    keys = key.split(".")
    value = self._persona
    for k in keys:
        if isinstance(value, dict):
            value = value.get(k, None)
        else:
            return None  # If the path is invalid, return None
    return value
def import_fragment(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def include_persona_definitions(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def internalize_goal(self, goal, max_content_length=None)

Internalizes a goal and updates its internal cognitive state.

Expand source code 
@config_manager.config_defaults(max_content_length="max_content_display_length")
def internalize_goal(
    self, goal, max_content_length=None
):
    """
    Internalizes a goal and updates its internal cognitive state.
    """
    return self._observe(
        stimulus={
            "type": "INTERNAL_GOAL_FORMULATION",
            "content": goal,
            "source": name_or_empty(self),
        },
        max_content_length=max_content_length,
    )
def iso_datetime(self) ‑> str

Returns the current datetime of the environment, if any.

Returns

datetime
The current datetime of the environment in ISO forat.
Expand source code 
def iso_datetime(self) -> str:
    """
    Returns the current datetime of the environment, if any.

    Returns:
        datetime: The current datetime of the environment in ISO forat.
    """
    if self.environment is not None and self.environment.current_datetime is not None:
        return self.environment.current_datetime.isoformat()
    else:
        return None
def last_remembered_action(self, ignore_done: bool = True)

Returns the last remembered action.

Args

ignore_done : bool
Whether to ignore the "DONE" action or not. Defaults to True.
Expand source code 
def last_remembered_action(self, ignore_done:bool=True):
    """
    Returns the last remembered action.

    Args:
        ignore_done (bool): Whether to ignore the "DONE" action or not. Defaults to True.
    """
    action = None 
    
    memory_items_list = self.episodic_memory.retrieve_last(include_omission_info=False, item_type="action")

    if len(memory_items_list) > 0:
        # iterate from last to first while the action type is not "DONE"
        for candidate_item in memory_items_list[::-1]:
            if candidate_item["content"]["action"]["type"] != "DONE":
                action = candidate_item["content"]["action"]
                break
            else:
                if ignore_done:
                    continue
                else:
                    action = candidate_item["content"]["action"]
                    break

    return action
def listen(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def listen_and_act(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def make_agent_accessible(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def make_agent_inaccessible(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def make_agents_accessible(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def make_all_agents_inaccessible(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def minibio(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def move_to(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def optimize_memory(self)
Expand source code 
def optimize_memory(self):
    pass #TODO
def pop_actions_and_get_contents_for(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def pop_and_display_latest_communications(self)

Pops the latest communications and displays them.

Expand source code 
def pop_and_display_latest_communications(self):
    """
    Pops the latest communications and displays them.
    """
    communications = self._displayed_communications_buffer
    self._displayed_communications_buffer = []

    for communication in communications:
        print(communication["rendering"])

    return communications
def pop_latest_actions(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def pp_current_interactions(self, simplified=True, skip_system=True, max_content_length=4000, first_n=None, last_n=None, include_omission_info: bool = True)

Pretty prints the current messages.

Expand source code 
def pp_current_interactions(
    self,
    simplified=True,
    skip_system=True,
    max_content_length=default["max_content_display_length"],
    first_n=None, 
    last_n=None, 
    include_omission_info:bool=True
):
    """
    Pretty prints the current messages.
    """
    print(
        self.pretty_current_interactions(
            simplified=simplified,
            skip_system=skip_system,
            max_content_length=max_content_length,
            first_n=first_n,
            last_n=last_n,
            include_omission_info=include_omission_info
        )
    )
def pp_last_interactions(self, n=3, simplified=True, skip_system=True, max_content_length=4000, include_omission_info: bool = True)

Pretty prints the last n messages. Useful to examine the conclusion of an experiment.

Expand source code 
def pp_last_interactions(
    self,
    n=3,
    simplified=True,
    skip_system=True,
    max_content_length=default["max_content_display_length"],
    include_omission_info:bool=True
):
    """
    Pretty prints the last n messages. Useful to examine the conclusion of an experiment.
    """
    print(
        self.pretty_current_interactions(
            simplified=simplified,
            skip_system=skip_system,
            max_content_length=max_content_length,
            first_n=None,
            last_n=n,
            include_omission_info=include_omission_info
        )
    )
def pretty_current_interactions(self, simplified=True, skip_system=True, max_content_length=4000, first_n=None, last_n=None, include_omission_info: bool = True)

Returns a pretty, readable, string with the current messages.

Expand source code 
def pretty_current_interactions(self, simplified=True, skip_system=True, max_content_length=default["max_content_display_length"], first_n=None, last_n=None, include_omission_info:bool=True):
  """
  Returns a pretty, readable, string with the current messages.
  """
  lines = [f"**** BEGIN SIMULATION TRAJECTORY FOR {self.name} ****"]
  last_step = 0
  for i, message in enumerate(self.episodic_memory.retrieve(first_n=first_n, last_n=last_n, include_omission_info=include_omission_info)):
    try:
        if not (skip_system and message['role'] == 'system'):
            msg_simplified_type = ""
            msg_simplified_content = ""
            msg_simplified_actor = ""

            last_step = i
            lines.append(f"Agent simulation trajectory event #{i}:")
            lines.append(self._pretty_timestamp(message['role'], message['simulation_timestamp']))

            if message["role"] == "system":
                msg_simplified_actor = "SYSTEM"
                msg_simplified_type = message["role"]
                msg_simplified_content = message["content"]

                lines.append(
                    f"[dim] {msg_simplified_type}: {msg_simplified_content}[/]"
                )

            elif message["role"] == "user":
                lines.append(
                    self._pretty_stimuli(
                        role=message["role"],
                        content=message["content"],
                        simplified=simplified,
                        max_content_length=max_content_length,
                    )
                )

            elif message["role"] == "assistant":
                lines.append(
                    self._pretty_action(
                        role=message["role"],
                        content=message["content"],
                        simplified=simplified,
                        max_content_length=max_content_length,
                    )
                )
            else:
                lines.append(f"{message['role']}: {message['content']}")
    except:
        # print(f"ERROR: {message}")
        continue

  lines.append(f"The last agent simulation trajectory event number was {last_step}, thus the current number of the NEXT POTENTIAL TRAJECTORY EVENT is {last_step + 1}.")
  lines.append(f"**** END SIMULATION TRAJECTORY FOR {self.name} ****\n\n")
  return "\n".join(lines)
def read_document_from_file(self, file_path: str)

Reads a document from a file and loads it into the semantic memory.

Expand source code 
def read_document_from_file(self, file_path:str):
    """
    Reads a document from a file and loads it into the semantic memory.
    """
    logger.info(f"Reading document from file: {file_path}")

    self.semantic_memory.add_document_path(file_path)
def read_document_from_web(self, web_url: str)

Reads a document from a web URL and loads it into the semantic memory.

Expand source code 
def read_document_from_web(self, web_url:str):
    """
    Reads a document from a web URL and loads it into the semantic memory.
    """
    logger.info(f"Reading document from web URL: {web_url}")

    self.semantic_memory.add_web_url(web_url)
def read_documents_from_folder(self, documents_path: str)

Reads documents from a directory and loads them into the semantic memory.

Expand source code 
def read_documents_from_folder(self, documents_path:str):
    """
    Reads documents from a directory and loads them into the semantic memory.
    """
    logger.info(f"Setting documents path to {documents_path} and loading documents.")

    self.semantic_memory.add_documents_path(documents_path)
def read_documents_from_web(self, web_urls: list)

Reads documents from web URLs and loads them into the semantic memory.

Expand source code 
def read_documents_from_web(self, web_urls:list):
    """
    Reads documents from web URLs and loads them into the semantic memory.
    """
    logger.info(f"Reading documents from the following web URLs: {web_urls}")

    self.semantic_memory.add_web_urls(web_urls)
def related_to(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def reset_prompt(self)
Expand source code 
def reset_prompt(self):

    # render the template with the current configuration
    self._init_system_message = self.generate_agent_system_prompt()

    # - reset system message
    # - make it clear that the provided events are past events and have already had their effects
    self.current_messages = [
        {"role": "system", "content": self._init_system_message},
        {"role": "system", "content": "The next messages refer to past interactions you had recently and are meant to help you contextualize your next actions. "\
                                    + "They are the most recent episodic memories you have, including stimuli and actions. "\
                                    + "Their effects already took place and led to your present cognitive state (described above), so you can use them in conjunction "\
                                    + "with your cognitive state to inform your next actions and perceptions. Please consider them and then proceed with your next actions right after. "}
    ]

    # sets up the actual interaction messages to use for prompting
    self.current_messages += self.retrieve_recent_memories()
def retrieve_memories(self, first_n: int, last_n: int, include_omission_info: bool = True, max_content_length: int = None) ‑> list
Expand source code 
def retrieve_memories(self, first_n: int, last_n: int, include_omission_info:bool=True, max_content_length:int=None) -> list:
    episodes = self.episodic_memory.retrieve(first_n=first_n, last_n=last_n, include_omission_info=include_omission_info)

    if max_content_length is not None:
        episodes = utils.truncate_actions_or_stimuli(episodes, max_content_length)

    return episodes
def retrieve_recent_memories(self, max_content_length: int = None) ‑> list
Expand source code 
def retrieve_recent_memories(self, max_content_length:int=None) -> list:
    episodes = self.episodic_memory.retrieve_recent()

    if max_content_length is not None:
        episodes = utils.truncate_actions_or_stimuli(episodes, max_content_length)

    return episodes
def retrieve_relevant_memories(self, relevance_target: str, top_k=20) ‑> list
Expand source code 
def retrieve_relevant_memories(self, relevance_target:str, top_k=20) -> list:
    relevant = self.semantic_memory.retrieve_relevant(relevance_target, top_k=top_k)

    return relevant
def retrieve_relevant_memories_for_current_context(self, top_k=7) ‑> list
Expand source code 
def retrieve_relevant_memories_for_current_context(self, top_k=7) -> list:
    # current context is composed of th recent memories, plus context, goals, attention, and emotions
    context = self._mental_state["context"]
    goals = self._mental_state["goals"]
    attention = self._mental_state["attention"]
    emotions = self._mental_state["emotions"]
    recent_memories = "\n".join([f"  - {m['content']}"  for m in self.retrieve_memories(first_n=10, last_n=20, max_content_length=500)])

    # put everything together in a nice markdown string to fetch relevant memories
    target = f"""
    Current Context: {context}
    Current Goals: {goals}
    Current Attention: {attention}
    Current Emotions: {emotions}
    Selected Episodic Memories (from oldest to newest):
    {recent_memories}
    """

    logger.debug(f"Retrieving relevant memories for contextual target: {target}")

    return self.retrieve_relevant_memories(target, top_k=top_k)
def save_specification(self, path, include_mental_faculties=True, include_memory=False, include_mental_state=False)

Saves the current configuration to a JSON file.

Expand source code 
def save_specification(self, path, include_mental_faculties=True, include_memory=False, include_mental_state=False):
    """
    Saves the current configuration to a JSON file.
    """
    
    suppress_attributes = []

    # should we include the mental faculties?
    if not include_mental_faculties:
        suppress_attributes.append("_mental_faculties")

    # should we include the memory?
    if not include_memory:
        suppress_attributes.append("episodic_memory")
        suppress_attributes.append("semantic_memory")

    # should we include the mental state?
    if not include_mental_state:
        suppress_attributes.append("_mental_state")
    

    self.to_json(suppress=suppress_attributes, file_path=path,
                 serialization_type_field_name="type")
def see(self, visual_description, source: Union[~Self, ForwardRef('TinyWorld')] = None, max_content_length=None)

Perceives a visual stimulus through a description and updates its internal cognitive state.

Args

visual_description : str
The description of the visual stimulus.
source : AgentOrWorld, optional
The source of the visual stimulus. Defaults to None.
Expand source code 
@config_manager.config_defaults(max_content_length="max_content_display_length")
def see(
    self,
    visual_description,
    source: AgentOrWorld = None,
    max_content_length=None,
):
    """
    Perceives a visual stimulus through a description and updates its internal cognitive state.

    Args:
        visual_description (str): The description of the visual stimulus.
        source (AgentOrWorld, optional): The source of the visual stimulus. Defaults to None.
    """
    return self._observe(
        stimulus={
            "type": "VISUAL",
            "content": visual_description,
            "source": name_or_empty(source),
        },
        max_content_length=max_content_length,
    )
def see_and_act(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result
def socialize(self, social_description: str, source: Union[~Self, ForwardRef('TinyWorld')] = None, max_content_length=None)

Perceives a social stimulus through a description and updates its internal cognitive state.

Args

social_description : str
The description of the social stimulus.
source : AgentOrWorld, optional
The source of the social stimulus. Defaults to None.
Expand source code 
@config_manager.config_defaults(max_content_length="max_content_display_length")
def socialize(
    self,
    social_description: str,
    source: AgentOrWorld = None,
    max_content_length=None,
):
    """
    Perceives a social stimulus through a description and updates its internal cognitive state.

    Args:
        social_description (str): The description of the social stimulus.
        source (AgentOrWorld, optional): The source of the social stimulus. Defaults to None.
    """
    return self._observe(
        stimulus={
            "type": "SOCIAL",
            "content": social_description,
            "source": name_or_empty(source),
        },
        max_content_length=max_content_length,
    )
def store_in_memory(self, value: Any) ‑> list
Expand source code 
def store_in_memory(self, value: Any) -> list:
    self.episodic_memory.store(value)
    
    self._current_episode_event_count += 1
    logger.debug(f"[{self.name}] Current episode event count: {self._current_episode_event_count}.")

    if self._current_episode_event_count >= self.MAX_EPISODE_LENGTH:
        # commit the current episode to memory, if it is long enough
        logger.warning(f"[{self.name}] Episode length exceeded {self.MAX_EPISODE_LENGTH} events. Committing episode to memory. Please check whether this was expected or not.")
        self.consolidate_episode_memories()
def summarize_relevant_memories_via_full_scan(self, relevance_target: str, item_type: str = None) ‑> str

Summarizes relevant memories for a given target by scanning the entire semantic memory.

Args

relevance_target : str
The target to retrieve relevant memories for.
item_type : str, optional
The type of items to summarize. Defaults to None.
max_summary_length : int, optional
The maximum length of the summary. Defaults to 1000.

Returns

str
The summary of relevant memories.
Expand source code 
def summarize_relevant_memories_via_full_scan(self, relevance_target:str, item_type: str = None) -> str:
    """
    Summarizes relevant memories for a given target by scanning the entire semantic memory.
    
    Args:
        relevance_target (str): The target to retrieve relevant memories for.
        item_type (str, optional): The type of items to summarize. Defaults to None.
        max_summary_length (int, optional): The maximum length of the summary. Defaults to 1000.
    
    Returns:
        str: The summary of relevant memories.
    """
    return self.semantic_memory.summarize_relevant_via_full_scan(relevance_target, item_type=item_type)
def think(self, thought, max_content_length=None)

Forces the agent to think about something and updates its internal cognitive state.

Expand source code 
@config_manager.config_defaults(max_content_length="max_content_display_length")
def think(self, thought, max_content_length=None):
    """
    Forces the agent to think about something and updates its internal cognitive state.

    """
    return self._observe(
        stimulus={
            "type": "THOUGHT",
            "content": thought,
            "source": name_or_empty(self),
        },
        max_content_length=max_content_length,
    )
def think_and_act(*args, **kwargs)
Expand source code 
def wrapper(*args, **kwargs):
    obj_under_transaction = args[0]
    simulation = current_simulation()
    obj_sim_id = obj_under_transaction.simulation_id if hasattr(obj_under_transaction, 'simulation_id') else None

    logger.debug(f"-----------------------------------------> Transaction: {func.__name__} with args {args[1:]} and kwargs {kwargs} under simulation {obj_sim_id}, parallel={parallel}.")
    
    parallel_id = str(threading.current_thread())
    
    transaction = Transaction(obj_under_transaction, simulation, func, *args, **kwargs)
    result = transaction.execute(begin_parallel=parallel, parallel_id=parallel_id)
    
    return result

Inherited members

class TinyToolUse (tools: list)

Allows the agent to use tools to accomplish tasks. Tool usage is one of the most important cognitive skills humans and primates have as we know.

Initializes the mental faculty.

Args

name : str
The name of the mental faculty.
requires_faculties : list
A list of mental faculties that this faculty requires to function properly.
Expand source code 
class TinyToolUse(TinyMentalFaculty):
    """
    Allows the agent to use tools to accomplish tasks. Tool usage is one of the most important cognitive skills
    humans and primates have as we know.
    """

    def __init__(self, tools:list) -> None:
        super().__init__("Tool Use")
    
        self.tools = tools
    
    def process_action(self, agent, action: dict) -> bool:
        for tool in self.tools:
            if tool.process_action(agent, action):
                return True
        
        return False
    
    def actions_definitions_prompt(self) -> str:
        # each tool should provide its own actions definitions prompt
        prompt = ""
        for tool in self.tools:
            prompt += tool.actions_definitions_prompt()
        
        return prompt
    
    def actions_constraints_prompt(self) -> str:
        # each tool should provide its own actions constraints prompt
        prompt = ""
        for tool in self.tools:
            prompt += tool.actions_constraints_prompt()
        
        return prompt

Ancestors

Inherited members