Update app.py

app.py

@@ -1,332 +1,94 @@
-from smolagents import tool
-import requests
-import json
+from smolagents import CodeAgent, tool
 import datetime
-import os
-import base64
-from typing import List, Optional, Dict, Any
-import json
-import io
-
-
-@tool
-def web_scrape(url: str) -> str:
-    """Scrapes the content from a specified URL.
-    
-    Args:
-        url: The URL to scrape content from.
-    """
-    try:
-        response = requests.get(url, headers={
-            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
-        })
-        response.raise_for_status()
-        return response.text
-    except Exception as e:
-        return f"Error scraping {url}: {str(e)}"
-
+import pytz
+import yaml
+from tools.final_answer import FinalAnswerTool
+from Gradio_UI import GradioUI
 
+# Define a simple tool that doesn't rely on any external libraries
 @tool
-def extract_structured_data(text: str, schema: str) -> str:
-    """Extracts structured data from text based on a provided schema.
+def text_analyzer(text: str) -> str:
+    """Analyzes text and returns statistics about it.
     
     Args:
-        text: The text to extract data from.
-        schema: JSON schema describing the data structure to extract.
+        text: The text to analyze.
     """
     try:
-        # In a real implementation, you might use regex, NLP, or ML models
-        # This is a placeholder for demonstrating the concept
-        return f"Extracted structured data according to schema: {schema}"
-    except Exception as e:
-        return f"Error extracting structured data: {str(e)}"
-
-
-@tool
-def data_visualization(data: str, chart_type: str, title: str = "Data Visualization") -> str:
-    """Creates a textual representation of data for visualization.
-    
-    Args:
-        data: JSON string with the data to visualize.
-        chart_type: Type of representation to create (summary, table, ascii).
-        title: Title for the visualization.
-    """
-    try:
-        # Parse the input data
-        data_parsed = json.loads(data)
+        # Simple word count
+        words = text.split()
+        word_count = len(words)
         
-        # Create appropriate representation
-        result = f"=== {title} ===\n\n"
+        # Character count
+        char_count = len(text)
         
-        if chart_type.lower() == 'summary':
-            # Provide summary statistics
-            if isinstance(data_parsed, list):
-                result += f"Number of items: {len(data_parsed)}\n"
-                if len(data_parsed) > 0 and isinstance(data_parsed[0], dict):
-                    keys = data_parsed[0].keys()
-                    result += f"Fields: {', '.join(keys)}\n\n"
-                    
-                    # Try to get numeric values for min/max/avg
-                    for key in keys:
-                        try:
-                            values = [item[key] for item in data_parsed if isinstance(item[key], (int, float))]
-                            if values:
-                                result += f"{key}:\n"
-                                result += f"  Min: {min(values)}\n"
-                                result += f"  Max: {max(values)}\n"
-                                result += f"  Avg: {sum(values)/len(values):.2f}\n\n"
-                        except (KeyError, TypeError):
-                            pass
-            else:
-                result += f"Data summary: {str(data_parsed)[:100]}...\n"
-                
-        elif chart_type.lower() == 'table':
-            # Create ASCII table
-            if isinstance(data_parsed, list) and len(data_parsed) > 0 and isinstance(data_parsed[0], dict):
-                # Get headers
-                headers = list(data_parsed[0].keys())
-                
-                # Calculate column widths
-                col_widths = [len(h) for h in headers]
-                for item in data_parsed:
-                    for i, h in enumerate(headers):
-                        if h in item:
-                            col_widths[i] = max(col_widths[i], len(str(item[h])))
-                
-                # Create header row
-                header_row = " | ".join(h.ljust(col_widths[i]) for i, h in enumerate(headers))
-                separator = "-+-".join("-" * w for w in col_widths)
-                
-                result += header_row + "\n"
-                result += separator + "\n"
-                
-                # Create data rows
-                for item in data_parsed[:10]:  # Limit to 10 rows for readability
-                    row = " | ".join(str(item.get(h, "")).ljust(col_widths[i]) for i, h in enumerate(headers))
-                    result += row + "\n"
-                
-                if len(data_parsed) > 10:
-                    result += f"\n... and {len(data_parsed) - 10} more rows"
-            else:
-                result += "Data is not in a format suitable for table display"
-                
-        elif chart_type.lower() == 'ascii':
-            # Create simple ASCII chart
-            if isinstance(data_parsed, list):
-                # Try to extract x and y values
-                x_values = []
-                y_values = []
-                
-                # Attempt to detect data structure
-                if len(data_parsed) > 0:
-                    if isinstance(data_parsed[0], dict) and len(data_parsed[0]) >= 2:
-                        # Use first two keys as x and y
-                        keys = list(data_parsed[0].keys())
-                        x_key, y_key = keys[0], keys[1]
-                        
-                        for item in data_parsed:
-                            try:
-                                x_values.append(str(item[x_key]))
-                                y_values.append(float(item[y_key]))
-                            except (KeyError, ValueError, TypeError):
-                                continue
-                    elif isinstance(data_parsed[0], (list, tuple)) and len(data_parsed[0]) >= 2:
-                        # Use first two elements as x and y
-                        for item in data_parsed:
-                            try:
-                                x_values.append(str(item[0]))
-                                y_values.append(float(item[1]))
-                            except (IndexError, ValueError, TypeError):
-                                continue
-                
-                if x_values and y_values:
-                    # Create a simple bar chart
-                    max_y = max(y_values)
-                    scale = 20 / max_y if max_y > 0 else 1
-                    
-                    result += f"Chart scale: Each * represents {1/scale:.2f} units\n\n"
-                    
-                    for i, (x, y) in enumerate(zip(x_values, y_values)):
-                        bar_length = int(y * scale)
-                        result += f"{x.ljust(10)}: {'*' * bar_length} ({y})\n"
-                else:
-                    result += "Could not extract plottable values from the data"
-            else:
-                result += "Data is not in a format suitable for ASCII chart display"
-        else:
-            return f"Unsupported visualization type: {chart_type}. Use 'summary', 'table', or 'ascii'."
+        # Unique words
+        unique_words = len(set(word.lower() for word in words))
         
-        return result
-    except Exception as e:
-        return f"Error creating visualization: {str(e)}"
-
-
-@tool
-def code_refactor(code: str, language: str, optimization: str) -> str:
-    """Refactors code based on specified optimization criteria.
-    
-    Args:
-        code: The source code to refactor.
-        language: Programming language of the code.
-        optimization: Type of optimization to perform (performance, readability, security).
-    """
-    try:
-        # In a real implementation, you'd use language-specific tools or ML models
-        # This is a placeholder for demonstrating the concept
-        if optimization.lower() == 'performance':
-            return f"Code refactored for performance: \n```{language}\n# Performance optimized\n{code}\n```"
-        elif optimization.lower() == 'readability':
-            return f"Code refactored for readability: \n```{language}\n# Readability optimized\n{code}\n```"
-        elif optimization.lower() == 'security':
-            return f"Code refactored for security: \n```{language}\n# Security optimized\n{code}\n```"
-        else:
-            return f"Unsupported optimization type: {optimization}"
-    except Exception as e:
-        return f"Error refactoring code: {str(e)}"
-
-
-@tool
-def api_interaction(endpoint: str, method: str = "GET", params: Optional[str] = None, headers: Optional[str] = None) -> str:
-    """Interacts with an API endpoint.
-    
-    Args:
-        endpoint: The API endpoint URL.
-        method: HTTP method (GET, POST, PUT, DELETE).
-        params: JSON string of parameters or data to send.
-        headers: JSON string of headers to include.
-    """
-    try:
-        # Parse headers and params if provided
-        headers_dict = json.loads(headers) if headers else {}
+        # Average word length
+        avg_word_length = sum(len(word) for word in words) / max(1, word_count)
         
-        if method.upper() == "GET":
-            params_dict = json.loads(params) if params else {}
-            response = requests.get(endpoint, params=params_dict, headers=headers_dict)
-        elif method.upper() == "POST":
-            data_dict = json.loads(params) if params else {}
-            response = requests.post(endpoint, json=data_dict, headers=headers_dict)
-        elif method.upper() == "PUT":
-            data_dict = json.loads(params) if params else {}
-            response = requests.put(endpoint, json=data_dict, headers=headers_dict)
-        elif method.upper() == "DELETE":
-            response = requests.delete(endpoint, headers=headers_dict)
-        else:
-            return f"Unsupported HTTP method: {method}"
+        # Most common words (top 5)
+        word_freq = {}
+        for word in words:
+            word_lower = word.lower()
+            word_freq[word_lower] = word_freq.get(word_lower, 0) + 1
         
-        response.raise_for_status()
+        common_words = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)[:5]
+        common_words_str = ", ".join(f"{word} ({count})" for word, count in common_words)
         
-        # Try to return JSON if possible, otherwise return text
-        try:
-            return json.dumps(response.json(), indent=2)
-        except:
-            return response.text
+        return f"""Text Analysis Results:
+- Word count: {word_count}
+- Character count: {char_count}
+- Unique words: {unique_words}
+- Average word length: {avg_word_length:.2f}
+- Most common words: {common_words_str}
+"""
     except Exception as e:
-        return f"Error interacting with API {endpoint}: {str(e)}"
-
+        return f"Error analyzing text: {str(e)}"
 
 @tool
-def natural_language_query(database_description: str, query: str) -> str:
-    """Translates a natural language query to structured data operations.
+def get_current_time_in_timezone(timezone: str) -> str:
+    """A tool that fetches the current local time in a specified timezone.
     
     Args:
-        database_description: Description of the database schema.
-        query: Natural language query about the data.
+        timezone: A string representing a valid timezone (e.g., 'America/New_York').
     """
     try:
-        # In a real implementation, you'd use NLP to SQL or similar technology
-        # This is a placeholder for demonstrating the concept
-        return f"Query translated and executed. Results for: {query}"
+        # Create timezone object
+        tz = pytz.timezone(timezone)
+        # Get current time in that timezone
+        local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
+        return f"The current local time in {timezone} is: {local_time}"
     except Exception as e:
-        return f"Error processing natural language query: {str(e)}"
-
+        return f"Error fetching time for timezone '{timezone}': {str(e)}"
 
-@tool
-def file_operations(operation: str, file_path: str, content: Optional[str] = None) -> str:
-    """Performs operations on files.
-    
-    Args:
-        operation: The operation to perform (read, write, append, list).
-        file_path: Path to the file or directory.
-        content: Content to write or append (only for write/append operations).
-    """
-    try:
-        if operation.lower() == 'read':
-            with open(file_path, 'r') as file:
-                return file.read()
-        elif operation.lower() == 'write':
-            if content is None:
-                return "Content must be provided for write operation"
-            with open(file_path, 'w') as file:
-                file.write(content)
-            return f"Content written to {file_path}"
-        elif operation.lower() == 'append':
-            if content is None:
-                return "Content must be provided for append operation"
-            with open(file_path, 'a') as file:
-                file.write(content)
-            return f"Content appended to {file_path}"
-        elif operation.lower() == 'list':
-            if os.path.isdir(file_path):
-                return str(os.listdir(file_path))
-            else:
-                return f"{file_path} is not a directory"
-        else:
-            return f"Unsupported file operation: {operation}"
-    except Exception as e:
-        return f"Error performing file operation: {str(e)}"
+# Set up the agent with minimal tools
+final_answer = FinalAnswerTool()
 
+with open("prompts.yaml", 'r') as stream:
+    prompt_templates = yaml.safe_load(stream)
 
-@tool
-def semantic_search(corpus: str, query: str, top_k: int = 3) -> str:
-    """Performs semantic search on a corpus of text.
+from smolagents import HfApiModel
     
-    Args:
-        corpus: The text corpus to search within (could be a large text or list of documents).
-        query: The search query.
-        top_k: Number of top results to return.
-    """
-    try:
-        # In a real implementation, you'd use embedding models and vector similarity
-        # This is a placeholder for demonstrating the concept
-        results = [
-            {"text": f"Result {i} for query: {query}", "score": (top_k - i) / top_k}
-            for i in range(1, top_k + 1)
-        ]
-        return json.dumps(results, indent=2)
-    except Exception as e:
-        return f"Error performing semantic search: {str(e)}"
-
-
-@tool
-def weather_forecast(location: str) -> str:
-    """Fetches weather forecast for a specified location.
-    
-    Args:
-        location: The location to get weather forecast for (city name or coordinates).
-    """
-    try:
-        # In a real implementation, you'd connect to a weather API
-        # This is a placeholder for demonstrating the concept
-        return f"Weather forecast for {location}: Sunny with a chance of AI"
-    except Exception as e:
-        return f"Error fetching weather forecast: {str(e)}"
-
-
-@tool 
-def task_scheduler(task: str, schedule_time: str, priority: int = 1) -> str:
-    """Schedules a task to be performed at a specified time.
-    
-    Args:
-        task: Description of the task to be scheduled.
-        schedule_time: Time to schedule the task (ISO format).
-        priority: Priority level of the task (1-5, where 1 is highest).
-    """
-    try:
-        # Parse the schedule time
-        schedule_datetime = datetime.datetime.fromisoformat(schedule_time)
-        
-        # In a real implementation, you'd connect to a scheduling system
-        # This is a placeholder for demonstrating the concept
-        return f"Task '{task}' scheduled for {schedule_datetime} with priority {priority}"
-    except Exception as e:
-        return f"Error scheduling task: {str(e)}"
+model = HfApiModel(
+    max_tokens=2096,
+    temperature=0.5,
+    model_id='Qwen/Qwen2.5-Coder-32B-Instruct',
+    custom_role_conversions=None,
+)
+
+# Create agent with minimal tools
+agent = CodeAgent(
+    model=model,
+    tools=[text_analyzer, get_current_time_in_timezone, final_answer],
+    max_steps=6,
+    verbosity_level=1,
+    grammar=None,
+    planning_interval=None,
+    name=None,
+    description=None,
+    prompt_templates=prompt_templates
+)
+
+# Launch the Gradio UI
+GradioUI(agent).launch()