Agents

This module contains the Agent class which can be either an AI architect that produces agents from scratch or an agent that is trained.

Agent

Bases: BaseModel

Agent class that represents either an AI architect or a trained agent.

Attributes:

Name	Type	Description
id	str	The unique identifier of the agent.
graph	Graph	The graph associated with the agent.
input_model	type[BaseModel | dict]	The input model of the agent.
output_model	type[BaseModel | dict]	The output model of the agent.

Source code in ebiose/agents/agent.py

class Agent(BaseModel):
    """Agent class that represents either an AI architect or a trained agent.

    Attributes:
            id: The unique identifier of the agent.
            graph: The graph associated with the agent.
            input_model: The input model of the agent.
            output_model: The output model of the agent.

    """

    id: str = Field(default_factory=lambda: "a-" + str(uuid4()))
    graph: Graph = Field(default=None)
    input_model: type[BaseModel | dict] = Field(default=None)
    output_model: type[BaseModel | dict] = Field(default=None)

An agent contains the graph Class which is, to date, the way Ebiose represents the workflow of an agent.

Bases: BaseModel

Graph used to represent the agent workflow on solving a problem.

The graph is formed by a series of nodes, each representing a distinct stage in the problem-solving process. The nodes are connected by edges, which signify the flow of information and decision-making within the graph.

Each node type has a specific role and function within the graph, contributing to the overall problem-solving capacity of the AI model.

Attributes:

Name	Type	Description
edges	list[Edge]	list of edges in the graph
nodes	list[NodeTypes]	list of nodes in the graph
description	str | None	Description of the graph purpose
shared_context_prompt	str	Prompt shared among all nodes in the graph

Source code in ebiose/agents/graph.py

class Graph(BaseModel):
    """Graph used to represent the agent workflow on solving a problem.

    The graph is formed by a series of nodes, each representing a distinct stage in the problem-solving
    process.
    The nodes are connected by edges, which signify the flow of information and decision-making
    within the graph.

    Each node type has a specific role and function within the graph, contributing to the overall
    problem-solving capacity of the AI model.


    Attributes:
            edges: list of edges in the graph
            nodes: list of nodes in the graph
            description: Description of the graph purpose
            shared_context_prompt: Prompt shared among all nodes in the graph

    """

    edges: list[Edge] = Field(
        default_factory=list,
        description="list of edges in the graph",
    )
    nodes: list[NodeTypes] = Field(
        default_factory=list,
        description="list of nodes in the graph",
    )

    description: str | None = ""
    shared_context_prompt: str = Field(min_length=1)

    @model_validator(mode="after")
    def validate_graph(self, info: ValidationInfo) -> Graph:
        """Validate the graph by checking placeholders and outgoing conditional edges."""
        msg = ""

        # Validate placeholders
        msg += self.__validate_placeholders(info)

        # Validate outgoing conditional edges
        msg += self.__validate_outgoing_conditional_edges()

        if msg != "":
            raise ValueError(msg)

        return self

    def __validate_placeholders(self, info: ValidationInfo) -> str:
        """Check if all and only the authorized placeholders in the Agentinput are in the graph."""
        if info.context is None:
            return ""
        msg = ""

        if (
            "input" in info.context
            and "authorized_placeholders" in info.context["input"]
        ):
            authorized_placeholders = set(
                info.context["input"]["authorized_placeholders"],
            )
            all_prompts = self.shared_context_prompt
            for node in self.nodes:
                if isinstance(node, LLMNode):
                    all_prompts += node.prompt
            used_placeholders = set(find_placeholders(all_prompts))
            # check for unauthorized placeholders
            unauthorized_placeholders = used_placeholders - authorized_placeholders
            if unauthorized_placeholders:
                unauthorized_placeholders_list = [
                    "{" + f"{ph}" + "}" for ph in unauthorized_placeholders
                ]

                formatted_placeholders = (
                    ", ".join(unauthorized_placeholders_list[:-1])
                    + f" and {unauthorized_placeholders_list[-1]}"
                    if len(unauthorized_placeholders_list) > 1
                    else unauthorized_placeholders_list[0]
                )
                msg += f"Unauthorized placeholders found: {formatted_placeholders}. "
                msg += f"Please remove {formatted_placeholders} from the prompts. "

            # check for missing placeholders
            missing_placeholders = authorized_placeholders - used_placeholders
            if missing_placeholders:
                missing_placeholders_list = [
                    "{" + f"{ph}" + "}" for ph in missing_placeholders
                ]
                formatted_placeholders = (
                    ", ".join(missing_placeholders_list[:-1])
                    + f" and {missing_placeholders_list[-1]}"
                    if len(missing_placeholders_list) > 1
                    else missing_placeholders_list[0]
                )
                msg += f"Missing placeholders: {formatted_placeholders}. "
                msg += f"Please add {formatted_placeholders} to the prompts."
        return msg

    def __validate_outgoing_conditional_edges(self) -> str:
        """Check for each node that ahs outgoing conditional edges, if it has at least two conditional edges."""
        nodes_with_errors = []
        for node in self.nodes:
            # Count the number of condtionnal outgoing edges and check if number is less than 2
            conditional_outgoing_edges = [
                edge
                for edge in self.edges
                if edge.start_node_id == node.id and edge.is_conditional()
            ]
            if len(conditional_outgoing_edges) == 1:
                nodes_with_errors.append(node.id)

        if nodes_with_errors:
            nodes_str = ", ".join(nodes_with_errors)
            return f"The following nodes have just one conditional outgoing edge: {nodes_str}. Consider adding a second conditional edge or removing the condition on the existing outgoing edge."

        return ""

    @field_validator("shared_context_prompt", mode="before")
    @classmethod
    def validate_shared_context_prompt(cls, shared_context_prompt: str) -> str:
        """Make sure that the shared context prompt is a string."""
        if not isinstance(shared_context_prompt, str):
            msg = "Prompt should be a string"
            raise TypeError(msg)

        return shared_context_prompt

    @field_validator("nodes", mode="before")
    @classmethod
    def validate_nodes(cls, nodes: list[dict]) -> list[NodeTypes]:
        """Validate the nodes in the graph and generate explicit errors for retries."""
        if not isinstance(nodes, list):
            msg = "Nodes should be a list"
            raise TypeError(msg)

        if isinstance(nodes[0], BaseModel):
            return nodes

        errors, validated_nodes = Graph.__validate_nodes(nodes)
        if errors:
            raise ValidationError.from_exception_data(
                title=cls.__name__,
                line_errors=errors,
            )

        return validated_nodes

    @classmethod
    def __validate_nodes(cls, nodes: list[dict]) -> tuple[list, list] | None:
        validated_nodes = []
        errors = []

        for index, node in enumerate(nodes):
            if not isinstance(node, dict):
                errors.append(
                    {
                        "loc": (index),
                        "type": "value_error",
                        "input": node,
                        "ctx": {
                            "error": ValueError(
                                f"Type of node should be a dict, got {type(node)}",
                            ),
                        },
                    },
                )

            if "id" in node and isinstance(node["id"], int):
                node["id"] = str(node["id"])

            node_type = node.get("type", None)

            if node_type is None:
                # missing node type
                errors.append(
                    {
                        "loc": (index, "type"),
                        "msg": "Field required",
                        "type": "missing",
                        "input": node,
                    },
                )
                continue

            if node_type not in node_types_map:
                # invalid node type
                errors.append(
                    {
                        "loc": (index, "type"),
                        "type": "value_error",
                        "input": node,
                        "ctx": {"error": ValueError(f"Invalid node type: {node_type}")},
                    },
                )
                continue

            try:
                validated_nodes.append(
                    node_types_map[node_type].model_validate(node),
                )
            except ValidationError as e:
                for error in e.errors():
                    error["loc"] = (index,) + error["loc"]
                    errors.append(error)

        return errors, validated_nodes

    def add_edge(self: Self, edge: Edge) -> None:
        """Add an edge to the graph.

        Args:
                edge: An instance of the Edge class
        """
        self.edges.append(edge)

    def add_node(self: Self, node: BaseNode) -> None:
        """Add a node to the graph.

        Args:
                node: An instance of the Node class
        """
        if (
            node not in self.nodes
        ):  # TODO(ISSUE) add the node in ascending order in respect to id and edit get_node to have a binary search O(log(n))
            self.nodes.append(node)

    def get_node(self: Self, node_id: str) -> BaseNode:
        """Get a node from the graph.

        Args:
                node_id: The id of the node

        Returns:
                The node with the given id
        """
        for node in self.nodes:
            if node.id == node_id:
                return node
        msg = f"Node with id {node_id} not found in the graph"
        raise ValueError(msg)

    def get_last_node_ids(self: Self) -> list[BaseNode]:
        """Get the ids of the last nodes in the graph.

        Returns:
                A list of ids of the nodes that have the EndNode as outgoing edges
        """
        return [
            edge.start_node_id
            for edge in self.edges
            if edge.end_node_id == self.get_end_node_id()
        ]

    def get_outgoing_nodes(
        self: Self,
        node_id: str,
        conditional: bool | None = None,
    ) -> list[BaseNode]:
        """Get the outgoing nodes of a node. By default, return all nodes connected to the node.

        Args:
                node_id: The id of the node
                conditional: If True, only return nodes that are connected by a conditional edge.If False, only return nodes that are connected by a non-conditional edge. If None, return all nodes.

        Returns:
                A list of nodes that are connected to the node
        """
        if conditional is None:
            return [
                self.get_node(edge.end_node_id)
                for edge in self.edges
                if edge.start_node_id == node_id
            ]
        if conditional:
            return [
                self.get_node(edge.end_node_id)
                for edge in self.edges
                if edge.start_node_id == node_id and edge.is_conditional()
            ]

        return [
            self.get_node(edge.end_node_id)
            for edge in self.edges
            if edge.start_node_id == node_id and not edge.is_conditional()
        ]

    def get_end_node_id(self) -> str:
        """Get the id of the end node in the graph.

        Returns: The id of the end node
        """
        for node in self.nodes:
            if isinstance(node, EndNode):
                return node.id
        msg = "End node not found in the graph"
        raise ValueError(msg)

    def get_outgoing_edges(
        self: Self,
        node_id: str,
        conditional: bool | None = None,
    ) -> list[Edge]:
        """Get the outgoing edges of a node. By default, return all edges connected to the node.

        node_id: The id of the node
        conditional: If True, only return edges that are conditional. If False, only return edges that are non-conditional. If None, return all edges.

        Returns: A list of edges that are connected to the node
        """
        if conditional is None:
            return [edge for edge in self.edges if edge.start_node_id == node_id]
        if conditional:
            return [
                edge
                for edge in self.edges
                if edge.start_node_id == node_id and edge.is_conditional()
            ]

        return [
            edge
            for edge in self.edges
            if edge.start_node_id == node_id and not edge.is_conditional()
        ]

    def to_mermaid_str(
        self: Self,
        orientation: Literal["TB", "TR", "TD", "RL", "LR"] = "TD",
    ) -> LiteralString | str:
        """Generate a mermaid string representation of the graph.

        Note:
                                The orientation of the graph can be set to one of the following values:

                                - **TB**: Top to bottom
                                - **TR**: Top to right
                                - **TD**: Top to down
                                - **RL**: Right to left
                                - **LR**: Left to right

        Args:
                                orientation: The orientation of the graph.
        """
        node_type_display_name = {
            "LLMNode": "({node_name})",
            "StartNode": "[{node_name}]",
            "EndNode": "[{node_name}]",
        }
        mermaid_str = f"graph {orientation}\n"

        for edge in self.edges:
            start_node = self.get_node(edge.start_node_id)
            end_node = self.get_node(edge.end_node_id)

            start_node_name = getattr(start_node, "name", start_node.id)
            end_node_name = getattr(end_node, "name", end_node.id)

            start_node_id = edge.start_node_id.replace(" ", "").title()
            end_node_id = edge.end_node_id.replace(" ", "").title()

            # Determine the appropriate brackets for the node type
            start_node_block = node_type_display_name.get(
                start_node.type,
                "[/{node_name}/]",
            ).format(node_name=start_node_name)
            end_node_block = node_type_display_name.get(
                end_node.type,
                "[/{node_name}/]",
            ).format(node_name=end_node_name)

            if edge.is_conditional():
                # replace the [ with #91; and ] with #93; to avoid mermaid syntax error
                condition = edge.condition.replace("[", "#91;").replace("]", "#93;")
                mermaid_str += f"\t{start_node_id}{start_node_block} -->|{condition}| {end_node_id}{end_node_block}\n"
            else:
                mermaid_str += f"\t{start_node_id}{start_node_block} --> {end_node_id}{end_node_block}\n"

        return mermaid_str

__validate_outgoing_conditional_edges()

Check for each node that ahs outgoing conditional edges, if it has at least two conditional edges.

Source code in ebiose/agents/graph.py

def __validate_outgoing_conditional_edges(self) -> str:
    """Check for each node that ahs outgoing conditional edges, if it has at least two conditional edges."""
    nodes_with_errors = []
    for node in self.nodes:
        # Count the number of condtionnal outgoing edges and check if number is less than 2
        conditional_outgoing_edges = [
            edge
            for edge in self.edges
            if edge.start_node_id == node.id and edge.is_conditional()
        ]
        if len(conditional_outgoing_edges) == 1:
            nodes_with_errors.append(node.id)

    if nodes_with_errors:
        nodes_str = ", ".join(nodes_with_errors)
        return f"The following nodes have just one conditional outgoing edge: {nodes_str}. Consider adding a second conditional edge or removing the condition on the existing outgoing edge."

    return ""

__validate_placeholders(info)

Check if all and only the authorized placeholders in the Agentinput are in the graph.

Source code in ebiose/agents/graph.py

def __validate_placeholders(self, info: ValidationInfo) -> str:
    """Check if all and only the authorized placeholders in the Agentinput are in the graph."""
    if info.context is None:
        return ""
    msg = ""

    if (
        "input" in info.context
        and "authorized_placeholders" in info.context["input"]
    ):
        authorized_placeholders = set(
            info.context["input"]["authorized_placeholders"],
        )
        all_prompts = self.shared_context_prompt
        for node in self.nodes:
            if isinstance(node, LLMNode):
                all_prompts += node.prompt
        used_placeholders = set(find_placeholders(all_prompts))
        # check for unauthorized placeholders
        unauthorized_placeholders = used_placeholders - authorized_placeholders
        if unauthorized_placeholders:
            unauthorized_placeholders_list = [
                "{" + f"{ph}" + "}" for ph in unauthorized_placeholders
            ]

            formatted_placeholders = (
                ", ".join(unauthorized_placeholders_list[:-1])
                + f" and {unauthorized_placeholders_list[-1]}"
                if len(unauthorized_placeholders_list) > 1
                else unauthorized_placeholders_list[0]
            )
            msg += f"Unauthorized placeholders found: {formatted_placeholders}. "
            msg += f"Please remove {formatted_placeholders} from the prompts. "

        # check for missing placeholders
        missing_placeholders = authorized_placeholders - used_placeholders
        if missing_placeholders:
            missing_placeholders_list = [
                "{" + f"{ph}" + "}" for ph in missing_placeholders
            ]
            formatted_placeholders = (
                ", ".join(missing_placeholders_list[:-1])
                + f" and {missing_placeholders_list[-1]}"
                if len(missing_placeholders_list) > 1
                else missing_placeholders_list[0]
            )
            msg += f"Missing placeholders: {formatted_placeholders}. "
            msg += f"Please add {formatted_placeholders} to the prompts."
    return msg

add_edge(edge)

Add an edge to the graph.

Parameters:

Name	Type	Description	Default
edge	Edge	An instance of the Edge class	required

Source code in ebiose/agents/graph.py

def add_edge(self: Self, edge: Edge) -> None:
    """Add an edge to the graph.

    Args:
            edge: An instance of the Edge class
    """
    self.edges.append(edge)

add_node(node)

Add a node to the graph.

Parameters:

Name	Type	Description	Default
node	BaseNode	An instance of the Node class	required

Source code in ebiose/agents/graph.py

def add_node(self: Self, node: BaseNode) -> None:
    """Add a node to the graph.

    Args:
            node: An instance of the Node class
    """
    if (
        node not in self.nodes
    ):  # TODO(ISSUE) add the node in ascending order in respect to id and edit get_node to have a binary search O(log(n))
        self.nodes.append(node)

get_end_node_id()

Get the id of the end node in the graph.

Returns: The id of the end node

Source code in ebiose/agents/graph.py

def get_end_node_id(self) -> str:
    """Get the id of the end node in the graph.

    Returns: The id of the end node
    """
    for node in self.nodes:
        if isinstance(node, EndNode):
            return node.id
    msg = "End node not found in the graph"
    raise ValueError(msg)

get_last_node_ids()

Get the ids of the last nodes in the graph.

Returns:

Type	Description
list[BaseNode]	A list of ids of the nodes that have the EndNode as outgoing edges

Source code in ebiose/agents/graph.py

def get_last_node_ids(self: Self) -> list[BaseNode]:
    """Get the ids of the last nodes in the graph.

    Returns:
            A list of ids of the nodes that have the EndNode as outgoing edges
    """
    return [
        edge.start_node_id
        for edge in self.edges
        if edge.end_node_id == self.get_end_node_id()
    ]

get_node(node_id)

Get a node from the graph.

Parameters:

Name	Type	Description	Default
node_id	str	The id of the node	required

Returns:

Type	Description
BaseNode	The node with the given id

Source code in ebiose/agents/graph.py

def get_node(self: Self, node_id: str) -> BaseNode:
    """Get a node from the graph.

    Args:
            node_id: The id of the node

    Returns:
            The node with the given id
    """
    for node in self.nodes:
        if node.id == node_id:
            return node
    msg = f"Node with id {node_id} not found in the graph"
    raise ValueError(msg)

get_outgoing_edges(node_id, conditional=None)

Get the outgoing edges of a node. By default, return all edges connected to the node.

node_id: The id of the node conditional: If True, only return edges that are conditional. If False, only return edges that are non-conditional. If None, return all edges.

Returns: A list of edges that are connected to the node

Source code in ebiose/agents/graph.py

def get_outgoing_edges(
    self: Self,
    node_id: str,
    conditional: bool | None = None,
) -> list[Edge]:
    """Get the outgoing edges of a node. By default, return all edges connected to the node.

    node_id: The id of the node
    conditional: If True, only return edges that are conditional. If False, only return edges that are non-conditional. If None, return all edges.

    Returns: A list of edges that are connected to the node
    """
    if conditional is None:
        return [edge for edge in self.edges if edge.start_node_id == node_id]
    if conditional:
        return [
            edge
            for edge in self.edges
            if edge.start_node_id == node_id and edge.is_conditional()
        ]

    return [
        edge
        for edge in self.edges
        if edge.start_node_id == node_id and not edge.is_conditional()
    ]

get_outgoing_nodes(node_id, conditional=None)

Get the outgoing nodes of a node. By default, return all nodes connected to the node.

Parameters:

Name	Type	Description	Default
node_id	str	The id of the node	required
conditional	bool | None	If True, only return nodes that are connected by a conditional edge.If False, only return nodes that are connected by a non-conditional edge. If None, return all nodes.	None

Returns:

Type	Description
list[BaseNode]	A list of nodes that are connected to the node

Source code in ebiose/agents/graph.py

def get_outgoing_nodes(
    self: Self,
    node_id: str,
    conditional: bool | None = None,
) -> list[BaseNode]:
    """Get the outgoing nodes of a node. By default, return all nodes connected to the node.

    Args:
            node_id: The id of the node
            conditional: If True, only return nodes that are connected by a conditional edge.If False, only return nodes that are connected by a non-conditional edge. If None, return all nodes.

    Returns:
            A list of nodes that are connected to the node
    """
    if conditional is None:
        return [
            self.get_node(edge.end_node_id)
            for edge in self.edges
            if edge.start_node_id == node_id
        ]
    if conditional:
        return [
            self.get_node(edge.end_node_id)
            for edge in self.edges
            if edge.start_node_id == node_id and edge.is_conditional()
        ]

    return [
        self.get_node(edge.end_node_id)
        for edge in self.edges
        if edge.start_node_id == node_id and not edge.is_conditional()
    ]

to_mermaid_str(orientation='TD')

Generate a mermaid string representation of the graph.

Note

The orientation of the graph can be set to one of the following values:

• TB: Top to bottom
• TR: Top to right
• TD: Top to down
• RL: Right to left
• LR: Left to right

Parameters:

Name	Type	Description	Default
orientation	Literal['TB', 'TR', 'TD', 'RL', 'LR']	The orientation of the graph.	'TD'

Source code in ebiose/agents/graph.py

def to_mermaid_str(
    self: Self,
    orientation: Literal["TB", "TR", "TD", "RL", "LR"] = "TD",
) -> LiteralString | str:
    """Generate a mermaid string representation of the graph.

    Note:
                            The orientation of the graph can be set to one of the following values:

                            - **TB**: Top to bottom
                            - **TR**: Top to right
                            - **TD**: Top to down
                            - **RL**: Right to left
                            - **LR**: Left to right

    Args:
                            orientation: The orientation of the graph.
    """
    node_type_display_name = {
        "LLMNode": "({node_name})",
        "StartNode": "[{node_name}]",
        "EndNode": "[{node_name}]",
    }
    mermaid_str = f"graph {orientation}\n"

    for edge in self.edges:
        start_node = self.get_node(edge.start_node_id)
        end_node = self.get_node(edge.end_node_id)

        start_node_name = getattr(start_node, "name", start_node.id)
        end_node_name = getattr(end_node, "name", end_node.id)

        start_node_id = edge.start_node_id.replace(" ", "").title()
        end_node_id = edge.end_node_id.replace(" ", "").title()

        # Determine the appropriate brackets for the node type
        start_node_block = node_type_display_name.get(
            start_node.type,
            "[/{node_name}/]",
        ).format(node_name=start_node_name)
        end_node_block = node_type_display_name.get(
            end_node.type,
            "[/{node_name}/]",
        ).format(node_name=end_node_name)

        if edge.is_conditional():
            # replace the [ with #91; and ] with #93; to avoid mermaid syntax error
            condition = edge.condition.replace("[", "#91;").replace("]", "#93;")
            mermaid_str += f"\t{start_node_id}{start_node_block} -->|{condition}| {end_node_id}{end_node_block}\n"
        else:
            mermaid_str += f"\t{start_node_id}{start_node_block} --> {end_node_id}{end_node_block}\n"

    return mermaid_str

validate_graph(info)

Validate the graph by checking placeholders and outgoing conditional edges.

Source code in ebiose/agents/graph.py

@model_validator(mode="after")
def validate_graph(self, info: ValidationInfo) -> Graph:
    """Validate the graph by checking placeholders and outgoing conditional edges."""
    msg = ""

    # Validate placeholders
    msg += self.__validate_placeholders(info)

    # Validate outgoing conditional edges
    msg += self.__validate_outgoing_conditional_edges()

    if msg != "":
        raise ValueError(msg)

    return self

validate_nodes(nodes) classmethod

Validate the nodes in the graph and generate explicit errors for retries.

Source code in ebiose/agents/graph.py

@field_validator("nodes", mode="before")
@classmethod
def validate_nodes(cls, nodes: list[dict]) -> list[NodeTypes]:
    """Validate the nodes in the graph and generate explicit errors for retries."""
    if not isinstance(nodes, list):
        msg = "Nodes should be a list"
        raise TypeError(msg)

    if isinstance(nodes[0], BaseModel):
        return nodes

    errors, validated_nodes = Graph.__validate_nodes(nodes)
    if errors:
        raise ValidationError.from_exception_data(
            title=cls.__name__,
            line_errors=errors,
        )

    return validated_nodes

validate_shared_context_prompt(shared_context_prompt) classmethod

Make sure that the shared context prompt is a string.

Source code in ebiose/agents/graph.py

@field_validator("shared_context_prompt", mode="before")
@classmethod
def validate_shared_context_prompt(cls, shared_context_prompt: str) -> str:
    """Make sure that the shared context prompt is a string."""
    if not isinstance(shared_context_prompt, str):
        msg = "Prompt should be a string"
        raise TypeError(msg)

    return shared_context_prompt