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 logging
from pydantic import BaseModel
import tinytroupe.utils as utils
logger = logging.getLogger("tinytroupe")
###########################################################################
# Types and constants
###########################################################################
from typing import Any, List, TypeVar, Union
Self = TypeVar("Self", bound="TinyPerson")
AgentOrWorld = Union[Self, Any]
###########################################################################
# 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
# New multi-action response formats (sequence per turn)
class CognitiveActionsModel(BaseModel):
actions: List[Action]
cognitive_state: CognitiveState
class CognitiveActionsModelWithReasoning(BaseModel):
reasoning: str
actions: List[Action]
cognitive_state: CognitiveState
###########################################################################
# Exposed API
###########################################################################
# from. grounding ... ---> not exposing this, clients should not need to know about detailed grounding mechanisms
from .memory import (
EpisodicConsolidator,
EpisodicMemory,
ReflectionConsolidator,
SemanticMemory,
)
from .mental_faculty import (
CustomMentalFaculty,
FilesAndWebGroundingFaculty,
RecallFaculty,
TinyToolUse,
)
from .tiny_person import TinyPerson
__all__ = [
"SemanticMemory",
"EpisodicMemory",
"EpisodicConsolidator",
"ReflectionConsolidator",
"CustomMentalFaculty",
"RecallFaculty",
"FilesAndWebGroundingFaculty",
"TinyToolUse",
"TinyPerson",
]Sub-modules
tinytroupe.agent.action_generatortinytroupe.agent.groundingtinytroupe.agent.memorytinytroupe.agent.mental_facultytinytroupe.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 promptAncestors
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." ) # we'll limit the length of memories to process at once, to avoid overflowing the input of embedding models, which # are much more limited than the LLM's. max_word_count = 1000 total_word_count = self.count_memory_content_words(memories) if total_word_count <= max_word_count: # Process all memories at once if under the limit result = self._consolidate(memories, timestamp, enriched_context, persona) logger.debug(f"Consolidated {len(memories)} memories into: {result}") return result else: # Break memories into batches and consolidate each batch logger.debug( f"Total word count {total_word_count} exceeds {max_word_count}, breaking into batches" ) consolidated_results = [] batch_memories = [] current_batch_word_count = 0 for memory in memories: memory_word_count = self.count_memory_content_words([memory]) # If adding this memory would exceed the limit, process the current batch if ( current_batch_word_count + memory_word_count > max_word_count and batch_memories ): batch_result = self._consolidate( batch_memories, timestamp, enriched_context, persona ) if ( isinstance(batch_result, dict) and "consolidation" in batch_result ): consolidated_results.extend(batch_result["consolidation"]) logger.debug( f"Consolidated batch of {len(batch_memories)} memories" ) # Start new batch with current memory batch_memories = [memory] current_batch_word_count = memory_word_count else: # Add memory to current batch batch_memories.append(memory) current_batch_word_count += memory_word_count # Process the final batch if it has memories if batch_memories: batch_result = self._consolidate( batch_memories, timestamp, enriched_context, persona ) if isinstance(batch_result, dict) and "consolidation" in batch_result: consolidated_results.extend(batch_result["consolidation"]) logger.debug( f"Consolidated final batch of {len(batch_memories)} memories" ) logger.debug( f"Consolidated {len(memories)} memories into {len(consolidated_results)} consolidated memories across multiple batches" ) return {"consolidation": consolidated_results} @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 implementationAncestors
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": "Internal memory mechanism note: there were other memories 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 + memoriesAncestors
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 # TODOAncestors
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) try: logger.debug(f"Storing engram in semantic memory: {engram_doc}") self.semantic_grounding_connector.add_document(engram_doc) except Exception as e: logger.error( f"Error storing engram in semantic memory: {e}. Ignoring and continuing." ) 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", 10 ) # The minimum number of messages in an episode before it is considered valid. MAX_EPISODE_LENGTH = config_manager.get( "max_episode_length", 15 ) # 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.", "working_semantic_memory": 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.v2.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", encoding="utf-8", errors="replace" ) 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 ) # include recent episodic memory as part of the system prompt try: working_episodic_memory_text = ( self._render_recent_episodic_memories_for_prompt() ) except Exception: working_episodic_memory_text = "(No recent episodic memories available)" template_variables["working_episodic_memory"] = textwrap.indent( working_episodic_memory_text.strip(), " " ) return chevron.render(agent_prompt_template, template_variables) def _render_recent_episodic_memories_for_prompt(self) -> str: """ Builds a concise text block describing recent episodic events (oldest to newest), suitable to be embedded inside the system prompt. """ max_len = config_manager.get("max_content_display_length") episodes = self.retrieve_recent_memories(max_content_length=max_len) if not episodes: return "(No recent episodic memories available)" lines = [] for msg in episodes: try: role = msg.get("role") timestamp = msg.get("simulation_timestamp", None) timestamp_str = f" [@{timestamp}]" if timestamp else "" if role == "user": for s in msg.get("content", {}).get("stimuli", []): src = s.get("source") or "ENV" typ = s.get("type", "?") cnt = utils.break_text_at_length( s.get("content", ""), max_length=max_len ) lines.append( f"- [STIMULUS:{typ}] from {src}{timestamp_str}: {cnt}" ) elif role == "assistant": action = msg.get("content", {}).get("action", {}) or {} typ = action.get("type", "?") tgt = action.get("target") or "" cnt = utils.break_text_at_length( action.get("content", "") or "", max_length=max_len ) target_str = f" to {tgt}" if tgt else "" lines.append(f"- [ACTION:{typ}]{target_str}{timestamp_str}: {cnt}") # system messages are ignored in the episodic memory prompt except Exception: continue return "\n".join(lines) if lines else "(No recent episodic memories available)" 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} ] # NOTE: Episodic memories are now part of the system message (see template's 'Episodic Memory' section). self.current_messages.append( { "role": "user", "content": "You can now produce your next sequence of actions for this turn, in order, and finish with DONE.", } ) ######################################################################### # 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", encoding="utf-8", errors="replace") 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, communication_display: bool = 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. max_content_length (int): The maximum length of the content to display. Defaults to None, which uses the global configuration value. communication_display (bool): Whether to display the communication or not, will override the global setting if provided. Defaults to None. """ # 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(): pass # Helper to persist a single action and its side-effects def _commit_action(action, role, content): # check similarity quickly and replace by DONE if excessively repetitive next_action_similarity = utils.next_action_jaccard_similarity(self, action) 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." ) action = {"type": "DONE", "content": "", "target": ""} # Store a system feedback about the replacement 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(), } ) # Build a minimal content record for memory (single action) content_for_memory = {"action": action} if isinstance(content, dict) and "cognitive_state" in content: content_for_memory["cognitive_state"] = content["cognitive_state"] # Commit to episodic memory self.store_in_memory( { "role": role, "content": content_for_memory, "type": "action", "simulation_timestamp": self.iso_datetime(), } ) # buffer action for environment consumption self._actions_buffer.append(action) # Update mental state if isinstance(content, dict) and "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), ) # Display contents.append( { "action": action, "cognitive_state": content_for_memory.get("cognitive_state", {}), } ) if utils.first_non_none( communication_display, TinyPerson.communication_display ): self._display_communication( role=role, content={ "action": action, "cognitive_state": content_for_memory.get( "cognitive_state", {} ), }, kind="action", simplified=True, max_content_length=max_content_length, ) # Side-effects via mental faculties for faculty in self._mental_faculties: faculty.process_action(self, action) # count self.actions_count += 1 @repeat_on_error(retries=5, exceptions=[KeyError, TypeError]) def aux_act_once_sequence(): # ensure we have the latest prompt self.reset_prompt() # Provide latest perceived stimuli explicitly in a user JSON message def _latest_stimuli_payload(): try: # Walk recent episodic memory from newest to oldest to find the last stimulus block recent = self.episodic_memory.retrieve_recent() for msg in reversed(recent): if msg.get("role") == "user" and msg.get("type") == "stimulus": # Content already has {"stimuli": [...]} as stored by _observe return msg.get("content") except Exception: return None return None stimuli_payload = _latest_stimuli_payload() if stimuli_payload: self.current_messages.append( { "role": "user", "content": stimuli_payload, # dict will be JSON-serialized by the generator } ) actions_or_action, role, content, all_negative_feedbacks = ( self.action_generator.generate_next_actions(self, self.current_messages) ) # Normalize to list actions = ( actions_or_action if isinstance(actions_or_action, list) else [actions_or_action] ) # Enforce a reasonable cap per turn if len(actions) > 0 and len(actions) > TinyPerson.MAX_ACTIONS_BEFORE_DONE: actions = actions[: TinyPerson.MAX_ACTIONS_BEFORE_DONE] if actions[-1] is not None and isinstance(actions[-1], dict) and actions[-1].get("type") != "DONE": actions[-1] = {"type": "DONE", "content": "", "target": ""} # Commit each action in order for action in actions: # Skip None or non-dict actions if action is None or not isinstance(action, dict): logger.warning(f"[{self.name}] Skipping invalid action: {action}") continue _commit_action(action, role, content) if action.get("type") == "DONE": break # Option 1: run N actions (may span multiple turns if model emits only one) if n is not None: remaining = n while remaining > 0: aux_pre_act() # run one inference producing a sequence, but stop after committing the requested number before = self.actions_count aux_act_once_sequence() produced = self.actions_count - before remaining -= produced if remaining <= 0: break # If the last committed action was DONE, the turn ended; next loop will start a new turn # Option 2: run until DONE for this turn elif until_done: aux_pre_act() aux_act_once_sequence() # End of turn => consolidate episode memories 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, communication_display: bool = 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. max_content_length (int, optional): The maximum length of the content to display. Defaults to None, which uses the global configuration value. communication_display (bool): Whether to display the communication or not, will override the global setting if provided. Defaults to None. """ return self._observe( stimulus={ "type": "CONVERSATION", "content": speech, "source": name_or_empty(source), }, max_content_length=max_content_length, communication_display=communication_display, ) @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, communication_display: bool = None ): """ Observes a stimulus and updates its internal cognitive state. Args: stimulus (dict): The stimulus to observe. It must contain a 'type' and 'content' keys. max_content_length (int, optional): The maximum length of the content to display. Defaults to None, which uses the global configuration value. communication_display (bool): Whether to display the communication or not, will override the global setting if provided. Defaults to 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 utils.first_non_none( communication_display, 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, communication_display: bool = None, ): """ Convenience method that combines the `listen` and `act` methods. """ self.listen( speech, max_content_length=max_content_length, communication_display=communication_display, ) return self.act( return_actions=return_actions, max_content_length=max_content_length, communication_display=communication_display, ) @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_working_semantic_memory = ( self.retrieve_relevant_memories_for_current_context() ) self._mental_state["working_semantic_memory"] = current_working_semantic_memory self.reset_prompt() ########################################################### # Memory management ########################################################### def store_in_memory(self, value: Any) -> None: """ Stores a value in episodic memory and manages episode length. Args: value: The memory item to store (e.g., action, stimulus, thought) Returns: None """ 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) -> bool: """ Applies all memory consolidation or transformation processes appropriate to the conclusion of one simulation episode. Returns: bool: True if memories were successfully consolidated, False otherwise. """ # 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(), ).get("consolidation", None) 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.warning( 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: """ Retrieves memories relevant to the current context by combining current state with recent memories. Args: top_k (int): Number of top relevant memories to retrieve. Defaults to 7. Returns: list: List of relevant memories for the current context. """ # Check if continuous contextual semantic memory retrieval is enabled if not config_manager.get( "enable_continuous_contextual_semantic_memory_retrieval" ): return [] # Extract current mental state components context = self._mental_state.get("context", "") goals = self._mental_state.get("goals", "") attention = self._mental_state.get("attention", "") emotions = self._mental_state.get("emotions", "") # Retrieve recent memories efficiently recent_memories_list = self.retrieve_memories( first_n=10, last_n=20, max_content_length=500 ) recent_memories = "\n".join( [f" - {m.get('content', '')}" for m in recent_memories_list] ) # Build contextual target for memory retrieval using textwrap.dedent for cleaner formatting target = textwrap.dedent( f""" Current Context: {context} Current Goals: {goals} Current Attention: {attention} Current Emotions: {emotions} Selected Episodic Memories (from oldest to newest): {recent_memories} """ ).strip() logger.debug( f"[{self.name}] 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. Returns: dict or None: The last remembered action, or None if no suitable action found. """ 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]: action_content = candidate_item.get("content", {}).get("action", {}) action_type = action_content.get("type", "") if not ignore_done or action_type != "DONE": action = action_content break return action ########################################################### # Communication display and action execution ########################################################### @config_manager.config_defaults(max_content_length="max_content_display_length") def _display_communication( self, role, content, kind, simplified=True, max_content_length=None ): """ 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"] residence = self.get("residence") or "an unspecified location" base_biography = f"{self.name} is a {self._persona['age']} year old {occupation}, {self._persona['nationality']}, currently living in {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 @config_manager.config_defaults(max_content_length="max_content_display_length") def pp_current_interactions( self, simplified=True, skip_system=True, max_content_length=None, 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, ) ) @config_manager.config_defaults(max_content_length="max_content_display_length") def pp_last_interactions( self, n=3, simplified=True, skip_system=True, max_content_length=None, 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, ) ) @config_manager.config_defaults(max_content_length="max_content_display_length") def pretty_current_interactions( self, simplified=True, skip_system=True, max_content_length=None, 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) @config_manager.config_defaults(max_content_length="max_content_display_length") def _pretty_stimuli( self, role, content, simplified=True, max_content_length=None, ) -> 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) @config_manager.config_defaults(max_content_length="max_content_display_length") def _pretty_action( self, role, content, simplified=True, max_content_length=None, ) -> 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 = {} ####################################################################### # Cost statistics methods ####################################################################### @staticmethod def get_global_cost_stats(): """ Gets global cost statistics for all agents based on the current OpenAI client stats. Returns: dict: A dictionary containing: - base_stats: Raw stats from the OpenAI client - total_agents: Total number of agents - per_agent: Average resources per agent """ from tinytroupe.clients import client base_stats = client().get_cost_stats() num_agents = len(TinyPerson.all_agents) result = {"base_stats": base_stats, "total_agents": num_agents} # Per-agent statistics if num_agents > 0: result["per_agent"] = { "input_tokens": base_stats["input_tokens"] / num_agents, "output_tokens": base_stats["output_tokens"] / num_agents, "total_tokens": base_stats["total_tokens"] / num_agents, "model_calls": base_stats["model_calls"] / num_agents, "cached_calls": base_stats["cached_calls"] / num_agents, } else: result["per_agent"] = None return result @staticmethod def pretty_print_global_cost_stats(): """ Pretty prints global cost statistics for all agents. """ stats = TinyPerson.get_global_cost_stats() print("\n" + "=" * 70) print("GLOBAL COST STATISTICS (ALL AGENTS)") print("=" * 70) print(f"\nGlobal Context:") print(f" Total agents: {stats['total_agents']}") base = stats["base_stats"] print(f"\nTotal Resources Used:") print(f" Input tokens: {base['input_tokens']:,}") print(f" Output tokens: {base['output_tokens']:,}") print(f" Total tokens: {base['total_tokens']:,}") print(f" Model calls: {base['model_calls']:,}") print(f" Cached calls: {base['cached_calls']:,}") if stats["per_agent"] is not None: pa = stats["per_agent"] print(f"\nAverage Per Agent:") print(f" Input tokens: {pa['input_tokens']:,.1f}") print(f" Output tokens: {pa['output_tokens']:,.1f}") print(f" Total tokens: {pa['total_tokens']:,.1f}") print(f" Model calls: {pa['model_calls']:,.1f}") print(f" Cached calls: {pa['cached_calls']:,.1f}") print("=" * 70 + "\n")Ancestors
Class variables
var MAX_ACTIONS_BEFORE_DONEvar MAX_ACTION_SIMILARITYvar MAX_EPISODE_LENGTHvar MIN_EPISODE_LENGTHvar PP_TEXT_WIDTHvar all_agentsvar communication_display : boolvar serializable_attributesvar 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 get_global_cost_stats()-
Gets global cost statistics for all agents based on the current OpenAI client stats.
Returns
dict- A dictionary containing: - base_stats: Raw stats from the OpenAI client - total_agents: Total number of agents - per_agent: Average resources per agent
Expand source code
@staticmethod def get_global_cost_stats(): """ Gets global cost statistics for all agents based on the current OpenAI client stats. Returns: dict: A dictionary containing: - base_stats: Raw stats from the OpenAI client - total_agents: Total number of agents - per_agent: Average resources per agent """ from tinytroupe.clients import client base_stats = client().get_cost_stats() num_agents = len(TinyPerson.all_agents) result = {"base_stats": base_stats, "total_agents": num_agents} # Per-agent statistics if num_agents > 0: result["per_agent"] = { "input_tokens": base_stats["input_tokens"] / num_agents, "output_tokens": base_stats["output_tokens"] / num_agents, "total_tokens": base_stats["total_tokens"] / num_agents, "model_calls": base_stats["model_calls"] / num_agents, "cached_calls": base_stats["cached_calls"] / num_agents, } else: result["per_agent"] = None return result 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:strordict- 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 pretty_print_global_cost_stats()-
Pretty prints global cost statistics for all agents.
Expand source code
@staticmethod def pretty_print_global_cost_stats(): """ Pretty prints global cost statistics for all agents. """ stats = TinyPerson.get_global_cost_stats() print("\n" + "=" * 70) print("GLOBAL COST STATISTICS (ALL AGENTS)") print("=" * 70) print(f"\nGlobal Context:") print(f" Total agents: {stats['total_agents']}") base = stats["base_stats"] print(f"\nTotal Resources Used:") print(f" Input tokens: {base['input_tokens']:,}") print(f" Output tokens: {base['output_tokens']:,}") print(f" Total tokens: {base['total_tokens']:,}") print(f" Model calls: {base['model_calls']:,}") print(f" Cached calls: {base['cached_calls']:,}") if stats["per_agent"] is not None: pa = stats["per_agent"] print(f"\nAverage Per Agent:") print(f" Input tokens: {pa['input_tokens']:,.1f}") print(f" Output tokens: {pa['output_tokens']:,.1f}") print(f" Total tokens: {pa['total_tokens']:,.1f}") print(f" Model calls: {pa['model_calls']:,.1f}") print(f" Cached calls: {pa['cached_calls']:,.1f}") print("=" * 70 + "\n") 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) ‑> bool-
Applies all memory consolidation or transformation processes appropriate to the conclusion of one simulation episode.
Returns
bool- True if memories were successfully consolidated, False otherwise.
Expand source code
def consolidate_episode_memories(self) -> bool: """ Applies all memory consolidation or transformation processes appropriate to the conclusion of one simulation episode. Returns: bool: True if memories were successfully consolidated, False otherwise. """ # 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(), ).get("consolidation", None) 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.warning( 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", encoding="utf-8", errors="replace" ) 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 ) # include recent episodic memory as part of the system prompt try: working_episodic_memory_text = ( self._render_recent_episodic_memories_for_prompt() ) except Exception: working_episodic_memory_text = "(No recent episodic memories available)" template_variables["working_episodic_memory"] = textwrap.indent( working_episodic_memory_text.strip(), " " ) 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.
Returns
dictorNone- The last remembered action, or None if no suitable action found.
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. Returns: dict or None: The last remembered action, or None if no suitable action found. """ 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]: action_content = candidate_item.get("content", {}).get("action", {}) action_type = action_content.get("type", "") if not ignore_done or action_type != "DONE": action = action_content 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=None, first_n=None, last_n=None, include_omission_info: bool = True)-
Pretty prints the current messages.
Expand source code
@config_manager.config_defaults(max_content_length="max_content_display_length") def pp_current_interactions( self, simplified=True, skip_system=True, max_content_length=None, 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=None, include_omission_info: bool = True)-
Pretty prints the last n messages. Useful to examine the conclusion of an experiment.
Expand source code
@config_manager.config_defaults(max_content_length="max_content_display_length") def pp_last_interactions( self, n=3, simplified=True, skip_system=True, max_content_length=None, 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=None, first_n=None, last_n=None, include_omission_info: bool = True)-
Returns a pretty, readable, string with the current messages.
Expand source code
@config_manager.config_defaults(max_content_length="max_content_display_length") def pretty_current_interactions( self, simplified=True, skip_system=True, max_content_length=None, 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) -
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} ] # NOTE: Episodic memories are now part of the system message (see template's 'Episodic Memory' section). self.current_messages.append( { "role": "user", "content": "You can now produce your next sequence of actions for this turn, in order, and finish with DONE.", } ) 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-
Retrieves memories relevant to the current context by combining current state with recent memories.
Args
top_k:int- Number of top relevant memories to retrieve. Defaults to 7.
Returns
list- List of relevant memories for the current context.
Expand source code
def retrieve_relevant_memories_for_current_context(self, top_k=7) -> list: """ Retrieves memories relevant to the current context by combining current state with recent memories. Args: top_k (int): Number of top relevant memories to retrieve. Defaults to 7. Returns: list: List of relevant memories for the current context. """ # Check if continuous contextual semantic memory retrieval is enabled if not config_manager.get( "enable_continuous_contextual_semantic_memory_retrieval" ): return [] # Extract current mental state components context = self._mental_state.get("context", "") goals = self._mental_state.get("goals", "") attention = self._mental_state.get("attention", "") emotions = self._mental_state.get("emotions", "") # Retrieve recent memories efficiently recent_memories_list = self.retrieve_memories( first_n=10, last_n=20, max_content_length=500 ) recent_memories = "\n".join( [f" - {m.get('content', '')}" for m in recent_memories_list] ) # Build contextual target for memory retrieval using textwrap.dedent for cleaner formatting target = textwrap.dedent( f""" Current Context: {context} Current Goals: {goals} Current Attention: {attention} Current Emotions: {emotions} Selected Episodic Memories (from oldest to newest): {recent_memories} """ ).strip() logger.debug( f"[{self.name}] 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, Any] = 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 -
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) ‑> None-
Stores a value in episodic memory and manages episode length.
Args
value- The memory item to store (e.g., action, stimulus, thought)
Returns
None
Expand source code
def store_in_memory(self, value: Any) -> None: """ Stores a value in episodic memory and manages episode length. Args: value: The memory item to store (e.g., action, stimulus, thought) Returns: None """ 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 promptAncestors
Inherited members