Spaces:
Sleeping
Sleeping
| """ | |
| Minimal Gradio app for CrewAI data analysis with file upload and parallel agent execution. | |
| """ | |
| import os | |
| import gradio as gr | |
| import traceback | |
| from crew import create_flow_crew, create_analyst_only_crew | |
| def process_file_and_analyze(file, user_query: str = "", engineer_result: str = None) -> tuple[str, str]: | |
| """ | |
| Process uploaded file and run all agents (Engineer, Analyst, Storyteller), then merge results. | |
| Used for the "Analyze Dataset" button. | |
| Args: | |
| file: Uploaded file object | |
| user_query: The user's analysis query/task (empty for general analysis) | |
| engineer_result: Previously computed engineer result (if available) | |
| Returns: | |
| tuple: (merged_results, engineer_result) - engineer_result is stored for reuse | |
| """ | |
| if file is None: | |
| return "Please upload a CSV file.", engineer_result or "" | |
| # Use default analysis if no query provided | |
| if not user_query or not user_query.strip(): | |
| user_query = "Provide a comprehensive analysis of the dataset including: top performers, key statistics, interesting patterns, and notable insights." | |
| try: | |
| # Get file path | |
| file_path = file.name if hasattr(file, 'name') else str(file) | |
| csv_path = file_path | |
| # Full analysis: run all agents | |
| crew = create_flow_crew(user_query.strip(), csv_path) | |
| result = crew.kickoff() | |
| merged_output = [] | |
| stored_engineer_result = "" | |
| # Get engineer result (first task) | |
| if hasattr(result, 'tasks_output') and result.tasks_output: | |
| if len(result.tasks_output) >= 1: | |
| engineer_output = str(result.tasks_output[0]) | |
| stored_engineer_result = engineer_output | |
| merged_output.append("## Engineer Agent Results") | |
| merged_output.append("") | |
| merged_output.append(engineer_output) | |
| merged_output.append("") | |
| merged_output.append("---") | |
| merged_output.append("") | |
| # Get analyst result (second task) | |
| if hasattr(result, 'tasks_output') and result.tasks_output: | |
| if len(result.tasks_output) >= 2: | |
| analyst_output = str(result.tasks_output[1]) | |
| merged_output.append("## Analyst Agent Results") | |
| merged_output.append("") | |
| merged_output.append(analyst_output) | |
| merged_output.append("") | |
| merged_output.append("---") | |
| merged_output.append("") | |
| # Get storyteller result (third task) | |
| if hasattr(result, 'tasks_output') and result.tasks_output: | |
| if len(result.tasks_output) >= 3: | |
| storyteller_output = str(result.tasks_output[2]) | |
| merged_output.append("## Storyteller Agent Results") | |
| merged_output.append("") | |
| merged_output.append(storyteller_output) | |
| merged_output.append("") | |
| # If we couldn't extract from tasks_output, use the full result | |
| if not merged_output: | |
| merged_output.append("## Complete Analysis Results") | |
| merged_output.append("") | |
| merged_output.append(str(result)) | |
| return "\n".join(merged_output), stored_engineer_result | |
| except Exception as e: | |
| error_trace = traceback.format_exc() | |
| error_msg = f"Error: {str(e)}\n\nTraceback:\n{error_trace}" | |
| print(error_msg) | |
| return error_msg, engineer_result or "" | |
| def process_question_only(file, user_query: str) -> str: | |
| """ | |
| Process a specific user question using only the Analyst agent (no Engineer, no Storyteller). | |
| Used for the "Analyze with Question" button. | |
| Args: | |
| file: Uploaded file object | |
| user_query: The user's specific analysis question | |
| Returns: | |
| str: Analyst results only | |
| """ | |
| if file is None: | |
| return "Please upload a CSV file." | |
| if not user_query or not user_query.strip(): | |
| return "Please enter a question." | |
| try: | |
| # Get file path | |
| file_path = file.name if hasattr(file, 'name') else str(file) | |
| csv_path = file_path | |
| # Run only analyst | |
| crew = create_analyst_only_crew(user_query.strip(), csv_path) | |
| result = crew.kickoff() | |
| # Get analyst result | |
| if hasattr(result, 'tasks_output') and result.tasks_output: | |
| if len(result.tasks_output) >= 1: | |
| analyst_output = str(result.tasks_output[0]) | |
| return analyst_output | |
| # Fallback to full result | |
| return str(result) | |
| except Exception as e: | |
| error_trace = traceback.format_exc() | |
| error_msg = f"Error: {str(e)}\n\nTraceback:\n{error_trace}" | |
| print(error_msg) | |
| return error_msg | |
| def create_app(): | |
| """Create and return the Gradio interface.""" | |
| with gr.Blocks(title="NBA Stats Analysis with CrewAI", theme=gr.themes.Soft()) as app: | |
| gr.Markdown(""" | |
| # NBA Stats Analysis with CrewAI | |
| Upload your NBA statistics CSV file to get comprehensive analysis with engaging storylines. | |
| **How it works:** | |
| - **Engineer Agent**: Examines and validates your dataset | |
| - **Analyst Agent**: Performs deep analysis (general or based on your question) | |
| - **Storyteller Agent**: Creates headlines and compelling storylines | |
| All agents work in parallel and results are merged for you! | |
| """) | |
| # Store engineer result in state | |
| engineer_state = gr.State(value="") | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| file_input = gr.File( | |
| label="Upload CSV File", | |
| file_types=[".csv"], | |
| type="filepath" | |
| ) | |
| analyze_btn = gr.Button( | |
| "Analyze Dataset", | |
| variant="primary", | |
| size="lg", | |
| visible=False | |
| ) | |
| gr.Markdown("### Ask a Specific Question") | |
| query_input = gr.Textbox( | |
| label="Your Analysis Question", | |
| placeholder="e.g., 'Who are the top 5 three-point shooters?' or 'Analyze the best players by assists'", | |
| lines=2 | |
| ) | |
| question_output = gr.Markdown( | |
| value="", | |
| label="Answer", | |
| visible=False | |
| ) | |
| query_btn = gr.Button( | |
| "Analyze with Question", | |
| variant="secondary", | |
| size="lg" | |
| ) | |
| with gr.Row(): | |
| with gr.Column(): | |
| status_output = gr.Markdown( | |
| value="", | |
| label="Agent Status", | |
| visible=False | |
| ) | |
| with gr.Row(): | |
| with gr.Column(): | |
| merged_output = gr.Markdown( | |
| value="**Ready to analyze!** Upload a CSV file above, then click 'Analyze Dataset' to get started.", | |
| label="Full Analysis Results" | |
| ) | |
| def show_loading_animation(is_question: bool = False): | |
| """Show loading animation while processing.""" | |
| if is_question: | |
| return """## Analysis in Progress... | |
| <div style="text-align: center; padding: 20px;"> | |
| <div style="font-size: 18px; margin-bottom: 15px;"> | |
| <strong>Analyzing your question...</strong> | |
| </div> | |
| <div style="display: flex; justify-content: center; max-width: 600px; margin: 0 auto;"> | |
| <div style="text-align: center; margin: 10px;"> | |
| <div style="font-size: 14px; font-weight: bold;">Analyst Agent</div> | |
| <div style="font-size: 12px; color: #666; margin-top: 5px;">Processing query...</div> | |
| </div> | |
| </div> | |
| <div style="margin-top: 25px; font-size: 14px; color: #888;"> | |
| This may take a moment... Please wait while the agent processes your question. | |
| </div> | |
| </div>""" | |
| else: | |
| return """## Analysis in Progress... | |
| <div style="text-align: center; padding: 20px;"> | |
| <div style="font-size: 18px; margin-bottom: 15px;"> | |
| <strong>Agents are working in parallel...</strong> | |
| </div> | |
| <div style="display: flex; justify-content: space-around; max-width: 600px; margin: 0 auto; flex-wrap: wrap;"> | |
| <div style="text-align: center; margin: 10px;"> | |
| <div style="font-size: 14px; font-weight: bold;">Engineer Agent</div> | |
| <div style="font-size: 12px; color: #666; margin-top: 5px;">Examining dataset...</div> | |
| </div> | |
| <div style="text-align: center; margin: 10px;"> | |
| <div style="font-size: 14px; font-weight: bold;">Analyst Agent</div> | |
| <div style="font-size: 12px; color: #666; margin-top: 5px;">Analyzing data...</div> | |
| </div> | |
| <div style="text-align: center; margin: 10px;"> | |
| <div style="font-size: 14px; font-weight: bold;">Storyteller Agent</div> | |
| <div style="font-size: 12px; color: #666; margin-top: 5px;">Creating storylines...</div> | |
| </div> | |
| </div> | |
| <div style="margin-top: 25px; font-size: 14px; color: #888;"> | |
| This may take a moment... Please wait while the agents process your data. | |
| </div> | |
| </div>""" | |
| def on_file_upload(file): | |
| """Handle file upload - show analyze button and reset state.""" | |
| if file is not None: | |
| return gr.update(visible=True), "" | |
| return gr.update(visible=False), "" | |
| def start_full_analysis(file, engineer_result: str = ""): | |
| """Start full analysis and show loading animation.""" | |
| loading_msg = show_loading_animation(is_question=False) | |
| return gr.update(visible=True, value=loading_msg), gr.update(value="") | |
| def complete_full_analysis(file, engineer_result: str = ""): | |
| """Complete full analysis and return results.""" | |
| result, new_engineer_result = process_file_and_analyze(file, "", engineer_result) | |
| if result.startswith("Error:") or result.startswith("Please upload"): | |
| result = f"### {result}" | |
| return result, gr.update(visible=False), new_engineer_result | |
| def start_question_analysis(file, user_query: str = ""): | |
| """Start question analysis and show loading animation.""" | |
| loading_msg = show_loading_animation(is_question=True) | |
| return gr.update(visible=True, value=loading_msg), gr.update(visible=True, value="") | |
| def complete_question_analysis(file, user_query: str = ""): | |
| """Complete question analysis and return results.""" | |
| result = process_question_only(file, user_query) | |
| if result.startswith("Error:") or result.startswith("Please"): | |
| result = f"### {result}" | |
| else: | |
| # Format the answer in a highlighted box | |
| result = f"""<div style="background-color: #f0f7ff; border: 2px solid #4a90e2; border-radius: 8px; padding: 15px; margin: 10px 0;"> | |
| {result} | |
| </div>""" | |
| return result, gr.update(visible=False) | |
| # When file is uploaded, show analyze button and reset engineer state | |
| file_input.change( | |
| fn=on_file_upload, | |
| inputs=[file_input], | |
| outputs=[analyze_btn, engineer_state] | |
| ) | |
| # Analyze button - runs general analysis (no query needed) | |
| analyze_btn.click( | |
| fn=start_full_analysis, | |
| inputs=[file_input, engineer_state], | |
| outputs=[status_output, merged_output] | |
| ).then( | |
| fn=complete_full_analysis, | |
| inputs=[file_input, engineer_state], | |
| outputs=[merged_output, status_output, engineer_state] | |
| ) | |
| # Query button - runs analysis with user's question (only Analyst) | |
| query_btn.click( | |
| fn=start_question_analysis, | |
| inputs=[file_input, query_input], | |
| outputs=[status_output, question_output] | |
| ).then( | |
| fn=complete_question_analysis, | |
| inputs=[file_input, query_input], | |
| outputs=[question_output, status_output] | |
| ) | |
| # Allow Enter key to submit query | |
| query_input.submit( | |
| fn=start_question_analysis, | |
| inputs=[file_input, query_input], | |
| outputs=[status_output, question_output] | |
| ).then( | |
| fn=complete_question_analysis, | |
| inputs=[file_input, query_input], | |
| outputs=[question_output, status_output] | |
| ) | |
| return app | |
| if __name__ == "__main__": | |
| try: | |
| print("Creating Gradio app...") | |
| app = create_app() | |
| print("Launching Gradio app...") | |
| # For Hugging Face Spaces, use default launch settings | |
| # Spaces will automatically handle server_name and port | |
| app.launch( | |
| show_error=True | |
| ) | |
| except Exception as e: | |
| print(f"Error launching app: {e}") | |
| traceback.print_exc() | |
| raise | |