Update app.py

app.py

@@ -1 +1,1048 @@
-from file_management import FileManagementAgent
+import logging
+import os
+import sys
+from pathlib import Path
+import json
+import io
+import uuid
+import traceback
+from typing import Dict, List, Any, Tuple, Optional
+from dataclasses import dataclass
+
+# Set UTF-8 encoding for Windows
+if sys.platform == 'win32':
+    os.environ["PYTHONIOENCODING"] = "utf-8"
+
+import gradio as gr
+import numpy as np
+import pandas as pd
+import matplotlib.pyplot as plt
+from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM
+from sklearn.datasets import load_iris
+import cv2
+from PIL import Image
+
+# Additional libraries for web research & scraping
+import wikipedia
+import requests
+from bs4 import BeautifulSoup
+
+# Configure logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+# ---------------------------
+# Agent Context & Memory System
+# ---------------------------
+@dataclass
+class AgentMemory:
+    short_term: List[Dict[str, Any]] = None
+    long_term: Dict[str, Any] = None
+
+    def __post_init__(self):
+        if self.short_term is None:
+            self.short_term = []
+        if self.long_term is None:
+            self.long_term = {}
+
+    def add_short_term(self, data: Dict[str, Any]) -> None:
+        self.short_term.append(data)
+        if len(self.short_term) > 10:
+            self.short_term.pop(0)
+
+    def add_long_term(self, key: str, value: Any) -> None:
+        self.long_term[key] = value
+
+    def get_recent_context(self, n: int = 3) -> List[Dict[str, Any]]:
+        return self.short_term[-n:] if len(self.short_term) >= n else self.short_term
+
+    def search_long_term(self, query: str) -> List[Tuple[str, Any]]:
+        results = []
+        for key, value in self.long_term.items():
+            if query.lower() in key.lower():
+                results.append((key, value))
+        return results
+
+# ---------------------------
+# Agent Hub
+# ---------------------------
+class AgentHub:
+    def __init__(self):
+        self.agents = {}
+        self.global_memory = AgentMemory()
+        self.session_id = str(uuid.uuid4())
+
+        try:
+            self.tokenizer = AutoTokenizer.from_pretrained("distilgpt2")
+            self.model = AutoModelForCausalLM.from_pretrained("distilgpt2")
+            self.generator = pipeline("text-generation", model=self.model, tokenizer=self.tokenizer)
+            logger.info("Initialized text generation pipeline with distilgpt2")
+        except Exception as e:
+            logger.error(f"Failed to initialize text generation: {e}")
+            self.generator = None
+
+        try:
+            self.summarizer = pipeline("summarization", model="facebook/bart-large-cnn")
+            logger.info("Initialized summarization pipeline")
+        except Exception as e:
+            logger.error(f"Failed to initialize summarizer: {e}")
+            self.summarizer = None
+
+    def register_agent(self, agent_id: str, agent_instance) -> None:
+        self.agents[agent_id] = agent_instance
+        logger.info(f"Registered agent: {agent_id}")
+
+    def get_agent(self, agent_id: str):
+        return self.agents.get(agent_id)
+
+    def broadcast(self, message: Dict[str, Any], exclude: Optional[List[str]] = None) -> Dict[str, List[Dict]]:
+        exclude = exclude or []
+        responses = {}
+        for agent_id, agent in self.agents.items():
+            if agent_id not in exclude:
+                try:
+                    response = agent.process_message(message)
+                    responses[agent_id] = response
+                except Exception as e:
+                    logger.error(f"Error in agent {agent_id}: {e}")
+                    responses[agent_id] = {"error": str(e)}
+        return responses
+
+    def chain_of_thought(self, initial_task: str, agent_sequence: List[str]) -> Dict[str, Any]:
+        results = {"final_output": None, "chain_outputs": [], "errors": []}
+        current_input = initial_task
+        for agent_id in agent_sequence:
+            agent = self.get_agent(agent_id)
+            if not agent:
+                error = f"Agent {agent_id} not found"
+                results["errors"].append(error)
+                logger.error(error)
+                continue
+            try:
+                output = agent.process_task(current_input)
+                step_result = {"agent": agent_id, "input": current_input, "output": output}
+                results["chain_outputs"].append(step_result)
+                if isinstance(output, dict) and "text" in output:
+                    current_input = output["text"]
+                elif isinstance(output, str):
+                    current_input = output
+                else:
+                    current_input = f"Result from {agent_id}: {type(output).__name__} object"
+            except Exception as e:
+                error = f"Error in agent {agent_id}: {str(e)}\n{traceback.format_exc()}"
+                results["errors"].append(error)
+                logger.error(error)
+        if results["chain_outputs"]:
+            last_output = results["chain_outputs"][-1]["output"]
+            results["final_output"] = last_output if isinstance(last_output, dict) else {"text": str(last_output)}
+        return results
+
+# ---------------------------
+# Intelligent Agent Base Class
+# ---------------------------
+class IntelligentAgent:
+    def __init__(self, agent_id: str, hub: AgentHub):
+        self.agent_id = agent_id
+        self.hub = hub
+        self.memory = AgentMemory()
+        logger.info(f"Initialized agent: {agent_id}")
+
+    def process_task(self, task: Any) -> Any:
+        raise NotImplementedError("Subclasses must implement process_task")
+
+    def process_message(self, message: Dict[str, Any]) -> Dict[str, Any]:
+        logger.info(f"Agent {self.agent_id} received message: {message}")
+        self.memory.add_short_term({"timestamp": pd.Timestamp.now(), "message": message})
+        return {"sender": self.agent_id, "received": True, "action": "acknowledge"}
+
+    def request_assistance(self, target_agent_id: str, data: Dict[str, Any]) -> Dict[str, Any]:
+        target_agent = self.hub.get_agent(target_agent_id)
+        if not target_agent:
+            logger.error(f"Agent {self.agent_id} requested unknown agent: {target_agent_id}")
+            return {"error": f"Agent {target_agent_id} not found"}
+        request = {"sender": self.agent_id, "type": "assistance_request", "data": data}
+        return target_agent.process_message(request)
+
+    def evaluate_result(self, result: Any) -> Dict[str, Any]:
+        success = result is not None
+        confidence = 0.8 if success else 0.2
+        return {"success": success, "confidence": confidence, "timestamp": pd.Timestamp.now().isoformat()}
+
+# ---------------------------
+# Specialized Agent Implementations
+# ---------------------------
+class WebResearchAgent(IntelligentAgent):
+    def __init__(self, hub: AgentHub):
+        super().__init__("web_research", hub)
+
+    def process_task(self, task: str) -> Dict[str, Any]:
+        logger.info(f"WebResearchAgent processing: {task}")
+        search_term = task
+        if self.hub.summarizer:
+            try:
+                keywords = task.split()
+                if len(keywords) > 5:
+                    summary = self.hub.summarizer(task, max_length=20, min_length=5, do_sample=False)
+                    search_term = summary[0]['summary_text']
+                else:
+                    search_term = task
+            except Exception as e:
+                logger.error(f"Summarization error in WebResearchAgent: {e}")
+                search_term = task
+        try:
+            search_results = wikipedia.search(search_term)
+            if not search_results:
+                result = {"text": f"No Wikipedia pages found for '{task}'."}
+                self.memory.add_short_term({"task": task, "result": result, "success": False})
+                return result
+            page_title = None
+            summary_text = None
+            error_details = []
+            for candidate in search_results[:3]:
+                try:
+                    summary_text = wikipedia.summary(candidate, sentences=5)
+                    page_title = candidate
+                    break
+                except (wikipedia.exceptions.DisambiguationError, wikipedia.exceptions.PageError) as e:
+                    error_details.append(f"{candidate}: {str(e)}")
+                    continue
+            if not summary_text:
+                result = {"text": f"Failed to get Wikipedia summary for '{task}'. Errors: {'; '.join(error_details)}", "search_results": search_results}
+                self.memory.add_short_term({"task": task, "result": result, "success": False})
+                return result
+            self.memory.add_long_term(f"research:{search_term}", {"page_title": page_title, "summary": summary_text, "timestamp": pd.Timestamp.now().isoformat()})
+            result = {"text": f"Research on '{page_title}':\n{summary_text}", "page_title": page_title, "related_topics": search_results[:5], "source": "Wikipedia"}
+            self.memory.add_short_term({"task": task, "result": result, "success": True})
+            return result
+        except Exception as e:
+            error_msg = f"Error in web research: {str(e)}"
+            logger.error(error_msg)
+            result = {"text": error_msg, "error": str(e)}
+            self.memory.add_short_term({"task": task, "result": result, "success": False})
+            return result
+
+class WebScraperAgent(IntelligentAgent):
+    def __init__(self, hub: AgentHub):
+        super().__init__("web_scraper", hub)
+
+    def process_task(self, task: str) -> Dict[str, Any]:
+        logger.info(f"WebScraperAgent processing URL: {task}")
+        if not task.startswith(('http://', 'https://')):
+            return {"text": "Invalid URL format. Please provide a URL starting with http:// or https://"}
+        try:
+            headers = {'User-Agent': 'Mozilla/5.0'}
+            response = requests.get(task, headers=headers, timeout=10)
+            if response.status_code != 200:
+                result = {"text": f"Error: received status code {response.status_code} from {task}"}
+                self.memory.add_short_term({"url": task, "result": result, "success": False})
+                return result
+            soup = BeautifulSoup(response.text, 'html.parser')
+            title = soup.title.string.strip() if soup.title and soup.title.string else "No title found"
+            main_content = soup.find('main') or soup.find(id='content') or soup.find(class_='content')
+            paras = main_content.find_all('p') if main_content else soup.find_all('p')
+            content = "\n".join([p.get_text().strip() for p in paras if len(p.get_text().strip()) > 50])
+            if len(content) > 2000 and self.hub.summarizer:
+                chunks = [content[i:i+1000] for i in range(0, len(content), 1000)]
+                summarized_chunks = []
+                for chunk in chunks:
+                    summary = self.hub.summarizer(chunk, max_length=100, min_length=30, do_sample=False)
+                    summarized_chunks.append(summary[0]['summary_text'])
+                content = "\n".join(summarized_chunks)
+            elif len(content) > 2000:
+                content = content[:2000] + "... (content truncated)"
+            links = []
+            for a in soup.find_all('a', href=True):
+                href = a['href']
+                if href.startswith('http') and len(links) < 5:
+                    links.append({"url": href, "text": a.get_text().strip() or href})
+            result = {"text": f"Content from {task}:\n\nTitle: {title}\n\n{content}", "title": title, "raw_content": content, "links": links, "source_url": task}
+            self.memory.add_short_term({"url": task, "result": result, "success": True})
+            self.memory.add_long_term(f"scraped:{task}", {"title": title, "content_preview": content[:200], "timestamp": pd.Timestamp.now().isoformat()})
+            return result
+        except requests.RequestException as e:
+            error_msg = f"Request error for {task}: {str(e)}"
+            logger.error(error_msg)
+            return {"text": error_msg, "error": str(e)}
+        except Exception as e:
+            error_msg = f"Error scraping {task}: {str(e)}"
+            logger.error(error_msg)
+            return {"text": error_msg, "error": str(e)}
+
+class TextProcessingAgent(IntelligentAgent):
+    def __init__(self, hub: AgentHub):
+        super().__init__("text_processing", hub)
+
+    def process_task(self, task: str) -> Dict[str, Any]:
+        logger.info(f"TextProcessingAgent processing text ({len(task)} chars)")
+        if not task or len(task) < 10:
+            return {"text": "Text too short to process meaningfully."}
+        results = {}
+        words = task.split()
+        sentences = task.split('. ')
+        results["statistics"] = {
+            "character_count": len(task),
+            "word_count": len(words),
+            "estimated_sentences": len(sentences),
+            "average_word_length": sum(len(word) for word in words) / len(words) if words else 0
+        }
+        if len(task) > 5000:
+            chunk_size = 500
+            chunking_strategy = "character_blocks"
+        elif len(words) > 200:
+            chunk_size = 50
+            chunking_strategy = "word_blocks"
+        else:
+            chunk_size = 5
+            chunking_strategy = "sentence_blocks"
+        if chunking_strategy == "character_blocks":
+            chunks = [task[i:i+chunk_size] for i in range(0, len(task), chunk_size)]
+        elif chunking_strategy == "word_blocks":
+            chunks = [' '.join(words[i:i+chunk_size]) for i in range(0, len(words), chunk_size)]
+        else:
+            chunks = ['. '.join(sentences[i:i+chunk_size]) + '.' for i in range(0, len(sentences), chunk_size)]
+        results["chunks"] = chunks
+        results["chunking_strategy"] = chunking_strategy
+        if self.hub.summarizer and len(task) > 200:
+            try:
+                task_for_summary = task[:1000] if len(task) > 1000 else task
+                summary = self.hub.summarizer(task_for_summary, max_length=100, min_length=30, do_sample=False)
+                results["summary"] = summary[0]['summary_text']
+            except Exception as e:
+                logger.error(f"Summarization error: {e}")
+                results["summary_error"] = str(e)
+        stop_words = set(['the', 'a', 'an', 'and', 'in', 'on', 'at', 'to', 'for', 'of', 'with'])
+        word_freq = {}
+        for word in words:
+            w = word.lower().strip('.,!?:;()-"\'')
+            if w and w not in stop_words and len(w) > 1:
+                word_freq[w] = word_freq.get(w, 0) + 1
+        results["frequent_words"] = sorted(word_freq.items(), key=lambda x: x[1], reverse=True)[:10]
+        positive_words = set(['good', 'great', 'excellent', 'positive', 'happy', 'best', 'better', 'success'])
+        negative_words = set(['bad', 'worst', 'terrible', 'negative', 'sad', 'problem', 'fail', 'issue'])
+        pos_count = sum(1 for word in words if word.lower().strip('.,!?:;()-"\'') in positive_words)
+        neg_count = sum(1 for word in words if word.lower().strip('.,!?:;()-"\'') in negative_words)
+        sentiment = "possibly positive" if pos_count > neg_count and pos_count > 2 else ("possibly negative" if neg_count > pos_count and neg_count > 2 else "neutral or mixed")
+        results["basic_sentiment"] = {"assessment": sentiment, "positive_word_count": pos_count, "negative_word_count": neg_count}
+        self.memory.add_short_term({"task_preview": task[:100] + "..." if len(task) > 100 else task, "word_count": results["statistics"]["word_count"], "result": results})
+        text_response = (
+            f"Text Analysis Results:\n- {results['statistics']['word_count']} words, {results['statistics']['character_count']} characters\n"
+            f"- Split into {len(chunks)} chunks using {chunking_strategy}\n"
+        )
+        if "summary" in results:
+            text_response += f"\nSummary:\n{results['summary']}\n"
+        if results["frequent_words"]:
+            text_response += "\nMost frequent words:\n"
+            for word, count in results["frequent_words"][:5]:
+                text_response += f"- {word}: {count} occurrences\n"
+        text_response += f"\nOverall tone appears {results['basic_sentiment']['assessment']}"
+        results["text"] = text_response
+        return results
+
+class DataAnalysisAgent(IntelligentAgent):
+    def __init__(self, hub: AgentHub):
+        super().__init__("data_analysis", hub)
+
+    def process_task(self, task: str) -> Dict[str, Any]:
+        logger.info(f"DataAnalysisAgent processing: {task}")
+        file_path = None
+        if "analyze" in task.lower() and ".csv" in task.lower():
+            for word in task.split():
+                if word.endswith('.csv'):
+                    file_path = word
+                    break
+        if not file_path or not Path(file_path).exists():
+            logger.info("No specific CSV file mentioned or file not found, creating sample data")
+            if "time series" in task.lower():
+                dates = pd.date_range(start='2023-01-01', periods=30, freq='D')
+                df = pd.DataFrame({'date': dates, 'value': np.random.normal(100, 15, 30), 'trend': np.linspace(0, 20, 30) + np.random.normal(0, 2, 30)})
+                file_path = "sample_timeseries.csv"
+            elif "sales" in task.lower():
+                products = ['ProductA', 'ProductB', 'ProductC', 'ProductD']
+                regions = ['North', 'South', 'East', 'West']
+                dates = pd.date_range(start='2023-01-01', periods=50, freq='D')
+                data = []
+                for _ in range(200):
+                    data.append({'date': np.random.choice(dates), 'product': np.random.choice(products), 'region': np.random.choice(regions), 'units_sold': np.random.randint(10, 100), 'revenue': np.random.uniform(100, 1000)})
+                df = pd.DataFrame(data)
+                file_path = "sample_sales.csv"
+            else:
+                df = pd.DataFrame({
+                    'A': np.random.normal(0, 1, 100),
+                    'B': np.random.normal(5, 2, 100),
+                    'C': np.random.uniform(-10, 10, 100),
+                    'D': np.random.randint(0, 5, 100),
+                    'label': np.random.choice(['X', 'Y', 'Z'], 100)
+                })
+                file_path = "sample_data.csv"
+            df.to_csv(file_path, index=False)
+            logger.info(f"Created sample data file: {file_path}")
+        else:
+            try:
+                df = pd.read_csv(file_path)
+                logger.info(f"Loaded existing file: {file_path}")
+            except Exception as e:
+                error_msg = f"Error loading CSV file {file_path}: {str(e)}"
+                logger.error(error_msg)
+                return {"text": error_msg, "error": str(e)}
+        analysis_results = {}
+        try:
+            numeric_cols = df.select_dtypes(include=[np.number]).columns
+            analysis_results["summary_stats"] = df[numeric_cols].describe().to_dict()
+            categorical_cols = df.select_dtypes(exclude=[np.number]).columns
+            for col in categorical_cols:
+                if df[col].nunique() < 10:
+                    analysis_results[f"{col}_distribution"] = df[col].value_counts().to_dict()
+        except Exception as e:
+            logger.error(f"Error in basic statistics: {e}")
+            analysis_results["stats_error"] = str(e)
+        try:
+            missing_values = df.isnull().sum().to_dict()
+            analysis_results["missing_values"] = {k: v for k, v in missing_values.items() if v > 0}
+        except Exception as e:
+            logger.error(f"Error in missing values analysis: {e}")
+            analysis_results["missing_values_error"] = str(e)
+        try:
+            if len(numeric_cols) > 1:
+                analysis_results["correlations"] = df[numeric_cols].corr().to_dict()
+        except Exception as e:
+            logger.error(f"Error in correlation analysis: {e}")
+            analysis_results["correlation_error"] = str(e)
+        try:
+            plt.figure(figsize=(10, 8))
+            categorical_cols = df.select_dtypes(exclude=[np.number]).columns
+            if len(numeric_cols) >= 2:
+                plt.subplot(2, 1, 1)
+                x_col, y_col = numeric_cols[0], numeric_cols[1]
+                sample_df = df.sample(1000) if len(df) > 1000 else df
+                if len(categorical_cols) > 0 and df[categorical_cols[0]].nunique() < 10:
+                    cat_col = categorical_cols[0]
+                    for category, group in sample_df.groupby(cat_col):
+                        plt.scatter(group[x_col], group[y_col], label=category, alpha=0.6)
+                    plt.legend()
+                else:
+                    plt.scatter(sample_df[x_col], sample_df[y_col], alpha=0.6)
+                plt.xlabel(x_col)
+                plt.ylabel(y_col)
+                plt.title(f"Scatter Plot: {x_col} vs {y_col}")
+                plt.subplot(2, 1, 2)
+                if 'date' in df.columns or any('time' in col.lower() for col in df.columns):
+                    date_col = [col for col in df.columns if 'date' in col.lower() or 'time' in col.lower()][0]
+                    value_col = numeric_cols[0] if numeric_cols[0] != date_col else numeric_cols[1]
+                    if not pd.api.types.is_datetime64_dtype(df[date_col]):
+                        df[date_col] = pd.to_datetime(df[date_col], errors='coerce')
+                    temp_df = df.dropna(subset=[date_col, value_col]).sort_values(date_col)
+                    plt.plot(temp_df[date_col], temp_df[value_col])
+                    plt.xlabel(date_col)
+                    plt.ylabel(value_col)
+                    plt.title(f"Time Series: {value_col} over {date_col}")
+                    plt.xticks(rotation=45)
+                else:
+                    plt.hist(df[numeric_cols[0]].dropna(), bins=20, alpha=0.7)
+                    plt.xlabel(numeric_cols[0])
+                    plt.ylabel('Frequency')
+                    plt.title(f"Distribution of {numeric_cols[0]}")
+            else:
+                if len(categorical_cols) > 0:
+                    cat_col = categorical_cols[0]
+                    df[cat_col].value_counts().plot(kind='bar')
+                    plt.xlabel(cat_col)
+                    plt.ylabel('Count')
+                    plt.title(f"Counts by {cat_col}")
+                    plt.xticks(rotation=45)
+                else:
+                    plt.hist(df[numeric_cols[0]].dropna(), bins=20)
+                    plt.xlabel(numeric_cols[0])
+                    plt.ylabel('Frequency')
+                    plt.title(f"Distribution of {numeric_cols[0]}")
+            plt.tight_layout()
+            viz_path = f"{Path(file_path).stem}_viz.png"
+            plt.savefig(viz_path)
+            plt.close()
+            analysis_results["visualization_path"] = viz_path
+            analysis_results["visualization_created"] = True
+            logger.info(f"Created visualization: {viz_path}")
+        except Exception as e:
+            logger.error(f"Error creating visualization: {e}")
+            analysis_results["visualization_error"] = str(e)
+            analysis_results["visualization_created"] = False
+        insights = []
+        try:
+            for col in numeric_cols:
+                q1 = df[col].quantile(0.25)
+                q3 = df[col].quantile(0.75)
+                iqr = q3 - q1
+                outlier_count = ((df[col] < (q1 - 1.5 * iqr)) | (df[col] > (q3 + 1.5 * iqr))).sum()
+                if outlier_count > 0:
+                    insights.append(f"Found {outlier_count} potential outliers in '{col}'")
+            if "correlations" in analysis_results:
+                for col1, corr_dict in analysis_results["correlations"].items():
+                    for col2, corr_val in corr_dict.items():
+                        if col1 != col2 and abs(corr_val) > 0.7:
+                            insights.append(f"Strong correlation ({corr_val:.2f}) between '{col1}' and '{col2}'")
+            for col in categorical_cols:
+                if df[col].nunique() < 10:
+                    value_counts = df[col].value_counts()
+                    most_common = value_counts.idxmax()
+                    most_common_pct = value_counts.max() / value_counts.sum() * 100
+                    if most_common_pct > 80:
+                        insights.append(f"Imbalanced category in '{col}': '{most_common}' accounts for {most_common_pct:.1f}% of data")
+            analysis_results["insights"] = insights
+        except Exception as e:
+            logger.error(f"Error extracting insights: {e}")
+            analysis_results["insights_error"] = str(e)
+        self.memory.add_short_term({"file": file_path, "columns": list(df.columns), "row_count": len(df), "analysis": analysis_results})
+        if "sample" in file_path:
+            self.memory.add_long_term(f"analysis:{file_path}", {"file": file_path, "type": "generated", "columns": list(df.columns), "row_count": len(df), "timestamp": pd.Timestamp.now().isoformat()})
+        column_list = ", ".join(df.columns[:5]) + (", ..." if len(df.columns) > 5 else "")
+        text_response = (
+            f"Data Analysis Results for {file_path}\n- Dataset: {len(df)} rows x {len(df.columns)} columns ({column_list})\n"
+        )
+        if "missing_values" in analysis_results and analysis_results["missing_values"]:
+            text_response += f"- Missing values found in {len(analysis_results['missing_values'])} columns\n"
+        if insights:
+            text_response += "\nKey Insights:\n"
+            for i, insight in enumerate(insights[:5], 1):
+                text_response += f"{i}. {insight}\n"
+            if len(insights) > 5:
+                text_response += f"... and {len(insights) - 5} more insights\n"
+        text_response += f"\nVisualization saved to {viz_path}" if analysis_results.get("visualization_created") else "\nNo visualization created"
+        analysis_results["text"] = text_response
+        analysis_results["dataframe_shape"] = df.shape
+        analysis_results["data_preview"] = df.head(5).to_dict()
+        return analysis_results
+
+class CodingAssistantAgent(IntelligentAgent):
+    def __init__(self, hub: AgentHub):
+        super().__init__("coding_assistant", hub)
+        self.code_snippets = {
+            "file_operations": {
+                "read_file": '''
+def read_file(file_path):
+    """Read a file and return its contents"""
+    with open(file_path, 'r') as file:
+        return file.read()
+''',
+                "write_file": '''
+def write_file(file_path, content):
+    """Write content to a file"""
+    with open(file_path, 'w') as file:
+        file.write(content)
+    return True
+'''
+            },
+            "data_processing": {
+                "pandas_read_csv": '''
+import pandas as pd
+def load_csv(file_path):
+    """Load a CSV file into a Pandas DataFrame"""
+    return pd.read_csv(file_path)
+''',
+                "pandas_basic_stats": '''
+def get_basic_stats(df):
+    """Get basic statistics for a DataFrame"""
+    numeric_stats = df.describe()
+    categorical_columns = df.select_dtypes(include=['object']).columns
+    categorical_stats = {col: df[col].value_counts().to_dict() for col in categorical_columns}
+    return {
+        'numeric': numeric_stats.to_dict(),
+        'categorical': categorical_stats
+    }
+'''
+            },
+            "visualization": {
+                "matplotlib_basic_plot": '''
+import matplotlib.pyplot as plt
+def create_basic_plot(data, x_col, y_col, title="Plot", kind="line"):
+    """Create a basic plot using matplotlib"""
+    plt.figure(figsize=(10, 6))
+    if kind == "line":
+        plt.plot(data[x_col], data[y_col])
+    elif kind == "scatter":
+        plt.scatter(data[x_col], data[y_col])
+    elif kind == "bar":
+        plt.bar(data[x_col], data[y_col])
+    plt.title(title)
+    plt.xlabel(x_col)
+    plt.ylabel(y_col)
+    plt.tight_layout()
+    plt.savefig(f"{title.lower().replace(' ', '_')}.png")
+    plt.close()
+    return f"{title.lower().replace(' ', '_')}.png"
+'''
+            },
+            "web_scraping": {
+                "requests_beautifulsoup": '''
+import requests
+from bs4 import BeautifulSoup
+def scrape_webpage(url):
+    """Scrape a webpage and extract text from paragraphs"""
+    try:
+        response = requests.get(url)
+        response.raise_for_status()
+        soup = BeautifulSoup(response.text, 'html.parser')
+        paragraphs = soup.find_all('p')
+        text = [p.get_text() for p in paragraphs]
+        return {
+            'title': soup.title.string if soup.title else "No title",
+            'text': text,
+            'url': url
+        }
+    except Exception as e:
+        return {'error': str(e), 'url': url}
+'''
+            },
+            "nlp": {
+                "basic_text_analysis": '''
+from collections import Counter
+import re
+def analyze_text(text):
+    """Perform basic text analysis"""
+    text = text.lower()
+    words = re.findall(r'\w+', text)
+    word_count = len(words)
+    unique_words = len(set(words))
+    stop_words = {'the', 'a', 'an', 'in', 'on', 'at', 'to', 'for', 'of', 'with', 'and', 'or'}
+    word_freq = Counter([w for w in words if w not in stop_words and len(w) > 1])
+    return {
+        'word_count': word_count,
+        'unique_words': unique_words,
+        'avg_word_length': sum(len(w) for w in words) / word_count if word_count else 0,
+        'most_common': word_freq.most_common(10)
+    }
+'''
+            },
+            "machine_learning": {
+                "basic_classifier": '''
+from sklearn.model_selection import train_test_split
+from sklearn.ensemble import RandomForestClassifier
+from sklearn.metrics import classification_report
+def train_basic_classifier(X, y, test_size=0.2, random_state=42):
+    """Train a basic RandomForest classifier"""
+    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=test_size, random_state=random_state)
+    model = RandomForestClassifier(n_estimators=100, random_state=random_state)
+    model.fit(X_train, y_train)
+    y_pred = model.predict(X_test)
+    report = classification_report(y_test, y_pred, output_dict=True)
+    return {
+        'model': model,
+        'accuracy': report['accuracy'],
+        'classification_report': report,
+        'feature_importance': dict(zip(range(X.shape[1]), model.feature_importances_))
+    }
+'''
+            }
+        }
+
+    def process_task(self, task: str) -> Dict[str, Any]:
+        logger.info(f"CodingAssistantAgent processing: {task}")
+        task_lower = task.lower()
+        keyword_mapping = {
+            "file": "file_operations",
+            "read file": "file_operations",
+            "write file": "file_operations",
+            "csv": "data_processing",
+            "data": "data_processing",
+            "pandas": "data_processing",
+            "dataframe": "data_processing",
+            "plot": "visualization",
+            "chart": "visualization",
+            "graph": "visualization",
+            "visualize": "visualization",
+            "matplotlib": "visualization",
+            "scrape": "web_scraping",
+            "web": "web_scraping",
+            "html": "web_scraping",
+            "beautifulsoup": "web_scraping",
+            "text analysis": "nlp",
+            "nlp": "nlp",
+            "natural language": "nlp",
+            "word count": "nlp",
+            "text processing": "nlp",
+            "machine learning": "machine_learning",
+            "ml": "machine_learning",
+            "model": "machine_learning",
+            "predict": "machine_learning",
+            "classifier": "machine_learning"
+        }
+        code_category = None
+        function_name = None
+        for keyword, category in keyword_mapping.items():
+            if keyword in task_lower:
+                code_category = category
+                for func_name in self.code_snippets.get(category, {}):
+                    natural_func = func_name.replace('_', ' ')
+                    if natural_func in task_lower:
+                        function_name = func_name
+                        break
+                break
+        if not code_category:
+            if any(word in task_lower for word in ["add", "sum", "calculate", "compute"]):
+                code_category = "data_processing"
+            elif any(word in task_lower for word in ["show", "display", "generate"]):
+                code_category = "visualization"
+        if code_category and not function_name and self.code_snippets.get(code_category):
+            function_name = next(iter(self.code_snippets[code_category]))
+        if not code_category:
+            function_parts = [word for word in task_lower.split() if word not in ["a", "the", "an", "to", "for", "function", "code", "create", "make"]]
+            func_name = "_".join(function_parts[:2]) if len(function_parts) >= 2 else "custom_function"
+            custom_code = f"""
+def {func_name}(input_data):
+    # Custom function based on your request: '{task}'
+    result = None
+    # TODO: Implement specific logic based on requirements
+    if isinstance(input_data, list):
+        result = len(input_data)
+    elif isinstance(input_data, str):
+        result = input_data.upper()
+    elif isinstance(input_data, (int, float)):
+        result = input_data * 2
+    return {{
+        'input': input_data,
+        'result': result,
+        'status': 'processed'
+    }}
+"""
+            result = {
+                "text": f"I've created a custom function template based on your request:\n\n```python\n{custom_code}\n```\n\nThis is a starting point you can customize further.",
+                "code": custom_code,
+                "language": "python",
+                "type": "custom"
+            }
+        else:
+            code_snippet = self.code_snippets[code_category][function_name]
+            result = {
+                "text": f"Here's a {code_category.replace('_', ' ')} function for {function_name.replace('_', ' ')}:\n\n```python\n{code_snippet}\n```\n\nYou can customize this code.",
+                "code": code_snippet,
+                "language": "python",
+                "category": code_category,
+                "function": function_name
+            }
+        self.memory.add_short_term({"task": task, "code_category": code_category, "function_provided": function_name, "timestamp": pd.Timestamp.now().isoformat()})
+        return result
+
+class ImageProcessingAgent(IntelligentAgent):
+    def __init__(self, hub: AgentHub):
+        super().__init__("image_processing", hub)
+
+    def process_task(self, task: Any) -> Dict[str, Any]:
+        logger.info("ImageProcessingAgent processing task")
+        image = None
+        task_type = None
+        if isinstance(task, Image.Image):
+            image = task
+            task_type = "direct_image"
+        elif isinstance(task, str):
+            if Path(task).exists() and Path(task).suffix.lower() in ['.jpg', '.jpeg', '.png', '.bmp', '.tiff']:
+                try:
+                    image = Image.open(task)
+                    task_type = "image_path"
+                except Exception as e:
+                    return {"text": f"Error loading image from {task}: {str(e)}", "error": str(e)}
+            else:
+                task_type = "text_instruction"
+        elif isinstance(task, dict) and 'image' in task:
+            if isinstance(task['image'], Image.Image):
+                image = task['image']
+            elif isinstance(task['image'], str) and Path(task['image']).exists():
+                try:
+                    image = Image.open(task['image'])
+                except Exception as e:
+                    return {"text": f"Error loading image from {task['image']}: {str(e)}", "error": str(e)}
+            task_type = "dict_with_image"
+        if task_type == "text_instruction" and not image:
+            return {"text": "Please provide an image to process along with instructions."}
+        if not image:
+            return {"text": "No valid image provided for processing."}
+        processing_type = "edge_detection"
+        if task_type in ["text_instruction", "dict_with_image"] and isinstance(task, dict):
+            instruction = task.get('instruction', '').lower()
+            if 'blur' in instruction or 'smooth' in instruction:
+                processing_type = "blur"
+            elif 'edge' in instruction or 'contour' in instruction:
+                processing_type = "edge_detection"
+            elif 'gray' in instruction or 'greyscale' in instruction or 'black and white' in instruction:
+                processing_type = "grayscale"
+            elif 'bright' in instruction or 'contrast' in instruction:
+                processing_type = "enhance"
+            elif 'resize' in instruction or 'scale' in instruction:
+                processing_type = "resize"
+        try:
+            img_array = np.array(image)
+            if img_array.ndim == 3 and img_array.shape[-1] == 4:
+                img_cv = cv2.cvtColor(img_array, cv2.COLOR_RGBA2BGR)
+            else:
+                img_cv = cv2.cvtColor(img_array, cv2.COLOR_RGB2BGR)
+            processed_img = None
+            processing_details = {"original_size": image.size}
+            if processing_type == "edge_detection":
+                gray = cv2.cvtColor(img_cv, cv2.COLOR_BGR2GRAY)
+                edges = cv2.Canny(gray, 100, 200)
+                processed_img = cv2.cvtColor(edges, cv2.COLOR_GRAY2BGR)
+                processing_details["processing"] = "Edge detection using Canny"
+            elif processing_type == "blur":
+                processed_img = cv2.GaussianBlur(img_cv, (7, 7), 0)
+                processing_details["processing"] = "Gaussian Blur"
+            elif processing_type == "grayscale":
+                processed_img = cv2.cvtColor(img_cv, cv2.COLOR_BGR2GRAY)
+                processed_img = cv2.cvtColor(processed_img, cv2.COLOR_GRAY2BGR)
+                processing_details["processing"] = "Grayscale conversion"
+            elif processing_type == "enhance":
+                lab = cv2.cvtColor(img_cv, cv2.COLOR_BGR2LAB)
+                l, a, b = cv2.split(lab)
+                clahe = cv2.createCLAHE(clipLimit=3.0, tileGridSize=(8,8))
+                cl = clahe.apply(l)
+                limg = cv2.merge((cl, a, b))
+                processed_img = cv2.cvtColor(limg, cv2.COLOR_LAB2BGR)
+                processing_details["processing"] = "Contrast enhancement"
+            elif processing_type == "resize":
+                processed_img = cv2.resize(img_cv, (image.size[0]//2, image.size[1]//2))
+                processing_details["processing"] = "Resized to half"
+            else:
+                processed_img = img_cv
+                processing_details["processing"] = "No processing applied"
+            processed_pil = Image.fromarray(cv2.cvtColor(processed_img, cv2.COLOR_BGR2RGB))
+            return {"text": f"Image processing completed with {processing_details['processing']}.", "image": processed_pil, "details": processing_details}
+        except Exception as e:
+            error_msg = f"Error processing image: {str(e)}\n{traceback.format_exc()}"
+            logger.error(error_msg)
+            return {"text": f"Error processing image: {str(e)}", "error": str(e)}
+
+class FileManagementAgent(IntelligentAgent):
+    def __init__(self, hub: AgentHub):
+        super().__init__("file_management", hub)
+
+    def process_task(self, task: str) -> Dict[str, Any]:
+        logger.info(f"FileManagementAgent processing: {task}")
+        task_lower = task.lower()
+        if any(word in task_lower for word in ["create", "make", "generate", "write"]):
+            operation = "create"
+        elif any(word in task_lower for word in ["read", "open", "show", "display", "content"]):
+            operation = "read"
+        elif any(word in task_lower for word in ["list", "find", "directory", "folder", "files in"]):
+            operation = "list"
+        elif any(word in task_lower for word in ["delete", "remove"]):
+            operation = "delete"
+        else:
+            operation = "unknown"
+        filename = None
+        file_extensions = ['.txt', '.json', '.csv', '.md', '.py', '.html', '.js', '.css']
+        words = task.split()
+        for word in words:
+            for ext in file_extensions:
+                if ext in word.lower():
+                    filename = word.strip(':"\'.,;')
+                    break
+            if filename:
+                break
+        if not filename:
+            file_keywords = ["file", "named", "called", "filename"]
+            for i, word in enumerate(words):
+                if word.lower() in file_keywords and i < len(words) - 1:
+                    potential_name = words[i+1].strip(':"\'.,;')
+                    if '.' not in potential_name:
+                        if "json" in task_lower:
+                            potential_name += ".json"
+                        elif "csv" in task_lower:
+                            potential_name += ".csv"
+                        elif "python" in task_lower or "py" in task_lower:
+                            potential_name += ".py"
+                        else:
+                            potential_name += ".txt"
+                    filename = potential_name
+                    break
+        if not filename:
+            if "json" in task_lower:
+                filename = f"data_{uuid.uuid4().hex[:6]}.json"
+            elif "csv" in task_lower:
+                filename = f"data_{uuid.uuid4().hex[:6]}.csv"
+            elif "python" in task_lower or "py" in task_lower:
+                filename = f"script_{uuid.uuid4().hex[:6]}.py"
+            elif "log" in task_lower:
+                filename = f"log_{uuid.uuid4().hex[:6]}.txt"
+            else:
+                filename = f"file_{uuid.uuid4().hex[:6]}.txt"
+        result = {}
+        if operation == "create":
+            if filename.endswith('.json'):
+                content = json.dumps({
+                    "name": "Sample Data",
+                    "description": task,
+                    "created": pd.Timestamp.now().isoformat(),
+                    "values": [1, 2, 3, 4, 5],
+                    "metadata": {"source": "FileManagementAgent", "version": "1.0"}
+                }, indent=2)
+            elif filename.endswith('.csv'):
+                content = "id,name,value,timestamp\n"
+                for i in range(5):
+                    content += f"{i+1},Item{i+1},{np.random.randint(1, 100)},{pd.Timestamp.now().isoformat()}\n"
+            elif filename.endswith('.py'):
+                content = f"""# Generated Python Script: {filename}
+# Created: {pd.Timestamp.now().isoformat()}
+# Description: {task}
+
+def main():
+    print("Hello from the FileManagementAgent!")
+    data = [1, 2, 3, 4, 5]
+    result = sum(data)
+    print(f"Sample calculation: sum(data) = {{result}}")
+    return result
+
+if __name__ == "__main__":
+    main()
+"""
+            else:
+                content = f"File created by FileManagementAgent\nCreated: {pd.Timestamp.now().isoformat()}\nBased on request: {task}\n\nThis is sample content."
+            try:
+                with open(filename, 'w', encoding='utf-8') as f:
+                    f.write(content)
+                result = {"text": f"Successfully created file: {filename}", "operation": "create", "filename": filename, "size": len(content), "preview": content[:200] + "..." if len(content) > 200 else content}
+                self.memory.add_short_term({"operation": "create", "filename": filename, "timestamp": pd.Timestamp.now().isoformat()})
+                self.memory.add_long_term(f"file:{filename}", {"operation": "create", "type": Path(filename).suffix, "timestamp": pd.Timestamp.now().isoformat()})
+            except Exception as e:
+                error_msg = f"Error creating file {filename}: {str(e)}"
+                logger.error(error_msg)
+                result = {"text": error_msg, "error": str(e)}
+        elif operation == "read":
+            if not filename:
+                result = {"text": "Please specify a filename to read."}
+            elif not Path(filename).exists():
+                result = {"text": f"File '{filename}' not found."}
+            else:
+                try:
+                    with open(filename, 'r', encoding='utf-8') as f:
+                        content = f.read()
+                    result = {"text": f"Content of {filename}:\n\n{content}", "operation": "read", "filename": filename, "content": content, "size": len(content)}
+                    self.memory.add_short_term({"operation": "read", "filename": filename, "timestamp": pd.Timestamp.now().isoformat()})
+                except Exception as e:
+                    error_msg = f"Error reading file {filename}: {str(e)}"
+                    logger.error(error_msg)
+                    result = {"text": error_msg, "error": str(e)}
+        elif operation == "list":
+            try:
+                directory = "."
+                for term in ["directory", "folder", "in"]:
+                    if term in task_lower:
+                        parts = task_lower.split(term)
+                        if len(parts) > 1:
+                            potential_dir = parts[1].strip().split()[0].strip(':"\'.,;')
+                            if Path(potential_dir).exists() and Path(potential_dir).is_dir():
+                                directory = potential_dir
+                extension_filter = None
+                for ext in file_extensions:
+                    if ext in task_lower:
+                        extension_filter = ext
+                        break
+                files = list(Path(directory).glob('*' + (extension_filter or '')))
+                file_groups = {}
+                for file in files:
+                    file_groups.setdefault(file.suffix, []).append({
+                        "name": file.name,
+                        "size": file.stat().st_size,
+                        "modified": pd.Timestamp(file.stat().st_mtime, unit='s').isoformat()
+                    })
+                response_text = f"Found {len(files)} files" + (f" with extension {extension_filter}" if extension_filter else "") + f" in {directory}:\n\n"
+                for ext, group in file_groups.items():
+                    response_text += f"{ext} files ({len(group)}):\n"
+                    for file_info in sorted(group, key=lambda x: x["name"]):
+                        size_kb = file_info["size"] / 1024
+                        response_text += f"- {file_info['name']} ({size_kb:.1f} KB, modified: {file_info['modified']})\n"
+                    response_text += "\n"
+                result = {"text": response_text, "operation": "list", "directory": directory, "file_count": len(files), "files": file_groups}
+                self.memory.add_short_term({"operation": "list", "directory": directory, "file_count": len(files), "timestamp": pd.Timestamp.now().isoformat()})
+            except Exception as e:
+                error_msg = f"Error listing files: {str(e)}"
+                logger.error(error_msg)
+                result = {"text": error_msg, "error": str(e)}
+        elif operation == "delete":
+            if not filename:
+                result = {"text": "Please specify a filename to delete."}
+            elif not Path(filename).exists():
+                result = {"text": f"File '{filename}' not found."}
+            else:
+                try:
+                    os.remove(filename)
+                    result = {"text": f"Successfully deleted file: {filename}", "operation": "delete", "filename": filename}
+                    self.memory.add_short_term({"operation": "delete", "filename": filename, "timestamp": pd.Timestamp.now().isoformat()})
+                    self.memory.add_long_term(f"file:{filename}", {"operation": "delete", "timestamp": pd.Timestamp.now().isoformat()})
+                except Exception as e:
+                    error_msg = f"Error deleting file {filename}: {str(e)}"
+                    logger.error(error_msg)
+                    result = {"text": error_msg, "error": str(e)}
+        else:
+            result = {"text": f"Unknown operation requested in task: {task}"}
+        return result
+
+# ---------------------------
+# Gradio Interface Setup
+# ---------------------------
+def create_agent_hub():
+    hub = AgentHub()
+    hub.register_agent("web_research", WebResearchAgent(hub))
+    hub.register_agent("web_scraper", WebScraperAgent(hub))
+    hub.register_agent("text_processing", TextProcessingAgent(hub))
+    hub.register_agent("data_analysis", DataAnalysisAgent(hub))
+    hub.register_agent("coding_assistant", CodingAssistantAgent(hub))
+    hub.register_agent("image_processing", ImageProcessingAgent(hub))
+    hub.register_agent("file_management", FileManagementAgent(hub))
+    return hub
+
+def create_gradio_interface():
+    hub = create_agent_hub()
+    def process_request(request_type, input_data, extra_data=""):
+        try:
+            if request_type == "chain":
+                agent_sequence = [agent.strip() for agent in extra_data.split(",") if agent.strip()]
+                return hub.chain_of_thought(input_data, agent_sequence)
+            else:
+                agent = hub.get_agent(request_type)
+                if not agent:
+                    return {"error": f"Unknown agent type: {request_type}"}
+                return agent.process_task(input_data)
+        except Exception as e:
+            logger.error(f"Error processing request: {e}")
+            return {"error": str(e)}
+    with gr.Blocks(title="SmolAgents Toolbelt") as interface:
+        gr.Markdown("# SmolAgents Toolbelt")
+        gr.Markdown("A collection of specialized agents for various tasks with evolved logic :contentReference[oaicite:0]{index=0}.")
+        with gr.Tabs():
+            with gr.Tab("Single Agent"):
+                agent_type = gr.Dropdown(
+                    choices=["web_research", "web_scraper", "text_processing", "data_analysis", "coding_assistant", "image_processing", "file_management"],
+                    label="Select Agent",
+                    value="web_research"
+                )
+                with gr.Row():
+                    input_text = gr.Textbox(label="Input", placeholder="Enter your request...")
+                    extra_input = gr.Textbox(label="Extra (e.g., image path or additional info)", placeholder="Optional extra input...")
+                output_text = gr.JSON(label="Output")
+                process_btn = gr.Button("Process")
+                process_btn.click(fn=process_request, inputs=[agent_type, input_text, extra_input], outputs=output_text)
+            with gr.Tab("Chain of Thought"):
+                chain_input = gr.Textbox(label="Input", placeholder="Enter your request for the chain...")
+                chain_sequence = gr.Textbox(label="Agent Sequence", placeholder="Comma-separated agent names (e.g., text_processing,data_analysis)")
+                chain_output = gr.JSON(label="Chain Output")
+                chain_type = gr.State("chain")
+                chain_btn = gr.Button("Process Chain")
+                chain_btn.click(fn=process_request, inputs=[chain_type, chain_input, chain_sequence], outputs=chain_output)
+            with gr.Tab("Help"):
+                gr.Markdown("""
+                ## Available Agents
+                
+                - **Web Research Agent**: Searches Wikipedia for information.
+                - **Web Scraper Agent**: Scrapes content from provided URLs.
+                - **Text Processing Agent**: Analyzes and processes text.
+                - **Data Analysis Agent**: Performs data analysis and visualization.
+                - **Coding Assistant Agent**: Generates code snippets.
+                - **Image Processing Agent**: Processes images based on instructions.
+                - **File Management Agent**: Handles file creation, reading, listing, and deletion.
+                
+                ### Usage
+                1. Select an agent (or choose 'Chain of Thought' for a sequence).
+                2. Enter your request.
+                3. For chains, provide a comma-separated list of agent IDs.
+                """)
+    return interface
+
+if __name__ == "__main__":
+    demo = create_gradio_interface()
+    demo.launch(server_name="0.0.0.0", server_port=7860, share=True)