refactor: improve flow handling, typing, and logging; update UI and tests

fix: refine nested flow conditionals and ensure router methods and routes are fully parsed
fix: improve docstrings, typing, and logging coverage across all events
feat: update flow.plot feature with new UI enhancements
chore: apply Ruff linting, reorganize imports, and remove deprecated utilities/files
chore: split constants and utils, clean JS comments, and add typing for linters
tests: strengthen test coverage for flow execution paths and router logic

Author: greysonlalonde

lib/crewai/pyproject.toml

@@ -23,7 +23,6 @@ dependencies = [
     "chromadb~=1.1.0",
     "tokenizers>=0.20.3",
     "openpyxl>=3.1.5",
-    "pyvis>=0.3.2",
     # Authentication and Security
     "python-dotenv>=1.1.1",
     "pyjwt>=2.9.0",

lib/crewai/src/crewai/events/event_listener.py

@@ -88,6 +88,7 @@ class EventListener(BaseEventListener):
     text_stream = StringIO()
     knowledge_retrieval_in_progress = False
     knowledge_query_in_progress = False
+    method_branches: dict[str, Any] = Field(default_factory=dict)
 
     def __new__(cls):
         if cls._instance is None:
@@ -101,6 +102,7 @@ def __init__(self):
             self._telemetry = Telemetry()
             self._telemetry.set_tracer()
             self.execution_spans = {}
+            self.method_branches = {}
             self._initialized = True
             self.formatter = ConsoleFormatter(verbose=True)
 
@@ -263,7 +265,8 @@ def on_lite_agent_execution_error(source, event: LiteAgentExecutionErrorEvent):
         @crewai_event_bus.on(FlowCreatedEvent)
         def on_flow_created(source, event: FlowCreatedEvent):
             self._telemetry.flow_creation_span(event.flow_name)
-            self.formatter.create_flow_tree(event.flow_name, str(source.flow_id))
+            tree = self.formatter.create_flow_tree(event.flow_name, str(source.flow_id))
+            self.formatter.current_flow_tree = tree
 
         @crewai_event_bus.on(FlowStartedEvent)
         def on_flow_started(source, event: FlowStartedEvent):
@@ -280,30 +283,36 @@ def on_flow_finished(source, event: FlowFinishedEvent):
 
         @crewai_event_bus.on(MethodExecutionStartedEvent)
         def on_method_execution_started(source, event: MethodExecutionStartedEvent):
-            self.formatter.update_method_status(
-                self.formatter.current_method_branch,
+            method_branch = self.method_branches.get(event.method_name)
+            updated_branch = self.formatter.update_method_status(
+                method_branch,
                 self.formatter.current_flow_tree,
                 event.method_name,
                 "running",
             )
+            self.method_branches[event.method_name] = updated_branch
 
         @crewai_event_bus.on(MethodExecutionFinishedEvent)
         def on_method_execution_finished(source, event: MethodExecutionFinishedEvent):
-            self.formatter.update_method_status(
-                self.formatter.current_method_branch,
+            method_branch = self.method_branches.get(event.method_name)
+            updated_branch = self.formatter.update_method_status(
+                method_branch,
                 self.formatter.current_flow_tree,
                 event.method_name,
                 "completed",
             )
+            self.method_branches[event.method_name] = updated_branch
 
         @crewai_event_bus.on(MethodExecutionFailedEvent)
         def on_method_execution_failed(source, event: MethodExecutionFailedEvent):
-            self.formatter.update_method_status(
-                self.formatter.current_method_branch,
+            method_branch = self.method_branches.get(event.method_name)
+            updated_branch = self.formatter.update_method_status(
+                method_branch,
                 self.formatter.current_flow_tree,
                 event.method_name,
                 "failed",
             )
+            self.method_branches[event.method_name] = updated_branch
 
         # ----------- TOOL USAGE EVENTS -----------
 

lib/crewai/src/crewai/events/utils/console_formatter.py

@@ -357,7 +357,14 @@ def create_flow_tree(self, flow_name: str, flow_id: str) -> Tree | None:
         return flow_tree
 
     def start_flow(self, flow_name: str, flow_id: str) -> Tree | None:
-        """Initialize a flow execution tree."""
+        """Initialize or update a flow execution tree."""
+        if self.current_flow_tree is not None:
+            for child in self.current_flow_tree.children:
+                if "Starting Flow" in str(child.label):
+                    child.label = Text("🚀 Flow Started", style="green")
+                    break
+            return self.current_flow_tree
+
         flow_tree = Tree("")
         flow_label = Text()
         flow_label.append("🌊 Flow: ", style="blue bold")
@@ -436,34 +443,46 @@ def update_method_status(
             prefix, style = "🔄 Running:", "yellow"
         elif status == "completed":
             prefix, style = "✅ Completed:", "green"
-            # Update initialization node when a method completes successfully
             for child in flow_tree.children:
                 if "Starting Flow" in str(child.label):
                     child.label = Text("Flow Method Step", style="white")
                     break
         else:
             prefix, style = "❌ Failed:", "red"
-            # Update initialization node on failure
             for child in flow_tree.children:
                 if "Starting Flow" in str(child.label):
                     child.label = Text("❌ Flow Step Failed", style="red")
                     break
 
-        if not method_branch:
-            # Find or create method branch
-            for branch in flow_tree.children:
-                if method_name in str(branch.label):
-                    method_branch = branch
-                    break
-            if not method_branch:
-                method_branch = flow_tree.add("")
+        if method_branch is not None:
+            if method_branch in flow_tree.children:
+                method_branch.label = Text(prefix, style=f"{style} bold") + Text(
+                    f" {method_name}", style=style
+                )
+                self.print(flow_tree)
+                self.print()
+                return method_branch
+
+        for branch in flow_tree.children:
+            label_str = str(branch.label)
+            if f" {method_name}" in label_str and (
+                "Running:" in label_str
+                or "Completed:" in label_str
+                or "Failed:" in label_str
+            ):
+                method_branch = branch
+                break
+
+        if method_branch is None:
+            method_branch = flow_tree.add("")
 
         method_branch.label = Text(prefix, style=f"{style} bold") + Text(
             f" {method_name}", style=style
         )
 
         self.print(flow_tree)
         self.print()
+
         return method_branch
 
     def get_llm_tree(self, tool_name: str):

lib/crewai/src/crewai/flow/__init__.py

@@ -1,5 +1,25 @@
+from crewai.flow.visualization import (
+    FlowStructure,
+    build_flow_structure,
+    print_structure_summary,
+    structure_to_dict,
+    visualize_flow_structure,
+)
 from crewai.flow.flow import Flow, and_, listen, or_, router, start
 from crewai.flow.persistence import persist
 
 
-__all__ = ["Flow", "and_", "listen", "or_", "persist", "router", "start"]
+__all__ = [
+    "Flow",
+    "FlowStructure",
+    "and_",
+    "build_flow_structure",
+    "listen",
+    "or_",
+    "persist",
+    "print_structure_summary",
+    "router",
+    "start",
+    "structure_to_dict",
+    "visualize_flow_structure",
+]