Refactor feedback collection flow to simplify UI integration and improve error handling

app.py

@@ -1,11 +1,8 @@
 import asyncio
-import os
 import gradio as gr
 from prefect import flow, get_run_logger, pause_flow_run, settings
 from prefect.context import get_run_context
 from prefect.input import RunInput
-from prefect.deployments import Deployment
-from prefect.server.schemas.schedules import IntervalSchedule
 
 
 class UserFeedback(RunInput):
@@ -18,47 +15,19 @@ class UserFeedback(RunInput):
 
 @flow(name="interactive-feedback-flow")
 async def collect_feedback(username: str):
-    """A flow that pauses to collect user feedback via the Prefect Cloud UI."""
+    """A flow that pauses to collect user feedback via the Prefect UI."""
     logger = get_run_logger()
 
-    # Ensure we're using Prefect Cloud
-    logger.info("Running with Prefect Cloud configuration")
-
     logger.info(f"Started feedback collection for user: {username}")
 
-    # Construct a message that would typically be sent as a notification
-    message = (
-        f"Collecting feedback from user: {username}"  # Get flow run context for UI URL
-    )
+    # Get flow run context for UI URL
     flow_run = get_run_context().flow_run
     flow_run_url = ""
 
-    if flow_run:
-        # For Prefect Cloud, use the PREFECT_API_URL to determine the Cloud UI URL
-        if os.environ.get("PREFECT_API_URL") and "api.prefect.cloud" in os.environ.get(
-            "PREFECT_API_URL", ""
-        ):
-            # Extract account_id and workspace_name from the API URL
-            # Format: https://api.prefect.cloud/api/accounts/{account_id}/workspaces/{workspace_name}/
-            api_url = os.environ["PREFECT_API_URL"]
-            parts = api_url.split("/")
-
-            if (
-                len(parts) >= 7
-            ):  # Ensure we have enough parts to extract account_id and workspace_name
-                account_id = parts[5]
-                workspace_name = parts[7]
-                cloud_ui_url = f"https://app.prefect.cloud/account/{account_id}/workspace/{workspace_name}"
-                flow_run_url = f"{cloud_ui_url}/flow-runs/flow-run/{flow_run.id}"
-            else:
-                # Fallback to standard URL formation if we can't parse the API URL
-                if settings.PREFECT_UI_URL:
-                    flow_run_url = f"{settings.PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run.id}"
-        else:
-            # Fallback for local Prefect or other setups
-            if settings.PREFECT_UI_URL:
-                flow_run_url = f"{settings.PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run.id}"
-
+    if flow_run and settings.PREFECT_UI_URL:
+        flow_run_url = (
+            f"{settings.PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run.id}"
+        )
         logger.info(f"Flow run UI available at: {flow_run_url}")
 
     # Pause the flow and wait for input
@@ -70,7 +39,7 @@ async def collect_feedback(username: str):
             comment="",
             improve_feature=False,
         ),
-        timeout=60,  # 5 minutes timeout
+        timeout=300,  # 5 minutes timeout
     )
 
     # Process the input received after the flow is resumed
@@ -79,17 +48,13 @@ async def collect_feedback(username: str):
     logger.info(f"Comment: {user_feedback.comment}")
     logger.info(f"Wants feature improvement: {user_feedback.improve_feature}")
 
-    # Return results to be displayed in Gradio
-    result = {
+    return {
         "username": username,
         "rating": user_feedback.rating,
         "comment": user_feedback.comment,
         "improve_feature": user_feedback.improve_feature,
-        "flow_run_url": flow_run_url,
     }
 
-    return result
-
 
 def run_feedback_flow(username):
     """Run the Prefect flow and handle the UI interaction."""
@@ -99,75 +64,29 @@ def run_feedback_flow(username):
 
         print(f"Starting feedback collection for user: {username}")
 
-        # Create a deployment of the flow to ensure it runs properly in Prefect Cloud
-        from prefect.deployments import Deployment
-
-        # Create or update the deployment
-        deployment = Deployment.build_from_flow(
-            flow=collect_feedback,
-            name="interactive-feedback-deployment",
-            version=None,  # Auto-version
-            work_queue_name="default",
-            tags=["interactive", "feedback"],
-            parameters={"username": username},
-            enforce_parameter_schema=False,
-        )
-
-        # Apply the deployment
-        deployment_id = deployment.apply()
-        print(f"Created deployment with ID: {deployment_id}")
+        # Start the flow in a separate thread to avoid blocking the Gradio interface
+        import threading
 
-        # Run the deployment
-        from prefect.deployments import run_deployment
-
-        flow_run = run_deployment(
-            name="interactive-feedback-flow/interactive-feedback-deployment",
-            parameters={"username": username},
-        )
+        def run_flow():
+            # Run the flow and get the result
+            result = asyncio.run(collect_feedback(username))
+            print(f"Flow completed with results: {result}")
 
-        print(f"Flow run created with ID: {flow_run.id}")
-
-        # Get the URL
-        flow_run_url = ""
-        if settings.PREFECT_UI_URL:
-            flow_run_url = (
-                f"{settings.PREFECT_UI_URL.value()}/flow-runs/flow-run/{flow_run.id}"
-            )
-        elif os.environ.get(
-            "PREFECT_API_URL"
-        ) and "api.prefect.cloud" in os.environ.get("PREFECT_API_URL", ""):
-            # Extract account_id and workspace_name from the API URL
-            api_url = os.environ["PREFECT_API_URL"]
-            parts = api_url.split("/")
-
-            if len(parts) >= 7:
-                account_id = parts[5]
-                workspace_name = parts[7]
-                flow_run_url = f"https://app.prefect.cloud/account/{account_id}/workspace/{workspace_name}/flow-runs/flow-run/{flow_run.id}"
-
-        print(f"Flow run UI available at: {flow_run_url}")
+        # Start the thread
+        threading.Thread(target=run_flow).start()
 
         return f"""
         Process started for {username}!
         
         IMPORTANT INSTRUCTIONS:
+        1. Look for a line in the terminal that says "Flow run UI available at: [URL]"
+        2. Copy that URL and open it in your browser
+        3. Click "Resume" in the Prefect UI
+        4. Fill out the feedback form and submit
         
-        Go to this URL to provide your feedback:
-        {flow_run_url}
-        
-        1. Click the "Resume" button in the Prefect UI
-        2. Fill out the feedback form and submit
-        3. Your input will be processed by the workflow
-        
-        Note: This window will not automatically update with the results.
-        The complete results will be available in the Prefect UI and terminal.
+        The results will be available in the terminal after completion.
         """
-
     except Exception as e:
-        import traceback
-
-        print(f"Error: {str(e)}")
-        print(traceback.format_exc())
         return f"An error occurred: {str(e)}"