Update smolagents_agent.py

smolagents_agent.py

@@ -1,339 +1,240 @@
 import os
 import re
-import math
-import pandas as pd
 from typing import Optional
 from dotenv import load_dotenv
-from smolagents import Tool, CodeAgent
-from smolagents import DuckDuckGoSearchTool, VisitWebpageTool, InferenceClientModel
-
+from smolagents import InferenceClientModel
 
 # Load environment variables
 load_dotenv()
 
 # -------------------------
-# Robust Custom Tools
-# -------------------------
-
-class RobustCalculatorTool(Tool):
-    name = "robust_calculator"
-    description = "Perform mathematical calculations with error handling"
-    inputs = {"expression": {"type": "string", "description": "Math expression"}}
-    output_type = "string"
-
-    def forward(self, expression: str) -> str:
-        try:
-            # Clean expression
-            expr = expression.strip()
-            expr = re.sub(r'\s+', '', expr)  # Remove all whitespace
-            
-            # Safe replacements only
-            replacements = {
-                'sin': 'math.sin', 'cos': 'math.cos', 'tan': 'math.tan',
-                'log': 'math.log', 'sqrt': 'math.sqrt', 'pi': 'math.pi', 'e': 'math.e',
-                '^': '**', 'ln': 'math.log'
-            }
-            
-            for old, new in replacements.items():
-                expr = expr.replace(old, new)
-            
-            # Very restrictive evaluation
-            allowed_names = {
-                'math': math, 'abs': abs, 'min': min, 'max': max, 
-                'round': round, 'sum': sum, 'pow': pow
-            }
-            
-            result = eval(expr, {"__builtins__": {}}, allowed_names)
-            return str(result)
-        except Exception as e:
-            return f"Calculation error: {str(e)}"
-
-class SafeWebSearchTool(Tool):
-    name = "safe_web_search"
-    description = "Safe web search with error handling"
-    inputs = {"query": {"type": "string", "description": "Search query"}}
-    output_type = "string"
-
-    def forward(self, query: str) -> str:
-        try:
-            search_tool = DuckDuckGoSearchTool()
-            result = search_tool.forward(query)
-            if isinstance(result, str):
-                # Limit result length
-                if len(result) > 500:
-                    result = result[:500] + "..."
-                return result
-            return "Search completed"
-        except Exception as e:
-            return f"Search error: {str(e)}"
-
-class SafeWikipediaTool(Tool):
-    name = "safe_wikipedia"
-    description = "Safe Wikipedia search with error handling"
-    inputs = {"query": {"type": "string", "description": "Search query"}}
-    output_type = "string"
-
-    def forward(self, query: str) -> str:
-        try:
-            import wikipedia
-            summary = wikipedia.summary(query, sentences=2)
-            if len(summary) > 300:
-                summary = summary[:300] + "..."
-            return summary
-        except Exception as e:
-            return f"Wikipedia error: {str(e)}"
-
-class SafeVisitWebpageTool(Tool):
-    name = "safe_visit_webpage"
-    description = "Safe webpage visit with error handling"
-    inputs = {"url": {"type": "string", "description": "URL to visit"}}
-    output_type = "string"
-
-    def forward(self, url: str) -> str:
-        try:
-            visit_tool = VisitWebpageTool()
-            result = visit_tool.forward(url)
-            if isinstance(result, str) and len(result) > 500:
-                return result[:500] + "..."
-            return str(result) if result else "Page content not available"
-        except Exception as e:
-            return f"Visit error: {str(e)}"
-
-class SafeImageProcessorTool(Tool):
-    name = "safe_image_processor"
-    description = "Safe image processing with error handling"
-    inputs = {
-        "image_path": {"type": "string", "description": "Image path or URL"},
-        "operation": {"type": "string", "description": "Operation type", "default": "analyze"}
-    }
-    output_type = "string"
-
-    def forward(self, image_path: str, operation: str = "analyze") -> str:
-        try:
-            if operation == "analyze":
-                return f"Image analysis: {image_path} - Processing image content for task completion"
-            else:
-                return f"Image {operation}: {image_path} - Operation completed"
-        except Exception as e:
-            return f"Image processing error: {str(e)}"
-
-# -------------------------
-# Comprehensive Error Handling Agent
+# Direct Tool-Based Agent (No Code Generation)
 # -------------------------
 
 class OptimizedSmolagentsGAIAgent:
+    """
+    A direct agent that uses tools without code generation.
+    This avoids all the syntax and runtime errors from generated code.
+    """
+    
     def __init__(self):
-        # Initialize robust tools
-        self.calculator = RobustCalculatorTool()
-        self.web_search = SafeWebSearchTool()
-        self.wikipedia = SafeWikipediaTool()
-        self.visit_webpage = SafeVisitWebpageTool()
-        self.image_processor = SafeImageProcessorTool()
-
-        self.tools = [
-            self.calculator,
-            self.web_search,
-            self.wikipedia,
-            self.visit_webpage,
-            self.image_processor
-        ]
-
-        # Initialize Hugging Face model with multiple fallbacks
+        # Initialize model
         self.model = self._initialize_model()
         
-        # Initialize agent with error handling
-        self._initialize_agent()
+        # Available tools
+        self.tools = {
+            'calculator': self._safe_calculate,
+            'web_search': self._safe_web_search,
+            'wikipedia': self._safe_wikipedia_search,
+            'visit_webpage': self._safe_visit_webpage,
+            'image_analysis': self._safe_image_analysis
+        }
 
     def _initialize_model(self):
-        """Initialize model with multiple fallback options and timeout handling"""
+        """Initialize model with multiple fallbacks"""
         hf_token = os.getenv("HF_TOKEN")
         if not hf_token:
             print("HF_TOKEN not found. Using fallback mode.")
             return None
         
-        # Multiple model options for reliability
+        # Try multiple models for reliability
         model_options = [
-            {"model_id": "allenai/Olmo-3-7B-Instruct", "timeout": 30},
-            {"model_id": "allenai/Olmo-3-7B-Think", "timeout": 30},
-            {"timeout": 30}  # Fallback without specific model
+            "allenai/Olmo-3-7B-Instruct",
+            "allenai/Olmo-3-7B-Think"
         ]
         
-        for config in model_options:
+        for model in model_options:
             try:
                 model = InferenceClientModel(
+                    model_id=model,
                     token=hf_token,
-                    **config
+                    timeout=30
                 )
-                print(f"Using model: {config.get('model', 'default')}")
+                print(f"Using model: {model}")
                 return model
             except Exception as e:
-                print(f"Failed to initialize {config.get('model', 'default')}: {e}")
+                print(f"Failed to initialize {model}: {e}")
                 continue
         
-        print("All model initialization attempts failed")
         return None
 
-    def _initialize_agent(self):
-        """Initialize agent with conservative settings"""
-        if self.model:
-            try:
-                self.agent = CodeAgent(
-                    tools=self.tools,
-                    model=self.model,
-                    max_steps=5,  # Reduced steps to prevent errors
-                    verbosity_level=0  # Minimal verbosity
-                )
-                print("Agent initialized successfully")
-            except Exception as e:
-                print(f"Agent initialization failed: {e}")
-                self.agent = None
-
-    def _generate_safe_prompt(self, question: str) -> str:
-        """Generate a safe, structured prompt to prevent code generation errors"""
+    def _classify_question(self, question: str) -> str:
+        """Classify question type for appropriate tool selection"""
+        q_lower = question.lower()
         
-        # Question classification for targeted prompts
-        question_lower = question.lower()
+        # Mathematical questions
+        if any(word in q_lower for word in ['calculate', 'compute', 'solve', '+', '-', '*', '/', '=']):
+            return 'math'
         
-        if any(word in question_lower for word in ['calculate', 'compute', 'solve', '+', '-', '*', '/']):
-            prompt = f"""Solve this mathematical problem step by step using the robust_calculator tool.
-
-Question: {question}
-
-Instructions:
-- Use only the robust_calculator tool for calculations
-- Provide the final numerical answer
-- Be precise and concise"""
-
-        elif any(word in question_lower for word in ['search', 'find', 'web', 'google']):
-            prompt = f"""Search for information using safe web search tools.
-
-Question: {question}
-
-Instructions:
-- Use safe_web_search tool for web searches
-- Use safe_wikipedia tool for Wikipedia information
-- Provide a concise, factual answer"""
-
-        elif any(word in question_lower for word in ['who is', 'what is', 'when', 'where']):
-            prompt = f"""Find factual information about this topic.
-
-Question: {question}
-
-Instructions:
-- Use safe_wikipedia tool for reliable information
-- Use safe_web_search tool if additional info is needed
-- Provide a clear, factual response"""
-
+        # Web search questions
+        elif any(word in q_lower for word in ['search', 'find', 'recent', 'current', 'today']):
+            return 'search'
+        
+        # Factual/Wikipedia questions
+        elif any(word in q_lower for word in ['who is', 'what is', 'when', 'where', 'history', 'biography']):
+            return 'wikipedia'
+        
+        # Webpage questions
         elif 'http' in question or 'www.' in question:
-            prompt = f"""Visit and analyze webpage content.
-
-Question: {question}
-
-Instructions:
-- Use safe_visit_webpage tool to access the URL
-- Extract relevant information
-- Provide a summary of findings"""
-
-        elif any(word in question_lower for word in ['image', 'picture', 'visual']):
-            prompt = f"""Analyze image content if provided.
-
-Question: {question}
-
-Instructions:
-- Use safe_image_processor tool for image analysis
-- Describe what you can determine from the image
-- Provide a focused answer"""
-
+            return 'webpage'
+        
+        # Image questions
+        elif any(word in q_lower for word in ['image', 'picture', 'photo', 'visual', 'chess']):
+            return 'image'
+        
+        # Default to search for general questions
         else:
-            # General question prompt
-            prompt = f"""Answer this question using available tools safely.
+            return 'search'
 
-Question: {question}
+    def _safe_calculate(self, question: str) -> str:
+        """Safe calculation using basic math"""
+        try:
+            # Extract math expressions
+            math_pattern = r'[\d\+\-\*\/\.\(\)\s]+'
+            match = re.search(math_pattern, question)
+            if match:
+                expr = match.group().strip()
+                # Only allow basic arithmetic
+                if re.match(r'^[\d\+\-\*\/\.\(\)\s]+$', expr):
+                    result = eval(expr)
+                    return f"The calculation result is: {result}"
+            return "No clear mathematical expression found in the question."
+        except Exception as e:
+            return f"Unable to calculate: {str(e)}"
 
-Instructions:
-- Use appropriate tools (robust_calculator, safe_web_search, safe_wikipedia)
-- Provide a clear, concise answer
-- Avoid generating complex code"""
+    def _safe_web_search(self, question: str) -> str:
+        """Safe web search with error handling"""
+        try:
+            from smolagents import DuckDuckGoSearchTool
+            search_tool = DuckDuckGoSearchTool()
+            
+            # Clean the query for search
+            query = re.sub(r'[^a-zA-Z0-9\s]', '', question)
+            if len(query) > 100:
+                query = query[:100]
+            
+            result = search_tool.forward(query)
+            if isinstance(result, str):
+                if len(result) > 300:
+                    result = result[:300] + "..."
+                return f"Search results for '{query}': {result}"
+            return "Search completed successfully."
+        except Exception as e:
+            return f"Web search error: {str(e)}"
 
-        return prompt
+    def _safe_wikipedia_search(self, question: str) -> str:
+        """Safe Wikipedia search with error handling"""
+        try:
+            import wikipedia
+            
+            # Extract search terms
+            if 'who is' in question.lower():
+                query = question.lower().replace('who is', '').strip()
+            elif 'what is' in question.lower():
+                query = question.lower().replace('what is', '').strip()
+            else:
+                query = question.strip()
+            
+            if query:
+                summary = wikipedia.summary(query, sentences=2)
+                if len(summary) > 200:
+                    summary = summary[:200] + "..."
+                return f"Information about '{query}': {summary}"
+            return "Unable to extract search terms from question."
+        except Exception as e:
+            return f"Wikipedia search error: {str(e)}"
 
-    def _safe_response_formatting(self, result: str) -> str:
-        """Safely format response to avoid HuggingFace parsing errors"""
-        
-        # If no code blocks, return as-is
-        if "```" not in result:
-            return result
-        
-        # Only wrap actual markdown code blocks, never natural language
+    def _safe_visit_webpage(self, question: str) -> str:
+        """Safe webpage visiting with error handling"""
         try:
-            import re
-            # Pattern for markdown code blocks only
-            code_pattern = r'```(?:python)?\n(.*?)\n```'
-            matches = re.findall(code_pattern, result, re.DOTALL)
+            from smolagents import VisitWebpageTool
+            visit_tool = VisitWebpageTool()
             
-            if matches:
-                formatted_result = result
-                for match in matches:
-                    code_content = match.strip()
-                    if code_content and not any(phrase in code_content.lower() for phrase in [
-                        'the', 'and', 'or', 'but', 'then', 'now', 'here', 'there'
-                    ]):
-                        # Only wrap if it looks like actual code
-                        wrapped_code = f"<code>{code_content}</code>"
-                        formatted_result = formatted_result.replace(f"```{code_content}```", wrapped_code)
-                return formatted_result
+            # Extract URL from question
+            url_pattern = r'https?://[^\s\)]+'
+            urls = re.findall(url_pattern, question)
             
-            return result  # Return original if no valid code blocks
+            if urls:
+                url = urls[0]
+                result = visit_tool.forward(url)
+                if isinstance(result, str):
+                    if len(result) > 200:
+                        result = result[:200] + "..."
+                    return f"Content from {url}: {result}"
+                return f"Successfully visited {url}"
+            return "No URL found in the question."
+        except Exception as e:
+            return f"Webpage visit error: {str(e)}"
+
+    def _safe_image_analysis(self, question: str) -> str:
+        """Safe image analysis with error handling"""
+        try:
+            # For chess questions
+            if 'chess' in question.lower():
+                return "Chess position analysis: This appears to be a chess-related question. Black's turn means black pieces need to make the next move. Without the actual board image, I cannot provide the specific move, but typical strategic considerations include developing pieces, controlling center, or_castling."
             
-        except Exception:
-            return result  # Return original if formatting fails
+            # For general image questions
+            elif any(word in question.lower() for word in ['image', 'picture', 'photo']):
+                return "Image analysis: The question references image content that I cannot directly access. For visual analysis tasks, please describe what you can see in the image or provide specific details about the visual elements."
+            
+            else:
+                return "Image processing: Unable to analyze image content directly. Please provide more details about what visual information you need."
+        except Exception as e:
+            return f"Image analysis error: {str(e)}"
+
+    def _generate_direct_answer(self, question: str, question_type: str) -> str:
+        """Generate direct answers without code generation"""
+        
+        if question_type == 'math':
+            return self._safe_calculate(question)
+        
+        elif question_type == 'search':
+            return self._safe_web_search(question)
+        
+        elif question_type == 'wikipedia':
+            return self._safe_wikipedia_search(question)
+        
+        elif question_type == 'webpage':
+            return self._safe_visit_webpage(question)
+        
+        elif question_type == 'image':
+            return self._safe_image_analysis(question)
+        
+        else:
+            # Default fallback
+            return self._safe_web_search(question)
 
     def process_question(self, question: str) -> str:
-        """Process question with comprehensive error handling"""
+        """Process question using direct tool approach (no code generation)"""
         
         # Handle no model case
         if not self.model:
             return "No language model available. Please set HF_TOKEN in environment variables."
         
-        # Handle no agent case
-        if not self.agent:
-            return "Agent not properly initialized. Please check configuration."
-        
         try:
-            # Generate safe prompt
-            safe_prompt = self._generate_safe_prompt(question)
+            # Classify question type
+            question_type = self._classify_question(question)
             
-            # Run agent with timeout protection
-            result = self.agent.run(safe_prompt)
-            result_str = str(result)
+            # Generate direct answer using appropriate tool
+            answer = self._generate_direct_answer(question, question_type)
             
-            # Safe response formatting
-            formatted_result = self._safe_response_formatting(result_str)
-            
-            return formatted_result
+            return answer
             
         except Exception as e:
             error_msg = str(e)
             
             # Specific error handling
-            if "timeout" in error_msg.lower() or "read timed out" in error_msg.lower():
+            if "timeout" in error_msg.lower():
                 return "Request timed out. The question may be too complex. Please try a simpler question."
             
-            elif "500" in error_msg and "internal server error" in error_msg.lower():
+            elif "500" in error_msg:
                 return "Server error occurred. This may be a temporary issue. Please try again later."
             
-            elif "syntax" in error_msg.lower() or "parsing" in error_msg.lower():
-                return "Unable to process this question format. Please try rephrasing the question."
-            
-            elif "tool" in error_msg.lower() and "not found" in error_msg.lower():
-                return "Required tools are not available for this question type."
-            
             else:
-                # General error handling
                 return f"Unable to process question: {error_msg[:200]}"
 
 # -------------------------
-# Test the comprehensive agent
+# Test the direct tool agent
 # -------------------------
 
 if __name__ == "__main__":
@@ -341,12 +242,13 @@ if __name__ == "__main__":
     
     test_questions = [
         "What is the capital of France?",
-        "Calculate 15 + 27 * 3",
-        "Search for information about Python programming",
-        "Who wrote Romeo and Juliet?"
+        "Calculate 15 + 27 * 3", 
+        "Who is Mercedes Sosa?",
+        "Review the chess position in the image",
+        "What does this webpage say: https://example.com"
     ]
     
-    print("=== COMPREHENSIVE SMOLAGENTS AGENT TEST ===\n")
+    print("=== DIRECT TOOL AGENT TEST ===\n")
     
     for question in test_questions:
         print(f"Q: {question}")