File size: 3,940 Bytes
1bf8909 183df63 1bf8909 183df63 1bf8909 32563aa 1bf8909 32563aa 1bf8909 32563aa 1bf8909 32563aa 1bf8909 32563aa 1bf8909 836766a 32563aa 836766a 32563aa 836766a 32563aa 836766a 1bf8909 836766a 32563aa 836766a 32563aa 1bf8909 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | import asyncio
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
class UserFeedback(RunInput):
"""Input model for capturing user feedback in the Prefect UI."""
rating: int
comment: str = ""
improve_feature: bool = False
@flow(name="interactive-feedback-flow")
async def collect_feedback(username: str):
"""A flow that pauses to collect user feedback via the Prefect UI."""
logger = get_run_logger()
logger.info(f"Started feedback collection for user: {username}")
# Get flow run context for UI URL
flow_run = get_run_context().flow_run
flow_run_url = ""
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
logger.info("Pausing flow to collect feedback...")
user_feedback = await pause_flow_run(
wait_for_input=UserFeedback.with_initial_data(
description="Please provide your feedback about our service.",
rating=5,
comment="",
improve_feature=False,
),
timeout=300, # 5 minutes timeout
)
# Process the input received after the flow is resumed
logger.info(f"Feedback received from {username}:")
logger.info(f"Rating: {user_feedback.rating}/5")
logger.info(f"Comment: {user_feedback.comment}")
logger.info(f"Wants feature improvement: {user_feedback.improve_feature}")
return {
"username": username,
"rating": user_feedback.rating,
"comment": user_feedback.comment,
"improve_feature": user_feedback.improve_feature,
}
def run_feedback_flow(username):
"""Run the Prefect flow and handle the UI interaction."""
try:
if not username:
return "Please enter a username before starting the process."
print(f"Starting feedback collection for user: {username}")
# Start the flow in a separate thread to avoid blocking the Gradio interface
import threading
def run_flow():
# Run the flow and get the result
result = asyncio.run(collect_feedback(username))
print(f"Flow completed with results: {result}")
# 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
The results will be available in the terminal after completion.
"""
except Exception as e:
return f"An error occurred: {str(e)}"
# Gradio interface
with gr.Blocks() as demo:
gr.Markdown("# Interactive Human-in-the-Loop Workflow with Prefect")
with gr.Row():
with gr.Column():
username_input = gr.Textbox(label="Your Name")
submit_btn = gr.Button("Start Feedback Process")
with gr.Column():
output_text = gr.Textbox(label="Result", lines=10)
flow_instructions = gr.Markdown(
"""
## Instructions:
1. Enter your name and click "Start Feedback Process"
2. The flow will pause waiting for your input
3. Go to the Prefect UI (URL will be printed in terminal)
4. Click "Resume" and provide your feedback
5. Results will appear here after completion
"""
)
submit_btn.click(fn=run_feedback_flow, inputs=username_input, outputs=output_text)
if __name__ == "__main__":
demo.launch()
|