rreverse

app.py

@@ -8,7 +8,6 @@ import aiohttp
 import time
 import random
 import json
-import re
 from smolagents import FinalAnswerTool, Tool, tool, OpenAIServerModel, DuckDuckGoSearchTool, CodeAgent, VisitWebpageTool
 
 
@@ -22,213 +21,182 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
 OPENAI_TOKEN = os.getenv("OPENAI_API_KEY")
 
-# --- Custom Tools for Better Reasoning ---
-
-class TrickQuestionDetector(Tool):
-    """Detects and handles trick questions"""
+# --- Custom Tools ---
+class KnowledgeBaseTool(Tool):
+    name = "knowledge_base"
+    description = "Access structured knowledge for common topics"
+    inputs = {"topic": {"type": "string", "description": "The topic to look up"}}
+    output_type = "string"
     
     def __init__(self):
         super().__init__()
-        self.description = "Analyze if a question is a trick question and provide guidance"
+        self.is_initialized = True
+        # Common knowledge base
+        self.knowledge = {
+            "olympics": "Olympic Games data: Countries, athletes, years, sports",
+            "countries": "Country codes: ISO, IOC, FIFA codes and country information",
+            "sports": "Sports history, rules, famous athletes and events",
+            "science": "Scientific facts, formulas, discoveries, and researchers",
+            "history": "Historical events, dates, people, and places",
+            "geography": "Countries, capitals, populations, and geographical features"
+        }
     
-    def __call__(self, question: str) -> str:
-        """Detect common trick question patterns"""
-        q_lower = question.lower()
-        
-        # Reverse text tricks
-        if question != question and any(c.isalpha() for c in question):
-            reversed_q = question[::-1]
-            if reversed_q.count(' ') > 0:
-                return f"TRICK DETECTED: This appears to be reversed text. Decoded: '{reversed_q}'"
-        
-        # Word puzzles
-        if 'rewsna' in question or 'tfel' in question:
-            return "TRICK DETECTED: Contains reversed words. Try reading backwards."
-        
-        # Contradictory statements
-        contradiction_words = ['impossible', 'never', 'always', 'none', 'all']
-        if sum(word in q_lower for word in contradiction_words) >= 2:
-            return "TRICK DETECTED: Contains contradictory terms. Look for logical impossibilities."
-        
-        # Mathematical tricks
-        if any(phrase in q_lower for phrase in ['how many', 'total', 'sum']) and 'zero' in q_lower:
-            return "TRICK DETECTED: Mathematical trick involving zero or impossible calculations."
-        
-        return "No obvious trick detected. Proceed with normal analysis."
-
-class StepByStepReasoner(Tool):
-    """Breaks down complex questions into steps"""
+    def forward(self, topic: str) -> str:
+        topic_lower = topic.lower()
+        for key, info in self.knowledge.items():
+            if key in topic_lower:
+                return f"Knowledge base: {info}. Use this context to answer questions about {topic}."
+        return f"No specific knowledge base entry for '{topic}'. Use general reasoning."
+
+class WikipediaSearchTool(Tool):
+    name = "wikipedia_search"
+    description = "Search Wikipedia for information"
+    inputs = {"query": {"type": "string", "description": "The search query for Wikipedia"}}
+    output_type = "string"
     
     def __init__(self):
         super().__init__()
-        self.description = "Break down complex questions into logical steps"
-    
-    def __call__(self, question: str) -> str:
-        """Break question into reasoning steps"""
-        steps = []
-        q_lower = question.lower()
-        
-        # Identify question components
-        if any(word in q_lower for word in ['who', 'what', 'when', 'where', 'why', 'how']):
-            steps.append("1. Identify the specific information being requested")
-        
-        if any(word in q_lower for word in ['between', 'from', 'to', 'during']):
-            steps.append("2. Note the time period or range specified")
-        
-        if any(word in q_lower for word in ['calculate', 'count', 'how many', 'total']):
-            steps.append("3. Determine what needs to be calculated or counted")
+        self.is_initialized = True
         
-        if any(word in q_lower for word in ['wikipedia', 'article', 'featured']):
-            steps.append("4. Consider Wikipedia-specific processes and history")
-        
-        if any(word in q_lower for word in ['only', 'single', 'one', 'unique']):
-            steps.append("5. Focus on finding the single/unique answer requested")
-        
-        steps.append("6. Verify the answer makes logical sense")
-        
-        return "REASONING STEPS:\n" + "\n".join(steps)
-
-class FactChecker(Tool):
-    """Validates factual claims and provides confidence levels"""
-    
-    def __init__(self):
-        super().__init__()
-        self.description = "Check factual accuracy and provide confidence assessment"
-    
-    def __call__(self, claim: str) -> str:
-        """Assess factual accuracy of a claim"""
-        confidence_indicators = {
-            'high': ['wikipedia', 'well-known', 'documented', 'official', 'verified'],
-            'medium': ['likely', 'probably', 'appears', 'seems', 'reported'],
-            'low': ['unclear', 'uncertain', 'possibly', 'might', 'could be']
-        }
-        
-        claim_lower = claim.lower()
-        
-        # Check for confidence indicators
-        high_conf = sum(1 for word in confidence_indicators['high'] if word in claim_lower)
-        medium_conf = sum(1 for word in confidence_indicators['medium'] if word in claim_lower)
-        low_conf = sum(1 for word in confidence_indicators['low'] if word in claim_lower)
+    def forward(self, query: str) -> str:
+        """Search Wikipedia with simple fallback."""
+        try:
+            import requests
+            wiki_url = "https://en.wikipedia.org/api/rest_v1/page/summary/" + query.replace(" ", "_")
+            response = requests.get(wiki_url, timeout=2)
+            if response.status_code == 200:
+                data = response.json()
+                if 'extract' in data and data['extract']:
+                    return f"Wikipedia: {data['extract'][:500]}"  # Limit length
+        except Exception as e:
+            print(f"Wikipedia search failed: {e}")
         
-        if high_conf > medium_conf and high_conf > low_conf:
-            return f"CONFIDENCE: HIGH - Claim appears to be well-documented: '{claim}'"
-        elif low_conf > high_conf:
-            return f"CONFIDENCE: LOW - Claim contains uncertainty markers: '{claim}'"
-        else:
-            return f"CONFIDENCE: MEDIUM - Standard factual claim: '{claim}'"
+        return f"Wikipedia search unavailable for '{query}'. Use your knowledge to answer."
 
-class AnswerValidator(Tool):
-    """Validates if an answer makes sense for the question"""
-    
+# --- Basic Agent Definition ---
+# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
+class SlpMultiAgent:
     def __init__(self):
-        super().__init__()
-        self.description = "Validate if an answer is reasonable for the given question"
-    
-    def __call__(self, question: str, answer: str) -> str:
-        """Check if answer is reasonable for the question"""
-        q_lower = question.lower()
-        a_lower = answer.lower()
-        
-        # Check for question-answer type matching
-        if 'who' in q_lower and not any(indicator in a_lower for indicator in ['person', 'user', 'editor', 'author', 'name']):
-            return "WARNING: 'Who' question but answer doesn't seem to identify a person"
+        print("BasicAgent initialized.")
         
-        if 'when' in q_lower and not any(indicator in a_lower for indicator in ['year', 'date', 'time', '20', '19']):
-            return "WARNING: 'When' question but answer doesn't contain time information"
+    async def __call__(self, question: str) -> str:
+        print(f"Agent received question (first 50 chars): {question[:50]}...")
+        fixed_answer = "This is a default answer."
+        print(f"Agent returning fixed answer: {fixed_answer}")
         
-        if 'how many' in q_lower and not any(char.isdigit() for char in answer):
-            return "WARNING: 'How many' question but answer contains no numbers"
+        # Truncate question to avoid exceeding model context length
+        MAX_QUESTION_LENGTH = 1000
+        short_question = question  # [:MAX_QUESTION_LENGTH]
         
-        if len(answer.strip()) < 3:
-            return "WARNING: Answer seems too short"
+        # Use cheaper, faster model
+        model = OpenAIServerModel(
+            model_id="gpt-3.5-turbo", 
+            temperature=0.0,  # Deterministic for consistency
+            max_tokens=400    # Reduced tokens for cost efficiency
+        )
         
-        if len(answer.strip()) > 200:
-            return "WARNING: Answer seems too long - may need to be more concise"
+        # Create only essential agents with reduced complexity
+        research_agent = CodeAgent(
+            tools=[KnowledgeBaseTool()],  # Remove Wikipedia to avoid timeouts
+            model=model,
+            additional_authorized_imports=["re", "datetime"],
+            max_steps=2,  # Reduced steps for cost
+            name="ResearchAgent",
+            verbosity_level=0,
+            description="Quick factual research and knowledge lookup."
+        )
         
-        return "VALIDATION: Answer format appears appropriate for question type"
+        solver_agent = CodeAgent(
+            tools=[],
+            model=model,
+            additional_authorized_imports=["math", "re", "collections", "itertools"],
+            max_steps=2,  # Reduced steps
+            name="SolverAgent",
+            verbosity_level=0,
+            description="Problem solving, calculations, and logical reasoning."
+        )
 
-# --- Enhanced Agent with Tools ---
-class SlpMultiAgent:
-    def __init__(self):
-        print("Enhanced Agent initialized with reasoning tools.")
-        self.trick_detector = TrickQuestionDetector()
-        self.step_reasoner = StepByStepReasoner()
-        self.fact_checker = FactChecker()
-        self.answer_validator = AnswerValidator()
+        manager_agent = CodeAgent(
+            model=OpenAIServerModel(
+                model_id="gpt-3.5-turbo",
+                temperature=0.0,
+                max_tokens=500
+            ),
+            tools=[KnowledgeBaseTool()],  # Remove Wikipedia to avoid timeouts
+            managed_agents=[research_agent, solver_agent],  # Only 2 agents
+            name="ManagerAgent",
+            description="Efficient manager for quick problem solving.",
+            additional_authorized_imports=["re", "math"],
+            planning_interval=1,  # Faster planning
+            verbosity_level=0,  # Reduce verbosity
+            max_steps=3,  # Further reduced steps to avoid timeouts
+            final_answer_checks=[check_reasoning]
+        )
         
-    async def __call__(self, question: str) -> str:
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
+        # Create a task for the agent run with retry mechanism for rate limits
+        max_retries = 3
+        result = None
         
-        # Step 1: Check for tricks
-        trick_analysis = self.trick_detector(question)
-        print(f"Trick analysis: {trick_analysis}")
+        for attempt in range(max_retries):
+            try:
+                loop = asyncio.get_event_loop()
+                result = await loop.run_in_executor(
+                    None,
+                    lambda: manager_agent.run(f"""
+                    Question: {short_question}
+                    
+                    You have knowledge_base() tool and two agents:
+                    - ResearchAgent: For factual questions
+                    - SolverAgent: For calculations and logic
+                    
+                    IMPORTANT: Always end with exactly this format:
+                    <code>
+                    final_answer("your direct answer")
+                    </code>
+                    
+                    Be concise and direct.
+                    """)
+                )
+                break  # Success, exit retry loop
+            except Exception as e:
+                print(f"Attempt {attempt+1}/{max_retries} failed: {e}")
+                if "rate limit" in str(e).lower() and attempt < max_retries - 1:
+                    # Add jitter to avoid synchronized retries
+                    wait_time = (attempt + 1) * 10 + random.uniform(0, 5)  
+                    print(f"Rate limit hit. Waiting {wait_time:.2f} seconds before retry...")
+                    await asyncio.sleep(wait_time)
+                elif attempt < max_retries - 1:
+                    await asyncio.sleep(5)  # Wait before general retry
+                else:
+                    print(f"All attempts failed. Returning default answer.")
+                    return "I apologize, but I'm currently experiencing technical difficulties. Please try again later."
         
-        # Step 2: Break down reasoning steps
-        reasoning_steps = self.step_reasoner(question)
-        print(f"Reasoning steps: {reasoning_steps}")
+        # If we couldn't get a result after all retries
+        if result is None:
+            return "I apologize, but I'm currently experiencing technical difficulties. Please try again later."
         
-        # Step 3: Enhanced model call with tool insights
-        model = OpenAIServerModel(
-            model_id="gpt-4o-mini",
-            temperature=0.1,
-            max_tokens=1000
-        )
         
-        try:
-            enhanced_prompt = f"""You are an expert problem solver. Analyze this question carefully:
-
-QUESTION: {question}
-
-TRICK ANALYSIS: {trick_analysis}
-
-{reasoning_steps}
-
-Instructions:
-1. If a trick was detected, handle it appropriately
-2. Follow the reasoning steps systematically
-3. Think through each step carefully
-4. Provide a clear, direct answer
-5. If unsure, state your uncertainty clearly
-
-Be precise and thorough in your analysis."""
-
-            messages = [
-                {
-                    "role": "system",
-                    "content": "You are an expert at solving complex and trick questions. Always think step by step and be very careful about the exact wording of questions."
-                },
-                {
-                    "role": "user", 
-                    "content": enhanced_prompt
-                }
-            ]
+        # Extract clean answer from result
+        if result and isinstance(result, str):
+            # Look for final_answer pattern
+            import re
+            final_answer_match = re.search(r'final_answer\(["\']([^"\']*)["\'\)]', result)  # Fixed regex
+            if final_answer_match:
+                clean_answer = final_answer_match.group(1)
+                return clean_answer
             
-            result = model(messages)
-            
-            if result:
-                # Step 4: Validate the answer
-                validation = self.answer_validator(question, result)
-                print(f"Answer validation: {validation}")
-                
-                # Clean up the result
-                lines = result.strip().split('\n')
-                for line in reversed(lines):
-                    line = line.strip()
-                    if line and len(line) > 5 and not line.startswith(('Step', 'Analysis', 'TRICK', 'REASONING')):
-                        # Remove common prefixes
-                        line = re.sub(r'^(Answer:|Final answer:|The answer is:?)\s*', '', line, flags=re.IGNORECASE)
-                        if line:
-                            return line
-                
-                return result
-            else:
-                return "I don't have enough information to answer this question accurately."
-                
-        except Exception as e:
-            print(f"Model call failed: {e}")
-            return "I apologize, but I'm currently experiencing technical difficulties."
-
+            # If no final_answer found, try to extract the last meaningful line
+            lines = result.strip().split('\n')
+            for line in reversed(lines):
+                line = line.strip()
+                if line and not line.startswith('#') and not line.startswith('###') and len(line) < 200:
+                    return line
+        
+        # Return the result from the agent
+        return result if result else "Unable to determine answer."
+    
 def check_reasoning(final_answer, agent_memory):
+    # Skip expensive validation to save costs
     return True
 
 
@@ -293,8 +261,8 @@ async def run_and_submit_all(profile):
     answers_payload = []
     print(f"Running agent on {len(questions_data)} questions...")
     
-    # Process questions with controlled concurrency
-    semaphore = asyncio.Semaphore(2)  # Process 2 questions at a time
+    # Process questions one at a time to avoid rate limits
+    semaphore = asyncio.Semaphore(1)  # Process 1 question at a time
     
     async def process_question(item):
         task_id = item.get("task_id")
@@ -304,16 +272,27 @@ async def run_and_submit_all(profile):
             return None
         
         async with semaphore:
-            try:
-                print(f"Processing task {task_id}")
-                submitted_answer = await agent(question_text)
-                return {"task_id": task_id, "submitted_answer": submitted_answer, 
-                        "log": {"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer}}
-            except Exception as e:
-                print(f"Error running agent on task {task_id}: {e}")
-                default_answer = "I don't have enough information to answer this question accurately."
-                return {"task_id": task_id, "submitted_answer": default_answer, 
-                        "log": {"Task ID": task_id, "Question": question_text, "Submitted Answer": default_answer}}
+            max_retries = 3
+            for attempt in range(max_retries):
+                try:
+                    print(f"Processing task {task_id}, attempt {attempt+1}/{max_retries}")
+                    submitted_answer = await agent(question_text)
+                    return {"task_id": task_id, "submitted_answer": submitted_answer, 
+                            "log": {"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer}}
+                except Exception as e:
+                    print(f"Error running agent on task {task_id}, attempt {attempt+1}: {e}")
+                    if "rate limit" in str(e).lower() and attempt < max_retries - 1:
+                        # Exponential backoff with jitter
+                        wait_time = (2 ** attempt) * 5 + random.uniform(0, 3)
+                        print(f"Rate limit hit. Waiting {wait_time:.2f} seconds before retry...")
+                        await asyncio.sleep(wait_time)
+                    elif attempt < max_retries - 1:
+                        await asyncio.sleep(5)  # Reduced wait time
+                    else:
+                        # All retries failed, return default answer
+                        default_answer = "This is a default answer."
+                        return {"task_id": task_id, "submitted_answer": default_answer, 
+                                "log": {"Task ID": task_id, "Question": question_text, "Submitted Answer": default_answer}}
     
     # Create tasks for all questions
     tasks = [process_question(item) for item in questions_data]