invoke-ai/InvokeAI · error · InvalidEdgeError

Edge already exists ( )

Error message

Edge already exists ({edge})

What it means

InvalidEdgeError raised by _validate_edge_destination_uniqueness() when a destination input field already has an incoming edge. Most input fields accept exactly one edge; only CollectInvocation's 'item' field may receive many. This prevents silently overwriting or duplicating a node's input.

Solutions

  1. Remove the existing edge into that field before adding the new one
  2. If you need to merge multiple outputs, route them through a CollectInvocation item field
  3. Make graph construction idempotent (check for the edge before adding)
  4. In the UI, delete the old connection first when rewiring

Example fix

// before
g.add_edge(a, "image", denoise, "image")
g.add_edge(b, "image", denoise, "image")  # duplicate destination
// after
g.add_edge(a, "image", denoise, "image")
g.delete_edge(a, "image", denoise, "image")
g.add_edge(b, "image", denoise, "image")
Defensive patterns

Strategy: validation

Validate before calling

def has_input_edge(graph, node_id: str, field: str) -> bool:
    return any(
        e.destination.node_id == node_id and e.destination.field == field
        for e in graph.edges
    )

if not has_input_edge(g, dst_id, field):
    g.add_edge(src_id, src_field, dst_id, field)

Type guard

def destination_is_free(graph, edge) -> bool:
    return not any(
        e.destination.node_id == edge.destination.node_id
        and e.destination.field == edge.destination.field
        for e in graph.edges
    )

Try / catch

from invokeai.app.services.shared.graph import InvalidEdgeError

try:
    g.add_edge(src, "image", dst, "image")
except InvalidEdgeError as e:
    if "already exists" in str(e):
        g.delete_edge(src, "image", dst, "image")
        g.add_edge(src, "image", dst, "image")  # replace

Prevention

When it happens

Trigger: Calling g.add_edge() connecting a second edge into the same (node_id, field) destination that is not a CollectInvocation item field.

Common situations: Re-running graph-building scripts that add edges twice, wiring two outputs into a single non-collect input, UI/script mix-ups after editing an existing workflow.

Related errors


AI-assisted analysis of invoke-ai/InvokeAI@0b6a024f2f (2026-08-29). Data as JSON: /api/errors/132ae43f16fe9d4b. Report an issue: GitHub.

Appendix: source

Thrown at invokeai/app/services/shared/graph.py:1924

        """Checks if the destination field for an edge is of type typing.Any"""
        return get_input_field_type(self.get_node(edge.destination.node_id), edge.destination.field) == Any

    def _is_destination_field_list_of_Any(self, edge: Edge) -> bool:
        """Checks if the destination field for an edge is of type typing.Any"""
        return get_input_field_type(self.get_node(edge.destination.node_id), edge.destination.field) == list[Any]

    def _get_edge_nodes(self, edge: Edge) -> tuple[BaseInvocation, BaseInvocation]:
        try:
            return self.get_node(edge.source.node_id), self.get_node(edge.destination.node_id)
        except NodeNotFoundError:
            raise InvalidEdgeError(f"One or both nodes don't exist ({edge})")

    def _validate_edge_destination_uniqueness(self, edge: Edge, destination_node: BaseInvocation) -> None:
        input_edges = self._get_input_edges(edge.destination.node_id, edge.destination.field)
        if len(input_edges) > 0 and (
            not isinstance(destination_node, CollectInvocation) or edge.destination.field != ITEM_FIELD
        ):
            raise InvalidEdgeError(f"Edge already exists ({edge})")

    def _validate_edge_would_not_create_cycle(self, edge: Edge) -> None:
        graph = self.nx_graph_flat()
        graph.add_edge(edge.source.node_id, edge.destination.node_id)
        if not nx.is_directed_acyclic_graph(graph):
            raise InvalidEdgeError(f"Edge creates a cycle in the graph ({edge})")

    def _validate_edge_field_compatibility(
        self, edge: Edge, source_node: BaseInvocation, destination_node: BaseInvocation
    ) -> None:
        if isinstance(destination_node, CallSavedWorkflowInvocation) and is_call_saved_workflow_dynamic_input(
            edge.destination.field
        ):
            return
        if not are_connections_compatible(source_node, edge.source.field, destination_node, edge.destination.field):
            raise InvalidEdgeError(f"Field types are incompatible ({edge})")

    def _validate_iterator_edge_rules(

View on GitHub (pinned to 0b6a024f2f)