Managers

UserManager

The UserManager is designed to manage, for each user:

• TreeManager
• ClientManager
• FrontendConfig

It contains methods for creating and updating these objects across a range of users, stored in a dictionary. It can be used as a dependency injection container for FastAPI. The user manager/tree manager is decoupled from the core of the Elysia package functionality. It is designed to be used to manage separate Elysia instances, set of trees (via tree managers), configs, etc.

Source code in elysia/api/services/user.py

class UserManager:
    """
    The UserManager is designed to manage, for each user:

    - TreeManager
    - ClientManager
    - FrontendConfig

    It contains methods for creating and updating these objects across a range of users, stored in a dictionary.
    It can be used as a dependency injection container for FastAPI.
    The user manager/tree manager is decoupled from the core of the Elysia package functionality.
    It is designed to be used to manage separate Elysia instances, set of trees (via tree managers), configs, etc.
    """

    def __init__(
        self,
        user_timeout: datetime.timedelta | int | None = None,
    ):
        """
        Args:
            user_timeout (datetime.timedelta | int | None): Optional.
                The length of time a user can be idle before being timed out.
                Defaults to 20 minutes or the value of the USER_TIMEOUT environment variable.
                If an integer is provided, it is interpreted as the number of minutes.
        """
        if user_timeout is None:
            self.user_timeout = datetime.timedelta(
                minutes=int(os.environ.get("USER_TIMEOUT", 20))
            )
        elif isinstance(user_timeout, int):
            self.user_timeout = datetime.timedelta(minutes=user_timeout)
        else:
            self.user_timeout = user_timeout

        self.manager_id = random.randint(0, 1000000)
        self.date_of_reset = None
        self.users = {}

    def user_exists(self, user_id: str):
        return user_id in self.users

    async def update_config(
        self,
        user_id: str,
        conversation_id: str | None = None,
        config_id: str | None = None,
        config_name: str | None = None,
        settings: dict[str, Any] | None = None,
        style: str | None = None,
        agent_description: str | None = None,
        end_goal: str | None = None,
        branch_initialisation: str | None = None,
    ):
        local_user = await self.get_user_local(user_id)
        local_user["tree_manager"].update_config(
            conversation_id,
            config_id,
            config_name,
            settings,
            style,
            agent_description,
            end_goal,
            branch_initialisation,
        )

        await local_user["client_manager"].reset_keys(
            wcd_url=local_user["tree_manager"].settings.WCD_URL,
            wcd_api_key=local_user["tree_manager"].settings.WCD_API_KEY,
            api_keys=local_user["tree_manager"].settings.API_KEYS,
        )

    async def update_frontend_config(
        self,
        user_id: str,
        config: dict[str, Any],
    ):
        local_user = await self.get_user_local(user_id)
        frontend_config: FrontendConfig = local_user["frontend_config"]
        await frontend_config.configure(**config)

    async def add_user_local(
        self,
        user_id: str,
        config: Config | None = None,
    ):
        """
        Add a user to the UserManager.

        Args:
            user_id (str): Required. The unique identifier for the user.
            config (Config): Required. The config for the user.
        """

        # add user if it doesn't exist
        if user_id not in self.users:
            self.users[user_id] = {}

            fe_config = await load_frontend_config_from_file(user_id, logger)

            self.users[user_id]["frontend_config"] = fe_config

            self.users[user_id]["tree_manager"] = TreeManager(
                user_id=user_id,
                config=config,
                tree_timeout=fe_config.config["tree_timeout"],
            )

            # client manager starts with env variables, when config is updated, api keys are updated
            self.users[user_id]["client_manager"] = ClientManager(
                logger=logger,
                client_timeout=fe_config.config["client_timeout"],
                settings=self.users[user_id]["tree_manager"].config.settings,
            )

    async def get_user_local(self, user_id: str):
        """
        Return a local user object.
        Will raise a ValueError if the user is not found.

        Args:
            user_id (str): Required. The unique identifier for the user.

        Returns:
            (dict): A local user object, containing a TreeManager ("tree_manager"),
                Frontend Config ("frontend_config") and ClientManager ("client_manager").
        """

        if user_id not in self.users:
            raise ValueError(
                f"User {user_id} not found. Please initialise a user first (by calling `add_user_local`)."
            )

        # update last request (adds last_request to user)
        await self.update_user_last_request(user_id)

        return self.users[user_id]

    async def get_tree(self, user_id: str, conversation_id: str):
        """
        Get a tree for a user.
        Will raise a ValueError if the user is not found, or the tree is not found.

        Args:
            user_id (str): Required. The unique identifier for the user.
            conversation_id (str): Required. The unique identifier for the conversation.

        Returns:
            (Tree): The tree.
        """
        local_user = await self.get_user_local(user_id)
        return local_user["tree_manager"].get_tree(conversation_id)

    async def check_all_trees_timeout(self):
        """
        Check all trees in all TreeManagers across all users and remove any that have not been active in the last tree_timeout.
        """
        for user_id in self.users:
            self.users[user_id]["tree_manager"].check_all_trees_timeout()

    def check_user_timeout(self, user_id: str):
        """
        Check if a user has been idle for the last user_timeout.

        Args:
            user_id (str): The user ID which contains the user.

        Returns:
            (bool): True if the user has been idle for the last user_timeout, False otherwise.
        """
        # if user not found, return True
        if user_id not in self.users:
            return True

        # Remove any trees that have not been active in the last user_timeout
        # if (
        #     "last_request" in self.users[user_id]
        #     and datetime.datetime.now() - self.users[user_id]["last_request"]
        #     > self.user_timeout
        # ):
        #     return True

        return False

    async def check_all_users_timeout(self):
        """
        Check all users in the UserManager and remove any that have not been active in the last user_timeout.
        """
        if self.user_timeout == datetime.timedelta(minutes=0):
            return

        for user_id in self.users:
            if self.check_user_timeout(user_id):
                del self.users[user_id]

    async def check_restart_clients(self):
        """
        Check all clients in all ClientManagers across all users and run the restart_client() method (for sync and async clients).
        The restart_client() methods will check if the client has been inactive for the last client_timeout minutes (set in init).
        """
        for user_id in self.users:
            if (
                "client_manager" in self.users[user_id]
                and self.users[user_id]["client_manager"].is_client
            ):
                await self.users[user_id]["client_manager"].restart_client()
                await self.users[user_id]["client_manager"].restart_async_client()

    async def close_all_clients(self):
        """
        Close all clients in all ClientManagers across all users.
        """
        for user_id in self.users:
            if "client_manager" in self.users[user_id]:
                await self.users[user_id]["client_manager"].close_clients()

    async def initialise_tree(
        self,
        user_id: str,
        conversation_id: str,
        low_memory: bool = False,
    ):
        """
        Initialises a tree for a user for an existing user at user_id.
        Requires a user to already exist in the UserManager, via `add_user_local`.
        This is a wrapper for the TreeManager.add_tree() method.

        Args:
            user_id (str): Required. The unique identifier for the user.
            conversation_id (str): Required. The unique identifier for a new conversation within the tree manager.
            low_memory (bool): Optional. Whether to use low memory mode for the tree.
                Controls the LM history being saved in the tree, and some other variables.
                Defaults to False.
        """
        # self.add_user_local(user_id)
        local_user = await self.get_user_local(user_id)
        tree_manager: TreeManager = local_user["tree_manager"]
        if not tree_manager.tree_exists(conversation_id):
            tree_manager.add_tree(
                conversation_id,
                low_memory,
            )
        return tree_manager.get_tree(conversation_id)

    async def save_tree(
        self,
        user_id: str,
        conversation_id: str,
        wcd_url: str | None = None,
        wcd_api_key: str | None = None,
    ):
        """
        Save a tree to a Weaviate instance (set in the frontend config).
        This is a wrapper for the TreeManager.save_tree_weaviate() method.

        Args:
            user_id (str): Required. The unique identifier for the user stored in the UserManager.
            conversation_id (str): Required. The unique identifier for the conversation for the user.
        """

        local_user = await self.get_user_local(user_id)
        tree_manager: TreeManager = local_user["tree_manager"]

        if wcd_url is None or wcd_api_key is None:
            save_location_client_manager = local_user[
                "frontend_config"
            ].save_location_client_manager
        else:
            save_location_client_manager = ClientManager(
                logger=logger,
                wcd_url=wcd_url,
                wcd_api_key=wcd_api_key,
            )

        await tree_manager.save_tree_weaviate(
            conversation_id, save_location_client_manager
        )

    async def check_tree_exists_weaviate(
        self,
        user_id: str,
        conversation_id: str,
        wcd_url: str | None = None,
        wcd_api_key: str | None = None,
    ):
        """
        Check if a tree exists in a Weaviate instance (set in the frontend config).

        Args:
            user_id (str): Required. The unique identifier for the user stored in the UserManager.
            conversation_id (str): Required. The unique identifier for the conversation for the user.
            wcd_url (str | None): Required. The URL of the Weaviate Cloud Database instance used to save the tree.
                Defaults to the value of the `wcd_url` setting in the frontend config.
            wcd_api_key (str | None): Required. The API key for the Weaviate Cloud Database instance used to save the tree.
                Defaults to the value of the `wcd_api_key` setting in the frontend config.

        Returns:
            (bool): True if the tree exists in the Weaviate instance, False otherwise.
        """

        local_user = await self.get_user_local(user_id)
        tree_manager: TreeManager = local_user["tree_manager"]

        if wcd_url is None or wcd_api_key is None:
            save_location_client_manager = local_user[
                "frontend_config"
            ].save_location_client_manager
        else:
            save_location_client_manager = ClientManager(
                logger=logger,
                wcd_url=wcd_url,
                wcd_api_key=wcd_api_key,
            )

        return await tree_manager.check_tree_exists_weaviate(
            conversation_id, save_location_client_manager
        )

    async def load_tree(
        self,
        user_id: str,
        conversation_id: str,
        wcd_url: str | None = None,
        wcd_api_key: str | None = None,
    ):
        """
        Load a tree from a Weaviate instance (set in the frontend config).
        This is a wrapper for the TreeManager.load_tree_weaviate() method.

        Args:
            user_id (str): Required. The unique identifier for the user stored in the UserManager.
            conversation_id (str): Required. The unique identifier for the conversation for the user.
            wcd_url (str | None): Required. The URL of the Weaviate Cloud Database instance used to save the tree.
                Defaults to the value of the `wcd_url` setting in the frontend config.
            wcd_api_key (str | None): Required. The API key for the Weaviate Cloud Database instance used to save the tree.
                Defaults to the value of the `wcd_api_key` setting in the frontend config.

        Returns:
            (list[dict]): A list of dictionaries, each containing a frontend payload that was used to generate the tree.
                The list is ordered by the time the payload was originally sent to the frontend (at the time it was saved).
        """

        local_user = await self.get_user_local(user_id)
        tree_manager: TreeManager = local_user["tree_manager"]

        if wcd_url is None or wcd_api_key is None:
            save_location_client_manager = local_user[
                "frontend_config"
            ].save_location_client_manager
        else:
            save_location_client_manager = ClientManager(
                logger=logger,
                wcd_url=wcd_url,
                wcd_api_key=wcd_api_key,
            )

        return await tree_manager.load_tree_weaviate(
            conversation_id, save_location_client_manager
        )

    async def delete_tree(
        self,
        user_id: str,
        conversation_id: str,
        wcd_url: str | None = None,
        wcd_api_key: str | None = None,
    ):
        """
        Delete a saved tree from a Weaviate instance (set in the frontend config).
        Also delete the tree from the local tree manager.
        This is a wrapper for the TreeManager.delete_tree_weaviate() method and the TreeManager.delete_tree_local() method.

        Args:
            user_id (str): Required. The unique identifier for the user stored in the UserManager.
            conversation_id (str): Required. The unique identifier for the conversation for the user.
            wcd_url (str | None): Required. The URL of the Weaviate Cloud Database instance used to save the tree.
                Defaults to the value of the `wcd_url` setting in the frontend config.
            wcd_api_key (str | None): Required. The API key for the Weaviate Cloud Database instance used to save the tree.
                Defaults to the value of the `wcd_api_key` setting in the frontend config.
        """

        local_user = await self.get_user_local(user_id)
        tree_manager: TreeManager = local_user["tree_manager"]

        if wcd_url is None or wcd_api_key is None:
            save_location_client_manager = local_user[
                "frontend_config"
            ].save_location_client_manager
        else:
            save_location_client_manager = ClientManager(
                logger=logger,
                wcd_url=wcd_url,
                wcd_api_key=wcd_api_key,
            )

        await tree_manager.delete_tree_weaviate(
            conversation_id, save_location_client_manager
        )
        tree_manager.delete_tree_local(conversation_id)

    async def get_saved_trees(
        self,
        user_id: str,
        wcd_url: str | None = None,
        wcd_api_key: str | None = None,
    ):
        """
        Get all saved trees from a Weaviate instance (set in the frontend config).

        Args:
            user_id (str): Required. The unique identifier for the user stored in the UserManager.
            wcd_url (str | None): Required. The URL of the Weaviate Cloud Database instance used to save the tree.
                Defaults to the value of the `wcd_url` setting in the frontend config.
            wcd_api_key (str | None): Required. The API key for the Weaviate Cloud Database instance used to save the tree.
                Defaults to the value of the `wcd_api_key` setting in the frontend config.

        Returns:
            (dict): A dictionary whose keys are the conversation IDs and whose values are dictionaries containing the title and last update time of the tree.
                E.g.
                ```
                {
                    "12345678-XXX-YYYY-ZZZZ": {
                        "title": "Query Request",
                        "last_update_time": "2025-06-07T10:06:47.376000Z"
                    }
                }
                ```
        """
        local_user = await self.get_user_local(user_id)

        if wcd_url is None or wcd_api_key is None:
            save_location_client_manager = local_user[
                "frontend_config"
            ].save_location_client_manager
        else:
            save_location_client_manager = ClientManager(
                logger=logger,
                wcd_url=wcd_url,
                wcd_api_key=wcd_api_key,
            )

        return await get_saved_trees_weaviate(
            "ELYSIA_TREES__", save_location_client_manager, user_id
        )

    async def update_user_last_request(self, user_id: str):
        self.users[user_id]["last_request"] = datetime.datetime.now()

    def check_tree_timeout(self, user_id: str, conversation_id: str):
        if user_id not in self.users:
            return True
        elif conversation_id not in self.users[user_id]["tree_manager"].trees:
            return True
        return False

    async def process_tree(
        self,
        query: str,
        user_id: str,
        conversation_id: str,
        query_id: str,
        training_route: str = "",
        collection_names: list[str] = [],
        save_trees_to_weaviate: bool | None = None,
        wcd_url: str | None = None,
        wcd_api_key: str | None = None,
    ):
        """
        Wrapper for the TreeManager.process_tree() method.
        Which itself is a wrapper for the Tree.async_run() method.
        This is an async generator which yields results from the tree.async_run() method.
        Automatically sends error payloads if the user or tree has been timed out.

        Args:
            query (str): Required. The user input/prompt to process in the decision tree.
            user_id (str): Required. The unique identifier for the user.
            conversation_id (str): Required. The conversation ID which contains the tree.
                This should be the same conversation ID as the one used to initialise the tree (see `initialise_tree`).
            query_id (str): Required. A unique identifier for the query.
            training_route (str): Optional. The training route, a string of the form "tool1/tool2/tool1" etc.
                See the `tree.async_run()` method for more details.
            collection_names (list[str]): Optional. A list of collection names to use in the query.
                If not supplied, all collections will be used.
            save_trees_to_weaviate (bool | None): Optional. Whether to save the trees to a Weaviate instance,
                after the process_tree() method has finished.
                Defaults to the value of the `save_trees_to_weaviate` setting in the frontend config.
            wcd_url (str | None): Required. The URL of the Weaviate Cloud Database instance used to save the tree.
                Defaults to the value of the `wcd_url` setting in the frontend config.
            wcd_api_key (str | None): Required. The API key for the Weaviate Cloud Database instance used to save the tree.
                Defaults to the value of the `wcd_api_key` setting in the frontend config.
        """

        if self.check_user_timeout(user_id):
            user_timeout_error = UserTimeoutError()
            error_payload = await user_timeout_error.to_frontend(
                user_id, conversation_id, query_id
            )
            yield error_payload
            return

        if self.check_tree_timeout(user_id, conversation_id):
            if await self.check_tree_exists_weaviate(user_id, conversation_id):
                await self.load_tree(user_id, conversation_id)
            else:
                tree_timeout_error = TreeTimeoutError()
                error_payload = await tree_timeout_error.to_frontend(
                    user_id, conversation_id, query_id
                )
                yield error_payload
                return

        local_user = await self.get_user_local(user_id)
        await self.update_user_last_request(user_id)

        tree_manager: TreeManager = local_user["tree_manager"]

        async for yielded_result in tree_manager.process_tree(
            query,
            conversation_id,
            query_id,
            training_route,
            collection_names,
            local_user["client_manager"],
        ):
            yield yielded_result
            await self.update_user_last_request(user_id)

        if save_trees_to_weaviate is None:
            frontend_config: FrontendConfig = local_user["frontend_config"]
            save_trees_to_weaviate = frontend_config.config["save_trees_to_weaviate"]

        if save_trees_to_weaviate:
            await self.save_tree(user_id, conversation_id, wcd_url, wcd_api_key)

__init__(user_timeout=None)

Parameters:

Name	Type	Description	Default
user_timeout	timedelta | int | None	Optional. The length of time a user can be idle before being timed out. Defaults to 20 minutes or the value of the USER_TIMEOUT environment variable. If an integer is provided, it is interpreted as the number of minutes.	None

Source code in elysia/api/services/user.py

def __init__(
    self,
    user_timeout: datetime.timedelta | int | None = None,
):
    """
    Args:
        user_timeout (datetime.timedelta | int | None): Optional.
            The length of time a user can be idle before being timed out.
            Defaults to 20 minutes or the value of the USER_TIMEOUT environment variable.
            If an integer is provided, it is interpreted as the number of minutes.
    """
    if user_timeout is None:
        self.user_timeout = datetime.timedelta(
            minutes=int(os.environ.get("USER_TIMEOUT", 20))
        )
    elif isinstance(user_timeout, int):
        self.user_timeout = datetime.timedelta(minutes=user_timeout)
    else:
        self.user_timeout = user_timeout

    self.manager_id = random.randint(0, 1000000)
    self.date_of_reset = None
    self.users = {}

add_user_local(user_id, config=None) async

Add a user to the UserManager.

Parameters:

Name	Type	Description	Default
user_id	str	Required. The unique identifier for the user.	required
config	Config	Required. The config for the user.	None

Source code in elysia/api/services/user.py

async def add_user_local(
    self,
    user_id: str,
    config: Config | None = None,
):
    """
    Add a user to the UserManager.

    Args:
        user_id (str): Required. The unique identifier for the user.
        config (Config): Required. The config for the user.
    """

    # add user if it doesn't exist
    if user_id not in self.users:
        self.users[user_id] = {}

        fe_config = await load_frontend_config_from_file(user_id, logger)

        self.users[user_id]["frontend_config"] = fe_config

        self.users[user_id]["tree_manager"] = TreeManager(
            user_id=user_id,
            config=config,
            tree_timeout=fe_config.config["tree_timeout"],
        )

        # client manager starts with env variables, when config is updated, api keys are updated
        self.users[user_id]["client_manager"] = ClientManager(
            logger=logger,
            client_timeout=fe_config.config["client_timeout"],
            settings=self.users[user_id]["tree_manager"].config.settings,
        )

check_all_trees_timeout() async

Check all trees in all TreeManagers across all users and remove any that have not been active in the last tree_timeout.

Source code in elysia/api/services/user.py

async def check_all_trees_timeout(self):
    """
    Check all trees in all TreeManagers across all users and remove any that have not been active in the last tree_timeout.
    """
    for user_id in self.users:
        self.users[user_id]["tree_manager"].check_all_trees_timeout()

check_all_users_timeout() async

Check all users in the UserManager and remove any that have not been active in the last user_timeout.

Source code in elysia/api/services/user.py

async def check_all_users_timeout(self):
    """
    Check all users in the UserManager and remove any that have not been active in the last user_timeout.
    """
    if self.user_timeout == datetime.timedelta(minutes=0):
        return

    for user_id in self.users:
        if self.check_user_timeout(user_id):
            del self.users[user_id]

check_restart_clients() async

Check all clients in all ClientManagers across all users and run the restart_client() method (for sync and async clients). The restart_client() methods will check if the client has been inactive for the last client_timeout minutes (set in init).

Source code in elysia/api/services/user.py

async def check_restart_clients(self):
    """
    Check all clients in all ClientManagers across all users and run the restart_client() method (for sync and async clients).
    The restart_client() methods will check if the client has been inactive for the last client_timeout minutes (set in init).
    """
    for user_id in self.users:
        if (
            "client_manager" in self.users[user_id]
            and self.users[user_id]["client_manager"].is_client
        ):
            await self.users[user_id]["client_manager"].restart_client()
            await self.users[user_id]["client_manager"].restart_async_client()

check_tree_exists_weaviate(user_id, conversation_id, wcd_url=None, wcd_api_key=None) async

Check if a tree exists in a Weaviate instance (set in the frontend config).

Parameters:

Name	Type	Description	Default
user_id	str	Required. The unique identifier for the user stored in the UserManager.	required
conversation_id	str	Required. The unique identifier for the conversation for the user.	required
wcd_url	str | None	Required. The URL of the Weaviate Cloud Database instance used to save the tree. Defaults to the value of the wcd_url setting in the frontend config.	None
wcd_api_key	str | None	Required. The API key for the Weaviate Cloud Database instance used to save the tree. Defaults to the value of the wcd_api_key setting in the frontend config.	None

Returns:

Type	Description
bool	True if the tree exists in the Weaviate instance, False otherwise.

Source code in elysia/api/services/user.py

async def check_tree_exists_weaviate(
    self,
    user_id: str,
    conversation_id: str,
    wcd_url: str | None = None,
    wcd_api_key: str | None = None,
):
    """
    Check if a tree exists in a Weaviate instance (set in the frontend config).

    Args:
        user_id (str): Required. The unique identifier for the user stored in the UserManager.
        conversation_id (str): Required. The unique identifier for the conversation for the user.
        wcd_url (str | None): Required. The URL of the Weaviate Cloud Database instance used to save the tree.
            Defaults to the value of the `wcd_url` setting in the frontend config.
        wcd_api_key (str | None): Required. The API key for the Weaviate Cloud Database instance used to save the tree.
            Defaults to the value of the `wcd_api_key` setting in the frontend config.

    Returns:
        (bool): True if the tree exists in the Weaviate instance, False otherwise.
    """

    local_user = await self.get_user_local(user_id)
    tree_manager: TreeManager = local_user["tree_manager"]

    if wcd_url is None or wcd_api_key is None:
        save_location_client_manager = local_user[
            "frontend_config"
        ].save_location_client_manager
    else:
        save_location_client_manager = ClientManager(
            logger=logger,
            wcd_url=wcd_url,
            wcd_api_key=wcd_api_key,
        )

    return await tree_manager.check_tree_exists_weaviate(
        conversation_id, save_location_client_manager
    )

check_user_timeout(user_id)

Check if a user has been idle for the last user_timeout.

Parameters:

Name	Type	Description	Default
user_id	str	The user ID which contains the user.	required

Returns:

Type	Description
bool	True if the user has been idle for the last user_timeout, False otherwise.

Source code in elysia/api/services/user.py

def check_user_timeout(self, user_id: str):
    """
    Check if a user has been idle for the last user_timeout.

    Args:
        user_id (str): The user ID which contains the user.

    Returns:
        (bool): True if the user has been idle for the last user_timeout, False otherwise.
    """
    # if user not found, return True
    if user_id not in self.users:
        return True

    # Remove any trees that have not been active in the last user_timeout
    # if (
    #     "last_request" in self.users[user_id]
    #     and datetime.datetime.now() - self.users[user_id]["last_request"]
    #     > self.user_timeout
    # ):
    #     return True

    return False

close_all_clients() async

Close all clients in all ClientManagers across all users.

Source code in elysia/api/services/user.py

async def close_all_clients(self):
    """
    Close all clients in all ClientManagers across all users.
    """
    for user_id in self.users:
        if "client_manager" in self.users[user_id]:
            await self.users[user_id]["client_manager"].close_clients()

delete_tree(user_id, conversation_id, wcd_url=None, wcd_api_key=None) async

Delete a saved tree from a Weaviate instance (set in the frontend config). Also delete the tree from the local tree manager. This is a wrapper for the TreeManager.delete_tree_weaviate() method and the TreeManager.delete_tree_local() method.

Parameters:

Name	Type	Description	Default
user_id	str	Required. The unique identifier for the user stored in the UserManager.	required
conversation_id	str	Required. The unique identifier for the conversation for the user.	required
wcd_url	str | None	Required. The URL of the Weaviate Cloud Database instance used to save the tree. Defaults to the value of the wcd_url setting in the frontend config.	None
wcd_api_key	str | None	Required. The API key for the Weaviate Cloud Database instance used to save the tree. Defaults to the value of the wcd_api_key setting in the frontend config.	None

Source code in elysia/api/services/user.py

async def delete_tree(
    self,
    user_id: str,
    conversation_id: str,
    wcd_url: str | None = None,
    wcd_api_key: str | None = None,
):
    """
    Delete a saved tree from a Weaviate instance (set in the frontend config).
    Also delete the tree from the local tree manager.
    This is a wrapper for the TreeManager.delete_tree_weaviate() method and the TreeManager.delete_tree_local() method.

    Args:
        user_id (str): Required. The unique identifier for the user stored in the UserManager.
        conversation_id (str): Required. The unique identifier for the conversation for the user.
        wcd_url (str | None): Required. The URL of the Weaviate Cloud Database instance used to save the tree.
            Defaults to the value of the `wcd_url` setting in the frontend config.
        wcd_api_key (str | None): Required. The API key for the Weaviate Cloud Database instance used to save the tree.
            Defaults to the value of the `wcd_api_key` setting in the frontend config.
    """

    local_user = await self.get_user_local(user_id)
    tree_manager: TreeManager = local_user["tree_manager"]

    if wcd_url is None or wcd_api_key is None:
        save_location_client_manager = local_user[
            "frontend_config"
        ].save_location_client_manager
    else:
        save_location_client_manager = ClientManager(
            logger=logger,
            wcd_url=wcd_url,
            wcd_api_key=wcd_api_key,
        )

    await tree_manager.delete_tree_weaviate(
        conversation_id, save_location_client_manager
    )
    tree_manager.delete_tree_local(conversation_id)

get_saved_trees(user_id, wcd_url=None, wcd_api_key=None) async

Get all saved trees from a Weaviate instance (set in the frontend config).

Parameters:

Name	Type	Description	Default
user_id	str	Required. The unique identifier for the user stored in the UserManager.	required
wcd_url	str | None	Required. The URL of the Weaviate Cloud Database instance used to save the tree. Defaults to the value of the wcd_url setting in the frontend config.	None
wcd_api_key	str | None	Required. The API key for the Weaviate Cloud Database instance used to save the tree. Defaults to the value of the wcd_api_key setting in the frontend config.	None

Returns:

Type	Description
dict	A dictionary whose keys are the conversation IDs and whose values are dictionaries containing the title and last update time of the tree. E.g. { "12345678-XXX-YYYY-ZZZZ": { "title": "Query Request", "last_update_time": "2025-06-07T10:06:47.376000Z" } }

Source code in elysia/api/services/user.py

async def get_saved_trees(
    self,
    user_id: str,
    wcd_url: str | None = None,
    wcd_api_key: str | None = None,
):
    """
    Get all saved trees from a Weaviate instance (set in the frontend config).

    Args:
        user_id (str): Required. The unique identifier for the user stored in the UserManager.
        wcd_url (str | None): Required. The URL of the Weaviate Cloud Database instance used to save the tree.
            Defaults to the value of the `wcd_url` setting in the frontend config.
        wcd_api_key (str | None): Required. The API key for the Weaviate Cloud Database instance used to save the tree.
            Defaults to the value of the `wcd_api_key` setting in the frontend config.

    Returns:
        (dict): A dictionary whose keys are the conversation IDs and whose values are dictionaries containing the title and last update time of the tree.
            E.g.
            ```
            {
                "12345678-XXX-YYYY-ZZZZ": {
                    "title": "Query Request",
                    "last_update_time": "2025-06-07T10:06:47.376000Z"
                }
            }
            ```
    """
    local_user = await self.get_user_local(user_id)

    if wcd_url is None or wcd_api_key is None:
        save_location_client_manager = local_user[
            "frontend_config"
        ].save_location_client_manager
    else:
        save_location_client_manager = ClientManager(
            logger=logger,
            wcd_url=wcd_url,
            wcd_api_key=wcd_api_key,
        )

    return await get_saved_trees_weaviate(
        "ELYSIA_TREES__", save_location_client_manager, user_id
    )

get_tree(user_id, conversation_id) async

Get a tree for a user. Will raise a ValueError if the user is not found, or the tree is not found.

Parameters:

Name	Type	Description	Default
user_id	str	Required. The unique identifier for the user.	required
conversation_id	str	Required. The unique identifier for the conversation.	required

Returns:

Type	Description
Tree	The tree.

Source code in elysia/api/services/user.py

async def get_tree(self, user_id: str, conversation_id: str):
    """
    Get a tree for a user.
    Will raise a ValueError if the user is not found, or the tree is not found.

    Args:
        user_id (str): Required. The unique identifier for the user.
        conversation_id (str): Required. The unique identifier for the conversation.

    Returns:
        (Tree): The tree.
    """
    local_user = await self.get_user_local(user_id)
    return local_user["tree_manager"].get_tree(conversation_id)

get_user_local(user_id) async

Return a local user object. Will raise a ValueError if the user is not found.

Parameters:

Name	Type	Description	Default
user_id	str	Required. The unique identifier for the user.	required

Returns:

Type	Description
dict	A local user object, containing a TreeManager ("tree_manager"), Frontend Config ("frontend_config") and ClientManager ("client_manager").

Source code in elysia/api/services/user.py

async def get_user_local(self, user_id: str):
    """
    Return a local user object.
    Will raise a ValueError if the user is not found.

    Args:
        user_id (str): Required. The unique identifier for the user.

    Returns:
        (dict): A local user object, containing a TreeManager ("tree_manager"),
            Frontend Config ("frontend_config") and ClientManager ("client_manager").
    """

    if user_id not in self.users:
        raise ValueError(
            f"User {user_id} not found. Please initialise a user first (by calling `add_user_local`)."
        )

    # update last request (adds last_request to user)
    await self.update_user_last_request(user_id)

    return self.users[user_id]

initialise_tree(user_id, conversation_id, low_memory=False) async

Initialises a tree for a user for an existing user at user_id. Requires a user to already exist in the UserManager, via add_user_local. This is a wrapper for the TreeManager.add_tree() method.

Parameters:

Name	Type	Description	Default
user_id	str	Required. The unique identifier for the user.	required
conversation_id	str	Required. The unique identifier for a new conversation within the tree manager.	required
low_memory	bool	Optional. Whether to use low memory mode for the tree. Controls the LM history being saved in the tree, and some other variables. Defaults to False.	False

Source code in elysia/api/services/user.py

async def initialise_tree(
    self,
    user_id: str,
    conversation_id: str,
    low_memory: bool = False,
):
    """
    Initialises a tree for a user for an existing user at user_id.
    Requires a user to already exist in the UserManager, via `add_user_local`.
    This is a wrapper for the TreeManager.add_tree() method.

    Args:
        user_id (str): Required. The unique identifier for the user.
        conversation_id (str): Required. The unique identifier for a new conversation within the tree manager.
        low_memory (bool): Optional. Whether to use low memory mode for the tree.
            Controls the LM history being saved in the tree, and some other variables.
            Defaults to False.
    """
    # self.add_user_local(user_id)
    local_user = await self.get_user_local(user_id)
    tree_manager: TreeManager = local_user["tree_manager"]
    if not tree_manager.tree_exists(conversation_id):
        tree_manager.add_tree(
            conversation_id,
            low_memory,
        )
    return tree_manager.get_tree(conversation_id)

load_tree(user_id, conversation_id, wcd_url=None, wcd_api_key=None) async

Load a tree from a Weaviate instance (set in the frontend config). This is a wrapper for the TreeManager.load_tree_weaviate() method.

Parameters:

Name	Type	Description	Default
user_id	str	Required. The unique identifier for the user stored in the UserManager.	required
conversation_id	str	Required. The unique identifier for the conversation for the user.	required
wcd_url	str | None	Required. The URL of the Weaviate Cloud Database instance used to save the tree. Defaults to the value of the wcd_url setting in the frontend config.	None
wcd_api_key	str | None	Required. The API key for the Weaviate Cloud Database instance used to save the tree. Defaults to the value of the wcd_api_key setting in the frontend config.	None

Returns:

Type	Description
list[dict]	A list of dictionaries, each containing a frontend payload that was used to generate the tree. The list is ordered by the time the payload was originally sent to the frontend (at the time it was saved).

Source code in elysia/api/services/user.py

async def load_tree(
    self,
    user_id: str,
    conversation_id: str,
    wcd_url: str | None = None,
    wcd_api_key: str | None = None,
):
    """
    Load a tree from a Weaviate instance (set in the frontend config).
    This is a wrapper for the TreeManager.load_tree_weaviate() method.

    Args:
        user_id (str): Required. The unique identifier for the user stored in the UserManager.
        conversation_id (str): Required. The unique identifier for the conversation for the user.
        wcd_url (str | None): Required. The URL of the Weaviate Cloud Database instance used to save the tree.
            Defaults to the value of the `wcd_url` setting in the frontend config.
        wcd_api_key (str | None): Required. The API key for the Weaviate Cloud Database instance used to save the tree.
            Defaults to the value of the `wcd_api_key` setting in the frontend config.

    Returns:
        (list[dict]): A list of dictionaries, each containing a frontend payload that was used to generate the tree.
            The list is ordered by the time the payload was originally sent to the frontend (at the time it was saved).
    """

    local_user = await self.get_user_local(user_id)
    tree_manager: TreeManager = local_user["tree_manager"]

    if wcd_url is None or wcd_api_key is None:
        save_location_client_manager = local_user[
            "frontend_config"
        ].save_location_client_manager
    else:
        save_location_client_manager = ClientManager(
            logger=logger,
            wcd_url=wcd_url,
            wcd_api_key=wcd_api_key,
        )

    return await tree_manager.load_tree_weaviate(
        conversation_id, save_location_client_manager
    )

process_tree(query, user_id, conversation_id, query_id, training_route='', collection_names=[], save_trees_to_weaviate=None, wcd_url=None, wcd_api_key=None) async

Wrapper for the TreeManager.process_tree() method. Which itself is a wrapper for the Tree.async_run() method. This is an async generator which yields results from the tree.async_run() method. Automatically sends error payloads if the user or tree has been timed out.

Parameters:

Name	Type	Description	Default
query	str	Required. The user input/prompt to process in the decision tree.	required
user_id	str	Required. The unique identifier for the user.	required
conversation_id	str	Required. The conversation ID which contains the tree. This should be the same conversation ID as the one used to initialise the tree (see initialise_tree).	required
query_id	str	Required. A unique identifier for the query.	required
training_route	str	Optional. The training route, a string of the form "tool1/tool2/tool1" etc. See the tree.async_run() method for more details.	''
collection_names	list[str]	Optional. A list of collection names to use in the query. If not supplied, all collections will be used.	[]
save_trees_to_weaviate	bool | None	Optional. Whether to save the trees to a Weaviate instance, after the process_tree() method has finished. Defaults to the value of the save_trees_to_weaviate setting in the frontend config.	None
wcd_url	str | None	Required. The URL of the Weaviate Cloud Database instance used to save the tree. Defaults to the value of the wcd_url setting in the frontend config.	None
wcd_api_key	str | None	Required. The API key for the Weaviate Cloud Database instance used to save the tree. Defaults to the value of the wcd_api_key setting in the frontend config.	None

Source code in elysia/api/services/user.py

async def process_tree(
    self,
    query: str,
    user_id: str,
    conversation_id: str,
    query_id: str,
    training_route: str = "",
    collection_names: list[str] = [],
    save_trees_to_weaviate: bool | None = None,
    wcd_url: str | None = None,
    wcd_api_key: str | None = None,
):
    """
    Wrapper for the TreeManager.process_tree() method.
    Which itself is a wrapper for the Tree.async_run() method.
    This is an async generator which yields results from the tree.async_run() method.
    Automatically sends error payloads if the user or tree has been timed out.

    Args:
        query (str): Required. The user input/prompt to process in the decision tree.
        user_id (str): Required. The unique identifier for the user.
        conversation_id (str): Required. The conversation ID which contains the tree.
            This should be the same conversation ID as the one used to initialise the tree (see `initialise_tree`).
        query_id (str): Required. A unique identifier for the query.
        training_route (str): Optional. The training route, a string of the form "tool1/tool2/tool1" etc.
            See the `tree.async_run()` method for more details.
        collection_names (list[str]): Optional. A list of collection names to use in the query.
            If not supplied, all collections will be used.
        save_trees_to_weaviate (bool | None): Optional. Whether to save the trees to a Weaviate instance,
            after the process_tree() method has finished.
            Defaults to the value of the `save_trees_to_weaviate` setting in the frontend config.
        wcd_url (str | None): Required. The URL of the Weaviate Cloud Database instance used to save the tree.
            Defaults to the value of the `wcd_url` setting in the frontend config.
        wcd_api_key (str | None): Required. The API key for the Weaviate Cloud Database instance used to save the tree.
            Defaults to the value of the `wcd_api_key` setting in the frontend config.
    """

    if self.check_user_timeout(user_id):
        user_timeout_error = UserTimeoutError()
        error_payload = await user_timeout_error.to_frontend(
            user_id, conversation_id, query_id
        )
        yield error_payload
        return

    if self.check_tree_timeout(user_id, conversation_id):
        if await self.check_tree_exists_weaviate(user_id, conversation_id):
            await self.load_tree(user_id, conversation_id)
        else:
            tree_timeout_error = TreeTimeoutError()
            error_payload = await tree_timeout_error.to_frontend(
                user_id, conversation_id, query_id
            )
            yield error_payload
            return

    local_user = await self.get_user_local(user_id)
    await self.update_user_last_request(user_id)

    tree_manager: TreeManager = local_user["tree_manager"]

    async for yielded_result in tree_manager.process_tree(
        query,
        conversation_id,
        query_id,
        training_route,
        collection_names,
        local_user["client_manager"],
    ):
        yield yielded_result
        await self.update_user_last_request(user_id)

    if save_trees_to_weaviate is None:
        frontend_config: FrontendConfig = local_user["frontend_config"]
        save_trees_to_weaviate = frontend_config.config["save_trees_to_weaviate"]

    if save_trees_to_weaviate:
        await self.save_tree(user_id, conversation_id, wcd_url, wcd_api_key)

save_tree(user_id, conversation_id, wcd_url=None, wcd_api_key=None) async

Save a tree to a Weaviate instance (set in the frontend config). This is a wrapper for the TreeManager.save_tree_weaviate() method.

Parameters:

Name	Type	Description	Default
user_id	str	Required. The unique identifier for the user stored in the UserManager.	required
conversation_id	str	Required. The unique identifier for the conversation for the user.	required

Source code in elysia/api/services/user.py

async def save_tree(
    self,
    user_id: str,
    conversation_id: str,
    wcd_url: str | None = None,
    wcd_api_key: str | None = None,
):
    """
    Save a tree to a Weaviate instance (set in the frontend config).
    This is a wrapper for the TreeManager.save_tree_weaviate() method.

    Args:
        user_id (str): Required. The unique identifier for the user stored in the UserManager.
        conversation_id (str): Required. The unique identifier for the conversation for the user.
    """

    local_user = await self.get_user_local(user_id)
    tree_manager: TreeManager = local_user["tree_manager"]

    if wcd_url is None or wcd_api_key is None:
        save_location_client_manager = local_user[
            "frontend_config"
        ].save_location_client_manager
    else:
        save_location_client_manager = ClientManager(
            logger=logger,
            wcd_url=wcd_url,
            wcd_api_key=wcd_api_key,
        )

    await tree_manager.save_tree_weaviate(
        conversation_id, save_location_client_manager
    )

TreeManager

Manages trees (different conversations) for a single user. Designed to be used with the Elysia API, or a manager for multiple Elysia decision trees.

You can initialise the TreeManager with particular config options. Or, upon adding a tree, you can set a specific style, description and end goal for that tree.

Each tree has separate elements: the tree itself, the last request time, and the asyncio event.

Source code in elysia/api/services/tree.py

class TreeManager:
    """
    Manages trees (different conversations) for a single user.
    Designed to be used with the Elysia API, or a manager for multiple Elysia decision trees.

    You can initialise the TreeManager with particular config options.
    Or, upon adding a tree, you can set a specific style, description and end goal for that tree.

    Each tree has separate elements: the tree itself, the last request time, and the asyncio event.
    """

    def __init__(
        self,
        user_id: str,
        config: Config | None = None,
        tree_timeout: datetime.timedelta | int | None = None,
    ):
        """
        Args:
            user_id (str): Required. A unique identifier for the user being managed within this TreeManager.
            config (Config | None): Optional. A config for all trees managed by this TreeManager.
                Defaults to a new config with default settings.
                If `settings` is not provided, it will be set to the default settings (smart_setup).
            tree_timeout (datetime.timedelta | int | None): Optional. A timeout for all trees managed by this TreeManager.
                Defaults to the value of the `TREE_TIMEOUT` environment variable.
                If an integer is passed, it will be interpreted as minutes.
                If set to 0, trees will not be automatically removed.
        """
        self.trees = {}
        self.user_id = user_id

        if tree_timeout is None:
            self.tree_timeout = datetime.timedelta(
                minutes=int(os.environ.get("TREE_TIMEOUT", 10))
            )
        elif isinstance(tree_timeout, int):
            self.tree_timeout = datetime.timedelta(minutes=tree_timeout)
        else:
            self.tree_timeout = tree_timeout

        if config is None:
            self.config = Config()
        else:
            self.config = config

        self.settings = self.config.settings

    def update_config(
        self,
        conversation_id: str | None = None,
        config_id: str | None = None,
        config_name: str | None = None,
        settings: dict[str, Any] | None = None,
        style: str | None = None,
        agent_description: str | None = None,
        end_goal: str | None = None,
        branch_initialisation: BranchInitType | None = None,
    ):
        if config_id is not None:
            self.config.id = config_id

        if config_name is not None:
            self.config.name = config_name

        if settings is not None:
            self.configure(conversation_id=conversation_id, replace=True, **settings)

        if style is not None:
            self.change_style(style, conversation_id)

        if agent_description is not None:
            self.change_agent_description(agent_description, conversation_id)

        if end_goal is not None:
            self.change_end_goal(end_goal, conversation_id)

        if branch_initialisation is not None:
            self.change_branch_initialisation(branch_initialisation, conversation_id)

    def add_tree(
        self,
        conversation_id: str,
        low_memory: bool = False,
    ):
        """
        Add a tree to the TreeManager.
        The decision tree can be initialised with specific config options, such as style, agent description and end goal.
        As well as a Settings object, which chooses options such as the LLM models, API keys and more.

        Args:
            conversation_id (str): Required. A unique identifier for the conversation.
            low_memory (bool): Optional. Whether to use low memory mode for the tree.
                Controls the LM history being saved in the tree, and some other variables.
                Defaults to False.
        """

        if not self.tree_exists(conversation_id):
            self.trees[conversation_id] = {
                "tree": Tree(
                    conversation_id=conversation_id,
                    user_id=self.user_id,
                    settings=self.settings,
                    style=self.config.style,
                    agent_description=self.config.agent_description,
                    end_goal=self.config.end_goal,
                    branch_initialisation=self.config.branch_initialisation,
                    low_memory=low_memory,
                    use_elysia_collections=self.config.use_elysia_collections,
                ),
                "last_request": datetime.datetime.now(),
                "event": asyncio.Event(),
            }
            self.trees[conversation_id]["event"].set()

    async def save_tree_weaviate(
        self, conversation_id: str, client_manager: ClientManager
    ):
        """
        Save a tree to Weaviate to collection ELYSIA_TREES__.
        Creates the collection if it doesn't exist.

        Args:
            conversation_id (str): The conversation ID which contains the tree.
            client_manager (ClientManager): The client manager to use for the tree.
        """
        tree: Tree = self.get_tree(conversation_id)
        await tree.export_to_weaviate("ELYSIA_TREES__", client_manager)

    async def check_tree_exists_weaviate(
        self, conversation_id: str, client_manager: ClientManager
    ):
        """
        Check if a tree exists in a Weaviate instance.
        The collection ELYSIA_TREES__ must exist, returns False if it doesn't.

        Args:
            conversation_id (str): The conversation ID which contains the tree.
            client_manager (ClientManager): The client manager to use for the tree.

        Returns:
            (bool): True if the tree exists in the Weaviate instance, False otherwise.
        """
        async with client_manager.connect_to_async_client() as client:
            if not await client.collections.exists("ELYSIA_TREES__"):
                return False

            collection = client.collections.get("ELYSIA_TREES__")
            uuid = generate_uuid5(conversation_id)
            return await collection.data.exists(uuid)

    async def load_tree_weaviate(
        self, conversation_id: str, client_manager: ClientManager
    ):
        """
        Load a tree from Weaviate.
        The conversation ID from the loaded tree is placed into the tree manager
        (possibly overwriting an existing tree with the same conversation ID).
        Then the tree itself is not returned - instead the list of frontend payloads
        that were yielded to the frontend by the tree is returned.

        Args:
            conversation_id (str): The conversation ID which contains the tree.
            client_manager (ClientManager): The client manager to use for the tree.

        Returns:
            (list): A list of dictionaries, each containing a frontend payload that was used to generate the tree.
                The list is ordered by the time the payload was originally sent to the frontend (at the time it was saved).
        """
        tree = await Tree.import_from_weaviate(
            "ELYSIA_TREES__", conversation_id, client_manager
        )
        if conversation_id not in self.trees:
            self.trees[conversation_id] = {
                "tree": None,
                "event": asyncio.Event(),
                "last_request": datetime.datetime.now(),
            }
        self.trees[conversation_id]["tree"] = tree
        self.trees[conversation_id]["event"].set()
        self.update_tree_last_request(conversation_id)
        return tree.returner.store

    async def delete_tree_weaviate(
        self, conversation_id: str, client_manager: ClientManager
    ):
        """
        Delete a tree from the stored trees in Weaviate.

        Args:
            conversation_id (str): The conversation ID of the tree to be deleted.
            client_manager (ClientManager): The client manager pointing to the Weaviate instance containing the tree.
        """
        await delete_tree_from_weaviate(
            conversation_id, "ELYSIA_TREES__", client_manager
        )

    def delete_tree_local(self, conversation_id: str):
        """
        Delete a tree from the TreeManager.

        Args:
            conversation_id (str): The conversation ID of the tree to be deleted.
        """
        if conversation_id in self.trees:
            del self.trees[conversation_id]

    def tree_exists(self, conversation_id: str):
        """
        Check if a tree exists in the TreeManager.

        Args:
            conversation_id (str): The conversation ID which may contain the tree.

        Returns:
            (bool): True if the tree exists, False otherwise.
        """
        return conversation_id in self.trees

    def get_tree(self, conversation_id: str):
        """
        Get a tree from the TreeManager.
        Will raise a ValueError if the tree is not found.

        Args:
            conversation_id (str): The conversation ID which contains the tree.

        Returns:
            (Tree): The tree associated with the conversation ID.
        """
        if conversation_id not in self.trees:
            raise ValueError(
                f"Tree {conversation_id} not found. Please initialise a tree first (by calling `add_tree`)."
            )

        return self.trees[conversation_id]["tree"]

    def get_event(self, conversation_id: str):
        """
        Get the asyncio.Event for a tree in the TreeManager.
        This is cleared when the tree is processing, and set when the tree is idle.
        This is used to block the API from sending multiple requests to the tree at once.

        Args:
            conversation_id (str): The conversation ID which contains the tree.

        Returns:
            (asyncio.Event): The event for the tree.
        """
        return self.trees[conversation_id]["event"]

    def configure(
        self, conversation_id: str | None = None, replace: bool = False, **kwargs: Any
    ):
        """
        Configure the settings for a tree in the TreeManager.

        Args:
            conversation_id (str | None): The conversation ID which contains the tree.
            replace (bool): Whether to override the current settings with the new settings.
                When this is True, all existing settings are removed, and only the new settings are used.
                Defaults to False.
            **kwargs (Any): The keyword arguments to pass to the Settings.configure() method.
        """
        if conversation_id is None:
            self.settings.configure(replace=replace, **kwargs)
            self.config.settings = self.settings
        else:
            self.get_tree(conversation_id).settings.configure(replace=replace, **kwargs)

    def change_style(self, style: str, conversation_id: str | None = None):
        """
        Change the style for a tree in the TreeManager.
        Or change the global style for all trees (if conversation_id is not supplied).
        And applies these changes to current trees with default settings.

        Args:
            style (str): The new style for the tree.
            conversation_id (str | None): Optional. The conversation ID which contains the tree.
                If not supplied, the style will be changed on all trees.
        """
        if conversation_id is None:
            self.config.style = style
            for conversation_id in self.trees:
                if not self.trees[conversation_id]["tree"]._config_modified:
                    self.trees[conversation_id]["tree"].change_style(style)
                    self.trees[conversation_id]["tree"]._config_modified = False
        else:
            if conversation_id not in self.trees:
                raise ValueError(f"Tree {conversation_id} not found")

            self.trees[conversation_id]["tree"].change_style(style)

    def change_agent_description(
        self, agent_description: str, conversation_id: str | None = None
    ):
        """
        Change the agent description for a tree in the TreeManager.
        Or change the global agent description for all trees (if conversation_id is not supplied).
        And applies these changes to current trees with default settings.

        Args:
            agent_description (str): The new agent description for the tree.
            conversation_id (str | None): Optional. The conversation ID which contains the tree.
                If not supplied, the agent description will be changed on all trees.
        """
        if conversation_id is None:
            self.config.agent_description = agent_description
            for conversation_id in self.trees:
                if not self.trees[conversation_id]["tree"]._config_modified:
                    self.trees[conversation_id]["tree"].change_agent_description(
                        agent_description
                    )
                    self.trees[conversation_id]["tree"]._config_modified = False
        else:
            if conversation_id not in self.trees:
                raise ValueError(f"Tree {conversation_id} not found")

            self.trees[conversation_id]["tree"].change_agent_description(
                agent_description
            )

    def change_end_goal(self, end_goal: str, conversation_id: str | None = None):
        """
        Change the end goal for a tree in the TreeManager.
        Or change the global end goal for all trees (if conversation_id is not supplied).
        And applies these changes to current trees with default settings.

        Args:
            end_goal (str): The new end goal for the tree.
            conversation_id (str | None): Optional. The conversation ID which contains the tree.
                If not supplied, the end goal will be changed on all trees.
        """
        if conversation_id is None:
            self.config.end_goal = end_goal
            for conversation_id in self.trees:
                if not self.trees[conversation_id]["tree"]._config_modified:
                    self.trees[conversation_id]["tree"].change_end_goal(end_goal)
                    self.trees[conversation_id]["tree"]._config_modified = False
        else:
            if conversation_id not in self.trees:
                raise ValueError(f"Tree {conversation_id} not found")

            self.trees[conversation_id]["tree"].change_end_goal(end_goal)

    def change_branch_initialisation(
        self, branch_initialisation: BranchInitType, conversation_id: str | None = None
    ):
        """
        Change the branch initialisation for a tree in the TreeManager.
        Or change the global branch initialisation for all trees (if conversation_id is not supplied).
        And applies these changes to current trees with default settings.

        Args:
            branch_initialisation (str): The new branch initialisation for the tree.
            conversation_id (str | None): Optional. The conversation ID which contains the tree.
                If not supplied, the branch initialisation will be changed on all trees.
        """
        if conversation_id is None:
            self.config.branch_initialisation = branch_initialisation
            for conversation_id in self.trees:
                if not self.trees[conversation_id]["tree"]._config_modified:
                    self.trees[conversation_id]["tree"].set_branch_initialisation(
                        branch_initialisation
                    )
                    self.trees[conversation_id]["tree"]._config_modified = False
        else:
            if conversation_id not in self.trees:
                raise ValueError(f"Tree {conversation_id} not found")

            self.trees[conversation_id]["tree"].set_branch_initialisation(
                branch_initialisation
            )

    async def process_tree(
        self,
        query: str,
        conversation_id: str,
        query_id: str | None = None,
        training_route: str = "",
        collection_names: list[str] = [],
        client_manager: ClientManager | None = None,
    ):
        """
        Process a tree in the TreeManager.
        This is an async generator which yields results from the tree.async_run() method.

        Args:
            query (str): Required. The user input/prompt to process in the decision tree.
            conversation_id (str): Required. The conversation ID which contains the tree.
            query_id (str): Optional. A unique identifier for the query.
                If not supplied, a random UUID will be generated.
            training_route (str): Optional. The training route, a string of the form "tool1/tool2/tool1" etc.
                See the `tree.async_run()` method for more details.
            collection_names (list[str]): Optional. A list of collection names to use in the query.
                If not supplied, all collections will be used.
            client_manager (ClientManager | None): Optional. A client manager to use for the query.
                If not supplied, a new ClientManager will be created.
        """

        if query_id is None:
            query_id = str(uuid.uuid4())

        tree: Tree = self.get_tree(conversation_id)
        self.update_tree_last_request(conversation_id)

        # wait for the tree to be idle
        await self.trees[conversation_id]["event"].wait()

        # clear the event, set it to working
        self.trees[conversation_id]["event"].clear()

        try:
            async for yielded_result in tree.async_run(
                query,
                collection_names=collection_names,
                client_manager=client_manager,
                query_id=query_id,
                training_route=training_route,
                close_clients_after_completion=False,
            ):
                yield yielded_result
                self.update_tree_last_request(conversation_id)

        finally:
            # set the event to idle
            self.trees[conversation_id]["event"].set()

    def check_tree_timeout(self, conversation_id: str):
        """
        Check if a tree has been idle for the last tree_timeout.

        Args:
            conversation_id (str): The conversation ID which contains the tree.

        Returns:
            (bool): True if the tree has been idle for the last tree_timeout, False otherwise.
        """
        # if tree not found, return True
        if conversation_id not in self.trees:
            return True

        # Remove any trees that have not been active in the last TREE_TIMEOUT minutes
        if (
            "last_request" in self.trees[conversation_id]
            and datetime.datetime.now() - self.trees[conversation_id]["last_request"]
            > self.tree_timeout
        ):
            return True

        return False

    def update_tree_last_request(self, conversation_id: str):
        self.trees[conversation_id]["last_request"] = datetime.datetime.now()

    def check_all_trees_timeout(self):
        """
        Check all trees in the TreeManager and remove any that have not been active in the last tree_timeout.
        """
        if self.tree_timeout == datetime.timedelta(minutes=0):
            return

        convs_to_remove = []
        for i, conversation_id in enumerate(self.trees):
            if self.check_tree_timeout(conversation_id):
                convs_to_remove.append(conversation_id)

        for conversation_id in convs_to_remove:
            del self.trees[conversation_id]

__init__(user_id, config=None, tree_timeout=None)

Parameters:

Name	Type	Description	Default
user_id	str	Required. A unique identifier for the user being managed within this TreeManager.	required
config	Config | None	Optional. A config for all trees managed by this TreeManager. Defaults to a new config with default settings. If settings is not provided, it will be set to the default settings (smart_setup).	None
tree_timeout	timedelta | int | None	Optional. A timeout for all trees managed by this TreeManager. Defaults to the value of the TREE_TIMEOUT environment variable. If an integer is passed, it will be interpreted as minutes. If set to 0, trees will not be automatically removed.	None

Source code in elysia/api/services/tree.py

def __init__(
    self,
    user_id: str,
    config: Config | None = None,
    tree_timeout: datetime.timedelta | int | None = None,
):
    """
    Args:
        user_id (str): Required. A unique identifier for the user being managed within this TreeManager.
        config (Config | None): Optional. A config for all trees managed by this TreeManager.
            Defaults to a new config with default settings.
            If `settings` is not provided, it will be set to the default settings (smart_setup).
        tree_timeout (datetime.timedelta | int | None): Optional. A timeout for all trees managed by this TreeManager.
            Defaults to the value of the `TREE_TIMEOUT` environment variable.
            If an integer is passed, it will be interpreted as minutes.
            If set to 0, trees will not be automatically removed.
    """
    self.trees = {}
    self.user_id = user_id

    if tree_timeout is None:
        self.tree_timeout = datetime.timedelta(
            minutes=int(os.environ.get("TREE_TIMEOUT", 10))
        )
    elif isinstance(tree_timeout, int):
        self.tree_timeout = datetime.timedelta(minutes=tree_timeout)
    else:
        self.tree_timeout = tree_timeout

    if config is None:
        self.config = Config()
    else:
        self.config = config

    self.settings = self.config.settings

add_tree(conversation_id, low_memory=False)

Add a tree to the TreeManager. The decision tree can be initialised with specific config options, such as style, agent description and end goal. As well as a Settings object, which chooses options such as the LLM models, API keys and more.

Parameters:

Name	Type	Description	Default
conversation_id	str	Required. A unique identifier for the conversation.	required
low_memory	bool	Optional. Whether to use low memory mode for the tree. Controls the LM history being saved in the tree, and some other variables. Defaults to False.	False

Source code in elysia/api/services/tree.py

def add_tree(
    self,
    conversation_id: str,
    low_memory: bool = False,
):
    """
    Add a tree to the TreeManager.
    The decision tree can be initialised with specific config options, such as style, agent description and end goal.
    As well as a Settings object, which chooses options such as the LLM models, API keys and more.

    Args:
        conversation_id (str): Required. A unique identifier for the conversation.
        low_memory (bool): Optional. Whether to use low memory mode for the tree.
            Controls the LM history being saved in the tree, and some other variables.
            Defaults to False.
    """

    if not self.tree_exists(conversation_id):
        self.trees[conversation_id] = {
            "tree": Tree(
                conversation_id=conversation_id,
                user_id=self.user_id,
                settings=self.settings,
                style=self.config.style,
                agent_description=self.config.agent_description,
                end_goal=self.config.end_goal,
                branch_initialisation=self.config.branch_initialisation,
                low_memory=low_memory,
                use_elysia_collections=self.config.use_elysia_collections,
            ),
            "last_request": datetime.datetime.now(),
            "event": asyncio.Event(),
        }
        self.trees[conversation_id]["event"].set()

change_agent_description(agent_description, conversation_id=None)

Change the agent description for a tree in the TreeManager. Or change the global agent description for all trees (if conversation_id is not supplied). And applies these changes to current trees with default settings.

Parameters:

Name	Type	Description	Default
agent_description	str	The new agent description for the tree.	required
conversation_id	str | None	Optional. The conversation ID which contains the tree. If not supplied, the agent description will be changed on all trees.	None

Source code in elysia/api/services/tree.py

def change_agent_description(
    self, agent_description: str, conversation_id: str | None = None
):
    """
    Change the agent description for a tree in the TreeManager.
    Or change the global agent description for all trees (if conversation_id is not supplied).
    And applies these changes to current trees with default settings.

    Args:
        agent_description (str): The new agent description for the tree.
        conversation_id (str | None): Optional. The conversation ID which contains the tree.
            If not supplied, the agent description will be changed on all trees.
    """
    if conversation_id is None:
        self.config.agent_description = agent_description
        for conversation_id in self.trees:
            if not self.trees[conversation_id]["tree"]._config_modified:
                self.trees[conversation_id]["tree"].change_agent_description(
                    agent_description
                )
                self.trees[conversation_id]["tree"]._config_modified = False
    else:
        if conversation_id not in self.trees:
            raise ValueError(f"Tree {conversation_id} not found")

        self.trees[conversation_id]["tree"].change_agent_description(
            agent_description
        )

change_branch_initialisation(branch_initialisation, conversation_id=None)

Change the branch initialisation for a tree in the TreeManager. Or change the global branch initialisation for all trees (if conversation_id is not supplied). And applies these changes to current trees with default settings.

Parameters:

Name	Type	Description	Default
branch_initialisation	str	The new branch initialisation for the tree.	required
conversation_id	str | None	Optional. The conversation ID which contains the tree. If not supplied, the branch initialisation will be changed on all trees.	None

Source code in elysia/api/services/tree.py

def change_branch_initialisation(
    self, branch_initialisation: BranchInitType, conversation_id: str | None = None
):
    """
    Change the branch initialisation for a tree in the TreeManager.
    Or change the global branch initialisation for all trees (if conversation_id is not supplied).
    And applies these changes to current trees with default settings.

    Args:
        branch_initialisation (str): The new branch initialisation for the tree.
        conversation_id (str | None): Optional. The conversation ID which contains the tree.
            If not supplied, the branch initialisation will be changed on all trees.
    """
    if conversation_id is None:
        self.config.branch_initialisation = branch_initialisation
        for conversation_id in self.trees:
            if not self.trees[conversation_id]["tree"]._config_modified:
                self.trees[conversation_id]["tree"].set_branch_initialisation(
                    branch_initialisation
                )
                self.trees[conversation_id]["tree"]._config_modified = False
    else:
        if conversation_id not in self.trees:
            raise ValueError(f"Tree {conversation_id} not found")

        self.trees[conversation_id]["tree"].set_branch_initialisation(
            branch_initialisation
        )

change_end_goal(end_goal, conversation_id=None)

Change the end goal for a tree in the TreeManager. Or change the global end goal for all trees (if conversation_id is not supplied). And applies these changes to current trees with default settings.

Parameters:

Name	Type	Description	Default
end_goal	str	The new end goal for the tree.	required
conversation_id	str | None	Optional. The conversation ID which contains the tree. If not supplied, the end goal will be changed on all trees.	None

Source code in elysia/api/services/tree.py

def change_end_goal(self, end_goal: str, conversation_id: str | None = None):
    """
    Change the end goal for a tree in the TreeManager.
    Or change the global end goal for all trees (if conversation_id is not supplied).
    And applies these changes to current trees with default settings.

    Args:
        end_goal (str): The new end goal for the tree.
        conversation_id (str | None): Optional. The conversation ID which contains the tree.
            If not supplied, the end goal will be changed on all trees.
    """
    if conversation_id is None:
        self.config.end_goal = end_goal
        for conversation_id in self.trees:
            if not self.trees[conversation_id]["tree"]._config_modified:
                self.trees[conversation_id]["tree"].change_end_goal(end_goal)
                self.trees[conversation_id]["tree"]._config_modified = False
    else:
        if conversation_id not in self.trees:
            raise ValueError(f"Tree {conversation_id} not found")

        self.trees[conversation_id]["tree"].change_end_goal(end_goal)

change_style(style, conversation_id=None)

Change the style for a tree in the TreeManager. Or change the global style for all trees (if conversation_id is not supplied). And applies these changes to current trees with default settings.

Parameters:

Name	Type	Description	Default
style	str	The new style for the tree.	required
conversation_id	str | None	Optional. The conversation ID which contains the tree. If not supplied, the style will be changed on all trees.	None

Source code in elysia/api/services/tree.py

def change_style(self, style: str, conversation_id: str | None = None):
    """
    Change the style for a tree in the TreeManager.
    Or change the global style for all trees (if conversation_id is not supplied).
    And applies these changes to current trees with default settings.

    Args:
        style (str): The new style for the tree.
        conversation_id (str | None): Optional. The conversation ID which contains the tree.
            If not supplied, the style will be changed on all trees.
    """
    if conversation_id is None:
        self.config.style = style
        for conversation_id in self.trees:
            if not self.trees[conversation_id]["tree"]._config_modified:
                self.trees[conversation_id]["tree"].change_style(style)
                self.trees[conversation_id]["tree"]._config_modified = False
    else:
        if conversation_id not in self.trees:
            raise ValueError(f"Tree {conversation_id} not found")

        self.trees[conversation_id]["tree"].change_style(style)

check_all_trees_timeout()

Check all trees in the TreeManager and remove any that have not been active in the last tree_timeout.

Source code in elysia/api/services/tree.py

def check_all_trees_timeout(self):
    """
    Check all trees in the TreeManager and remove any that have not been active in the last tree_timeout.
    """
    if self.tree_timeout == datetime.timedelta(minutes=0):
        return

    convs_to_remove = []
    for i, conversation_id in enumerate(self.trees):
        if self.check_tree_timeout(conversation_id):
            convs_to_remove.append(conversation_id)

    for conversation_id in convs_to_remove:
        del self.trees[conversation_id]

check_tree_exists_weaviate(conversation_id, client_manager) async

Check if a tree exists in a Weaviate instance. The collection ELYSIA_TREES__ must exist, returns False if it doesn't.

Parameters:

Name	Type	Description	Default
conversation_id	str	The conversation ID which contains the tree.	required
client_manager	ClientManager	The client manager to use for the tree.	required

Returns:

Type	Description
bool	True if the tree exists in the Weaviate instance, False otherwise.

Source code in elysia/api/services/tree.py

async def check_tree_exists_weaviate(
    self, conversation_id: str, client_manager: ClientManager
):
    """
    Check if a tree exists in a Weaviate instance.
    The collection ELYSIA_TREES__ must exist, returns False if it doesn't.

    Args:
        conversation_id (str): The conversation ID which contains the tree.
        client_manager (ClientManager): The client manager to use for the tree.

    Returns:
        (bool): True if the tree exists in the Weaviate instance, False otherwise.
    """
    async with client_manager.connect_to_async_client() as client:
        if not await client.collections.exists("ELYSIA_TREES__"):
            return False

        collection = client.collections.get("ELYSIA_TREES__")
        uuid = generate_uuid5(conversation_id)
        return await collection.data.exists(uuid)

check_tree_timeout(conversation_id)

Check if a tree has been idle for the last tree_timeout.

Parameters:

Name	Type	Description	Default
conversation_id	str	The conversation ID which contains the tree.	required

Returns:

Type	Description
bool	True if the tree has been idle for the last tree_timeout, False otherwise.

Source code in elysia/api/services/tree.py

def check_tree_timeout(self, conversation_id: str):
    """
    Check if a tree has been idle for the last tree_timeout.

    Args:
        conversation_id (str): The conversation ID which contains the tree.

    Returns:
        (bool): True if the tree has been idle for the last tree_timeout, False otherwise.
    """
    # if tree not found, return True
    if conversation_id not in self.trees:
        return True

    # Remove any trees that have not been active in the last TREE_TIMEOUT minutes
    if (
        "last_request" in self.trees[conversation_id]
        and datetime.datetime.now() - self.trees[conversation_id]["last_request"]
        > self.tree_timeout
    ):
        return True

    return False

configure(conversation_id=None, replace=False, **kwargs)

Configure the settings for a tree in the TreeManager.

Parameters:

Name	Type	Description	Default
conversation_id	str | None	The conversation ID which contains the tree.	None
replace	bool	Whether to override the current settings with the new settings. When this is True, all existing settings are removed, and only the new settings are used. Defaults to False.	False
**kwargs	Any	The keyword arguments to pass to the Settings.configure() method.	{}

Source code in elysia/api/services/tree.py

def configure(
    self, conversation_id: str | None = None, replace: bool = False, **kwargs: Any
):
    """
    Configure the settings for a tree in the TreeManager.

    Args:
        conversation_id (str | None): The conversation ID which contains the tree.
        replace (bool): Whether to override the current settings with the new settings.
            When this is True, all existing settings are removed, and only the new settings are used.
            Defaults to False.
        **kwargs (Any): The keyword arguments to pass to the Settings.configure() method.
    """
    if conversation_id is None:
        self.settings.configure(replace=replace, **kwargs)
        self.config.settings = self.settings
    else:
        self.get_tree(conversation_id).settings.configure(replace=replace, **kwargs)

delete_tree_local(conversation_id)

Delete a tree from the TreeManager.

Parameters:

Name	Type	Description	Default
conversation_id	str	The conversation ID of the tree to be deleted.	required

Source code in elysia/api/services/tree.py

def delete_tree_local(self, conversation_id: str):
    """
    Delete a tree from the TreeManager.

    Args:
        conversation_id (str): The conversation ID of the tree to be deleted.
    """
    if conversation_id in self.trees:
        del self.trees[conversation_id]

delete_tree_weaviate(conversation_id, client_manager) async

Delete a tree from the stored trees in Weaviate.

Parameters:

Name	Type	Description	Default
conversation_id	str	The conversation ID of the tree to be deleted.	required
client_manager	ClientManager	The client manager pointing to the Weaviate instance containing the tree.	required

Source code in elysia/api/services/tree.py

async def delete_tree_weaviate(
    self, conversation_id: str, client_manager: ClientManager
):
    """
    Delete a tree from the stored trees in Weaviate.

    Args:
        conversation_id (str): The conversation ID of the tree to be deleted.
        client_manager (ClientManager): The client manager pointing to the Weaviate instance containing the tree.
    """
    await delete_tree_from_weaviate(
        conversation_id, "ELYSIA_TREES__", client_manager
    )

get_event(conversation_id)

Get the asyncio.Event for a tree in the TreeManager. This is cleared when the tree is processing, and set when the tree is idle. This is used to block the API from sending multiple requests to the tree at once.

Parameters:

Name	Type	Description	Default
conversation_id	str	The conversation ID which contains the tree.	required

Returns:

Type	Description
Event	The event for the tree.

Source code in elysia/api/services/tree.py

def get_event(self, conversation_id: str):
    """
    Get the asyncio.Event for a tree in the TreeManager.
    This is cleared when the tree is processing, and set when the tree is idle.
    This is used to block the API from sending multiple requests to the tree at once.

    Args:
        conversation_id (str): The conversation ID which contains the tree.

    Returns:
        (asyncio.Event): The event for the tree.
    """
    return self.trees[conversation_id]["event"]

get_tree(conversation_id)

Get a tree from the TreeManager. Will raise a ValueError if the tree is not found.

Parameters:

Name	Type	Description	Default
conversation_id	str	The conversation ID which contains the tree.	required

Returns:

Type	Description
Tree	The tree associated with the conversation ID.

Source code in elysia/api/services/tree.py

def get_tree(self, conversation_id: str):
    """
    Get a tree from the TreeManager.
    Will raise a ValueError if the tree is not found.

    Args:
        conversation_id (str): The conversation ID which contains the tree.

    Returns:
        (Tree): The tree associated with the conversation ID.
    """
    if conversation_id not in self.trees:
        raise ValueError(
            f"Tree {conversation_id} not found. Please initialise a tree first (by calling `add_tree`)."
        )

    return self.trees[conversation_id]["tree"]

load_tree_weaviate(conversation_id, client_manager) async

Load a tree from Weaviate. The conversation ID from the loaded tree is placed into the tree manager (possibly overwriting an existing tree with the same conversation ID). Then the tree itself is not returned - instead the list of frontend payloads that were yielded to the frontend by the tree is returned.

Parameters:

Name	Type	Description	Default
conversation_id	str	The conversation ID which contains the tree.	required
client_manager	ClientManager	The client manager to use for the tree.	required

Returns:

Type	Description
list	A list of dictionaries, each containing a frontend payload that was used to generate the tree. The list is ordered by the time the payload was originally sent to the frontend (at the time it was saved).

Source code in elysia/api/services/tree.py

async def load_tree_weaviate(
    self, conversation_id: str, client_manager: ClientManager
):
    """
    Load a tree from Weaviate.
    The conversation ID from the loaded tree is placed into the tree manager
    (possibly overwriting an existing tree with the same conversation ID).
    Then the tree itself is not returned - instead the list of frontend payloads
    that were yielded to the frontend by the tree is returned.

    Args:
        conversation_id (str): The conversation ID which contains the tree.
        client_manager (ClientManager): The client manager to use for the tree.

    Returns:
        (list): A list of dictionaries, each containing a frontend payload that was used to generate the tree.
            The list is ordered by the time the payload was originally sent to the frontend (at the time it was saved).
    """
    tree = await Tree.import_from_weaviate(
        "ELYSIA_TREES__", conversation_id, client_manager
    )
    if conversation_id not in self.trees:
        self.trees[conversation_id] = {
            "tree": None,
            "event": asyncio.Event(),
            "last_request": datetime.datetime.now(),
        }
    self.trees[conversation_id]["tree"] = tree
    self.trees[conversation_id]["event"].set()
    self.update_tree_last_request(conversation_id)
    return tree.returner.store

process_tree(query, conversation_id, query_id=None, training_route='', collection_names=[], client_manager=None) async

Process a tree in the TreeManager. This is an async generator which yields results from the tree.async_run() method.

Parameters:

Name	Type	Description	Default
query	str	Required. The user input/prompt to process in the decision tree.	required
conversation_id	str	Required. The conversation ID which contains the tree.	required
query_id	str	Optional. A unique identifier for the query. If not supplied, a random UUID will be generated.	None
training_route	str	Optional. The training route, a string of the form "tool1/tool2/tool1" etc. See the tree.async_run() method for more details.	''
collection_names	list[str]	Optional. A list of collection names to use in the query. If not supplied, all collections will be used.	[]
client_manager	ClientManager | None	Optional. A client manager to use for the query. If not supplied, a new ClientManager will be created.	None

Source code in elysia/api/services/tree.py

async def process_tree(
    self,
    query: str,
    conversation_id: str,
    query_id: str | None = None,
    training_route: str = "",
    collection_names: list[str] = [],
    client_manager: ClientManager | None = None,
):
    """
    Process a tree in the TreeManager.
    This is an async generator which yields results from the tree.async_run() method.

    Args:
        query (str): Required. The user input/prompt to process in the decision tree.
        conversation_id (str): Required. The conversation ID which contains the tree.
        query_id (str): Optional. A unique identifier for the query.
            If not supplied, a random UUID will be generated.
        training_route (str): Optional. The training route, a string of the form "tool1/tool2/tool1" etc.
            See the `tree.async_run()` method for more details.
        collection_names (list[str]): Optional. A list of collection names to use in the query.
            If not supplied, all collections will be used.
        client_manager (ClientManager | None): Optional. A client manager to use for the query.
            If not supplied, a new ClientManager will be created.
    """

    if query_id is None:
        query_id = str(uuid.uuid4())

    tree: Tree = self.get_tree(conversation_id)
    self.update_tree_last_request(conversation_id)

    # wait for the tree to be idle
    await self.trees[conversation_id]["event"].wait()

    # clear the event, set it to working
    self.trees[conversation_id]["event"].clear()

    try:
        async for yielded_result in tree.async_run(
            query,
            collection_names=collection_names,
            client_manager=client_manager,
            query_id=query_id,
            training_route=training_route,
            close_clients_after_completion=False,
        ):
            yield yielded_result
            self.update_tree_last_request(conversation_id)

    finally:
        # set the event to idle
        self.trees[conversation_id]["event"].set()

save_tree_weaviate(conversation_id, client_manager) async

Save a tree to Weaviate to collection ELYSIA_TREES__. Creates the collection if it doesn't exist.

Parameters:

Name	Type	Description	Default
conversation_id	str	The conversation ID which contains the tree.	required
client_manager	ClientManager	The client manager to use for the tree.	required

Source code in elysia/api/services/tree.py

async def save_tree_weaviate(
    self, conversation_id: str, client_manager: ClientManager
):
    """
    Save a tree to Weaviate to collection ELYSIA_TREES__.
    Creates the collection if it doesn't exist.

    Args:
        conversation_id (str): The conversation ID which contains the tree.
        client_manager (ClientManager): The client manager to use for the tree.
    """
    tree: Tree = self.get_tree(conversation_id)
    await tree.export_to_weaviate("ELYSIA_TREES__", client_manager)

tree_exists(conversation_id)

Check if a tree exists in the TreeManager.

Parameters:

Name	Type	Description	Default
conversation_id	str	The conversation ID which may contain the tree.	required

Returns:

Type	Description
bool	True if the tree exists, False otherwise.

Source code in elysia/api/services/tree.py

def tree_exists(self, conversation_id: str):
    """
    Check if a tree exists in the TreeManager.

    Args:
        conversation_id (str): The conversation ID which may contain the tree.

    Returns:
        (bool): True if the tree exists, False otherwise.
    """
    return conversation_id in self.trees