agentsociety.memory.memory

Module Contents

Classes

KVMemory

MemoryNode

A data class representing a memory node.

StreamMemory

A class used to store and manage time-ordered stream information.

Memory

A class to manage different types of memory (status and stream).

Data

__all__

API

agentsociety.memory.memory.__all__

[‘KVMemory’, ‘StreamMemory’, ‘Memory’]

class agentsociety.memory.memory.KVMemory(memory_config: agentsociety.agent.memory_config_generator.MemoryConfig, embedding: fastembed.SparseTextEmbedding)

Initialization

Initialize the KVMemory with a unified memory configuration.

  • Args:

    • memory_config (MemoryConfig): The unified memory configuration.

    • embedding (SparseTextEmbedding): The embedding object.

async initialize_embeddings() → None

Initialize embeddings for all fields that require them.

_generate_semantic_text(key: str, value: Any) → str

Generate semantic text for a given key and value.

async search(query: str, top_k: int = 3, filter: Optional[dict] = None) → str

Search for relevant memories based on the provided query.

  • Args:

    • query (str): The text query to search for.

    • top_k (int, optional): Number of top relevant memories to return. Defaults to 3.

    • filter (Optional[dict], optional): Additional filters for the search. Defaults to None.

  • Returns:

    • str: Formatted string of the search results.

should_embed(key: str) → bool
async get(key: Any, default_value: Optional[Any] = None) → Any

Retrieve a value from the memory.

  • Args:

    • key (Any): The key to retrieve.

    • default_value (Optional[Any], optional): Default value if the key is not found. Defaults to None.

  • Returns:

    • Any: The retrieved value or the default value if the key is not found.

  • Raises:

    • KeyError: If the key is not found in any of the memory sections and no default value is provided.

async update(key: Any, value: Any, mode: Union[Literal[replace], Literal[merge]] = 'replace') → None

Update a value in the memory and refresh embeddings if necessary.

  • Args:

    • key (Any): The key to update.

    • value (Any): The new value to set.

    • mode (Union[Literal[“replace”], Literal[“merge”]], optional): Update mode. Defaults to “replace”.

  • Raises:

    • ValueError: If an invalid update mode is provided.

    • KeyError: If the key is not found in any of the memory sections.

async export(keys: list[str]) → dict[str, Any]

Export the memory of a given keys.

class agentsociety.memory.memory.MemoryNode

A data class representing a memory node.

  • Attributes:

    • topic: The topic associated with the memory node.

    • day: Day of the event or memory.

    • t: Time stamp or order.

    • location: Location where the event occurred.

    • description: Description of the memory.

    • cognition_id: ID related to cognitive memory (optional).

    • id: Unique ID for this memory node (optional).

topic: str

None

day: int

None

t: int

None

location: str

None

description: str

None

cognition_id: Optional[int]

None

id: Optional[int]

None

class agentsociety.memory.memory.StreamMemory(environment: Optional[agentsociety.environment.Environment], status_memory: agentsociety.memory.memory.KVMemory, embedding: fastembed.SparseTextEmbedding, max_len: int = 1000)

A class used to store and manage time-ordered stream information.

  • Attributes:

    • _memories: A deque to store memory nodes with a maximum length limit.

    • _memory_id_counter: An internal counter to generate unique IDs for each new memory node.

    • _vectorstore: The Faiss query object for search functionality.

    • _status_memory: The status memory object.

    • _environment: The environment object.

Initialization

Initialize an instance of StreamMemory.

  • Args:

    • environment (Environment): The environment object.

    • status_memory (KVMemory): The status memory object.

    • embedding (SparseTextEmbedding): The embedding object.

    • max_len (int): Maximum length of the deque. Default is 1000.

async add(topic: str, description: str) → int

A generic method for adding a memory node and returning the memory node ID.

  • Args:

    • topic (str): The topic associated with the memory node.

    • description (str): Description of the memory.

  • Returns:

    • int: The unique ID of the newly added memory node.

Retrieve the related cognition memory node by its ID.

  • Args:

    • memory_id (int): The ID of the memory to find related cognition for.

  • Returns:

    • Optional[MemoryNode]: The related cognition memory node, if found; otherwise, None.

async format_memory(memories: list[agentsociety.memory.memory.MemoryNode]) → str

Format a list of memory nodes into a readable string representation.

  • Args:

    • memories (list[MemoryNode]): List of MemoryNode objects to format.

  • Returns:

    • str: A formatted string containing the details of each memory node.

async get_by_ids(memory_ids: list[int]) → str

Retrieve memories by specified IDs

async search(query: str, topic: Optional[str] = None, top_k: int = 3, day_range: Optional[tuple[int, int]] = None, time_range: Optional[tuple[int, int]] = None) → str

Search stream memory with optional filters and return formatted results.

  • Args:

    • query (str): The text to use for searching.

    • topic (Optional[str], optional): Filter memories by this topic. Defaults to None.

    • top_k (int, optional): Number of top relevant memories to return. Defaults to 3.

    • day_range (Optional[tuple[int, int]], optional): Tuple of start and end days for filtering. Defaults to None.

    • time_range (Optional[tuple[int, int]], optional): Tuple of start and end times for filtering. Defaults to None.

  • Returns:

    • str: Formatted string of the search results.

async search_today(query: str = '', topic: Optional[str] = None, top_k: int = 100) → str

Search all memory events from today

  • Args:

    • query (str): Optional query text, returns all memories of the day if empty. Defaults to “”.

    • topic (Optional[str]): Optional memory topic for filtering specific types of memories. Defaults to None.

    • top_k (int): Number of most relevant memories to return. Defaults to 100.

  • Returns:

    • str: Formatted text of today’s memories.

async add_cognition_to_memory(memory_ids: list[int], cognition: str) → None

Add cognition to existing memory nodes.

  • Args:

    • memory_ids (list[int]): List of IDs of the memories to which cognition will be added.

    • cognition (str): Description of the cognition to add.

async get_all() → list[dict]

Retrieve all stream memory nodes as dictionaries.

  • Returns:

    • list[dict]: List of all memory nodes as dictionaries.

class agentsociety.memory.memory.Memory(environment: Optional[agentsociety.environment.Environment], embedding: fastembed.SparseTextEmbedding, memory_config: agentsociety.agent.memory_config_generator.MemoryConfig)

A class to manage different types of memory (status and stream).

  • Attributes:

    • _status (KVMemory): Stores status-related data.

    • _stream (StreamMemory): Stores stream-related data.

Initialization

Initializes the Memory with a unified memory configuration.

  • Args:

    • environment (Environment): The environment object.

    • embedding (SparseTextEmbedding): The embedding object.

    • memory_config (MemoryConfig): The unified memory configuration.

property status: agentsociety.memory.memory.KVMemory
property stream: agentsociety.memory.memory.StreamMemory
async initialize_embeddings()

Initialize embeddings within the status memory.

  • Description:

    • Asynchronously initializes embeddings for the status memory component, which prepares the system for performing searches.