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

refactor: (codelash) ⚡️ Speed up function find_all_cycle_edges by 17% (#5389)

Browse source 
⚡️ Speed up function `find_all_cycle_edges` by 17%
Here is the optimized version of the given program. The major optimization here is to avoid unnecessary list concatenations in the DFS recursion by using a more efficient approach for aggregating cycle edges.

Co-authored-by: codeflash-ai[bot] <148906541+codeflash-ai[bot]@users.noreply.github.com>
This commit is contained in:
Saurabh Misra 2025-02-11 03:45:02 -08:00 • committed by GitHub
commit 10aea2e9b6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 6 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -417,25 +417,25 @@ def find_all_cycle_edges(entry_point: str, edges: list[tuple[str, str]]) -> list
graph[u].append(v) graph[u].append(v)
# Utility function to perform DFS # Utility function to perform DFS
def dfs(v, visited, rec_stack): def dfs(v, visited, rec_stack, cycle_edges):
visited.add(v) visited.add(v)
rec_stack.add(v) rec_stack.add(v)
cycle_edges = []
for neighbor in graph[v]: for neighbor in graph[v]:
if neighbor not in visited: if neighbor not in visited:
cycle_edges += dfs(neighbor, visited, rec_stack) dfs(neighbor, visited, rec_stack, cycle_edges)
elif neighbor in rec_stack: elif neighbor in rec_stack:
cycle_edges.append((v, neighbor)) # This edge causes a cycle cycle_edges.append((v, neighbor)) # This edge causes a cycle
rec_stack.remove(v) rec_stack.remove(v)
return cycle_edges
visited: set[str] = set() visited: set[str] = set()
rec_stack: set[str] = set() rec_stack: set[str] = set()
cycle_edges: list[tuple[str, str]] = []
return dfs(entry_point, visited, rec_stack) dfs(entry_point, visited, rec_stack, cycle_edges)
return cycle_edges
def should_continue(yielded_counts: dict[str, int], max_iterations: int | None) -> bool: def should_continue(yielded_counts: dict[str, int], max_iterations: int | None) -> bool: