docs: add comprehensive docstrings and type hints to flow utils (#1819)

Co-Authored-By: Joe Moura <joao@crewai.com>

src/crewai/flow/utils.py

@@ -1,9 +1,25 @@
+"""
+Utility functions for flow visualization and dependency analysis.
+
+This module provides core functionality for analyzing and manipulating flow structures,
+including node level calculation, ancestor tracking, and return value analysis.
+Functions in this module are primarily used by the visualization system to create
+accurate and informative flow diagrams.
+
+Example
+-------
+>>> flow = Flow()
+>>> node_levels = calculate_node_levels(flow)
+>>> ancestors = build_ancestor_dict(flow)
+"""
+
 import ast
 import inspect
 import textwrap
+from typing import Any, Dict, List, Optional, Set, Union
 
 
-def get_possible_return_constants(function):
+def get_possible_return_constants(function: Any) -> Optional[List[str]]:
     try:
         source = inspect.getsource(function)
     except OSError:
@@ -77,7 +93,30 @@ def visit_Return(self, node):
     return list(return_values) if return_values else None
 
 
-def calculate_node_levels(flow):
+def calculate_node_levels(flow: Any) -> Dict[str, int]:
+    """
+    Calculate the hierarchical level of each node in the flow.
+
+    Performs a breadth-first traversal of the flow graph to assign levels
+    to nodes, starting with start methods at level 0.
+
+    Parameters
+    ----------
+    flow : Any
+        The flow instance containing methods, listeners, and router configurations.
+
+    Returns
+    -------
+    Dict[str, int]
+        Dictionary mapping method names to their hierarchical levels.
+
+    Notes
+    -----
+    - Start methods are assigned level 0
+    - Each subsequent connected node is assigned level = parent_level + 1
+    - Handles both OR and AND conditions for listeners
+    - Processes router paths separately
+    """
     levels = {}
     queue = []
     visited = set()
@@ -140,7 +179,20 @@ def calculate_node_levels(flow):
     return levels
 
 
-def count_outgoing_edges(flow):
+def count_outgoing_edges(flow: Any) -> Dict[str, int]:
+    """
+    Count the number of outgoing edges for each method in the flow.
+
+    Parameters
+    ----------
+    flow : Any
+        The flow instance to analyze.
+
+    Returns
+    -------
+    Dict[str, int]
+        Dictionary mapping method names to their outgoing edge count.
+    """
     counts = {}
     for method_name in flow._methods:
         counts[method_name] = 0
@@ -152,7 +204,20 @@ def count_outgoing_edges(flow):
     return counts
 
 
-def build_ancestor_dict(flow):
+def build_ancestor_dict(flow: Any) -> Dict[str, Set[str]]:
+    """
+    Build a dictionary mapping each node to its ancestor nodes.
+
+    Parameters
+    ----------
+    flow : Any
+        The flow instance to analyze.
+
+    Returns
+    -------
+    Dict[str, Set[str]]
+        Dictionary mapping each node to a set of its ancestor nodes.
+    """
     ancestors = {node: set() for node in flow._methods}
     visited = set()
     for node in flow._methods:
@@ -161,7 +226,31 @@ def build_ancestor_dict(flow):
     return ancestors
 
 
-def dfs_ancestors(node, ancestors, visited, flow):
+def dfs_ancestors(
+    node: str,
+    ancestors: Dict[str, Set[str]],
+    visited: Set[str],
+    flow: Any
+) -> None:
+    """
+    Perform depth-first search to build ancestor relationships.
+
+    Parameters
+    ----------
+    node : str
+        Current node being processed.
+    ancestors : Dict[str, Set[str]]
+        Dictionary tracking ancestor relationships.
+    visited : Set[str]
+        Set of already visited nodes.
+    flow : Any
+        The flow instance being analyzed.
+
+    Notes
+    -----
+    This function modifies the ancestors dictionary in-place to build
+    the complete ancestor graph.
+    """
     if node in visited:
         return
     visited.add(node)
@@ -185,11 +274,47 @@ def dfs_ancestors(node, ancestors, visited, flow):
                     dfs_ancestors(listener_name, ancestors, visited, flow)
 
 
-def is_ancestor(node, ancestor_candidate, ancestors):
+def is_ancestor(node: str, ancestor_candidate: str, ancestors: Dict[str, Set[str]]) -> bool:
+    """
+    Check if one node is an ancestor of another.
+
+    Parameters
+    ----------
+    node : str
+        The node to check ancestors for.
+    ancestor_candidate : str
+        The potential ancestor node.
+    ancestors : Dict[str, Set[str]]
+        Dictionary containing ancestor relationships.
+
+    Returns
+    -------
+    bool
+        True if ancestor_candidate is an ancestor of node, False otherwise.
+    """
     return ancestor_candidate in ancestors.get(node, set())
 
 
-def build_parent_children_dict(flow):
+def build_parent_children_dict(flow: Any) -> Dict[str, List[str]]:
+    """
+    Build a dictionary mapping parent nodes to their children.
+
+    Parameters
+    ----------
+    flow : Any
+        The flow instance to analyze.
+
+    Returns
+    -------
+    Dict[str, List[str]]
+        Dictionary mapping parent method names to lists of their child method names.
+
+    Notes
+    -----
+    - Maps listeners to their trigger methods
+    - Maps router methods to their paths and listeners
+    - Children lists are sorted for consistent ordering
+    """
     parent_children = {}
 
     # Map listeners to their trigger methods
@@ -214,7 +339,24 @@ def build_parent_children_dict(flow):
     return parent_children
 
 
-def get_child_index(parent, child, parent_children):
+def get_child_index(parent: str, child: str, parent_children: Dict[str, List[str]]) -> int:
+    """
+    Get the index of a child node in its parent's sorted children list.
+
+    Parameters
+    ----------
+    parent : str
+        The parent node name.
+    child : str
+        The child node name to find the index for.
+    parent_children : Dict[str, List[str]]
+        Dictionary mapping parents to their children lists.
+
+    Returns
+    -------
+    int
+        Zero-based index of the child in its parent's sorted children list.
+    """
     children = parent_children.get(parent, [])
     children.sort()
     return children.index(child)

src/crewai/flow/visualization_utils.py

@@ -1,5 +1,23 @@
+"""
+Utilities for creating visual representations of flow structures.
+
+This module provides functions for generating network visualizations of flows,
+including node placement, edge creation, and visual styling. It handles the
+conversion of flow structures into visual network graphs with appropriate
+styling and layout.
+
+Example
+-------
+>>> flow = Flow()
+>>> net = Network(directed=True)
+>>> node_positions = compute_positions(flow, node_levels)
+>>> add_nodes_to_network(net, flow, node_positions, node_styles)
+>>> add_edges(net, flow, node_positions, colors)
+"""
+
 import ast
 import inspect
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 from .utils import (
     build_ancestor_dict,
@@ -9,8 +27,25 @@
 )
 
 
-def method_calls_crew(method):
-    """Check if the method calls `.crew()`."""
+def method_calls_crew(method: Any) -> bool:
+    """
+    Check if the method contains a call to `.crew()`.
+
+    Parameters
+    ----------
+    method : Any
+        The method to analyze for crew() calls.
+
+    Returns
+    -------
+    bool
+        True if the method calls .crew(), False otherwise.
+
+    Notes
+    -----
+    Uses AST analysis to detect method calls, specifically looking for
+    attribute access of 'crew'.
+    """
     try:
         source = inspect.getsource(method)
         source = inspect.cleandoc(source)
@@ -34,7 +69,34 @@ def visit_Call(self, node):
     return visitor.found
 
 
-def add_nodes_to_network(net, flow, node_positions, node_styles):
+def add_nodes_to_network(
+    net: Any,
+    flow: Any,
+    node_positions: Dict[str, Tuple[float, float]],
+    node_styles: Dict[str, Dict[str, Any]]
+) -> None:
+    """
+    Add nodes to the network visualization with appropriate styling.
+
+    Parameters
+    ----------
+    net : Any
+        The pyvis Network instance to add nodes to.
+    flow : Any
+        The flow instance containing method information.
+    node_positions : Dict[str, Tuple[float, float]]
+        Dictionary mapping node names to their (x, y) positions.
+    node_styles : Dict[str, Dict[str, Any]]
+        Dictionary containing style configurations for different node types.
+
+    Notes
+    -----
+    Node types include:
+    - Start methods
+    - Router methods
+    - Crew methods
+    - Regular methods
+    """
     def human_friendly_label(method_name):
         return method_name.replace("_", " ").title()
 
@@ -73,7 +135,31 @@ def human_friendly_label(method_name):
         )
 
 
-def compute_positions(flow, node_levels, y_spacing=150, x_spacing=150):
+def compute_positions(
+    flow: Any,
+    node_levels: Dict[str, int],
+    y_spacing: float = 150,
+    x_spacing: float = 150
+) -> Dict[str, Tuple[float, float]]:
+    """
+    Compute the (x, y) positions for each node in the flow graph.
+
+    Parameters
+    ----------
+    flow : Any
+        The flow instance to compute positions for.
+    node_levels : Dict[str, int]
+        Dictionary mapping node names to their hierarchical levels.
+    y_spacing : float, optional
+        Vertical spacing between levels, by default 150.
+    x_spacing : float, optional
+        Horizontal spacing between nodes, by default 150.
+
+    Returns
+    -------
+    Dict[str, Tuple[float, float]]
+        Dictionary mapping node names to their (x, y) coordinates.
+    """
     level_nodes = {}
     node_positions = {}
 
@@ -90,7 +176,32 @@ def compute_positions(flow, node_levels, y_spacing=150, x_spacing=150):
     return node_positions
 
 
-def add_edges(net, flow, node_positions, colors):
+def add_edges(
+    net: Any,
+    flow: Any,
+    node_positions: Dict[str, Tuple[float, float]],
+    colors: Dict[str, str]
+) -> None:
+    """
+    Add edges to the network visualization with appropriate styling.
+
+    Parameters
+    ----------
+    net : Any
+        The pyvis Network instance to add edges to.
+    flow : Any
+        The flow instance containing edge information.
+    node_positions : Dict[str, Tuple[float, float]]
+        Dictionary mapping node names to their positions.
+    colors : Dict[str, str]
+        Dictionary mapping edge types to their colors.
+
+    Notes
+    -----
+    - Handles both normal listener edges and router edges
+    - Applies appropriate styling (color, dashes) based on edge type
+    - Adds curvature to edges when needed (cycles or multiple children)
+    """
     ancestors = build_ancestor_dict(flow)
     parent_children = build_parent_children_dict(flow)