Docstring, Error Handling, and Type Hints Improvements (#1828)

* docs: add comprehensive docstrings to Flow class and methods

- Added NumPy-style docstrings to all decorator functions
- Added detailed documentation to Flow class methods
- Included parameter types, return types, and examples
- Enhanced documentation clarity and completeness

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

* feat: add secure path handling utilities

- Add path_utils.py with safe path handling functions
- Implement path validation and security checks
- Integrate secure path handling in flow_visualizer.py
- Add path validation in html_template_handler.py
- Add comprehensive error handling for path operations

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

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

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

* fix: add type annotations and fix import sorting

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

* fix: add type annotations to flow utils and visualization utils

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

* fix: resolve import sorting and type annotation issues

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

* fix: properly initialize and update edge_smooth variable

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

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
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,11 +93,34 @@ def get_possible_return_constants(function):
     return list(return_values) if return_values else None
 
 
-def calculate_node_levels(flow):
-    levels = {}
-    queue = []
-    visited = set()
-    pending_and_listeners = {}
+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: Dict[str, int] = {}
+    queue: List[str] = []
+    visited: Set[str] = set()
+    pending_and_listeners: Dict[str, Set[str]] = {}
 
     # Make all start methods at level 0
     for method_name, method in flow._methods.items():
@@ -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,16 +204,53 @@ def count_outgoing_edges(flow):
     return counts
 
 
-def build_ancestor_dict(flow):
-    ancestors = {node: set() for node in flow._methods}
-    visited = set()
+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: Dict[str, Set[str]] = {node: set() for node in flow._methods}
+    visited: Set[str] = set()
     for node in flow._methods:
         if node not in visited:
             dfs_ancestors(node, ancestors, visited, 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,12 +274,48 @@ 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):
-    parent_children = {}
+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: Dict[str, List[str]] = {}
 
     # Map listeners to their trigger methods
     for listener_name, (_, trigger_methods) in flow._listeners.items():
@@ -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)