jyapayne/langflow
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

refactor: add graph utility tests and refactor sorting methods (#5538)

Browse source 
* refactor: turn sorting methods into functions in a separate module

- Added `layered_topological_sort` function to perform layered topological sorting of graph vertices, accommodating cycles and input vertex checks.
- Introduced `refine_layers` function to ensure proper dependency ordering among vertices.
- Implemented helper functions for sorting layers by dependency and filtering vertices based on predecessors.
- Enhanced utility functions to support better graph traversal and layer management.

This update improves the graph processing capabilities, allowing for more efficient handling of complex graph structures.

* feat(tests): enhance graph utility tests with cycle detection and sorting functionality

- Added a new fixture `graph_with_loop` to simulate a graph containing cycles for testing purposes.
- Improved the `test_large_graph_efficiency` to validate cycle detection in large graphs.
- Introduced multiple tests for sorting vertices in graphs with cycles, ensuring correct order and handling of input vertices.
- Enhanced assertions to provide clearer error messages for failed tests, improving debugging experience.

These changes strengthen the testing framework for graph utilities, ensuring robust handling of complex graph structures.

* refactor(graph): remove unused parent_node_map from Graph class initialization

- Eliminated the `parent_node_map` parameter from the Graph class constructor, streamlining the graph initialization process.
- This change enhances code clarity and reduces unnecessary complexity in graph management.

This update contributes to cleaner and more maintainable graph-related code.

* refactor(graph): optimize dependency sorting and vertex filtering

- Improved the `_max_dependency_index` function by utilizing `index_map.get()` for cleaner code and better handling of missing successors.
- Enhanced the `_sort_single_layer_by_dependency` function with a caching mechanism to avoid redundant calculations, improving performance during vertex sorting.
- Updated `filter_vertices_up_to_vertex` to use a set for `vertices_ids`, optimizing membership checks and enhancing efficiency in vertex filtering.

These changes contribute to more efficient graph processing and improved code readability.

* chore: remove unused 'parent_node_map' parameter

* [autofix.ci] apply automated fixes

* fix: replace old method call with a new func

* test: enhance assertions for file existence in webhook tests

* refactor(graph): enhance component ID retrieval and chat input sorting

- Updated `find_start_component_id` to accept an optional `is_webhook` parameter, allowing for dynamic priority input selection based on the flow type.
- Improved `sort_chat_inputs_first` to handle chat input positioning more efficiently, ensuring only one chat input exists and adjusting its position within the layers as needed.
- These changes enhance the flexibility and efficiency of graph processing, particularly for webhook flows.

* test(graph): update assertions in sort_chat_inputs_first test for accuracy

- Modified assertions in the `test_chat_inputs_at_start` function to reflect the correct expected output of the `sort_chat_inputs_first` utility.
- Adjusted the expected length and order of the result to ensure accurate validation of chat input sorting functionality.

These changes enhance the reliability of the test suite for graph utilities, ensuring that the sorting logic is correctly validated.

* test(chat): update assertion in consume_and_assert_stream for accurate ID validation

- Modified the assertion in the `consume_and_assert_stream` function to include an additional expected ID, ensuring the test accurately reflects the current output of the chat endpoint.
- This change enhances the reliability of the test suite by validating the correct behavior of the chat input sorting functionality.

* test(endpoints): update assertion in test_get_vertices for accurate ID validation

- Modified the assertion in the `test_get_vertices` function to include an additional expected ID, "Webhook", alongside "ChatInput".
- This change ensures the test accurately reflects the current output of the endpoint, enhancing the reliability of the test suite for endpoint functionality.

---------

Co-authored-by: italojohnny <italojohnnydosanjos@gmail.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This commit is contained in:
Gabriel Luiz Freitas Almeida 2025-01-09 17:15:24 -03:00 • committed by GitHub
commit 9c23759c7d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 1940 additions and 946 deletions
Download patch file Download diff file Expand all files Collapse all files

231
src/backend/base/langflow/graph/graph/base.py
View file

@ -26,9 +26,9 @@ from langflow.graph.graph.utils import (
find_all_cycle_edges,
find_cycle_vertices,
find_start_component_id,
get_sorted_vertices,
process_flow,
should_continue,
sort_up_to_vertex,
)
from langflow.graph.schema import InterfaceComponentTypes, RunOutputs
from langflow.graph.vertex.base import Vertex, VertexStates
@ -1872,168 +1872,25 @@ class Graph:
f"{edges_repr}"
)
def layered_topological_sort(
self,
vertices: list[Vertex],
*,
filter_graphs: bool = False,
) -> list[list[str]]:
"""Performs a layered topological sort of the vertices in the graph."""
vertices_ids = {vertex.id for vertex in vertices}
# Queue for vertices with no incoming edges
in_degree_map = self.in_degree_map.copy()
if self.is_cyclic and all(in_degree_map.values()):
# This means we have a cycle because all vertex have in_degree_map > 0
# because of this we set the queue to start on the ._start if it exists
if self._start is not None:
queue = deque([self._start._id])
else:
# Find the chat input component
chat_input = find_start_component_id(vertices_ids)
if chat_input is None:
msg = "No input component found and no start component provided"
raise ValueError(msg)
queue = deque([chat_input])
else:
queue = deque(
vertex.id
for vertex in vertices
# if filter_graphs then only vertex.is_input will be considered
if in_degree_map[vertex.id] == 0 and (not filter_graphs or vertex.is_input)
)
layers: list[list[str]] = []
visited = set(queue)
def get_vertex_predecessors_ids(self, vertex_id: str) -> list[str]:
"""Get the predecessor IDs of a vertex."""
return [v.id for v in self.get_predecessors(self.get_vertex(vertex_id))]
current_layer = 0
while queue:
layers.append([]) # Start a new layer
layer_size = len(queue)
for _ in range(layer_size):
vertex_id = queue.popleft()
visited.add(vertex_id)
def get_vertex_successors_ids(self, vertex_id: str) -> list[str]:
"""Get the successor IDs of a vertex."""
return [v.id for v in self.get_vertex(vertex_id).successors]
layers[current_layer].append(vertex_id)
for neighbor in self.successor_map[vertex_id]:
# only vertices in `vertices_ids` should be considered
# because vertices by have been filtered out
# in a previous step. All dependencies of theirs
# will be built automatically if required
if neighbor not in vertices_ids:
continue
def get_vertex_input_status(self, vertex_id: str) -> bool:
"""Check if a vertex is an input vertex."""
return self.get_vertex(vertex_id).is_input
in_degree_map[neighbor] -= 1 # 'remove' edge
if in_degree_map[neighbor] == 0 and neighbor not in visited:
queue.append(neighbor)
def get_parent_map(self) -> dict[str, str | None]:
"""Get the parent node map for all vertices."""
return {vertex.id: vertex.parent_node_id for vertex in self.vertices}
# if > 0 it might mean not all predecessors have added to the queue
# so we should process the neighbors predecessors
elif in_degree_map[neighbor] > 0:
for predecessor in self.predecessor_map[neighbor]:
if predecessor not in queue and predecessor not in visited:
queue.append(predecessor)
current_layer += 1 # Next layer
return self.refine_layers(layers)
def refine_layers(self, initial_layers):
# Map each vertex to its current layer
vertex_to_layer = {}
for layer_index, layer in enumerate(initial_layers):
for vertex in layer:
vertex_to_layer[vertex] = layer_index
# Build the adjacency list for reverse lookup (dependencies)
refined_layers = [[] for _ in initial_layers] # Start with empty layers
new_layer_index_map = defaultdict(int)
# Map each vertex to its new layer index
# by finding the lowest layer index of its dependencies
# and subtracting 1
# If a vertex has no dependencies, it will be placed in the first layer
# If a vertex has dependencies, it will be placed in the lowest layer index of its dependencies
# minus 1
for vertex_id, deps in self.successor_map.items():
indexes = [vertex_to_layer[dep] for dep in deps if dep in vertex_to_layer]
new_layer_index = max(min(indexes, default=0) - 1, 0)
new_layer_index_map[vertex_id] = new_layer_index
for layer_index, layer in enumerate(initial_layers):
for vertex_id in layer:
# Place the vertex in the highest possible layer where its dependencies are met
new_layer_index = new_layer_index_map[vertex_id]
if new_layer_index > layer_index:
refined_layers[new_layer_index].append(vertex_id)
vertex_to_layer[vertex_id] = new_layer_index
else:
refined_layers[layer_index].append(vertex_id)
# Remove empty layers if any
return [layer for layer in refined_layers if layer]
def sort_chat_inputs_first(self, vertices_layers: list[list[str]]) -> list[list[str]]:
chat_inputs = []
for layer in vertices_layers:
for vertex_id in layer:
if "ChatInput" in vertex_id and self.get_predecessors(self.get_vertex(vertex_id)):
return vertices_layers
if "ChatInput" in vertex_id:
chat_inputs.append(vertex_id)
if not chat_inputs:
return vertices_layers
chat_inputs_first = []
for layer in vertices_layers:
layer_chat_inputs_first = [vertex_id for vertex_id in layer if "ChatInput" in vertex_id]
chat_inputs_first.extend(layer_chat_inputs_first)
layer[:] = [v for v in layer if v not in layer_chat_inputs_first]
return [chat_inputs_first, *vertices_layers]
def sort_layer_by_dependency(self, vertices_layers: list[list[str]]) -> list[list[str]]:
"""Sorts the vertices in each layer by dependency, ensuring no vertex depends on a subsequent vertex."""
sorted_layers = []
for layer in vertices_layers:
sorted_layer = self._sort_single_layer_by_dependency(layer)
sorted_layers.append(sorted_layer)
return sorted_layers
def _sort_single_layer_by_dependency(self, layer: list[str]) -> list[str]:
"""Sorts a single layer by dependency using a stable sorting method."""
# Build a map of each vertex to its index in the layer for quick lookup.
index_map = {vertex: index for index, vertex in enumerate(layer)}
# Create a sorted copy of the layer based on dependency order.
return sorted(layer, key=lambda vertex: self._max_dependency_index(vertex, index_map), reverse=True)
def _max_dependency_index(self, vertex_id: str, index_map: dict[str, int]) -> int:
"""Finds the highest index a given vertex's dependencies occupy in the same layer."""
vertex = self.get_vertex(vertex_id)
max_index = -1
for successor in vertex.successors: # Assuming vertex.successors is a list of successor vertex identifiers.
if successor.id in index_map:
max_index = max(max_index, index_map[successor.id])
return max_index
def __to_dict(self) -> dict[str, dict[str, list[str]]]:
"""Converts the graph to a dictionary."""
result: dict = {}
for vertex in self.vertices:
vertex_id = vertex.id
sucessors = [i.id for i in self.get_all_successors(vertex)]
predecessors = [i.id for i in self.get_predecessors(vertex)]
result |= {vertex_id: {"successors": sucessors, "predecessors": predecessors}}
return result
def __filter_vertices(self, vertex_id: str, *, is_start: bool = False):
dictionaryized_graph = self.__to_dict()
parent_node_map = {vertex.id: vertex.parent_node_id for vertex in self.vertices}
vertex_ids = sort_up_to_vertex(
graph=dictionaryized_graph, vertex_id=vertex_id, parent_node_map=parent_node_map, is_start=is_start
)
return [self.get_vertex(vertex_id) for vertex_id in vertex_ids]
def get_vertex_ids(self) -> list[str]:
"""Get all vertex IDs in the graph."""
return [vertex.id for vertex in self.vertices]
def sort_vertices(
self,
@ -2042,39 +1899,27 @@ class Graph:
) -> list[str]:
"""Sorts the vertices in the graph."""
self.mark_all_vertices("ACTIVE")
if stop_component_id in self.cycle_vertices:
# Make the stop into a start because we are in a cycle and
# we cannot know where is the input or output
start_component_id = stop_component_id
stop_component_id = None
if stop_component_id is not None:
self.stop_vertex = stop_component_id
vertices = self.__filter_vertices(stop_component_id)
elif start_component_id:
vertices = self.__filter_vertices(start_component_id, is_start=True)
first_layer, remaining_layers = get_sorted_vertices(
vertices_ids=self.get_vertex_ids(),
cycle_vertices=self.cycle_vertices,
stop_component_id=stop_component_id,
start_component_id=start_component_id,
graph_dict=self.__to_dict(),
in_degree_map=self.in_degree_map,
successor_map=self.successor_map,
predecessor_map=self.predecessor_map,
is_input_vertex=self.get_vertex_input_status,
get_vertex_predecessors=self.get_vertex_predecessors_ids,
get_vertex_successors=self.get_vertex_successors_ids,
is_cyclic=self.is_cyclic,
)
else:
vertices = self.vertices
# without component_id we are probably running in the chat
# so we want to pick only graphs that start with ChatInput or
# TextInput
vertices_layers = self.layered_topological_sort(vertices)
vertices_layers = self.sort_by_avg_build_time(vertices_layers)
# Sort the chat inputs first to speed up sending the User message to the UI
vertices_layers = self.sort_chat_inputs_first(vertices_layers)
# Now we should sort each layer in a way that we make sure
# vertex V does not depend on vertex V+1
vertices_layers = self.sort_layer_by_dependency(vertices_layers)
self.increment_run_count()
self._sorted_vertices_layers = vertices_layers
first_layer = vertices_layers[0]
# save the only the rest
self.vertices_layers = vertices_layers[1:]
self.vertices_to_run = set(chain.from_iterable(vertices_layers))
self._sorted_vertices_layers = [first_layer, *remaining_layers]
self.vertices_layers = remaining_layers
self.vertices_to_run = set(chain.from_iterable([first_layer, *remaining_layers]))
self.build_run_map()
# Return just the first layer
self._first_layer = first_layer
return first_layer
@ -2195,3 +2040,13 @@ class Graph:
predecessor_map[edge.target_id].append(edge.source_id)
successor_map[edge.source_id].append(edge.target_id)
return predecessor_map, successor_map
def __to_dict(self) -> dict[str, dict[str, list[str]]]:
"""Converts the graph to a dictionary."""
result: dict = {}
for vertex in self.vertices:
vertex_id = vertex.id
sucessors = [i.id for i in self.get_all_successors(vertex)]
predecessors = [i.id for i in self.get_predecessors(vertex)]
result |= {vertex_id: {"successors": sucessors, "predecessors": predecessors}}
return result

493
src/backend/base/langflow/graph/graph/utils.py
View file

@ -1,21 +1,29 @@
import copy
from collections import defaultdict, deque
from collections.abc import Callable
from typing import Any
import networkx as nx
PRIORITY_LIST_OF_INPUTS = ["webhook", "chat"]
MAX_CYCLE_APPEARANCES = 2
def find_start_component_id(vertices):
def find_start_component_id(vertices, *, is_webhook: bool = False):
"""Finds the component ID from a list of vertices based on a priority list of input types.
Args:
vertices (list): A list of vertex IDs.
is_webhook (bool, optional): Whether the flow is being run as a webhook. Defaults to False.
Returns:
str or None: The component ID that matches the highest priority input type, or None if no match is found.
"""
for input_type_str in PRIORITY_LIST_OF_INPUTS:
# Set priority list based on whether this is a webhook flow
priority_inputs = ["webhook"] if is_webhook else PRIORITY_LIST_OF_INPUTS
# Check input types in priority order
for input_type_str in priority_inputs:
component_id = next((vertex_id for vertex_id in vertices if input_type_str in vertex_id.lower()), None)
if component_id:
return component_id
@ -440,3 +448,484 @@ def find_cycle_vertices(edges):
cycle_vertices.update(component)
return sorted(cycle_vertices)
def layered_topological_sort(
vertices_ids: set[str],
in_degree_map: dict[str, int],
successor_map: dict[str, list[str]],
predecessor_map: dict[str, list[str]],
start_id: str | None = None,
cycle_vertices: set[str] | None = None,
is_input_vertex: Callable[[str], bool] | None = None,
*,
is_cyclic: bool = False,
) -> list[list[str]]:
"""Performs a layered topological sort of the vertices in the graph.
Args:
vertices_ids: Set of vertex IDs to sort
in_degree_map: Map of vertex IDs to their in-degree
successor_map: Map of vertex IDs to their successors
predecessor_map: Map of vertex IDs to their predecessors
is_cyclic: Whether the graph is cyclic
start_id: ID of the start vertex (if any)
cycle_vertices: Set of vertices that form a cycle
is_input_vertex: Function to check if a vertex is an input vertex
Returns:
List of layers, where each layer is a list of vertex IDs
"""
# Queue for vertices with no incoming edges
cycle_vertices = cycle_vertices or set()
in_degree_map = in_degree_map.copy()
if is_cyclic and all(in_degree_map.values()):
# This means we have a cycle because all vertex have in_degree_map > 0
# because of this we set the queue to start on the start_id if it exists
if start_id is not None:
queue = deque([start_id])
# Reset in_degree for start_id to allow cycle traversal
in_degree_map[start_id] = 0
else:
# Find the chat input component
chat_input = find_start_component_id(vertices_ids)
if chat_input is None:
# If no input component is found, start with any vertex
queue = deque([next(iter(vertices_ids))])
in_degree_map[next(iter(vertices_ids))] = 0
else:
queue = deque([chat_input])
# Reset in_degree for chat_input to allow cycle traversal
in_degree_map[chat_input] = 0
else:
# Start with vertices that have no incoming edges or are input vertices
queue = deque(
vertex_id
for vertex_id in vertices_ids
if in_degree_map[vertex_id] == 0 or (is_input_vertex and is_input_vertex(vertex_id))
)
layers: list[list[str]] = []
visited = set()
cycle_counts = {vertex: 0 for vertex in vertices_ids}
current_layer = 0
# Process the first layer separately to avoid duplicates
if queue:
layers.append([]) # Start the first layer
first_layer_vertices = set()
layer_size = len(queue)
for _ in range(layer_size):
vertex_id = queue.popleft()
if vertex_id not in first_layer_vertices:
first_layer_vertices.add(vertex_id)
visited.add(vertex_id)
cycle_counts[vertex_id] += 1
layers[current_layer].append(vertex_id)
for neighbor in successor_map[vertex_id]:
# only vertices in `vertices_ids` should be considered
# because vertices by have been filtered out
# in a previous step. All dependencies of theirs
# will be built automatically if required
if neighbor not in vertices_ids:
continue
in_degree_map[neighbor] -= 1 # 'remove' edge
if in_degree_map[neighbor] == 0:
queue.append(neighbor)
# if > 0 it might mean not all predecessors have added to the queue
# so we should process the neighbors predecessors
elif in_degree_map[neighbor] > 0:
for predecessor in predecessor_map[neighbor]:
if (
predecessor not in queue
and predecessor not in first_layer_vertices
and (in_degree_map[predecessor] == 0 or predecessor in cycle_vertices)
):
queue.append(predecessor)
current_layer += 1 # Next layer
# Process remaining layers normally, allowing cycle vertices to appear multiple times
while queue:
layers.append([]) # Start a new layer
layer_size = len(queue)
for _ in range(layer_size):
vertex_id = queue.popleft()
if vertex_id not in visited or (is_cyclic and cycle_counts[vertex_id] < MAX_CYCLE_APPEARANCES):
if vertex_id not in visited:
visited.add(vertex_id)
cycle_counts[vertex_id] += 1
layers[current_layer].append(vertex_id)
for neighbor in successor_map[vertex_id]:
# only vertices in `vertices_ids` should be considered
# because vertices by have been filtered out
# in a previous step. All dependencies of theirs
# will be built automatically if required
if neighbor not in vertices_ids:
continue
in_degree_map[neighbor] -= 1 # 'remove' edge
if in_degree_map[neighbor] == 0 and neighbor not in visited:
queue.append(neighbor)
# # If this is a cycle vertex, reset its in_degree to allow it to appear again
# if neighbor in cycle_vertices and neighbor in visited:
# in_degree_map[neighbor] = len(predecessor_map[neighbor])
# if > 0 it might mean not all predecessors have added to the queue
# so we should process the neighbors predecessors
elif in_degree_map[neighbor] > 0:
for predecessor in predecessor_map[neighbor]:
if predecessor not in queue and (
predecessor not in visited
or (is_cyclic and cycle_counts[predecessor] < MAX_CYCLE_APPEARANCES)
):
queue.append(predecessor)
current_layer += 1 # Next layer
# Remove empty layers
return [layer for layer in layers if layer]
def refine_layers(
initial_layers: list[list[str]],
successor_map: dict[str, list[str]],
) -> list[list[str]]:
"""Refines the layers of vertices to ensure proper dependency ordering.
Args:
initial_layers: Initial layers of vertices
successor_map: Map of vertex IDs to their successors
Returns:
Refined layers with proper dependency ordering
"""
# Map each vertex to its current layer
vertex_to_layer: dict[str, int] = {}
for layer_index, layer in enumerate(initial_layers):
for vertex in layer:
vertex_to_layer[vertex] = layer_index
refined_layers: list[list[str]] = [[] for _ in initial_layers] # Start with empty layers
new_layer_index_map = defaultdict(int)
# Map each vertex to its new layer index
# by finding the lowest layer index of its dependencies
# and subtracting 1
# If a vertex has no dependencies, it will be placed in the first layer
# If a vertex has dependencies, it will be placed in the lowest layer index of its dependencies
# minus 1
for vertex_id, deps in successor_map.items():
indexes = [vertex_to_layer[dep] for dep in deps if dep in vertex_to_layer]
new_layer_index = max(min(indexes, default=0) - 1, 0)
new_layer_index_map[vertex_id] = new_layer_index
for layer_index, layer in enumerate(initial_layers):
for vertex_id in layer:
# Place the vertex in the highest possible layer where its dependencies are met
new_layer_index = new_layer_index_map[vertex_id]
if new_layer_index > layer_index:
refined_layers[new_layer_index].append(vertex_id)
vertex_to_layer[vertex_id] = new_layer_index
else:
refined_layers[layer_index].append(vertex_id)
# Remove empty layers if any
return [layer for layer in refined_layers if layer]
def _max_dependency_index(
vertex_id: str,
index_map: dict[str, int],
get_vertex_successors: Callable[[str], list[str]],
) -> int:
"""Finds the highest index a given vertex's dependencies occupy in the same layer.
Args:
vertex_id: ID of the vertex to check
index_map: Map of vertex IDs to their indices in the layer
get_vertex_successors: Function to get the successor IDs of a vertex
Returns:
The highest index of the vertex's dependencies
"""
max_index = -1
for successor_id in get_vertex_successors(vertex_id):
successor_index = index_map.get(successor_id, -1)
max_index = max(successor_index, max_index)
return max_index
def _sort_single_layer_by_dependency(
layer: list[str],
get_vertex_successors: Callable[[str], list[str]],
) -> list[str]:
"""Sorts a single layer by dependency using a stable sorting method.
Args:
layer: List of vertex IDs in the layer
get_vertex_successors: Function to get the successor IDs of a vertex
Returns:
Sorted list of vertex IDs
"""
# Build a map of each vertex to its index in the layer for quick lookup.
index_map = {vertex: index for index, vertex in enumerate(layer)}
dependency_cache: dict[str, int] = {}
def max_dependency_index(vertex: str) -> int:
if vertex in dependency_cache:
return dependency_cache[vertex]
max_index = index_map[vertex]
for successor in get_vertex_successors(vertex):
if successor in index_map:
max_index = max(max_index, max_dependency_index(successor))
dependency_cache[vertex] = max_index
return max_index
return sorted(layer, key=max_dependency_index, reverse=True)
def sort_layer_by_dependency(
vertices_layers: list[list[str]],
get_vertex_successors: Callable[[str], list[str]],
) -> list[list[str]]:
"""Sorts the vertices in each layer by dependency, ensuring no vertex depends on a subsequent vertex.
Args:
vertices_layers: List of layers, where each layer is a list of vertex IDs
get_vertex_successors: Function to get the successor IDs of a vertex
Returns:
Sorted layers
"""
return [_sort_single_layer_by_dependency(layer, get_vertex_successors) for layer in vertices_layers]
def sort_chat_inputs_first(
vertices_layers: list[list[str]],
get_vertex_predecessors: Callable[[str], list[str]],
) -> list[list[str]]:
"""Sorts the vertices so that chat inputs come first in the layers.
Only one chat input is allowed in the entire graph.
Args:
vertices_layers: List of layers, where each layer is a list of vertex IDs
get_vertex_predecessors: Function to get the predecessor IDs of a vertex
Returns:
Sorted layers with single chat input first
Raises:
ValueError: If there are multiple chat inputs in the graph
"""
chat_input = None
chat_input_layer_idx = None
# Find chat input and validate only one exists
for layer_idx, layer in enumerate(vertices_layers):
for vertex_id in layer:
if "ChatInput" in vertex_id and get_vertex_predecessors(vertex_id):
return vertices_layers
if "ChatInput" in vertex_id:
if chat_input is not None:
msg = "Only one chat input is allowed in the graph"
raise ValueError(msg)
chat_input = vertex_id
chat_input_layer_idx = layer_idx
if not chat_input:
return vertices_layers
# If chat input already in first layer, just move it to index 0
if chat_input_layer_idx == 0:
first_layer = vertices_layers[0]
first_layer.remove(chat_input)
first_layer.insert(0, chat_input)
return vertices_layers
# Otherwise create new layers with chat input first
result_layers = []
for layer in vertices_layers:
layer_vertices = [v for v in layer if v != chat_input]
if layer_vertices:
result_layers.append(layer_vertices)
return [[chat_input], *result_layers]
def get_sorted_vertices(
vertices_ids: list[str],
cycle_vertices: set[str],
stop_component_id: str | None = None,
start_component_id: str | None = None,
graph_dict: dict[str, Any] | None = None,
in_degree_map: dict[str, int] | None = None,
successor_map: dict[str, list[str]] | None = None,
predecessor_map: dict[str, list[str]] | None = None,
is_input_vertex: Callable[[str], bool] | None = None,
get_vertex_predecessors: Callable[[str], list[str]] | None = None,
get_vertex_successors: Callable[[str], list[str]] | None = None,
*,
is_cyclic: bool = False,
) -> tuple[list[str], list[list[str]]]:
"""Get sorted vertices in a graph.
Args:
vertices_ids: List of vertex IDs to sort
cycle_vertices: Set of vertices that form a cycle
stop_component_id: ID of the stop component (if any)
start_component_id: ID of the start component (if any)
graph_dict: Dictionary containing graph information
in_degree_map: Map of vertex IDs to their in-degree
successor_map: Map of vertex IDs to their successors
predecessor_map: Map of vertex IDs to their predecessors
is_input_vertex: Function to check if a vertex is an input vertex
get_vertex_predecessors: Function to get predecessors of a vertex
get_vertex_successors: Function to get successors of a vertex
is_cyclic: Whether the graph is cyclic
Returns:
Tuple of (first layer vertices, remaining layer vertices)
"""
# Handle cycles by converting stop to start
if stop_component_id in cycle_vertices:
start_component_id = stop_component_id
stop_component_id = None
# Build in_degree_map if not provided
if in_degree_map is None:
in_degree_map = {}
for vertex_id in vertices_ids:
if get_vertex_predecessors is not None:
in_degree_map[vertex_id] = len(get_vertex_predecessors(vertex_id))
else:
in_degree_map[vertex_id] = 0
# Build successor_map if not provided
if successor_map is None:
successor_map = {}
for vertex_id in vertices_ids:
if get_vertex_successors is not None:
successor_map[vertex_id] = get_vertex_successors(vertex_id)
else:
successor_map[vertex_id] = []
# Build predecessor_map if not provided
if predecessor_map is None:
predecessor_map = {}
for vertex_id in vertices_ids:
if get_vertex_predecessors is not None:
predecessor_map[vertex_id] = get_vertex_predecessors(vertex_id)
else:
predecessor_map[vertex_id] = []
# If we have a stop component, we need to filter out all vertices
# that are not predecessors of the stop component
if stop_component_id is not None:
filtered_vertices = filter_vertices_up_to_vertex(
vertices_ids,
stop_component_id,
get_vertex_predecessors=get_vertex_predecessors,
get_vertex_successors=get_vertex_successors,
graph_dict=graph_dict,
)
vertices_ids = list(filtered_vertices)
# Get the layers
layers = layered_topological_sort(
vertices_ids=set(vertices_ids),
in_degree_map=in_degree_map,
successor_map=successor_map,
predecessor_map=predecessor_map,
start_id=start_component_id,
is_input_vertex=is_input_vertex,
cycle_vertices=cycle_vertices,
is_cyclic=is_cyclic,
)
# Split into first layer and remaining layers
if not layers:
return [], []
first_layer = layers[0]
remaining_layers = layers[1:]
# If we have a stop component, we need to filter out all vertices
# that are not predecessors of the stop component
if stop_component_id is not None and remaining_layers and stop_component_id not in remaining_layers[-1]:
remaining_layers[-1].append(stop_component_id)
# Sort chat inputs first and sort each layer by dependencies
all_layers = [first_layer, *remaining_layers]
if get_vertex_predecessors is not None:
all_layers = sort_chat_inputs_first(all_layers, get_vertex_predecessors)
if get_vertex_successors is not None:
all_layers = sort_layer_by_dependency(all_layers, get_vertex_successors)
if not all_layers:
return [], []
return all_layers[0], all_layers[1:]
def filter_vertices_up_to_vertex(
vertices_ids: list[str],
vertex_id: str,
get_vertex_predecessors: Callable[[str], list[str]] | None = None,
get_vertex_successors: Callable[[str], list[str]] | None = None,
graph_dict: dict[str, Any] | None = None,
) -> set[str]:
"""Filter vertices up to a given vertex.
Args:
vertices_ids: List of vertex IDs to filter
vertex_id: ID of the vertex to filter up to
get_vertex_predecessors: Function to get predecessors of a vertex
get_vertex_successors: Function to get successors of a vertex
graph_dict: Dictionary containing graph information
parent_node_map: Map of vertex IDs to their parent node IDs
Returns:
Set of vertex IDs that are predecessors of the given vertex
"""
vertices_set = set(vertices_ids)
if vertex_id not in vertices_set:
return set()
# Build predecessor map if not provided
if get_vertex_predecessors is None:
if graph_dict is None:
return set()
def get_vertex_predecessors(v):
return graph_dict[v]["predecessors"]
# Build successor map if not provided
if get_vertex_successors is None:
if graph_dict is None:
return set()
def get_vertex_successors(v):
return graph_dict[v]["successors"]
# Start with the target vertex
filtered_vertices = {vertex_id}
queue = deque([vertex_id])
# Process vertices in breadth-first order
while queue:
current_vertex = queue.popleft()
for predecessor in get_vertex_predecessors(current_vertex):
if predecessor in vertices_set and predecessor not in filtered_vertices:
filtered_vertices.add(predecessor)
queue.append(predecessor)
return filtered_vertices

8
src/backend/base/langflow/services/socket/utils.py
View file

@ -8,6 +8,7 @@ from sqlmodel import select
from langflow.api.utils import format_elapsed_time
from langflow.api.v1.schemas import ResultDataResponse, VertexBuildResponse
from langflow.graph.graph.base import Graph
from langflow.graph.graph.utils import layered_topological_sort
from langflow.graph.utils import log_vertex_build
from langflow.graph.vertex.base import Vertex
from langflow.services.database.models.flow.model import Flow
@ -32,7 +33,12 @@ async def get_vertices(sio, sid, flow_id, chat_service) -> None:
graph = Graph.from_payload(flow.data)
chat_service.set_cache(flow_id, graph)
vertices = graph.layered_topological_sort(graph.vertices)
vertices = layered_topological_sort(
set(graph.get_vertex_ids()),
graph.in_degree_map,
graph.successor_map,
graph.predecessor_map,
)
# Emit the vertices to the client
await sio.emit("vertices_order", data=vertices, to=sid)