smolagents_agent.py

import os
import re
from typing import Optional
from dotenv import load_dotenv
from smolagents import InferenceClientModel


# Load environment variables
load_dotenv()

# -------------------------
# 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 model
        self.model = self._initialize_model()
        
        # 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 fallbacks"""
        hf_token = os.getenv("HF_TOKEN")
        if not hf_token:
            print("HF_TOKEN not found. Using fallback mode.")
            return None
        
        # Try multiple models for reliability
        model_options = [
            "allenai/Olmo-3-7B-Instruct",
            "allenai/Olmo-3-7B-Think"
        ]
        
        for model in model_options:
            try:
                model = InferenceClientModel(
                    model_id=model,
                    token=hf_token,
                    timeout=30
                )
                print(f"Using model: {model}")
                return model
            except Exception as e:
                print(f"Failed to initialize {model}: {e}")
                continue
        
        return None

    def _classify_question(self, question: str) -> str:
        """Classify question type for appropriate tool selection"""
        q_lower = question.lower()
        
        # Mathematical questions
        if any(word in q_lower for word in ['calculate', 'compute', 'solve', '+', '-', '*', '/', '=']):
            return 'math'
        
        # 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:
            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:
            return 'search'

    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)}"

    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)}"

    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_visit_webpage(self, question: str) -> str:
        """Safe webpage visiting with error handling"""
        try:
            from smolagents import VisitWebpageTool
            visit_tool = VisitWebpageTool()
            
            # Extract URL from question
            url_pattern = r'https?://[^\s\)]+'
            urls = re.findall(url_pattern, question)
            
            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."
            
            # 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 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."
        
        try:
            # Classify question type
            question_type = self._classify_question(question)
            
            # Generate direct answer using appropriate tool
            answer = self._generate_direct_answer(question, question_type)
            
            return answer
            
        except Exception as e:
            error_msg = str(e)
            
            # Specific error handling
            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:
                return "Server error occurred. This may be a temporary issue. Please try again later."
            
            else:
                return f"Unable to process question: {error_msg[:200]}"

# -------------------------
# Test the direct tool agent
# -------------------------

if __name__ == "__main__":
    agent = OptimizedSmolagentsGAIAgent()
    
    test_questions = [
        "What is the capital of France?",
        "Calculate 15 + 27 * 3", 
        "Who is Mercedes Sosa?",
        "Review the chess position in the image",
        "What does this webpage say: https://example.com"
    ]
    
    print("=== DIRECT TOOL AGENT TEST ===\n")
    
    for question in test_questions:
        print(f"Q: {question}")
        answer = agent.process_question(question)
        print(f"A: {answer[:200]}...")
        print("-" * 50)