core / action / action_library

core.action.action_library

Updated ActionLibrary that delegates database operations to the new DatabaseInterface.

Created on Thu Mar 27 21:29:03 2025 Author: zfoong

ActionLibrary

Manages storing, retrieving, and modifying actions via DatabaseInterface.

Source code in core\action\action_library.py

class ActionLibrary:
    """
    Manages storing, retrieving, and modifying actions via DatabaseInterface.
    """

    def __init__(self, llm_interface, db_interface: DatabaseInterface):
        """
        Initialize the library responsible for persisting actions.

        Args:
            llm_interface: LLM client used elsewhere for generating actions.
            db_interface: Database gateway that handles MongoDB/ChromaDB storage.
        """
        self.llm_interface = llm_interface
        self.db_interface = db_interface

    def store_action(self, action: Action):
        """
        Persist an action definition and stamp its update time.

        Args:
            action: Action instance to serialize and store.
        """
        action_dict = action.to_dict()
        action_dict["updatedAt"] = datetime.datetime.utcnow().isoformat()
        self.db_interface.store_action(action_dict)

    def retrieve_action(self, action_name: str) -> Optional[Action]:
        """
        Fetch a single action by name.

        Args:
            action_name: Case-insensitive name of the action to retrieve.

        Returns:
            Optional[Action]: Hydrated action instance if found, otherwise ``None``.
        """
        action_data = self.db_interface.get_action(action_name)
        if action_data:
            return Action.from_dict(action_data)
        return None

    def retrieve_default_action(self) -> List[Action]:
        """
        Retrieve actions marked as defaults.
        These actions are always available to the agents regardless of the mode.

        Returns:
            List[Action]: All default actions stored in the database.
        """
        docs = self.db_interface.list_actions(default=True)
        return [Action.from_dict(doc) for doc in docs]

    def get_default_action_names(self) -> set[str]:
        return {
            action.name
            for action in self.retrieve_default_action()
        }

    def search_action(self, query: str, top_k=50) -> List[str]:
        """
        Search for actions using vector similarity.

        Args:
            query: Natural-language description of the desired action.
            top_k: Maximum number of action names to return.

        Returns:
            List[str]: Ranked list of matching action names.
        """
        return self.db_interface.search_actions(query, top_k)

    def delete_action(self, action_name: str):
        """Deletes an action from both MongoDB and ChromaDB."""
        self.db_interface.delete_action(action_name)

__init__(llm_interface, db_interface)

Initialize the library responsible for persisting actions.

Parameters:

Name	Type	Description	Default
llm_interface		LLM client used elsewhere for generating actions.	required
db_interface	DatabaseInterface	Database gateway that handles MongoDB/ChromaDB storage.	required

Source code in core\action\action_library.py

def __init__(self, llm_interface, db_interface: DatabaseInterface):
    """
    Initialize the library responsible for persisting actions.

    Args:
        llm_interface: LLM client used elsewhere for generating actions.
        db_interface: Database gateway that handles MongoDB/ChromaDB storage.
    """
    self.llm_interface = llm_interface
    self.db_interface = db_interface

store_action(action)

Persist an action definition and stamp its update time.

Parameters:

Name	Type	Description	Default
action	Action	Action instance to serialize and store.	required

Source code in core\action\action_library.py

def store_action(self, action: Action):
    """
    Persist an action definition and stamp its update time.

    Args:
        action: Action instance to serialize and store.
    """
    action_dict = action.to_dict()
    action_dict["updatedAt"] = datetime.datetime.utcnow().isoformat()
    self.db_interface.store_action(action_dict)

retrieve_action(action_name)

Fetch a single action by name.

Parameters:

Name	Type	Description	Default
action_name	str	Case-insensitive name of the action to retrieve.	required

Returns:

Type	Description
Optional[Action]	Optional[Action]: Hydrated action instance if found, otherwise None.

Source code in core\action\action_library.py

def retrieve_action(self, action_name: str) -> Optional[Action]:
    """
    Fetch a single action by name.

    Args:
        action_name: Case-insensitive name of the action to retrieve.

    Returns:
        Optional[Action]: Hydrated action instance if found, otherwise ``None``.
    """
    action_data = self.db_interface.get_action(action_name)
    if action_data:
        return Action.from_dict(action_data)
    return None

retrieve_default_action()

Retrieve actions marked as defaults. These actions are always available to the agents regardless of the mode.

Returns:

Type	Description
List[Action]	List[Action]: All default actions stored in the database.

Source code in core\action\action_library.py

def retrieve_default_action(self) -> List[Action]:
    """
    Retrieve actions marked as defaults.
    These actions are always available to the agents regardless of the mode.

    Returns:
        List[Action]: All default actions stored in the database.
    """
    docs = self.db_interface.list_actions(default=True)
    return [Action.from_dict(doc) for doc in docs]

search_action(query, top_k=50)

Search for actions using vector similarity.

Parameters:

Name	Type	Description	Default
query	str	Natural-language description of the desired action.	required
top_k		Maximum number of action names to return.	50

Returns:

Type	Description
List[str]	List[str]: Ranked list of matching action names.

Source code in core\action\action_library.py

def search_action(self, query: str, top_k=50) -> List[str]:
    """
    Search for actions using vector similarity.

    Args:
        query: Natural-language description of the desired action.
        top_k: Maximum number of action names to return.

    Returns:
        List[str]: Ranked list of matching action names.
    """
    return self.db_interface.search_actions(query, top_k)

delete_action(action_name)

Deletes an action from both MongoDB and ChromaDB.

Source code in core\action\action_library.py

def delete_action(self, action_name: str):
    """Deletes an action from both MongoDB and ChromaDB."""
    self.db_interface.delete_action(action_name)