andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-01-11 00:58:30 +00:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

fix: type annotations and import sorting in flow modules

Browse Source 
Co-Authored-By: Joe Moura <joao@crewai.com>
This commit is contained in:
Devin AI
2024-12-30 20:59:06 +00:00
parent 1ece057ffc
commit 4d93a60806
2 changed files with 9 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/crewai/flow/flow_visual_utils.py
View File

@@ -136,8 +136,8 @@ def build_ancestor_dict(flow: Flow[Any]) -> Dict[str, Set[str]]:
dict[str, set[str]]: Dictionary mapping each method name to a set dict[str, set[str]]: Dictionary mapping each method name to a set
of its ancestor method names of its ancestor method names
""" """
ancestors = {node: set() for node in flow._methods} ancestors: Dict[str, Set[str]] = {node: set() for node in flow._methods}
visited = set() visited: Set[str] = set()
for node in flow._methods: for node in flow._methods:
if node not in visited: if node not in visited:
dfs_ancestors(node, ancestors, visited, flow) dfs_ancestors(node, ancestors, visited, flow)

14
src/crewai/flow/visualization_utils.py
View File

@@ -2,7 +2,7 @@ import ast
import inspect import inspect
import os import os
from pathlib import Path from pathlib import Path
from typing import Any, Callable, Dict, Optional, Tuple, cast from typing import Any, Callable, Dict, List, Optional, Set, Tuple, cast
from pyvis.network import Network from pyvis.network import Network
@@ -205,8 +205,8 @@ def compute_positions(flow: Flow[Any], node_levels: Dict[str, int],
Positions are calculated to maintain clear hierarchical structure while Positions are calculated to maintain clear hierarchical structure while
ensuring optimal spacing and readability of the flow diagram. ensuring optimal spacing and readability of the flow diagram.
""" """
level_nodes = {} level_nodes: Dict[int, List[str]] = {}
node_positions = {} node_positions: Dict[str, Tuple[float, float]] = {}
for method_name, level in node_levels.items(): for method_name, level in node_levels.items():
level_nodes.setdefault(level, []).append(method_name) level_nodes.setdefault(level, []).append(method_name)
@@ -306,9 +306,9 @@ def add_edges(net: Network, flow: Flow[Any],
else: else:
edge_smooth = {"type": "cubicBezier"} edge_smooth = {"type": "cubicBezier"}
else: else:
edge_smooth = False edge_smooth: Dict[str, Any] = {"type": "straight"}
edge_style = { edge_style: Dict[str, Any] = {
"color": edge_color, "color": edge_color,
"width": 2, "width": 2,
"arrows": "to", "arrows": "to",
@@ -369,9 +369,9 @@ def add_edges(net: Network, flow: Flow[Any],
else: else:
edge_smooth = {"type": "cubicBezier"} edge_smooth = {"type": "cubicBezier"}
else: else:
edge_smooth = False edge_smooth: Dict[str, Any] = {"type": "straight"}
edge_style = { edge_style: Dict[str, Any] = {
"color": colors["router_edge"], "color": colors["router_edge"],
"width": 2, "width": 2,
"arrows": "to", "arrows": "to",