| import gradio as gr |
| import pandas as pd |
| import aiohttp |
| import asyncio |
| import json |
| import io |
| import os |
| from typing import Optional, Tuple |
|
|
| class DataAnalyzer: |
| def __init__(self): |
| self.api_base_url = "https://llm.chutes.ai/v1/chat/completions" |
| |
| async def analyze_with_chutes(self, api_token: str, data_summary: str, user_question: str = None) -> str: |
| """Send data to Chutes API for analysis""" |
| headers = { |
| "Authorization": f"Bearer {api_token}", |
| "Content-Type": "application/json" |
| } |
| |
| # Create the prompt based on whether it's initial analysis or follow-up question |
| if user_question: |
| prompt = f"""Based on this dataset summary: |
| {data_summary} |
| User question: {user_question} |
| Please provide a detailed answer based on the data.""" |
| else: |
| prompt = f"""Analyze the following dataset and provide comprehensive insights: |
| {data_summary} |
| Please provide: |
| 1. Key statistical insights |
| 2. Notable patterns or trends |
| 3. Data quality observations |
| 4. Business recommendations |
| 5. Potential areas for further analysis |
| Keep the analysis clear, actionable, and data-driven.""" |
|
|
| body = { |
| "model": "openai/gpt-oss-20b", |
| "messages": [ |
| { |
| "role": "user", |
| "content": prompt |
| } |
| ], |
| "stream": True, |
| "max_tokens": 2048, |
| "temperature": 0.3 # Lower temperature for more consistent analysis |
| } |
| |
| try: |
| async with aiohttp.ClientSession() as session: |
| async with session.post(self.api_base_url, headers=headers, json=body) as response: |
| if response.status != 200: |
| return f"Error: API request failed with status {response.status}" |
| |
| full_response = "" |
| async for line in response.content: |
| line = line.decode("utf-8").strip() |
| if line.startswith("data: "): |
| data = line[6:] |
| if data == "[DONE]": |
| break |
| try: |
| chunk_data = json.loads(data) |
| if "choices" in chunk_data and len(chunk_data["choices"]) > 0: |
| delta = chunk_data["choices"][0].get("delta", {}) |
| content = delta.get("content", "") |
| if content: |
| full_response += content |
| except json.JSONDecodeError: |
| continue |
| |
| return full_response if full_response else "No response received from the model." |
| |
| except Exception as e: |
| return f"Error connecting to Chutes API: {str(e)}" |
|
|
| def process_file(self, file_path: str) -> Tuple[pd.DataFrame, str]: |
| """Process uploaded CSV or Excel file""" |
| try: |
| file_extension = os.path.splitext(file_path)[1].lower() |
| |
| if file_extension == '.csv': |
| df = pd.read_csv(file_path) |
| elif file_extension in ['.xlsx', '.xls']: |
| df = pd.read_excel(file_path) |
| else: |
| raise ValueError("Unsupported file format. Please upload CSV or Excel files.") |
| |
| # Generate comprehensive data summary |
| summary = self.generate_data_summary(df) |
| return df, summary |
| |
| except Exception as e: |
| raise Exception(f"Error processing file: {str(e)}") |
| |
| def generate_data_summary(self, df: pd.DataFrame) -> str: |
| """Generate a comprehensive summary of the dataset""" |
| summary = [] |
| |
| # Basic info |
| summary.append(f"Dataset Overview:") |
| summary.append(f"- Shape: {df.shape[0]} rows × {df.shape[1]} columns") |
| summary.append(f"- Total cells: {df.shape[0] * df.shape[1]:,}") |
| |
| # Column information |
| summary.append(f"\nColumn Information:") |
| for i, (col, dtype) in enumerate(df.dtypes.items()): |
| null_count = df[col].isnull().sum() |
| null_pct = (null_count / len(df)) * 100 |
| summary.append(f"- {col} ({dtype}): {null_count} nulls ({null_pct:.1f}%)") |
| |
| # Numerical columns statistics |
| numeric_cols = df.select_dtypes(include=['number']).columns |
| if len(numeric_cols) > 0: |
| summary.append(f"\nNumerical Columns Summary:") |
| for col in numeric_cols: |
| stats = df[col].describe() |
| summary.append(f"- {col}: Mean={stats['mean']:.2f}, Std={stats['std']:.2f}, Range=[{stats['min']:.2f}, {stats['max']:.2f}]") |
| |
| # Categorical columns |
| categorical_cols = df.select_dtypes(include=['object', 'category']).columns |
| if len(categorical_cols) > 0: |
| summary.append(f"\nCategorical Columns Summary:") |
| for col in categorical_cols: |
| unique_count = df[col].nunique() |
| most_common = df[col].mode().iloc[0] if len(df[col].mode()) > 0 else "N/A" |
| summary.append(f"- {col}: {unique_count} unique values, Most common: '{most_common}'") |
| |
| # Sample data |
| summary.append(f"\nFirst 5 rows preview:") |
| summary.append(df.head().to_string()) |
| |
| return "\n".join(summary) |
|
|
| # Initialize the analyzer |
| analyzer = DataAnalyzer() |
|
|
| async def analyze_data(file, api_key, user_question=""): |
| """Main function to analyze uploaded data""" |
| if not file: |
| return "Please upload a CSV or Excel file.", "", "" |
| |
| if not api_key: |
| return "Please enter your Chutes API key.", "", "" |
| |
| try: |
| # Process the uploaded file |
| df, data_summary = analyzer.process_file(file.name) |
| |
| # Get AI analysis |
| ai_analysis = await analyzer.analyze_with_chutes(api_key, data_summary, user_question) |
| |
| # Format the complete response |
| response = f"""## 📊 Data Analysis Complete! |
| ### 📈 Dataset Overview: |
| {data_summary} |
| ### 🤖 AI Insights & Recommendations: |
| {ai_analysis} |
| """ |
| |
| return response, data_summary, df.head(10).to_html() |
| |
| except Exception as e: |
| return f"Error: {str(e)}", "", "" |
|
|
| def sync_analyze_data(file, api_key, user_question=""): |
| """Synchronous wrapper for the async analyze function""" |
| return asyncio.run(analyze_data(file, api_key, user_question)) |
|
|
| # Create the Gradio interface |
| with gr.Blocks(title="📊 Smart Data Analyzer", theme=gr.themes.Ocean()) as app: |
| gr.Markdown(""" |
| # 📊 Smart Data Analyzer |
| ### Upload your CSV/Excel file and get instant AI-powered insights using OpenAI's gpt-oss-20b model via Chutes! |
| """) |
| |
| with gr.Row(): |
| with gr.Column(scale=1): |
| # File upload |
| file_input = gr.File( |
| label="📁 Upload CSV or Excel File", |
| file_types=[".csv", ".xlsx", ".xls"], |
| file_count="single" |
| ) |
| |
| # API key input |
| api_key_input = gr.Textbox( |
| label="🔑 Chutes API Key", |
| placeholder="Enter your Chutes API token here...", |
| type="password", |
| lines=1 |
| ) |
| |
| # Optional question input |
| question_input = gr.Textbox( |
| label="❓ Ask a Specific Question (Optional)", |
| placeholder="e.g., What are the sales trends? Which region performs best?", |
| lines=2 |
| ) |
| |
| # Analyze button |
| analyze_btn = gr.Button("🚀 Analyze Data", variant="primary", size="lg") |
| |
| with gr.Column(scale=2): |
| # Results display |
| analysis_output = gr.Markdown( |
| label="📋 Analysis Results", |
| value="Upload a file and click 'Analyze Data' to see insights..." |
| ) |
| |
| # Additional outputs (hidden by default) |
| with gr.Accordion("📊 Data Preview", open=False): |
| data_preview = gr.HTML(label="First 10 Rows") |
| |
| with gr.Accordion("🔍 Raw Data Summary", open=False): |
| raw_summary = gr.Textbox(label="Dataset Summary", lines=10) |
| |
| # Event handlers |
| analyze_btn.click( |
| fn=sync_analyze_data, |
| inputs=[file_input, api_key_input, question_input], |
| outputs=[analysis_output, raw_summary, data_preview] |
| ) |
| |
| # Example section |
| gr.Markdown(""" |
| ### 💡 Tips for Best Results: |
| - **File Size**: Keep files under 10MB for fastest processing |
| - **API Key**: Get your free Chutes API key from [chutes.ai](https://chutes.ai) |
| - **Questions**: Be specific! Ask about trends, patterns, outliers, or recommendations |
| - **Formats**: Supports CSV, XLSX, and XLS files |
| |
| ### 🎯 Example Questions to Ask: |
| - "What are the key trends in this sales data?" |
| - "Which products are underperforming?" |
| - "Are there any seasonal patterns?" |
| - "What recommendations do you have based on this data?" |
| """) |
|
|
| # Launch the application |
| if __name__ == "__main__": |
| app.launch( |
| share=True |
| ) |
