Objects

Completed

Bases: Update

Completed message to be sent to the frontend (tree is complete all recursions).

Source code in elysia/objects.py

class Completed(Update):
    """
    Completed message to be sent to the frontend (tree is complete all recursions).
    """

    def __init__(self):
        Update.__init__(self, "completed", {})

Error

Bases: Update

Error objects are used to communicate errors to the decision agent/tool calls. When yielded, Error objects are automatically saved inside the TreeData object. When calling the same tool again, the saved Error object is automatically loaded into any tool calls made with the same tool name. All errors are shown to the decision agent to help decide whether the tool should be called again (retried), or a different tool should be called.

Source code in elysia/objects.py

class Error(Update):
    """
    Error objects are used to communicate errors to the decision agent/tool calls.
    When yielded, Error objects are automatically saved inside the TreeData object.
    When calling the same tool again, the saved Error object is automatically loaded into any tool calls made with the same tool name.
    All errors are shown to the decision agent to help decide whether the tool should be called again (retried), or a different tool should be called.
    """

    def __init__(self, feedback: str = "", error_message: str = ""):
        """
        Args:
            feedback (str): The feedback to display to the decision agent.
                Usually this will be the error message, but it could be formatted more specifically.
        """
        self.feedback = feedback
        self.error_message = error_message

        if feedback == "":
            self.feedback = "An unknown issue occurred."

        Update.__init__(
            self,
            "self_healing_error",
            {"feedback": self.feedback, "error_message": self.error_message},
        )

__init__(feedback='', error_message='')

Parameters:

Name	Type	Description	Default
feedback	str	The feedback to display to the decision agent. Usually this will be the error message, but it could be formatted more specifically.	''

Source code in elysia/objects.py

def __init__(self, feedback: str = "", error_message: str = ""):
    """
    Args:
        feedback (str): The feedback to display to the decision agent.
            Usually this will be the error message, but it could be formatted more specifically.
    """
    self.feedback = feedback
    self.error_message = error_message

    if feedback == "":
        self.feedback = "An unknown issue occurred."

    Update.__init__(
        self,
        "self_healing_error",
        {"feedback": self.feedback, "error_message": self.error_message},
    )

Result

Bases: Return

Result objects are returned to the frontend. These are displayed on the frontend. E.g. a table, a chart, a text response, etc.

You can yield a Result directly, and specify the type and name of the result.

Source code in elysia/objects.py

class Result(Return):
    """
    Result objects are returned to the frontend.
    These are displayed on the frontend.
    E.g. a table, a chart, a text response, etc.

    You can yield a `Result` directly, and specify the `type` and `name` of the result.
    """

    def __init__(
        self,
        objects: list[dict],
        metadata: dict = {},
        payload_type: str = "default",
        name: str = "default",
        mapping: dict | None = None,
        llm_message: str | None = None,
        unmapped_keys: list[str] = ["_REF_ID"],
        display: bool = True,
    ):
        """
        Args:
            objects (list[dict]): The objects attached to this result.
            payload_type: (str): Identifier for the type of result.
            metadata (dict): The metadata attached to this result,
                for example, query used, time taken, any other information not directly within the objects
            name (str): The name of the result, e.g. this could be the name of the collection queried.
                Used to index the result in the environment.
            mapping (dict | None): A mapping of the objects to the frontend.
                Essentially, if the objects are going to be displayed on the frontend, the frontend has specific keys that it wants the objects to have.
                This is a dictionary that maps from frontend keys to the object keys.
            llm_message (str | None): A message to be displayed to the LLM to help it parse the result.
                You can use the following placeholders that will automatically be replaced with the correct values:

                - {type}: The type of the object
                - {name}: The name of the object
                - {num_objects}: The number of objects in the object
                - {metadata_key}: Any key in the metadata dictionary
            unmapped_keys (list[str]): A list of keys that are not mapped to the frontend.
            display (bool): Whether to display the result on the frontend when yielding this object.
                Defaults to `True`.
        """
        Return.__init__(self, "result", payload_type)
        self.objects = objects
        self.metadata = metadata
        self.name = name
        self.mapping = mapping
        self.llm_message = llm_message
        self.unmapped_keys = unmapped_keys
        self.display = display

    def __len__(self):
        return len(self.objects)

    def format_llm_message(self):
        """
        llm_message is a string that is used to help the LLM parse the output of the tool.
        It is a placeholder for the actual message that will be displayed to the user.

        You can use the following placeholders:

        - {payload_type}: The type of the object
        - {name}: The name of the object
        - {num_objects}: The number of objects in the object
        - {metadata_key}: Any key in the metadata dictionary
        """
        if self.llm_message is None:
            return ""

        return self.llm_message.format_map(
            {
                "payload_type": self.payload_type,
                "name": self.name,
                "num_objects": len(self),
                **self.metadata,
            }
        )

    def do_mapping(self, objects: list[dict]):

        if self.mapping is None:
            return objects

        output_objects = []
        for obj in objects:
            output_objects.append(
                {
                    key: obj[self.mapping[key]]
                    for key in self.mapping
                    if self.mapping[key] != ""
                }
            )
            output_objects[-1].update(
                {key: obj[key] for key in self.unmapped_keys if key in obj}
            )

        return output_objects

    def to_json(self, mapping: bool = False):
        """
        Convert the result to a list of dictionaries.

        Args:
            mapping (bool): Whether to map the objects to the frontend.
                This will use the `mapping` dictionary created on initialisation of the `Result` object.
                If `False`, the objects are returned as is.
                Defaults to `False`.

        Returns:
            (list[dict]): A list of dictionaries, which can be serialised to JSON.
        """
        from elysia.util.parsing import format_dict_to_serialisable

        assert all(
            isinstance(obj, dict) for obj in self.objects
        ), "All objects must be dictionaries"

        if mapping and self.mapping is not None:
            output_objects = self.do_mapping(self.objects)
        else:
            output_objects = self.objects

        for object in output_objects:
            format_dict_to_serialisable(object)

        return output_objects

    async def to_frontend(
        self,
        user_id: str,
        conversation_id: str,
        query_id: str,
    ):
        """
        Convert the result to a frontend payload.
        This is a wrapper around the `to_json` method.
        But the objects and metadata are inside a `payload` key, which also includes a `type` key,
        being the frontend identifier for the type of payload being sent.
        (e.g. `ticket`, `product`, `message`, etc.)

        The outside of the payload also contains the user_id, conversation_id, and query_id.

        Args:
            user_id (str): The user ID.
            conversation_id (str): The conversation ID.
            query_id (str): The query ID.

        Returns:
            (dict): The frontend payload, which is a dictionary with the following structure:
                ```python
                {
                    "type": "result",
                    "user_id": str,
                    "conversation_id": str,
                    "query_id": str,
                    "id": str,
                    "payload": dict = {
                        "type": str,
                        "objects": list[dict],
                        "metadata": dict,
                    },
                }
                ```
        """
        if not self.display:
            return

        objects = self.to_json(mapping=True)
        if len(objects) == 0:
            return

        payload = {
            "type": self.payload_type,
            "objects": objects,
            "metadata": self.metadata,
        }

        return {
            "type": self.frontend_type,
            "user_id": user_id,
            "conversation_id": conversation_id,
            "query_id": query_id,
            "id": self.frontend_type[:3] + "-" + str(uuid.uuid4()),
            "payload": payload,
        }

    def llm_parse(self):
        """
        This method is called when the result is displayed to the LLM.
        It is used to display custom information to the LLM about the result.
        If `llm_message` was defined, then the llm message is formatted using the placeholders.
        Otherwise a default message is used.

        Returns:
            (str): The formatted llm message.
        """

        if self.llm_message is not None:
            return self.format_llm_message()
        else:
            # Default message
            return f"Displayed: A {self.payload_type} object with {len(self.objects)} objects."

__init__(objects, metadata={}, payload_type='default', name='default', mapping=None, llm_message=None, unmapped_keys=['_REF_ID'], display=True)

Parameters:

Name	Type	Description	Default
objects	list[dict]	The objects attached to this result.	required
payload_type	str	(str): Identifier for the type of result.	'default'
metadata	dict	The metadata attached to this result, for example, query used, time taken, any other information not directly within the objects	{}
name	str	The name of the result, e.g. this could be the name of the collection queried. Used to index the result in the environment.	'default'
mapping	dict | None	A mapping of the objects to the frontend. Essentially, if the objects are going to be displayed on the frontend, the frontend has specific keys that it wants the objects to have. This is a dictionary that maps from frontend keys to the object keys.	None
llm_message	str | None	A message to be displayed to the LLM to help it parse the result. You can use the following placeholders that will automatically be replaced with the correct values: {type}: The type of the object {name}: The name of the object {num_objects}: The number of objects in the object {metadata_key}: Any key in the metadata dictionary	None
unmapped_keys	list[str]	A list of keys that are not mapped to the frontend.	['_REF_ID']
display	bool	Whether to display the result on the frontend when yielding this object. Defaults to True.	True

Source code in elysia/objects.py

def __init__(
    self,
    objects: list[dict],
    metadata: dict = {},
    payload_type: str = "default",
    name: str = "default",
    mapping: dict | None = None,
    llm_message: str | None = None,
    unmapped_keys: list[str] = ["_REF_ID"],
    display: bool = True,
):
    """
    Args:
        objects (list[dict]): The objects attached to this result.
        payload_type: (str): Identifier for the type of result.
        metadata (dict): The metadata attached to this result,
            for example, query used, time taken, any other information not directly within the objects
        name (str): The name of the result, e.g. this could be the name of the collection queried.
            Used to index the result in the environment.
        mapping (dict | None): A mapping of the objects to the frontend.
            Essentially, if the objects are going to be displayed on the frontend, the frontend has specific keys that it wants the objects to have.
            This is a dictionary that maps from frontend keys to the object keys.
        llm_message (str | None): A message to be displayed to the LLM to help it parse the result.
            You can use the following placeholders that will automatically be replaced with the correct values:

            - {type}: The type of the object
            - {name}: The name of the object
            - {num_objects}: The number of objects in the object
            - {metadata_key}: Any key in the metadata dictionary
        unmapped_keys (list[str]): A list of keys that are not mapped to the frontend.
        display (bool): Whether to display the result on the frontend when yielding this object.
            Defaults to `True`.
    """
    Return.__init__(self, "result", payload_type)
    self.objects = objects
    self.metadata = metadata
    self.name = name
    self.mapping = mapping
    self.llm_message = llm_message
    self.unmapped_keys = unmapped_keys
    self.display = display

format_llm_message()

llm_message is a string that is used to help the LLM parse the output of the tool. It is a placeholder for the actual message that will be displayed to the user.

You can use the following placeholders:

• {payload_type}: The type of the object
• {name}: The name of the object
• {num_objects}: The number of objects in the object
• {metadata_key}: Any key in the metadata dictionary

Source code in elysia/objects.py

def format_llm_message(self):
    """
    llm_message is a string that is used to help the LLM parse the output of the tool.
    It is a placeholder for the actual message that will be displayed to the user.

    You can use the following placeholders:

    - {payload_type}: The type of the object
    - {name}: The name of the object
    - {num_objects}: The number of objects in the object
    - {metadata_key}: Any key in the metadata dictionary
    """
    if self.llm_message is None:
        return ""

    return self.llm_message.format_map(
        {
            "payload_type": self.payload_type,
            "name": self.name,
            "num_objects": len(self),
            **self.metadata,
        }
    )

llm_parse()

This method is called when the result is displayed to the LLM. It is used to display custom information to the LLM about the result. If llm_message was defined, then the llm message is formatted using the placeholders. Otherwise a default message is used.

Returns:

Type	Description
str	The formatted llm message.

Source code in elysia/objects.py

def llm_parse(self):
    """
    This method is called when the result is displayed to the LLM.
    It is used to display custom information to the LLM about the result.
    If `llm_message` was defined, then the llm message is formatted using the placeholders.
    Otherwise a default message is used.

    Returns:
        (str): The formatted llm message.
    """

    if self.llm_message is not None:
        return self.format_llm_message()
    else:
        # Default message
        return f"Displayed: A {self.payload_type} object with {len(self.objects)} objects."

to_frontend(user_id, conversation_id, query_id) async

Convert the result to a frontend payload. This is a wrapper around the to_json method. But the objects and metadata are inside a payload key, which also includes a type key, being the frontend identifier for the type of payload being sent. (e.g. ticket, product, message, etc.)

The outside of the payload also contains the user_id, conversation_id, and query_id.

Parameters:

Name	Type	Description	Default
user_id	str	The user ID.	required
conversation_id	str	The conversation ID.	required
query_id	str	The query ID.	required

Returns:

Type	Description
dict	The frontend payload, which is a dictionary with the following structure: { "type": "result", "user_id": str, "conversation_id": str, "query_id": str, "id": str, "payload": dict = { "type": str, "objects": list[dict], "metadata": dict, }, }

Source code in elysia/objects.py

async def to_frontend(
    self,
    user_id: str,
    conversation_id: str,
    query_id: str,
):
    """
    Convert the result to a frontend payload.
    This is a wrapper around the `to_json` method.
    But the objects and metadata are inside a `payload` key, which also includes a `type` key,
    being the frontend identifier for the type of payload being sent.
    (e.g. `ticket`, `product`, `message`, etc.)

    The outside of the payload also contains the user_id, conversation_id, and query_id.

    Args:
        user_id (str): The user ID.
        conversation_id (str): The conversation ID.
        query_id (str): The query ID.

    Returns:
        (dict): The frontend payload, which is a dictionary with the following structure:
            ```python
            {
                "type": "result",
                "user_id": str,
                "conversation_id": str,
                "query_id": str,
                "id": str,
                "payload": dict = {
                    "type": str,
                    "objects": list[dict],
                    "metadata": dict,
                },
            }
            ```
    """
    if not self.display:
        return

    objects = self.to_json(mapping=True)
    if len(objects) == 0:
        return

    payload = {
        "type": self.payload_type,
        "objects": objects,
        "metadata": self.metadata,
    }

    return {
        "type": self.frontend_type,
        "user_id": user_id,
        "conversation_id": conversation_id,
        "query_id": query_id,
        "id": self.frontend_type[:3] + "-" + str(uuid.uuid4()),
        "payload": payload,
    }

to_json(mapping=False)

Convert the result to a list of dictionaries.

Parameters:

Name	Type	Description	Default
mapping	bool	Whether to map the objects to the frontend. This will use the mapping dictionary created on initialisation of the Result object. If False, the objects are returned as is. Defaults to False.	False

Returns:

Type	Description
list[dict]	A list of dictionaries, which can be serialised to JSON.

Source code in elysia/objects.py

def to_json(self, mapping: bool = False):
    """
    Convert the result to a list of dictionaries.

    Args:
        mapping (bool): Whether to map the objects to the frontend.
            This will use the `mapping` dictionary created on initialisation of the `Result` object.
            If `False`, the objects are returned as is.
            Defaults to `False`.

    Returns:
        (list[dict]): A list of dictionaries, which can be serialised to JSON.
    """
    from elysia.util.parsing import format_dict_to_serialisable

    assert all(
        isinstance(obj, dict) for obj in self.objects
    ), "All objects must be dictionaries"

    if mapping and self.mapping is not None:
        output_objects = self.do_mapping(self.objects)
    else:
        output_objects = self.objects

    for object in output_objects:
        format_dict_to_serialisable(object)

    return output_objects

Retrieval

Bases: Result

Store of returned objects from a query/aggregate/any displayed results.

Source code in elysia/objects.py

class Retrieval(Result):
    """
    Store of returned objects from a query/aggregate/any displayed results.
    """

    def __init__(
        self,
        objects: list[dict],
        metadata: dict = {},
        payload_type: str = "default",
        name: str | None = None,
        mapping: dict | None = None,
        unmapped_keys: list[str] = [
            "uuid",
            "ELYSIA_SUMMARY",
            "collection_name",
            "_REF_ID",
        ],
        display: bool = True,
    ) -> None:
        if name is None and "collection_name" in metadata:
            result_name = metadata["collection_name"]
        elif name is None:
            result_name = "default"
        else:
            result_name = name

        Result.__init__(
            self,
            objects=objects,
            payload_type=payload_type,
            metadata=metadata,
            name=result_name,
            mapping=mapping,
            unmapped_keys=unmapped_keys,
            display=display,
        )

    def add_summaries(self, summaries: list[str] = []) -> None:
        for i, obj in enumerate(self.objects):
            if i < len(summaries):
                obj["ELYSIA_SUMMARY"] = summaries[i]
            else:
                obj["ELYSIA_SUMMARY"] = ""

    def llm_parse(self) -> str:
        out = ""
        count = len(self.objects)

        if "collection_name" in self.metadata:
            if count != 0:
                out += f"\nQueried collection: '{self.metadata['collection_name']}' and returned {count} objects, "
            else:
                out += f"\nQueried collection: '{self.metadata['collection_name']}' but no objects were returned."
                out += f" Since it had no objects, judge the query that was created, and evaluate whether it was appropriate for the collection, the user prompt, and the data available."
                out += f" If it seemed innappropriate, you can choose to try again if you think it can still be completed (or there is more to do)."
        if "return_type" in self.metadata:
            out += f"\nReturned with type '{self.metadata['return_type']}', "
        if "output_type" in self.metadata:
            out += f"\noutputted '{'itemised summaries' if self.metadata['output_type'] == 'summary' else 'original objects'}'\n"
        if "query_text" in self.metadata:
            out += f"\nSearch terms: '{self.metadata['query_text']}'"
        if "query_type" in self.metadata:
            out += f"\nType of query: '{self.metadata['query_type']}'"
        if "impossible" in self.metadata:
            out += f"\nImpossible prompt: '{self.metadata['impossible']}'"
            if "collection_name" in self.metadata:
                out += f"\nThis attempt at querying the collection: {self.metadata['collection_name']} was deemed impossible."
            if "impossible_reason" in self.metadata:
                out += f"\nReasoning for impossibility: {self.metadata['impossible_reason']}"
        if "query_output" in self.metadata:
            out += f"\nThe query used was:\n{self.metadata['query_output']}"
        return out

    async def to_frontend(
        self,
        user_id: str,
        conversation_id: str,
        query_id: str,
    ) -> dict | None:
        objects = self.to_json(mapping=True)
        if len(objects) == 0:
            return

        payload = {
            "type": self.payload_type,
            "objects": objects,
            "metadata": self.metadata,
        }

        if "code" in self.metadata:
            payload["code"] = self.metadata["code"]

        return {
            "type": self.frontend_type,
            "user_id": user_id,
            "conversation_id": conversation_id,
            "query_id": query_id,
            "id": self.frontend_type[:3] + "-" + str(uuid.uuid4()),
            "payload": payload,
        }

Return

A Return object is something that is returned to the frontend. This kind of object is frontend-aware.

Source code in elysia/objects.py

class Return:
    """
    A Return object is something that is returned to the frontend.
    This kind of object is frontend-aware.
    """

    def __init__(self, frontend_type: str, payload_type: str):
        self.frontend_type = frontend_type  # frontend identifier
        self.payload_type = payload_type  # backend identifier

Status

Bases: Update

Status message to be sent to the frontend for real-time updates in words.

Source code in elysia/objects.py

class Status(Update):
    """
    Status message to be sent to the frontend for real-time updates in words.
    """

    def __init__(self, text: str):
        self.text = text
        Update.__init__(self, "status", {"text": text})

Text

Bases: Return

Frontend Return Type 1: Text Objects is usually a one-element list containing a dict with a "text" key only. But is not limited to this.

Source code in elysia/objects.py

class Text(Return):
    """
    Frontend Return Type 1: Text
    Objects is usually a one-element list containing a dict with a "text" key only.
    But is not limited to this.
    """

    def __init__(
        self,
        payload_type: str,
        objects: list[dict],
        metadata: dict = {},
        display: bool = True,
    ):
        Return.__init__(self, "text", payload_type)
        self.objects = objects
        self.metadata = metadata
        self.text = self._concat_text(self.objects)
        self.display = display

    def _concat_text(self, objects: list[dict]):
        text = ""
        for i, obj in enumerate(objects):
            if "text" in obj:
                spacer = ""
                if (
                    not obj["text"].endswith(" ")
                    and not obj["text"].endswith("\n")
                    and i != len(objects) - 1
                ):
                    spacer += " "

                if (
                    i != len(objects) - 1
                    and "text" in objects[i + 1]
                    and objects[i + 1]["text"].startswith("* ")
                    and not obj["text"].endswith("\n")
                ):
                    spacer += "\n"

                text += obj["text"] + spacer

        text = text.replace("_REF_ID", "")
        text = text.replace("REF_ID", "")
        text = text.replace("  ", " ")
        return text

    def to_json(self):
        return {
            "type": self.payload_type,
            "metadata": self.metadata,
            "objects": self.objects,
        }

    async def to_frontend(
        self,
        user_id: str,
        conversation_id: str,
        query_id: str,
    ):
        if not self.display:
            return

        return {
            "type": self.frontend_type,
            "id": self.frontend_type[:3] + "-" + str(uuid.uuid4()),
            "user_id": user_id,
            "conversation_id": conversation_id,
            "query_id": query_id,
            "payload": self.to_json(),
        }

Tool

The generic Tool class, which will be a superclass of any tools used by Elysia.

Source code in elysia/objects.py

class Tool(metaclass=ToolMeta):
    """
    The generic Tool class, which will be a superclass of any tools used by Elysia.
    """

    @classmethod
    def get_metadata(cls) -> dict[str, str | bool | dict | None]:
        """Get tool metadata without instantiation."""
        return {
            "name": getattr(cls, "_tool_name", None),
            "description": getattr(cls, "_tool_description", None),
            "inputs": getattr(cls, "_tool_inputs", None),
            "end": getattr(cls, "_tool_end", False),
        }

    def __init__(
        self,
        name: str,
        description: str,
        status: str = "",
        inputs: dict = {},
        end: bool = False,
        **kwargs,
    ):
        """
        Args:
            name (str): The name of the tool. Required.
            description (str): A detailed description of the tool to give to the LLM. Required.
            status (str): A status update message to display while the tool is running. Optional, defaults to "Running {name}...".
            inputs (dict): A dictionary of inputs for the tool, with the following structure:
                ```python
                {
                    input_name: {
                        "description": str,
                        "type": str,
                        "default": str,
                        "required": bool
                    }
                }
                ```
            end (bool): Whether the tool is an end tool. Optional, defaults to False.
        """
        self.name = name
        self.description = description
        self.inputs = inputs
        self.end = end

        if status == "":
            self.status = f"Running {self.name}..."
        else:
            self.status = status

    def get_default_inputs(self) -> dict:
        return {
            key: value["default"] if "default" in value else None
            for key, value in self.inputs.items()
        }

    async def run_if_true(
        self,
        tree_data,
        base_lm,
        complex_lm,
        client_manager,
    ) -> tuple[bool, dict]:
        """
        This method is called to check if the tool should be run automatically.
        If this returns `True`, then the tool is immediately called at the start of the tree.
        Otherwise it does not appear in the decision tree.
        You must also return a dictionary of inputs for the tool, which will be used to call the tool if `True` is returned.
        If the inputs are None or an empty dictionary, then the default inputs will be used.

        This must be an async function.

        Args:
            tree_data (TreeData): The tree data object.
            base_lm (LM): The base language model, a dspy.LM object.
            complex_lm (LM): The complex language model, a dspy.LM object.
            client_manager (ClientManager): The client manager, a way of interfacing with a Weaviate client.

        Returns:
            bool: True if the tool should be run automatically, False otherwise.
            dict: A dictionary of inputs for the tool, with the following structure:
                ```python
                {
                    input_name: input_value
                }
                ```
        """
        return False, {}

    async def is_tool_available(
        self,
        tree_data,
        base_lm,
        complex_lm,
        client_manager,
    ) -> bool:
        """
        This method is called to check if the tool is available.
        If this returns `True`, then the tool is available to be used by the LLM.
        Otherwise it does not appear in the decision tree.

        The difference between this and `run_if_true` is that `run_if_true` will _always_ run the __call__ method if it returns `True`,
        even if the LLM does not choose to use the tool.

        This must be an async function.

        Args:
            tree_data (TreeData): The tree data object.
            base_lm (LM): The base language model, a dspy.LM object.
            complex_lm (LM): The complex language model, a dspy.LM object.
            client_manager (ClientManager): The client manager, a way of interfacing with a Weaviate client.

        Returns:
            bool: True if the tool is available, False otherwise.
        """
        return True

    async def __call__(
        self, tree_data, inputs, base_lm, complex_lm, client_manager, **kwargs
    ) -> AsyncGenerator[Any, None]:
        """
        This method is called to run the tool.

        This must be an async generator, so objects must be yielded and not returned.

        Args:
            tree_data (TreeData): The data from the decision tree, includes the environment, user prompt, etc.
                See the `TreeData` class for more details.
            base_lm (LM): The base language model, a dspy.LM object.
            complex_lm (LM): The complex language model, a dspy.LM object.
            client_manager (ClientManager): The client manager, a way of interfacing with a Weaviate client.
        """
        yield None

__call__(tree_data, inputs, base_lm, complex_lm, client_manager, **kwargs) async

This method is called to run the tool.

This must be an async generator, so objects must be yielded and not returned.

Parameters:

Name	Type	Description	Default
tree_data	TreeData	The data from the decision tree, includes the environment, user prompt, etc. See the TreeData class for more details.	required
base_lm	LM	The base language model, a dspy.LM object.	required
complex_lm	LM	The complex language model, a dspy.LM object.	required
client_manager	ClientManager	The client manager, a way of interfacing with a Weaviate client.	required

Source code in elysia/objects.py

async def __call__(
    self, tree_data, inputs, base_lm, complex_lm, client_manager, **kwargs
) -> AsyncGenerator[Any, None]:
    """
    This method is called to run the tool.

    This must be an async generator, so objects must be yielded and not returned.

    Args:
        tree_data (TreeData): The data from the decision tree, includes the environment, user prompt, etc.
            See the `TreeData` class for more details.
        base_lm (LM): The base language model, a dspy.LM object.
        complex_lm (LM): The complex language model, a dspy.LM object.
        client_manager (ClientManager): The client manager, a way of interfacing with a Weaviate client.
    """
    yield None

__init__(name, description, status='', inputs={}, end=False, **kwargs)

Parameters:

Name	Type	Description	Default
name	str	The name of the tool. Required.	required
description	str	A detailed description of the tool to give to the LLM. Required.	required
status	str	A status update message to display while the tool is running. Optional, defaults to "Running {name}...".	''
inputs	dict	A dictionary of inputs for the tool, with the following structure: { input_name: { "description": str, "type": str, "default": str, "required": bool } }	{}
end	bool	Whether the tool is an end tool. Optional, defaults to False.	False

Source code in elysia/objects.py

def __init__(
    self,
    name: str,
    description: str,
    status: str = "",
    inputs: dict = {},
    end: bool = False,
    **kwargs,
):
    """
    Args:
        name (str): The name of the tool. Required.
        description (str): A detailed description of the tool to give to the LLM. Required.
        status (str): A status update message to display while the tool is running. Optional, defaults to "Running {name}...".
        inputs (dict): A dictionary of inputs for the tool, with the following structure:
            ```python
            {
                input_name: {
                    "description": str,
                    "type": str,
                    "default": str,
                    "required": bool
                }
            }
            ```
        end (bool): Whether the tool is an end tool. Optional, defaults to False.
    """
    self.name = name
    self.description = description
    self.inputs = inputs
    self.end = end

    if status == "":
        self.status = f"Running {self.name}..."
    else:
        self.status = status

get_metadata() classmethod

Get tool metadata without instantiation.

Source code in elysia/objects.py

@classmethod
def get_metadata(cls) -> dict[str, str | bool | dict | None]:
    """Get tool metadata without instantiation."""
    return {
        "name": getattr(cls, "_tool_name", None),
        "description": getattr(cls, "_tool_description", None),
        "inputs": getattr(cls, "_tool_inputs", None),
        "end": getattr(cls, "_tool_end", False),
    }

is_tool_available(tree_data, base_lm, complex_lm, client_manager) async

This method is called to check if the tool is available. If this returns True, then the tool is available to be used by the LLM. Otherwise it does not appear in the decision tree.

The difference between this and run_if_true is that run_if_true will always run the call method if it returns True, even if the LLM does not choose to use the tool.

This must be an async function.

Parameters:

Name	Type	Description	Default
tree_data	TreeData	The tree data object.	required
base_lm	LM	The base language model, a dspy.LM object.	required
complex_lm	LM	The complex language model, a dspy.LM object.	required
client_manager	ClientManager	The client manager, a way of interfacing with a Weaviate client.	required

Returns:

Name	Type	Description
bool	bool	True if the tool is available, False otherwise.

Source code in elysia/objects.py

async def is_tool_available(
    self,
    tree_data,
    base_lm,
    complex_lm,
    client_manager,
) -> bool:
    """
    This method is called to check if the tool is available.
    If this returns `True`, then the tool is available to be used by the LLM.
    Otherwise it does not appear in the decision tree.

    The difference between this and `run_if_true` is that `run_if_true` will _always_ run the __call__ method if it returns `True`,
    even if the LLM does not choose to use the tool.

    This must be an async function.

    Args:
        tree_data (TreeData): The tree data object.
        base_lm (LM): The base language model, a dspy.LM object.
        complex_lm (LM): The complex language model, a dspy.LM object.
        client_manager (ClientManager): The client manager, a way of interfacing with a Weaviate client.

    Returns:
        bool: True if the tool is available, False otherwise.
    """
    return True

run_if_true(tree_data, base_lm, complex_lm, client_manager) async

This method is called to check if the tool should be run automatically. If this returns True, then the tool is immediately called at the start of the tree. Otherwise it does not appear in the decision tree. You must also return a dictionary of inputs for the tool, which will be used to call the tool if True is returned. If the inputs are None or an empty dictionary, then the default inputs will be used.

This must be an async function.

Parameters:

Name	Type	Description	Default
tree_data	TreeData	The tree data object.	required
base_lm	LM	The base language model, a dspy.LM object.	required
complex_lm	LM	The complex language model, a dspy.LM object.	required
client_manager	ClientManager	The client manager, a way of interfacing with a Weaviate client.	required

Returns:

Name	Type	Description
bool	bool	True if the tool should be run automatically, False otherwise.
dict	dict	A dictionary of inputs for the tool, with the following structure: { input_name: input_value }

Source code in elysia/objects.py

async def run_if_true(
    self,
    tree_data,
    base_lm,
    complex_lm,
    client_manager,
) -> tuple[bool, dict]:
    """
    This method is called to check if the tool should be run automatically.
    If this returns `True`, then the tool is immediately called at the start of the tree.
    Otherwise it does not appear in the decision tree.
    You must also return a dictionary of inputs for the tool, which will be used to call the tool if `True` is returned.
    If the inputs are None or an empty dictionary, then the default inputs will be used.

    This must be an async function.

    Args:
        tree_data (TreeData): The tree data object.
        base_lm (LM): The base language model, a dspy.LM object.
        complex_lm (LM): The complex language model, a dspy.LM object.
        client_manager (ClientManager): The client manager, a way of interfacing with a Weaviate client.

    Returns:
        bool: True if the tool should be run automatically, False otherwise.
        dict: A dictionary of inputs for the tool, with the following structure:
            ```python
            {
                input_name: input_value
            }
            ```
    """
    return False, {}

ToolMeta

Bases: type

Metaclass that extracts tool metadata from init method.

Source code in elysia/objects.py

class ToolMeta(type):
    """Metaclass that extracts tool metadata from __init__ method."""

    _tool_name: str | None = None
    _tool_description: str | None = None
    _tool_inputs: dict | None = None
    _tool_end: bool | None = None

    @staticmethod
    def _convert_ast_dict(ast_dict: ast.Dict) -> dict:
        out = {}
        for i in range(len(ast_dict.keys)):
            k: ast.Constant = ast_dict.keys[i]  # type: ignore
            v: ast.Dict | ast.List | ast.Constant = ast_dict.values[i]  # type: ignore

            if isinstance(v, ast.Dict):
                out[k.value] = ToolMeta._convert_ast_dict(v)
            elif isinstance(v, ast.List):
                out[k.value] = ToolMeta._convert_ast_list(v)
            elif isinstance(v, ast.Constant):
                out[k.value] = v.value
        return out

    @staticmethod
    def _convert_ast_list(ast_list: ast.List) -> list:
        out = []
        for v in ast_list.values:  # type: ignore
            if isinstance(v, ast.Dict):
                out.append(ToolMeta._convert_ast_dict(v))
            elif isinstance(v, ast.List):
                out.append(ToolMeta._convert_ast_list(v))
            elif isinstance(v, ast.Constant):
                out.append(v.value)
        return out

    def __new__(cls, name, bases, namespace):
        new_class = super().__new__(cls, name, bases, namespace)

        # Skip the base Tool class itself
        if name == "Tool":
            return new_class

        # Extract metadata from __init__ method if it exists
        if "__init__" in namespace:
            try:
                init_method = namespace["__init__"]
                source = inspect.getsource(init_method)
                tree = ast.parse(source.strip())

                # Find the super().__init__() call and extract arguments
                for node in ast.walk(tree):
                    if (
                        isinstance(node, ast.Call)
                        and isinstance(node.func, ast.Attribute)
                        and node.func.attr == "__init__"
                        and isinstance(node.func.value, ast.Call)
                        and isinstance(node.func.value.func, ast.Name)
                        and node.func.value.func.id == "super"
                    ):

                        # Extract keyword arguments
                        for keyword in node.keywords:
                            if keyword.arg == "name" and isinstance(
                                keyword.value, ast.Constant
                            ):
                                new_class._tool_name = keyword.value.value
                            elif keyword.arg == "description" and isinstance(
                                keyword.value, ast.Constant
                            ):
                                new_class._tool_description = keyword.value.value
                            elif keyword.arg == "inputs" and isinstance(
                                keyword.value, ast.Dict
                            ):
                                new_class._tool_inputs = ToolMeta._convert_ast_dict(
                                    keyword.value
                                )
                            elif keyword.arg == "end" and isinstance(
                                keyword.value, ast.Constant
                            ):
                                new_class._tool_end = keyword.value.value
                        break
            except Exception:
                # If parsing fails, just continue without setting class attributes
                pass

        return new_class

Update

Bases: Return

Frontend Return Type 2: Update An update is something that is not displayed on the frontend, but gives information to the frontend. E.g. a warning, error, status message, etc.

Source code in elysia/objects.py

class Update(Return):
    """
    Frontend Return Type 2: Update
    An update is something that is not _displayed_ on the frontend, but gives information to the frontend.
    E.g. a warning, error, status message, etc.
    """

    def __init__(self, frontend_type: str, object: dict):
        Return.__init__(self, frontend_type, "update")
        self.object = object

    def to_json(self):
        return self.object

    async def to_frontend(self, user_id: str, conversation_id: str, query_id: str):
        return {
            "type": self.frontend_type,
            "user_id": user_id,
            "conversation_id": conversation_id,
            "query_id": query_id,
            "id": self.frontend_type[:3] + "-" + str(uuid.uuid4()),
            "payload": self.to_json(),
        }

Warning

Bases: Update

Warning message to be sent to the frontend.

Source code in elysia/objects.py

class Warning(Update):
    """
    Warning message to be sent to the frontend.
    """

    def __init__(self, text: str):
        self.text = text
        Update.__init__(self, "warning", {"text": text})

tool(function=None, *, status=None, end=False, tree=None, branch_id=None)

tool(function: Callable, *, status: str | None = None, end: bool = False, tree: Tree | None = None, branch_id: str | None = None) -> Tool

tool(function: None = None, *, status: str | None = None, end: bool = False, tree: Tree | None = None, branch_id: str | None = None) -> Callable[[Callable], Tool]

Create a tool from a python function. Use this decorator to create a tool from a function. The function must be an async function or async generator function.

Parameters:

Name	Type	Description	Default
function	Callable	The function to create a tool from.	None
status	str | None	The status message to display while the tool is running. Optional, defaults to None, which will use the default status message "Running {tool_name}...".	None
end	bool	Whether the tool can be at the end of the decision tree. Set to True when this tool is allowed to end the conversation. Optional, defaults to False.	False
tree	Tree | None	The tree to add the tool to. Optional, defaults to None, which will not add the tool to the tree.	None
branch_id	str | None	The branch of the tree to add the tool to. Optional, defaults to None, which will add the tool to the root branch.	None

Returns:

Type	Description
Tool	The tool object which can be added to the tree (via tree.add_tool(...)).

Source code in elysia/objects.py

def tool(
    function: Callable | None = None,
    *,
    status: str | None = None,
    end: bool = False,
    tree: "Tree | None" = None,
    branch_id: str | None = None,
) -> Callable[[Callable], Tool] | Tool:
    """
    Create a tool from a python function.
    Use this decorator to create a tool from a function.
    The function must be an async function or async generator function.

    Args:
        function (Callable): The function to create a tool from.
        status (str | None): The status message to display while the tool is running.
            Optional, defaults to None, which will use the default status message "Running {tool_name}...".
        end (bool): Whether the tool can be at the end of the decision tree.
            Set to True when this tool is allowed to end the conversation.
            Optional, defaults to False.
        tree (Tree | None): The tree to add the tool to.
            Optional, defaults to None, which will not add the tool to the tree.
        branch_id (str | None): The branch of the tree to add the tool to.
            Optional, defaults to None, which will add the tool to the root branch.

    Returns:
        (Tool): The tool object which can be added to the tree (via `tree.add_tool(...)`).
    """

    def decorator(function: Callable) -> Tool:
        async_function = inspect.iscoroutinefunction(function)
        async_generator_function = inspect.isasyncgenfunction(function)

        if not async_function and not async_generator_function:
            raise TypeError(
                "The provided function must be an async function or async generator function."
            )

        if "inputs" in list(function.__annotations__.keys()):
            raise TypeError(
                "The `inputs` argument is reserved in tool functions, please choose another name."
            )

        sig = inspect.signature(function)
        defaults_mapping = {
            k: v.default
            for k, v in sig.parameters.items()
            if v.default is not inspect.Parameter.empty
        }

        def list_to_list_of_dicts(result: list) -> list[dict]:
            objects = []
            for obj in result:
                if isinstance(obj, dict):
                    objects.append(obj)
                elif isinstance(obj, int | float | bool):
                    objects.append(
                        {
                            "tool_result": obj,
                        }
                    )
                elif isinstance(obj, list):
                    objects.append(list_to_list_of_dicts(obj))
                else:
                    objects.append(
                        {
                            "tool_result": obj,
                        }
                    )
            return objects

        def return_mapping(result, inputs: dict):
            if isinstance(result, Result | Text | Update | Status | Error):
                return result
            elif isinstance(result, str):
                return Response(result)
            elif isinstance(result, int | float | bool):
                return Result(
                    objects=[
                        {
                            "tool_result": result,
                        }
                    ],
                    metadata={
                        "tool_name": function.__name__,
                        "inputs_used": inputs,
                    },
                )
            elif isinstance(result, dict):
                return Result(
                    objects=[result],
                    metadata={
                        "tool_name": function.__name__,
                        "inputs_used": inputs,
                    },
                )
            elif isinstance(result, list):
                return Result(
                    objects=list_to_list_of_dicts(result),
                    metadata={
                        "tool_name": function.__name__,
                        "inputs_used": inputs,
                    },
                )
            else:
                return Result(
                    objects=[
                        {
                            "tool_result": result,
                        }
                    ],
                    metadata={
                        "tool_name": function.__name__,
                        "inputs_used": inputs,
                    },
                )

        class ToolClass(Tool):
            def __init__(self, **kwargs):
                super().__init__(
                    name=function.__name__,
                    description=function.__doc__ or "",
                    status=(
                        status
                        if status is not None
                        else f"Running {function.__name__}..."
                    ),
                    inputs={
                        input_key: {
                            "description": "",
                            "type": input_value,
                            "default": defaults_mapping.get(input_key, None),
                            "required": defaults_mapping.get(input_key, None)
                            is not None,
                        }
                        for input_key, input_value in function.__annotations__.items()
                        if input_key
                        not in [
                            "tree_data",
                            "base_lm",
                            "complex_lm",
                            "client_manager",
                            "return",
                        ]
                    },
                    end=end,
                )
                self._original_function = function

            async def __call__(
                self, tree_data, inputs, base_lm, complex_lm, client_manager, **kwargs
            ):

                if async_function:
                    tool_output = [
                        await function(
                            **inputs,
                            **{
                                k: v
                                for k, v in {
                                    "tree_data": tree_data,
                                    "base_lm": base_lm,
                                    "complex_lm": complex_lm,
                                    "client_manager": client_manager,
                                    **kwargs,
                                }.items()
                                if k in function.__annotations__
                            },
                        )
                    ]
                elif async_generator_function:
                    results = []
                    async for result in function(
                        **inputs,
                        **{
                            k: v
                            for k, v in {
                                "tree_data": tree_data,
                                "base_lm": base_lm,
                                "complex_lm": complex_lm,
                                "client_manager": client_manager,
                                **kwargs,
                            }.items()
                            if k in function.__annotations__
                        },
                    ):
                        results.append(result)
                    tool_output = results

                for result in tool_output:
                    mapped_result = return_mapping(result, inputs)
                    yield mapped_result

        tool_class = ToolClass()

        if tree is not None:
            tree.add_tool(tool_class, branch_id=branch_id)

        return tool_class

    if function is not None:
        return decorator(function)
    else:
        return decorator

CollectionData

Store of data about the Weaviate collections that are used in the tree. These are the output of the preprocess function.

You do not normally need to interact with this class directly. Instead, do so via the TreeData class, which has corresponding methods to get the data in a variety of formats. (Such as via the output_full_metadata method.)

Source code in elysia/tree/objects.py

class CollectionData:
    """
    Store of data about the Weaviate collections that are used in the tree.
    These are the output of the `preprocess` function.

    You do not normally need to interact with this class directly.
    Instead, do so via the `TreeData` class, which has corresponding methods to get the data in a variety of formats.
    (Such as via the `output_full_metadata` method.)
    """

    def __init__(
        self,
        collection_names: list[str],
        metadata: dict[str, Any] = {},
        logger: Logger | None = None,
    ):
        self.collection_names = collection_names
        self.metadata = metadata
        self.logger = logger

    async def set_collection_names(
        self, collection_names: list[str], client_manager: ClientManager
    ):
        temp_metadata = {}

        # check if any of these collections exist in the full metadata (cached)
        collections_to_get = []
        for collection_name in collection_names:
            if collection_name not in self.metadata:
                collections_to_get.append(collection_name)

        self.removed_collections = []
        self.incorrect_collections = []

        metadata_name = "ELYSIA_METADATA__"

        async with client_manager.connect_to_async_client() as client:
            # check if the metadata collection exists
            if not await client.collections.exists(metadata_name):
                self.removed_collections.extend(collections_to_get)
            else:
                metadata_collection = client.collections.get(metadata_name)
                filters = (
                    Filter.any_of(
                        [
                            Filter.by_property("name").equal(collection_name)
                            for collection_name in collections_to_get
                        ]
                    )
                    if len(collections_to_get) >= 1
                    else None
                )
                metadata = await metadata_collection.query.fetch_objects(
                    filters=filters,
                    limit=9999,
                )
                metadata_map = {
                    metadata_obj.properties["name"]: metadata_obj.properties
                    for metadata_obj in metadata.objects
                }

                for collection_name in collections_to_get:

                    if not await client.collections.exists(collection_name):
                        self.incorrect_collections.append(collection_name)
                        continue

                    if collection_name not in metadata_map:
                        self.removed_collections.append(collection_name)
                    else:
                        properties = metadata_map[collection_name]
                        format_dict_to_serialisable(properties)  # type: ignore
                        temp_metadata[collection_name] = properties

        if len(self.removed_collections) > 0 and self.logger:
            self.logger.warning(
                "The following collections have not been pre-processed for Elysia. "
                f"{self.removed_collections}. "
                "Ignoring these collections for now."
            )

        if len(self.incorrect_collections) > 0 and self.logger:
            self.logger.warning(
                "The following collections cannot be found in this Weaviate cluster. "
                f"{self.incorrect_collections}. "
                "These are being ignored for now. Please check that the collection names are correct."
            )

        self.collection_names = [
            collection_name
            for collection_name in collection_names
            if collection_name not in self.removed_collections
            and collection_name not in self.incorrect_collections
        ]

        # add to cached full metadata
        self.metadata = {
            **self.metadata,
            **{
                collection_name: temp_metadata[collection_name]
                for collection_name in temp_metadata
                if collection_name not in self.metadata
            },
        }

        return self.collection_names

    def output_full_metadata(
        self, collection_names: list[str] | None = None, with_mappings: bool = False
    ):
        if collection_names is None:
            collection_names = self.collection_names

        if with_mappings:
            return {
                collection_name: self.metadata[collection_name]
                for collection_name in collection_names
            }
        else:
            return {
                collection_name: {
                    k: v
                    for k, v in self.metadata[collection_name].items()
                    if k != "mappings"
                }
                for collection_name in collection_names
            }

    def output_collection_summaries(self, collection_names: list[str] | None = None):
        if collection_names is None:
            return {
                collection_name: self.metadata[collection_name]["summary"]
                for collection_name in self.collection_names
            }
        else:
            return {
                collection_name: self.metadata[collection_name]["summary"]
                for collection_name in collection_names
            }

    def output_mapping_lists(self):
        return {
            collection_name: list(self.metadata[collection_name]["mappings"].keys())
            for collection_name in self.collection_names
        }

    def output_mappings(self):
        return {
            collection_name: self.metadata[collection_name]["mappings"]
            for collection_name in self.collection_names
        }

    def to_json(self):
        return {
            "collection_names": self.collection_names,
            "metadata": self.metadata,
        }

    @classmethod
    def from_json(cls, json_data: dict, logger: Logger | None = None):
        return cls(
            collection_names=json_data["collection_names"],
            metadata=json_data["metadata"],
            logger=logger,
        )

Environment

Store of all objects across different types of queries and responses.

The environment is how the tree stores all objects across different types of queries and responses. This includes things like retrieved objects, retrieved metadata, retrieved summaries, etc.

This is persistent across the tree, so that all agents are aware of the same objects.

The environment is formatted and keyed as follows:

{
    "tool_name": {
        "result_name": [
            {
                "metadata": dict,
                "objects": list[dict],
            },
            ...
        ]
    }
}

Where "tool_name" is the name of the function that the result belongs to, e.g. if the result came from a tool called "query", then "tool_name" is "query".

"result_name" is the name of the Result object, which can be customised, e.g. if the result comes from a specific collection, then "result_name" is the name of the collection.

"metadata" is the metadata of the result, e.g. the time taken to retrieve the result, the query used, etc.

"objects" is the list of objects retrieved from the result. This is a list of dictionaries, where each dictionary is an object. It is important that objects should have be list of dictionaries. e.g. each object that was returned from a retrieval, where the fields of each dictionary are the fields of the object returned.

Each list under result_name is a dictionary with both metadata and objects keys. This is if, for example, you retrieve multiple objects from the same collection, each one is stored with different metadata. Because, for example, the query used to retrieve each object may be different (and stored differently in the metadata).

The environment is initialised with a default "SelfInfo.generic" key, which is a list of one object, containing information about Elysia itself.

You can use various methods to add, remove, replace, and find objects in the environment. See the methods below for more information.

Within the environment, there is a variable called hidden_environment, which is a dictionary of key-value pairs. This is used to store information that is not shown to the LLM, but is instead a 'store' of data that can be used across tools.

Source code in elysia/tree/objects.py

class Environment:
    """
    Store of all objects across different types of queries and responses.

    The environment is how the tree stores all objects across different types of queries and responses.
    This includes things like retrieved objects, retrieved metadata, retrieved summaries, etc.

    This is persistent across the tree, so that all agents are aware of the same objects.

    The environment is formatted and keyed as follows:
    ```python
    {
        "tool_name": {
            "result_name": [
                {
                    "metadata": dict,
                    "objects": list[dict],
                },
                ...
            ]
        }
    }
    ```

    Where `"tool_name"` is the name of the function that the result belongs to,
    e.g. if the result came from a tool called `"query"`, then `"tool_name"` is `"query"`.

    `"result_name"` is the name of the Result object, which can be customised,
    e.g. if the result comes from a specific collection, then `"result_name"` is the name of the collection.

    `"metadata"` is the metadata of the result,
    e.g. the time taken to retrieve the result, the query used, etc.

    `"objects"` is the list of objects retrieved from the result. This is a list of dictionaries, where each dictionary is an object.
    It is important that `objects` should have be list of dictionaries.
    e.g. each object that was returned from a retrieval, where the fields of each dictionary are the fields of the object returned.

    Each list under `result_name` is a dictionary with both `metadata` and `objects` keys.
    This is if, for example, you retrieve multiple objects from the same collection, each one is stored with different metadata.
    Because, for example, the query used to retrieve each object may be different (and stored differently in the metadata).

    The environment is initialised with a default "SelfInfo.generic" key, which is a list of one object, containing information about Elysia itself.

    You can use various methods to add, remove, replace, and find objects in the environment. See the methods below for more information.

    Within the environment, there is a variable called `hidden_environment`, which is a dictionary of key-value pairs.
    This is used to store information that is not shown to the LLM, but is instead a 'store' of data that can be used across tools.
    """

    def __init__(
        self,
        environment: dict[str, dict[str, Any]] | None = None,
        self_info: bool = True,
        hidden_environment: dict[str, Any] = {},
    ):
        if environment is None:
            environment = {}
        self.environment = environment
        self.hidden_environment = hidden_environment
        self.self_info = self_info
        if self_info:
            self.environment["SelfInfo"] = {}
            self.environment["SelfInfo"]["info"] = [
                {
                    "name": "Elysia",
                    "description": "An agentic RAG service in Weaviate.",
                    "purpose": remove_whitespace(
                        """Elysia is an agentic retrieval augmented generation (RAG) service, where users can query from Weaviate collections,
                    and the assistant will retrieve the most relevant information and answer the user's question. This includes a variety
                    of different ways to query, such as by filtering, sorting, querying multiple collections, and providing summaries
                    and textual responses.

                    Elysia will dynamically display retrieved objects from the collections in the frontend.
                    Elysia works via a tree-based approach, where the user's question is used to generate a tree of potential
                    queries to retrieve the most relevant information.
                    Each end of the tree connects to a separate agent that will perform a specific task, such as retrieval, aggregation, or generation, or more.

                    The tree itself has decision nodes that determine the next step in the query.
                    The decision nodes are decided via a decision-agent, which decides the task.

                    The agents communicate via a series of different prompts, which are stored in the prompt-library.
                    The decision-agent prompts are designed to be as general as possible, so that they can be used for a variety of different tasks.
                    Some of these variables include conversation history, retrieved objects, the user's original question, train of thought via model reasoning, and more.

                    The backend of Elysia is built with a range of libraries. Elysia is built with a lot of base python, and was constructed from the ground up to keep it as modular as possible.
                    The LLM components are built with DSPy, a library for training and running LLMs.
                    """
                    ),
                }
            ]

    def is_empty(self):
        """
        Check if the environment is empty.

        The "SelfInfo" key is not counted towards the empty environment.

        If the `.remove` method has been used, this is accounted for (e.g. empty lists count towards an empty environment).
        """
        empty = True
        for tool_key in self.environment.keys():
            if tool_key == "SelfInfo":
                continue

            for result_key in self.environment[tool_key].keys():
                if len(self.environment[tool_key][result_key]) > 0:
                    empty = False
                    break
        return empty

    def add(self, tool_name: str, result: Result, include_duplicates: bool = False):
        """
        Adds a result to the environment.
        Is called automatically by the tree when a result is returned from an agent.

        You can also add a result to the environment manually by using this method.
        In this case, you must be adding a 'Result' object (which has an implicit 'name' attribute used to key the environment).
        If you want to add something manually, you can use the `add_objects` method.

        Each item is added with a `_REF_ID` attribute, which is a unique identifier used to identify the object in the environment.
        If duplicate objects are detected, they are added with a duplicate `_REF_ID` entry detailing that they are a duplicate,
        as well as the `_REF_ID` of the original object.

        Args:
            tool_name (str): The name of the tool called that the result belongs to.
            result (Result): The result to add to the environment.
            include_duplicates (bool): Optional. Whether to include duplicate objects in the environment.
                Defaults to `False`, which still adds the duplicate object, but with a duplicate `_REF_ID` entry (and no repeating properties).
                If `True`, the duplicate object is added with a new `_REF_ID` entry, and the repeated properties are added to the object.
        """
        objects = result.to_json()
        name = result.name
        metadata = result.metadata
        if tool_name not in self.environment:
            self.environment[tool_name] = {}

        self.add_objects(tool_name, name, objects, metadata, include_duplicates)

    def add_objects(
        self,
        tool_name: str,
        name: str,
        objects: list[dict],
        metadata: dict = {},
        include_duplicates: bool = False,
    ):
        """
        Adds an object to the environment.
        Is not called automatically by the tree, so you must manually call this method.
        This is useful if you want to add an object to the environment manually that doesn't come from a Result object.

        Each item is added with a `_REF_ID` attribute, which is a unique identifier used to identify the object in the environment.
        If duplicate objects are detected, they are added with a duplicate `_REF_ID` entry detailing that they are a duplicate,
        as well as the `_REF_ID` of the original object.

        Args:
            tool_name (str): The name of the tool called that the result belongs to.
            name (str): The name of the result.
            objects (list[dict]): The objects to add to the environment.
            metadata (dict): Optional. The metadata of the objects to add to the environment.
                Defaults to an empty dictionary.
            include_duplicates (bool): Optional. Whether to include duplicate objects in the environment.
                Defaults to `False`, which still adds the duplicate object, but with a duplicate `_REF_ID` entry (and no repeating properties).
                If `True`, the duplicate object is added with a new `_REF_ID` entry, and the repeated properties are added to the object.

        """
        if tool_name not in self.environment:
            self.environment[tool_name] = {}

        if name not in self.environment[tool_name]:
            self.environment[tool_name][name] = []

        if len(objects) > 0:
            self.environment[tool_name][name].append(
                {
                    "metadata": metadata,
                    "objects": [],
                }
            )

            for i, obj in enumerate(objects):
                # check if the object is already in the environment
                obj_found = False
                where_obj = None
                for env_item in self.environment[tool_name][name]:
                    if obj in env_item["objects"]:
                        obj_found = True
                        where_obj = env_item["objects"].index(obj)
                        _REF_ID = env_item["objects"][where_obj]["_REF_ID"]
                        break

                if obj_found and not include_duplicates:
                    self.environment[tool_name][name][-1]["objects"].append(
                        {
                            "object_info": f"[repeat]",
                            "_REF_ID": _REF_ID,
                        }
                    )
                elif "_REF_ID" not in obj:
                    _REF_ID = f"{tool_name}_{name}_{len(self.environment[tool_name][name])}_{i}"
                    self.environment[tool_name][name][-1]["objects"].append(
                        {
                            "_REF_ID": _REF_ID,
                            **obj,
                        }
                    )
                else:
                    self.environment[tool_name][name][-1]["objects"].append(obj)

    def remove(self, tool_name: str, name: str, index: int | None = None):
        """
        Replaces the list of objects for the given `tool_name` and `name` with an empty list.

        Args:
            tool_name (str): The name of the tool called that the result belongs to.
            name (str): The name of the result.
            index (int | None): The index of the object to remove.
                If `None`, the entire list corresponding to `tool_name`/`name` is deleted.
                If an integer, the object at the given index is removed.
                Defaults to `None`.
                If `index=-1`, the last object is removed.
        """
        if tool_name in self.environment:
            if name in self.environment[tool_name]:
                if index is None:
                    self.environment[tool_name][name] = []
                else:
                    self.environment[tool_name][name].pop(index)

    def replace(
        self,
        tool_name: str,
        name: str,
        objects: list[dict],
        metadata: dict = {},
        index: int | None = None,
    ):
        """
        Replaces the list of objects for the given `tool_name` and `name` with the given list of objects.

        Args:
            tool_name (str): The name of the tool called that the result belongs to.
            name (str): The name of the result.
            objects (list[dict]): The objects to replace the existing objects with.
            metadata (dict): The metadata of the objects to replace the existing objects with.
            index (int | None): The index of the object to replace.
                If `None`, the entire list corresponding to `tool_name`/`name` is deleted and replaced with the new objects.
                If an integer, the object at the given index is replaced with the new objects.
                Defaults to `None`.
        """
        if tool_name in self.environment:
            if name in self.environment[tool_name]:
                if index is None:
                    self.environment[tool_name][name] = [
                        {
                            "metadata": metadata,
                            "objects": objects,
                        }
                    ]
                else:
                    self.environment[tool_name][name][index] = {
                        "metadata": metadata,
                        "objects": objects,
                    }

    def find(self, tool_name: str, name: str, index: int | None = None):
        """
        Finds a corresponding list of objects in the environment.
        Keyed via `tool_name` and `name`. See the base class description for more information on how the environment is keyed.

        Args:
            tool_name (str): The name of the tool called that the result belongs to.
            name (str): The name of the result.
            index (int | None): The index of the object to find.
                If `None`, the entire list corresponding to `tool_name`/`name` is returned.
                If an integer, the object at the given index is returned.

        Returns:
            (list[dict]): if `index` is `None` - The list of objects for the given `tool_name` and `name`.
            (dict): if `index` is an integer - The object at the given `index` for the given `tool_name` and `name`.
            (None): If the `tool_name` or `name` is not found in the environment.
        """

        if tool_name not in self.environment:
            return None
        if name not in self.environment[tool_name]:
            return None

        if index is None:
            return self.environment[tool_name][name]
        else:
            return self.environment[tool_name][name][index]

    def to_json(self, remove_unserialisable: bool = False):
        """
        Converts the environment to a JSON serialisable format.
        Used to access specific objects from the environment.
        """

        env_copy = deepcopy(self.environment)
        hidden_env_copy = deepcopy(self.hidden_environment)

        # Check if environment and hidden_environment are JSON serialisable
        for tool_name in env_copy:
            if tool_name != "SelfInfo":
                for name in self.environment[tool_name]:
                    for obj_metadata in self.environment[tool_name][name]:
                        format_dict_to_serialisable(
                            obj_metadata["metadata"], remove_unserialisable
                        )
                        for obj in obj_metadata["objects"]:
                            format_dict_to_serialisable(obj, remove_unserialisable)

        format_dict_to_serialisable(hidden_env_copy, remove_unserialisable)

        return {
            "environment": env_copy,
            "hidden_environment": hidden_env_copy,
            "self_info": self.self_info,
        }

    @classmethod
    def from_json(cls, json_data: dict):
        return cls(
            environment=json_data["environment"],
            hidden_environment=json_data["hidden_environment"],
            self_info=json_data["self_info"],
        )

add(tool_name, result, include_duplicates=False)

Adds a result to the environment. Is called automatically by the tree when a result is returned from an agent.

You can also add a result to the environment manually by using this method. In this case, you must be adding a 'Result' object (which has an implicit 'name' attribute used to key the environment). If you want to add something manually, you can use the add_objects method.

Each item is added with a _REF_ID attribute, which is a unique identifier used to identify the object in the environment. If duplicate objects are detected, they are added with a duplicate _REF_ID entry detailing that they are a duplicate, as well as the _REF_ID of the original object.

Parameters:

Name	Type	Description	Default
tool_name	str	The name of the tool called that the result belongs to.	required
result	Result	The result to add to the environment.	required
include_duplicates	bool	Optional. Whether to include duplicate objects in the environment. Defaults to False, which still adds the duplicate object, but with a duplicate _REF_ID entry (and no repeating properties). If True, the duplicate object is added with a new _REF_ID entry, and the repeated properties are added to the object.	False

Source code in elysia/tree/objects.py

def add(self, tool_name: str, result: Result, include_duplicates: bool = False):
    """
    Adds a result to the environment.
    Is called automatically by the tree when a result is returned from an agent.

    You can also add a result to the environment manually by using this method.
    In this case, you must be adding a 'Result' object (which has an implicit 'name' attribute used to key the environment).
    If you want to add something manually, you can use the `add_objects` method.

    Each item is added with a `_REF_ID` attribute, which is a unique identifier used to identify the object in the environment.
    If duplicate objects are detected, they are added with a duplicate `_REF_ID` entry detailing that they are a duplicate,
    as well as the `_REF_ID` of the original object.

    Args:
        tool_name (str): The name of the tool called that the result belongs to.
        result (Result): The result to add to the environment.
        include_duplicates (bool): Optional. Whether to include duplicate objects in the environment.
            Defaults to `False`, which still adds the duplicate object, but with a duplicate `_REF_ID` entry (and no repeating properties).
            If `True`, the duplicate object is added with a new `_REF_ID` entry, and the repeated properties are added to the object.
    """
    objects = result.to_json()
    name = result.name
    metadata = result.metadata
    if tool_name not in self.environment:
        self.environment[tool_name] = {}

    self.add_objects(tool_name, name, objects, metadata, include_duplicates)

add_objects(tool_name, name, objects, metadata={}, include_duplicates=False)

Adds an object to the environment. Is not called automatically by the tree, so you must manually call this method. This is useful if you want to add an object to the environment manually that doesn't come from a Result object.

Each item is added with a _REF_ID attribute, which is a unique identifier used to identify the object in the environment. If duplicate objects are detected, they are added with a duplicate _REF_ID entry detailing that they are a duplicate, as well as the _REF_ID of the original object.

Parameters:

Name	Type	Description	Default
tool_name	str	The name of the tool called that the result belongs to.	required
name	str	The name of the result.	required
objects	list[dict]	The objects to add to the environment.	required
metadata	dict	Optional. The metadata of the objects to add to the environment. Defaults to an empty dictionary.	{}
include_duplicates	bool	Optional. Whether to include duplicate objects in the environment. Defaults to False, which still adds the duplicate object, but with a duplicate _REF_ID entry (and no repeating properties). If True, the duplicate object is added with a new _REF_ID entry, and the repeated properties are added to the object.	False

Source code in elysia/tree/objects.py

def add_objects(
    self,
    tool_name: str,
    name: str,
    objects: list[dict],
    metadata: dict = {},
    include_duplicates: bool = False,
):
    """
    Adds an object to the environment.
    Is not called automatically by the tree, so you must manually call this method.
    This is useful if you want to add an object to the environment manually that doesn't come from a Result object.

    Each item is added with a `_REF_ID` attribute, which is a unique identifier used to identify the object in the environment.
    If duplicate objects are detected, they are added with a duplicate `_REF_ID` entry detailing that they are a duplicate,
    as well as the `_REF_ID` of the original object.

    Args:
        tool_name (str): The name of the tool called that the result belongs to.
        name (str): The name of the result.
        objects (list[dict]): The objects to add to the environment.
        metadata (dict): Optional. The metadata of the objects to add to the environment.
            Defaults to an empty dictionary.
        include_duplicates (bool): Optional. Whether to include duplicate objects in the environment.
            Defaults to `False`, which still adds the duplicate object, but with a duplicate `_REF_ID` entry (and no repeating properties).
            If `True`, the duplicate object is added with a new `_REF_ID` entry, and the repeated properties are added to the object.

    """
    if tool_name not in self.environment:
        self.environment[tool_name] = {}

    if name not in self.environment[tool_name]:
        self.environment[tool_name][name] = []

    if len(objects) > 0:
        self.environment[tool_name][name].append(
            {
                "metadata": metadata,
                "objects": [],
            }
        )

        for i, obj in enumerate(objects):
            # check if the object is already in the environment
            obj_found = False
            where_obj = None
            for env_item in self.environment[tool_name][name]:
                if obj in env_item["objects"]:
                    obj_found = True
                    where_obj = env_item["objects"].index(obj)
                    _REF_ID = env_item["objects"][where_obj]["_REF_ID"]
                    break

            if obj_found and not include_duplicates:
                self.environment[tool_name][name][-1]["objects"].append(
                    {
                        "object_info": f"[repeat]",
                        "_REF_ID": _REF_ID,
                    }
                )
            elif "_REF_ID" not in obj:
                _REF_ID = f"{tool_name}_{name}_{len(self.environment[tool_name][name])}_{i}"
                self.environment[tool_name][name][-1]["objects"].append(
                    {
                        "_REF_ID": _REF_ID,
                        **obj,
                    }
                )
            else:
                self.environment[tool_name][name][-1]["objects"].append(obj)

find(tool_name, name, index=None)

Finds a corresponding list of objects in the environment. Keyed via tool_name and name. See the base class description for more information on how the environment is keyed.

Parameters:

Name	Type	Description	Default
tool_name	str	The name of the tool called that the result belongs to.	required
name	str	The name of the result.	required
index	int | None	The index of the object to find. If None, the entire list corresponding to tool_name/name is returned. If an integer, the object at the given index is returned.	None

Returns:

Type	Description
list[dict]	if index is None - The list of objects for the given tool_name and name.
dict	if index is an integer - The object at the given index for the given tool_name and name.
None	If the tool_name or name is not found in the environment.

Source code in elysia/tree/objects.py

def find(self, tool_name: str, name: str, index: int | None = None):
    """
    Finds a corresponding list of objects in the environment.
    Keyed via `tool_name` and `name`. See the base class description for more information on how the environment is keyed.

    Args:
        tool_name (str): The name of the tool called that the result belongs to.
        name (str): The name of the result.
        index (int | None): The index of the object to find.
            If `None`, the entire list corresponding to `tool_name`/`name` is returned.
            If an integer, the object at the given index is returned.

    Returns:
        (list[dict]): if `index` is `None` - The list of objects for the given `tool_name` and `name`.
        (dict): if `index` is an integer - The object at the given `index` for the given `tool_name` and `name`.
        (None): If the `tool_name` or `name` is not found in the environment.
    """

    if tool_name not in self.environment:
        return None
    if name not in self.environment[tool_name]:
        return None

    if index is None:
        return self.environment[tool_name][name]
    else:
        return self.environment[tool_name][name][index]

is_empty()

Check if the environment is empty.

The "SelfInfo" key is not counted towards the empty environment.

If the .remove method has been used, this is accounted for (e.g. empty lists count towards an empty environment).

Source code in elysia/tree/objects.py

def is_empty(self):
    """
    Check if the environment is empty.

    The "SelfInfo" key is not counted towards the empty environment.

    If the `.remove` method has been used, this is accounted for (e.g. empty lists count towards an empty environment).
    """
    empty = True
    for tool_key in self.environment.keys():
        if tool_key == "SelfInfo":
            continue

        for result_key in self.environment[tool_key].keys():
            if len(self.environment[tool_key][result_key]) > 0:
                empty = False
                break
    return empty

remove(tool_name, name, index=None)

Replaces the list of objects for the given tool_name and name with an empty list.

Parameters:

Name	Type	Description	Default
tool_name	str	The name of the tool called that the result belongs to.	required
name	str	The name of the result.	required
index	int | None	The index of the object to remove. If None, the entire list corresponding to tool_name/name is deleted. If an integer, the object at the given index is removed. Defaults to None. If index=-1, the last object is removed.	None

Source code in elysia/tree/objects.py

def remove(self, tool_name: str, name: str, index: int | None = None):
    """
    Replaces the list of objects for the given `tool_name` and `name` with an empty list.

    Args:
        tool_name (str): The name of the tool called that the result belongs to.
        name (str): The name of the result.
        index (int | None): The index of the object to remove.
            If `None`, the entire list corresponding to `tool_name`/`name` is deleted.
            If an integer, the object at the given index is removed.
            Defaults to `None`.
            If `index=-1`, the last object is removed.
    """
    if tool_name in self.environment:
        if name in self.environment[tool_name]:
            if index is None:
                self.environment[tool_name][name] = []
            else:
                self.environment[tool_name][name].pop(index)

replace(tool_name, name, objects, metadata={}, index=None)

Replaces the list of objects for the given tool_name and name with the given list of objects.

Parameters:

Name	Type	Description	Default
tool_name	str	The name of the tool called that the result belongs to.	required
name	str	The name of the result.	required
objects	list[dict]	The objects to replace the existing objects with.	required
metadata	dict	The metadata of the objects to replace the existing objects with.	{}
index	int | None	The index of the object to replace. If None, the entire list corresponding to tool_name/name is deleted and replaced with the new objects. If an integer, the object at the given index is replaced with the new objects. Defaults to None.	None

Source code in elysia/tree/objects.py

def replace(
    self,
    tool_name: str,
    name: str,
    objects: list[dict],
    metadata: dict = {},
    index: int | None = None,
):
    """
    Replaces the list of objects for the given `tool_name` and `name` with the given list of objects.

    Args:
        tool_name (str): The name of the tool called that the result belongs to.
        name (str): The name of the result.
        objects (list[dict]): The objects to replace the existing objects with.
        metadata (dict): The metadata of the objects to replace the existing objects with.
        index (int | None): The index of the object to replace.
            If `None`, the entire list corresponding to `tool_name`/`name` is deleted and replaced with the new objects.
            If an integer, the object at the given index is replaced with the new objects.
            Defaults to `None`.
    """
    if tool_name in self.environment:
        if name in self.environment[tool_name]:
            if index is None:
                self.environment[tool_name][name] = [
                    {
                        "metadata": metadata,
                        "objects": objects,
                    }
                ]
            else:
                self.environment[tool_name][name][index] = {
                    "metadata": metadata,
                    "objects": objects,
                }

to_json(remove_unserialisable=False)

Converts the environment to a JSON serialisable format. Used to access specific objects from the environment.

Source code in elysia/tree/objects.py

def to_json(self, remove_unserialisable: bool = False):
    """
    Converts the environment to a JSON serialisable format.
    Used to access specific objects from the environment.
    """

    env_copy = deepcopy(self.environment)
    hidden_env_copy = deepcopy(self.hidden_environment)

    # Check if environment and hidden_environment are JSON serialisable
    for tool_name in env_copy:
        if tool_name != "SelfInfo":
            for name in self.environment[tool_name]:
                for obj_metadata in self.environment[tool_name][name]:
                    format_dict_to_serialisable(
                        obj_metadata["metadata"], remove_unserialisable
                    )
                    for obj in obj_metadata["objects"]:
                        format_dict_to_serialisable(obj, remove_unserialisable)

    format_dict_to_serialisable(hidden_env_copy, remove_unserialisable)

    return {
        "environment": env_copy,
        "hidden_environment": hidden_env_copy,
        "self_info": self.self_info,
    }

TreeData

Store of data across the tree. This includes things like conversation history, actions, decisions, etc. These data are given to ALL agents, so every agent is aware of the stage of the decision processes.

This also contains functions that process the data into an LLM friendly format, such as a string with extra description. E.g. the number of trees completed is converted into a string (i/N) and additional warnings if i is close to N.

The TreeData has the following objects:

• collection_data (CollectionData): The collection metadata/schema, which contains information about the collections used in the tree. This is the store of data that is saved by the preprocess function, and retrieved on initialisation of this object.
• atlas (Atlas): The atlas, described in the Atlas class.
• user_prompt (str): The user's prompt.
• conversation_history (list[dict]): The conversation history stored in the current tree, of the form:

  [
      {
          "role": "user" | "assistant",
          "content": str,
      },
      ...
  ]
  

• environment (Environment): The environment, described in the Environment class.
• tasks_completed (list[dict]): The tasks completed as a list of dictionaries. This is separate from the environment, as it separates what tasks were completed in each prompt in which order.
• num_trees_completed (int): The current level of the decision tree, how many iterations have been completed so far.
• recursion_limit (int): The maximum number of iterations allowed in the decision tree.
• errors (dict): A dictionary of self-healing errors that have occurred in the tree. Keyed by the function name that caused the error.

In general, you should not initialise this class directly. But you can access the data in this class to access the relevant data from the tree (in e.g. tool construction/usage).

Source code in elysia/tree/objects.py

class TreeData:
    """
    Store of data across the tree.
    This includes things like conversation history, actions, decisions, etc.
    These data are given to ALL agents, so every agent is aware of the stage of the decision processes.

    This also contains functions that process the data into an LLM friendly format,
    such as a string with extra description.
    E.g. the number of trees completed is converted into a string (i/N)
         and additional warnings if i is close to N.

    The TreeData has the following objects:

    - collection_data (CollectionData): The collection metadata/schema, which contains information about the collections used in the tree.
        This is the store of data that is saved by the `preprocess` function, and retrieved on initialisation of this object.
    - atlas (Atlas): The atlas, described in the Atlas class.
    - user_prompt (str): The user's prompt.
    - conversation_history (list[dict]): The conversation history stored in the current tree, of the form:
        ```python
        [
            {
                "role": "user" | "assistant",
                "content": str,
            },
            ...
        ]
        ```
    - environment (Environment): The environment, described in the Environment class.
    - tasks_completed (list[dict]): The tasks completed as a list of dictionaries.
        This is separate from the environment, as it separates what tasks were completed in each prompt in which order.
    - num_trees_completed (int): The current level of the decision tree, how many iterations have been completed so far.
    - recursion_limit (int): The maximum number of iterations allowed in the decision tree.
    - errors (dict): A dictionary of self-healing errors that have occurred in the tree. Keyed by the function name that caused the error.

    In general, you should not initialise this class directly.
    But you can access the data in this class to access the relevant data from the tree (in e.g. tool construction/usage).
    """

    def __init__(
        self,
        collection_data: CollectionData,
        atlas: Atlas,
        user_prompt: str | None = None,
        conversation_history: list[dict] | None = None,
        environment: Environment | None = None,
        tasks_completed: list[dict] | None = None,
        num_trees_completed: int | None = None,
        recursion_limit: int | None = None,
        settings: Settings | None = None,
    ):
        if settings is None:
            self.settings = environment_settings
        else:
            self.settings = settings

        # -- Base Data --
        if user_prompt is None:
            self.user_prompt = ""
        else:
            self.user_prompt = user_prompt

        if conversation_history is None:
            self.conversation_history = []
        else:
            self.conversation_history = conversation_history

        if environment is None:
            self.environment = Environment()
        else:
            self.environment = environment

        if tasks_completed is None:
            self.tasks_completed = []
        else:
            self.tasks_completed = tasks_completed

        if num_trees_completed is None:
            self.num_trees_completed = 0
        else:
            self.num_trees_completed = num_trees_completed

        if recursion_limit is None:
            self.recursion_limit = 3
        else:
            self.recursion_limit = recursion_limit

        # -- Atlas --
        self.atlas = atlas

        # -- Collection Data --
        self.collection_data = collection_data
        self.collection_names = []

        # -- Errors --
        self.errors: dict[str, list[str]] = {}
        self.current_task = None

    def set_property(self, property: str, value: Any):
        self.__dict__[property] = value

    def update_string(self, property: str, value: str):
        if property not in self.__dict__:
            self.__dict__[property] = ""
        self.__dict__[property] += value

    def update_list(self, property: str, value: Any):
        if property not in self.__dict__:
            self.__dict__[property] = []
        self.__dict__[property].append(value)

    def update_dict(self, property: str, key: str, value: Any):
        if property not in self.__dict__:
            self.__dict__[property] = {}
        self.__dict__[property][key] = value

    def delete_from_dict(self, property: str, key: str):
        if property in self.__dict__ and key in self.__dict__[property]:
            del self.__dict__[property][key]

    def soft_reset(self):
        self.previous_reasoning = {}

    def _update_task(self, task_dict, key, value):
        if value is not None:
            if key in task_dict:
                # If key already exists, append to it
                if isinstance(value, str):
                    task_dict[key] += "\n" + value
                elif isinstance(value, int) or isinstance(value, float):
                    task_dict[key] += value
                elif isinstance(value, list):
                    task_dict[key].extend(value)
                elif isinstance(value, dict):
                    task_dict[key].update(value)
                elif isinstance(value, bool):
                    task_dict[key] = value
            elif key not in task_dict:
                # If key does not exist, create it
                task_dict[key] = value

    def update_tasks_completed(
        self, prompt: str, task: str, num_trees_completed: int, **kwargs
    ):
        # search to see if the current prompt already has an entry for this task
        prompt_found = False
        task_found = False
        iteration_found = False

        for i, task_prompt in enumerate(self.tasks_completed):
            if task_prompt["prompt"] == prompt:
                prompt_found = True
                for j, task_j in enumerate(task_prompt["task"]):
                    if task_j["task"] == task:
                        task_found = True
                        task_i = j  # position of the task in the prompt

                    if task_j["iteration"] == num_trees_completed:
                        iteration_found = True

        # If the prompt is not found, add it to the list
        if not prompt_found:
            self.tasks_completed.append({"prompt": prompt, "task": [{}]})
            self.tasks_completed[-1]["task"][0]["task"] = task
            self.tasks_completed[-1]["task"][0]["iteration"] = num_trees_completed
            for kwarg in kwargs:
                self._update_task(
                    self.tasks_completed[-1]["task"][0], kwarg, kwargs[kwarg]
                )
            return

        # If the prompt is found but the task is not, add it to the list
        if prompt_found and not task_found:
            self.tasks_completed[-1]["task"].append({})
            self.tasks_completed[-1]["task"][-1]["task"] = task
            self.tasks_completed[-1]["task"][-1]["iteration"] = num_trees_completed
            for kwarg in kwargs:
                self._update_task(
                    self.tasks_completed[-1]["task"][-1], kwarg, kwargs[kwarg]
                )
            return

        # task already exists in this query, but the iteration is new
        if prompt_found and task_found and not iteration_found:
            self.tasks_completed[-1]["task"].append({})
            self.tasks_completed[-1]["task"][-1]["task"] = task
            self.tasks_completed[-1]["task"][-1]["iteration"] = num_trees_completed
            for kwarg in kwargs:
                self._update_task(
                    self.tasks_completed[-1]["task"][-1], kwarg, kwargs[kwarg]
                )
            return

        # If the prompt is found and the task is found, update the task
        if prompt_found and task_found:
            for kwarg in kwargs:
                self._update_task(
                    self.tasks_completed[-1]["task"][task_i], kwarg, kwargs[kwarg]
                )

    def set_current_task(self, task: str):
        self.current_task = task

    def get_errors(self):
        if self.current_task == "elysia_decision_node":
            return self.errors
        elif self.current_task is None or self.current_task not in self.errors:
            return []
        else:
            return self.errors[self.current_task]

    def clear_error(self, task: str):
        if task in self.errors:
            self.errors[task] = []

    def tasks_completed_string(self):
        """
        Output a nicely formatted string of the tasks completed so far, designed to be used in the LLM prompt.
        This is where the outputs of the `llm_message` fields are displayed.
        You can use this if you are interfacing with LLMs in tools, to help it understand the context of the tasks completed so far.

        Returns:
            (str): A separated and formatted string of the tasks completed so far in an LLM-parseable format.
        """
        out = ""
        for j, task_prompt in enumerate(self.tasks_completed):
            out += f"<prompt_{j+1}>\n"
            out += f"Prompt: {task_prompt['prompt']}\n"

            for i, task in enumerate(task_prompt["task"]):
                out += f"<task_{i+1}>\n"

                if "action" in task and task["action"]:
                    out += f"Chosen action: {task['task']} (this does not mean it has been completed, only that it was chosen) "
                    out += "(Use the environment to judge if a task is completed)\n"
                else:
                    out += f"Chosen subcategory: {task['task']} (this action has not been completed, this is only a subcategory)\n"

                for key in task:
                    if key != "task" and key != "action":
                        out += f"{key.capitalize()}: {task[key]}\n"

                out += f"</task_{i+1}>\n"
            out += f"</prompt_{j+1}>\n"

        return out

    async def set_collection_names(
        self, collection_names: list[str], client_manager: ClientManager
    ):
        self.collection_names = await self.collection_data.set_collection_names(
            collection_names, client_manager
        )
        return self.collection_names

    def tree_count_string(self):
        out = f"{self.num_trees_completed+1}/{self.recursion_limit}"
        if self.num_trees_completed == self.recursion_limit - 1:
            out += " (this is the last decision you can make before being cut off)"
        if self.num_trees_completed >= self.recursion_limit:
            out += " (recursion limit reached, write your full chat response accordingly - the decision process has been cut short, and it is likely the user's question has not been fully answered and you either haven't been able to do it or it was impossible)"
        return out

    def output_collection_metadata(
        self, collection_names: list[str] | None = None, with_mappings: bool = False
    ):
        """
        Outputs the full metadata for the given collection names.

        Args:
            with_mappings (bool): Whether to output the mappings for the collections as well as the other metadata.

        Returns:
            dict (dict[str, dict]): A dictionary of collection names to their metadata.
                The metadata are of the form:
                ```python
                {
                    # summary statistics of each field in the collection
                    "fields": list = [
                        field_name_1: dict = {
                            "description": str,
                            "range": list[float],
                            "type": str,
                            "groups": dict[str, str],
                            "mean": float
                        },
                        field_name_2: dict = {
                            ... # same fields as above
                        },
                        ...
                    ],

                    # mapping_1, mapping_2 etc refer to frontend-specific types that the AI has deemed appropriate for this data
                    # then the dict is to map the frontend fields to the data fields
                    "mappings": dict = {
                        mapping_1: dict = {
                            "frontend_field_1": "data_field_1",
                            "frontend_field_2": "data_field_2",
                            ...
                        },
                        mapping_2: dict = {
                            ... # same fields as above
                        },
                        ...,
                    },

                    # number of items in collection (float but just for consistency)
                    "length": float,

                    # AI generated summary of the dataset
                    "summary": str,

                    # name of collection
                    "name": str,

                    # what named vectors are available and their properties (if any)
                    "named_vectors": list = [
                        {
                            "name": str,
                            "enabled": bool,
                            "source_properties": list,
                            "description": str # defaults to empty
                        },
                        ...
                    ],

                    # some config settings relevant for queries
                    "index_properties": {
                        "isNullIndexed": bool,
                        "isLengthIndexed": bool,
                        "isTimestampIndexed": bool,
                    },
                }
                ```
                If `with_mappings` is `False`, then the mappings are not included.
                Each key in the outer level dictionary is a collection name.

        """

        if collection_names is None:
            collection_names = self.collection_names

        return self.collection_data.output_full_metadata(
            collection_names, with_mappings
        )

    def output_collection_return_types(self) -> dict[str, list[str]]:
        """
        Outputs the return types for the collections in the tree data.
        Essentially, this is a list of the keys that can be used to map the objects to the frontend.

        Returns:
            (dict): A dictionary of collection names to their return types.
                ```python
                {
                    collection_name_1: list[str],
                    collection_name_2: list[str],
                    ...,
                }
                ```
                Each of these lists is a list of the keys that can be used to map the objects to the frontend.
        """
        collection_return_types = self.collection_data.output_mapping_lists()
        out = {
            collection_name: collection_return_types[collection_name]
            for collection_name in self.collection_names
        }
        return out

    def to_json(self, remove_unserialisable: bool = False):
        out = {
            k: v
            for k, v in self.__dict__.items()
            if k not in ["collection_data", "atlas", "environment", "settings"]
        }
        out["collection_data"] = self.collection_data.to_json()
        out["atlas"] = self.atlas.model_dump()
        out["environment"] = self.environment.to_json(remove_unserialisable)
        out["settings"] = self.settings.to_json()
        return out

    @classmethod
    def from_json(cls, json_data: dict):
        settings = Settings.from_json(json_data["settings"])
        logger = settings.logger
        collection_data = CollectionData.from_json(json_data["collection_data"], logger)
        atlas = Atlas.model_validate(json_data["atlas"])
        environment = Environment.from_json(json_data["environment"])

        tree_data = cls(
            collection_data=collection_data,
            atlas=atlas,
            environment=environment,
            settings=settings,
        )
        for item in json_data:
            if item not in ["collection_data", "atlas", "environment", "settings"]:
                tree_data.set_property(item, json_data[item])
        return tree_data

output_collection_metadata(collection_names=None, with_mappings=False)

Outputs the full metadata for the given collection names.

Parameters:

Name	Type	Description	Default
with_mappings	bool	Whether to output the mappings for the collections as well as the other metadata.	False

Returns:

Name	Type	Description
dict	dict[str, dict]	A dictionary of collection names to their metadata. The metadata are of the form: { # summary statistics of each field in the collection "fields": list = [ field_name_1: dict = { "description": str, "range": list[float], "type": str, "groups": dict[str, str], "mean": float }, field_name_2: dict = { ... # same fields as above }, ... ], # mapping_1, mapping_2 etc refer to frontend-specific types that the AI has deemed appropriate for this data # then the dict is to map the frontend fields to the data fields "mappings": dict = { mapping_1: dict = { "frontend_field_1": "data_field_1", "frontend_field_2": "data_field_2", ... }, mapping_2: dict = { ... # same fields as above }, ..., }, # number of items in collection (float but just for consistency) "length": float, # AI generated summary of the dataset "summary": str, # name of collection "name": str, # what named vectors are available and their properties (if any) "named_vectors": list = [ { "name": str, "enabled": bool, "source_properties": list, "description": str # defaults to empty }, ... ], # some config settings relevant for queries "index_properties": { "isNullIndexed": bool, "isLengthIndexed": bool, "isTimestampIndexed": bool, }, } If with_mappings is False, then the mappings are not included. Each key in the outer level dictionary is a collection name.

Source code in elysia/tree/objects.py

def output_collection_metadata(
    self, collection_names: list[str] | None = None, with_mappings: bool = False
):
    """
    Outputs the full metadata for the given collection names.

    Args:
        with_mappings (bool): Whether to output the mappings for the collections as well as the other metadata.

    Returns:
        dict (dict[str, dict]): A dictionary of collection names to their metadata.
            The metadata are of the form:
            ```python
            {
                # summary statistics of each field in the collection
                "fields": list = [
                    field_name_1: dict = {
                        "description": str,
                        "range": list[float],
                        "type": str,
                        "groups": dict[str, str],
                        "mean": float
                    },
                    field_name_2: dict = {
                        ... # same fields as above
                    },
                    ...
                ],

                # mapping_1, mapping_2 etc refer to frontend-specific types that the AI has deemed appropriate for this data
                # then the dict is to map the frontend fields to the data fields
                "mappings": dict = {
                    mapping_1: dict = {
                        "frontend_field_1": "data_field_1",
                        "frontend_field_2": "data_field_2",
                        ...
                    },
                    mapping_2: dict = {
                        ... # same fields as above
                    },
                    ...,
                },

                # number of items in collection (float but just for consistency)
                "length": float,

                # AI generated summary of the dataset
                "summary": str,

                # name of collection
                "name": str,

                # what named vectors are available and their properties (if any)
                "named_vectors": list = [
                    {
                        "name": str,
                        "enabled": bool,
                        "source_properties": list,
                        "description": str # defaults to empty
                    },
                    ...
                ],

                # some config settings relevant for queries
                "index_properties": {
                    "isNullIndexed": bool,
                    "isLengthIndexed": bool,
                    "isTimestampIndexed": bool,
                },
            }
            ```
            If `with_mappings` is `False`, then the mappings are not included.
            Each key in the outer level dictionary is a collection name.

    """

    if collection_names is None:
        collection_names = self.collection_names

    return self.collection_data.output_full_metadata(
        collection_names, with_mappings
    )

output_collection_return_types()

Outputs the return types for the collections in the tree data. Essentially, this is a list of the keys that can be used to map the objects to the frontend.

Returns:

Type	Description
dict	A dictionary of collection names to their return types. { collection_name_1: list[str], collection_name_2: list[str], ..., } Each of these lists is a list of the keys that can be used to map the objects to the frontend.

Source code in elysia/tree/objects.py

def output_collection_return_types(self) -> dict[str, list[str]]:
    """
    Outputs the return types for the collections in the tree data.
    Essentially, this is a list of the keys that can be used to map the objects to the frontend.

    Returns:
        (dict): A dictionary of collection names to their return types.
            ```python
            {
                collection_name_1: list[str],
                collection_name_2: list[str],
                ...,
            }
            ```
            Each of these lists is a list of the keys that can be used to map the objects to the frontend.
    """
    collection_return_types = self.collection_data.output_mapping_lists()
    out = {
        collection_name: collection_return_types[collection_name]
        for collection_name in self.collection_names
    }
    return out

tasks_completed_string()

Output a nicely formatted string of the tasks completed so far, designed to be used in the LLM prompt. This is where the outputs of the llm_message fields are displayed. You can use this if you are interfacing with LLMs in tools, to help it understand the context of the tasks completed so far.

Returns:

Type	Description
str	A separated and formatted string of the tasks completed so far in an LLM-parseable format.

Source code in elysia/tree/objects.py

def tasks_completed_string(self):
    """
    Output a nicely formatted string of the tasks completed so far, designed to be used in the LLM prompt.
    This is where the outputs of the `llm_message` fields are displayed.
    You can use this if you are interfacing with LLMs in tools, to help it understand the context of the tasks completed so far.

    Returns:
        (str): A separated and formatted string of the tasks completed so far in an LLM-parseable format.
    """
    out = ""
    for j, task_prompt in enumerate(self.tasks_completed):
        out += f"<prompt_{j+1}>\n"
        out += f"Prompt: {task_prompt['prompt']}\n"

        for i, task in enumerate(task_prompt["task"]):
            out += f"<task_{i+1}>\n"

            if "action" in task and task["action"]:
                out += f"Chosen action: {task['task']} (this does not mean it has been completed, only that it was chosen) "
                out += "(Use the environment to judge if a task is completed)\n"
            else:
                out += f"Chosen subcategory: {task['task']} (this action has not been completed, this is only a subcategory)\n"

            for key in task:
                if key != "task" and key != "action":
                    out += f"{key.capitalize()}: {task[key]}\n"

            out += f"</task_{i+1}>\n"
        out += f"</prompt_{j+1}>\n"

    return out