Update app.py

app.py

@@ -3,500 +3,22 @@ import gradio as gr
 import requests
 import inspect
 import pandas as pd
-import re
-import base64
-from io import BytesIO
-from typing import Dict, List, Any, Optional, Tuple
-from langchain_core.messages import AIMessage, HumanMessage
-from langchain_openai import ChatOpenAI
-from langgraph.graph import StateGraph, END
-from pydantic import BaseModel, Field
-import json
-import math
-from dotenv import load_dotenv
-from PIL import Image
-import pytesseract
-import youtube_dl
-from youtube_transcript_api import YouTubeTranscriptApi
-
-# Load environment variables from .env file if present
-load_dotenv()
+from langgraph_agent import LangGraphAgent
 
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
-OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
 
 # --- Basic Agent Definition ---
 # ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
 class BasicAgent:
-    """
-    An agent designed for the Gaia Benchmark, implementing a multi-step reasoning approach.
-    
-    This agent:
-    1. Analyzes the question to determine its type
-    2. Performs step-by-step reasoning
-    3. Uses tools when needed (calculator, web search, image analysis, file processing)
-    4. Formats the final answer according to Gaia Benchmark requirements
-    """
-    
-    def __init__(self, model_name: str = "gpt-4.1", api_key: str = None):
-        """
-        Initialize the BasicAgent with the specified LLM.
-        
-        Args:
-            model_name: The name of the model to use (defaults to gpt-4.1)
-            api_key: OpenAI API key (defaults to environment variable)
-        """
-        # Use the provided API key or fall back to the environment variable
-        api_key = api_key or OPENAI_API_KEY
-        
-        if not api_key:
-            print("Warning: No OpenAI API key provided. Make sure to set OPENAI_API_KEY environment variable.")
-            
-        self.llm = ChatOpenAI(
-            model=model_name,
-            temperature=0,
-            api_key=api_key
-        )
-        
-        # Initialize vision model if needed for image analysis
-        try:
-            self.vision_model = ChatOpenAI(
-                model="gpt-4-vision-preview", 
-                temperature=0,
-                api_key=api_key
-            )
-            print("Vision model initialized successfully")
-        except Exception as e:
-            print(f"Warning: Could not initialize vision model: {e}")
-            self.vision_model = None
-            
-        print(f"BasicAgent initialized with model: {model_name}")
-    
-    def analyze_question(self, question: str):
-        """Analyze the question to determine its type and approach."""
-        prompt = [
-            HumanMessage(content=f"""
-                You are an expert problem analyzer. Examine the following question to determine its type 
-                and the tools needed to answer it effectively.
-                
-                Question: {question}
-                
-                Determine what tools are needed for this question:
-                1. Mathematical calculation? If yes, extract the mathematical expression.
-                2. Web search for factual information? If yes, determine the search query.
-                3. Image processing? Is there a reference to an image or video that needs analysis?
-                4. File processing? Is there a reference to an attached file (Excel, code, etc.)?
-                5. YouTube video? Is there a YouTube URL that needs transcript analysis?
-                6. Complex reasoning? Does this question require multi-step logical reasoning?
-                7. Backward text? Does the question contain text that needs to be reversed?
-                
-                Respond in JSON format:
-                {{
-                    "thought": "your analysis of the question",
-                    "need_calculator": true/false,
-                    "math_expression": "expression to calculate (if needed, otherwise null)",
-                    "need_websearch": true/false,
-                    "search_query": "search query (if needed, otherwise null)",
-                    "need_image_processing": true/false,
-                    "need_file_processing": true/false,
-                    "file_type": "excel/code/other (if needed, otherwise null)",
-                    "has_youtube_url": true/false,
-                    "youtube_url": "URL (if found, otherwise null)",
-                    "need_complex_reasoning": true/false,
-                    "has_backward_text": true/false,
-                    "backward_text": "the text to reverse (if needed, otherwise null)"
-                }}
-            """)
-        ]
-        
-        try:
-            response = self.llm.invoke(prompt)
-            try:
-                result = json.loads(response.content)
-                thought = f"Analysis: {result.get('thought', 'No analysis provided')}"
-                
-                # Extract all the analysis results
-                need_calculator = result.get('need_calculator', False)
-                math_expression = result.get('math_expression')
-                need_websearch = result.get('need_websearch', False)
-                search_query = result.get('search_query')
-                need_image_processing = result.get('need_image_processing', False)
-                need_file_processing = result.get('need_file_processing', False)
-                file_type = result.get('file_type')
-                has_youtube_url = result.get('has_youtube_url', False)
-                youtube_url = result.get('youtube_url')
-                need_complex_reasoning = result.get('need_complex_reasoning', False)
-                has_backward_text = result.get('has_backward_text', False)
-                backward_text = result.get('backward_text')
-                
-                return {
-                    "thought": thought,
-                    "need_calculator": need_calculator,
-                    "math_expression": math_expression,
-                    "need_websearch": need_websearch,
-                    "search_query": search_query,
-                    "need_image_processing": need_image_processing,
-                    "need_file_processing": need_file_processing,
-                    "file_type": file_type,
-                    "has_youtube_url": has_youtube_url,
-                    "youtube_url": youtube_url,
-                    "need_complex_reasoning": need_complex_reasoning,
-                    "has_backward_text": has_backward_text,
-                    "backward_text": backward_text
-                }
-            except json.JSONDecodeError as e:
-                # Fallback in case of JSON parsing error
-                return {
-                    "thought": f"Couldn't parse the analysis response as JSON: {e}",
-                    "need_calculator": False,
-                    "math_expression": None,
-                    "need_websearch": False,
-                    "search_query": None,
-                    "need_image_processing": False,
-                    "need_file_processing": False,
-                    "file_type": None,
-                    "has_youtube_url": False,
-                    "youtube_url": None,
-                    "need_complex_reasoning": False,
-                    "has_backward_text": False,
-                    "backward_text": None
-                }
-        except Exception as e:
-            # Fallback for LLM invocation error
-            return {
-                "thought": f"Error invoking language model: {str(e)}",
-                "need_calculator": False,
-                "math_expression": None,
-                "need_websearch": False,
-                "search_query": None,
-                "need_image_processing": False,
-                "need_file_processing": False,
-                "file_type": None,
-                "has_youtube_url": False,
-                "youtube_url": None,
-                "need_complex_reasoning": False,
-                "has_backward_text": False,
-                "backward_text": None
-            }
-    
-    def solve_math(self, math_expression):
-        """Evaluate the mathematical expression."""
-        try:
-            # Simple math expression evaluator
-            result = eval(math_expression)
-            return f"Calculated {math_expression} = {result}", result
-        except Exception as e:
-            return f"Error calculating expression: {e}", None
-    
-    def web_search(self, query):
-        """Perform a web search for factual information."""
-        try:
-            # Use DuckDuckGo search API (no key required)
-            search_url = f"https://api.duckduckgo.com/?q={query}&format=json"
-            response = requests.get(search_url, timeout=10)
-            response.raise_for_status()
-            search_data = response.json()
-            
-            # Extract and format the search results
-            results = []
-            
-            # Abstract (main result)
-            if search_data.get('Abstract'):
-                results.append(f"Main result: {search_data.get('Abstract')}")
-            
-            # Related topics
-            for topic in search_data.get('RelatedTopics', [])[:5]:  # Limit to first 5 results
-                if 'Text' in topic:
-                    results.append(topic['Text'])
-            
-            if not results:
-                return f"No search results found for '{query}'", None
-            
-            search_results = "\n".join(results)
-            return f"Search results for '{query}':\n{search_results}", search_results
-        except requests.exceptions.RequestException as e:
-            return f"Error during web search: {e}", None
-        except Exception as e:
-            return f"Error processing search results: {e}", None
-    
-    def process_youtube_video(self, youtube_url):
-        """Extract and process information from a YouTube video."""
-        try:
-            # Extract video ID from URL
-            video_id = None
-            if "youtu.be" in youtube_url:
-                video_id = youtube_url.split("/")[-1].split("?")[0]
-            elif "youtube.com/watch" in youtube_url:
-                video_id = re.search(r"v=([^&]+)", youtube_url).group(1)
-            
-            if not video_id:
-                return f"Could not extract video ID from URL: {youtube_url}", None
-            
-            # Get video transcript
-            try:
-                transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
-                transcript_text = " ".join([item['text'] for item in transcript_list])
-                
-                # Get video metadata
-                with youtube_dl.YoutubeDL({'quiet': True}) as ydl:
-                    info = ydl.extract_info(youtube_url, download=False)
-                    title = info.get('title', 'Unknown title')
-                    description = info.get('description', 'No description')
-                    
-                video_info = {
-                    'title': title,
-                    'description': description,
-                    'transcript': transcript_text
-                }
-                
-                return f"Successfully extracted information from YouTube video: {title}", video_info
-            except Exception as e:
-                # If transcript fails, try to get just metadata
-                try:
-                    with youtube_dl.YoutubeDL({'quiet': True}) as ydl:
-                        info = ydl.extract_info(youtube_url, download=False)
-                        title = info.get('title', 'Unknown title')
-                        description = info.get('description', 'No description')
-                        
-                    video_info = {
-                        'title': title,
-                        'description': description,
-                        'transcript': "Transcript unavailable"
-                    }
-                    
-                    return f"Extracted partial information from YouTube video (no transcript): {title}", video_info
-                except Exception as e2:
-                    return f"Error getting YouTube video information: {e}, then {e2}", None
-        except Exception as e:
-            return f"Error processing YouTube video: {e}", None
-    
-    def process_backward_text(self, backward_text):
-        """Reverse backward text to get the intended meaning."""
-        try:
-            forward_text = backward_text[::-1]
-            return f"Reversed text: '{backward_text}' → '{forward_text}'", forward_text
-        except Exception as e:
-            return f"Error reversing text: {e}", None
-    
-    def reasoning(self, question, context=""):
-        """Perform step-by-step reasoning to solve the problem."""
-        prompt = [
-            HumanMessage(content=f"""
-                You are a general AI assistant. Solve the following question using careful step-by-step reasoning.
-                
-                Question: {question}
-                
-                {context}
-                
-                Provide a detailed step-by-step solution showing your thought process. Be methodical 
-                and thorough. DO NOT include a final answer yet, just your reasoning.
-            """)
-        ]
-        
-        try:
-            response = self.llm.invoke(prompt)
-            return response.content
-        except Exception as e:
-            return f"Error during reasoning step: {str(e)}"
-    
-    def complex_reasoning(self, question, context=""):
-        """Perform more intensive reasoning for complex questions."""
-        prompt = [
-            HumanMessage(content=f"""
-                You are an advanced reasoning agent. Solve the following complex question using multi-step reasoning.
-                Break down the problem into logical steps and consider different approaches.
-                
-                Question: {question}
-                
-                {context}
-                
-                Start by analyzing what the question is really asking for. Consider different interpretations if ambiguous.
-                Identify key entities, relationships, or constraints mentioned in the question.
-                Work through the solution step-by-step, making your reasoning explicit at each stage.
-                If appropriate, use diagrams, tables, or mathematical expressions to organize your reasoning.
-                
-                Provide your detailed multi-step reasoning. DO NOT include a final answer yet.
-            """)
-        ]
-        
-        try:
-            response = self.llm.invoke(prompt)
-            first_pass = response.content
-            
-            # Follow-up with a more focused analysis
-            prompt = [
-                HumanMessage(content=f"""
-                    You are an advanced reasoning agent. Review and refine the following reasoning:
-                    
-                    Question: {question}
-                    
-                    Initial reasoning:
-                    {first_pass}
-                    
-                    Identify any gaps, inconsistencies, or areas where the reasoning could be improved.
-                    Apply critical thinking to ensure the logic is sound and addresses all aspects of the question.
-                    Offer improved or alternative reasoning paths if appropriate.
-                    
-                    Provide your refined reasoning. DO NOT include a final answer yet.
-                """)
-            ]
-            
-            try:
-                response = self.llm.invoke(prompt)
-                return f"{first_pass}\n\nRefined analysis:\n{response.content}"
-            except Exception as e:
-                return first_pass
-        except Exception as e:
-            return f"Error during complex reasoning step: {str(e)}"
-    
-    def format_answer(self, question, thoughts, answer_context=None):
-        """Format the final answer according to Gaia Benchmark requirements."""
-        all_thoughts = "\n".join(thoughts)
-        
-        context_prompt = ""
-        if answer_context:
-            context_prompt = f"""
-            Additional context for final answer formulation:
-            {answer_context}
-            """
-        
-        prompt = [
-            HumanMessage(content=f"""
-                You are a general AI assistant. Based on the following reasoning, provide a concise 
-                final answer to the question. The final answer format is extremely important.
-                
-                Question: {question}
-                
-                Reasoning:
-                {all_thoughts}
-                
-                {context_prompt}
-                
-                Format your answer following these strict rules:
-                - For numbers: No commas in large numbers, no units unless explicitly requested.
-                - For strings: Minimal, no articles, no abbreviations.
-                - For lists: Comma-separated values.
-                - Use digits in words unless otherwise stated.
-                - If the answer is a country code, use the standard 3-letter format (e.g., "USA", "GBR").
-                - For names, provide exactly what was asked for (first name, last name, or full name).
-                
-                IMPORTANT: Your answer must be concise, precise and EXACT. Do not include explanations.
-                Do not include a prefix like "FINAL ANSWER:" in your response. Just provide the exact answer.
-                
-                For example, if the answer is "5", just respond with "5".
-                If the answer is "New York", just respond with "New York".
-            """)
-        ]
-        
-        try:
-            response = self.llm.invoke(prompt)
-            answer = response.content.strip()
-            
-            # Remove any "FINAL ANSWER:" prefix if it exists
-            if "FINAL ANSWER:" in answer:
-                parts = answer.split("FINAL ANSWER:")
-                answer = parts[1].strip()
-            
-            return answer
-        except Exception as e:
-            return f"Error formatting answer: {str(e)}"
-        
+    def __init__(self):
+        print("BasicAgent initialized.")
     def __call__(self, question: str) -> str:
-        """
-        Process the question and return a formatted response according to Gaia requirements.
-        """
         print(f"Agent received question (first 50 chars): {question[:50]}...")
-        
-        # Initialize an empty list to collect thoughts
-        thoughts = []
-        
-        try:
-            # Step 1: Analyze the question
-            analysis = self.analyze_question(question)
-            thoughts.append(analysis["thought"])
-            
-            # Initialize context variables
-            all_context = ""
-            answer_context = ""
-            
-            # Step 2: Process backward text if needed
-            if analysis["has_backward_text"] and analysis["backward_text"]:
-                backward_thought, forward_text = self.process_backward_text(analysis["backward_text"])
-                thoughts.append(backward_thought)
-                if forward_text:
-                    all_context += f"\nBackward text analysis:\n{backward_text} (backward) = {forward_text} (forward)\n"
-                    # Replace the backward text in the question with forward text for further processing
-                    question = question.replace(analysis["backward_text"], forward_text)
-            
-            # Step 3: Calculate if needed
-            if analysis["need_calculator"] and analysis["math_expression"]:
-                calc_thought, calc_result = self.solve_math(analysis["math_expression"])
-                thoughts.append(calc_thought)
-                if calc_result is not None:
-                    all_context += f"\nMathematical calculation:\nExpression: {analysis['math_expression']}\nResult: {calc_result}\n"
-            
-            # Step 4: Perform web search if needed
-            if analysis["need_websearch"] and analysis["search_query"]:
-                search_thought, search_result = self.web_search(analysis["search_query"])
-                thoughts.append(search_thought)
-                if search_result:
-                    all_context += f"\nWeb search results:\nQuery: {analysis['search_query']}\nResults: {search_result}\n"
-            
-            # Step 5: Process YouTube video if present
-            if analysis["has_youtube_url"] and analysis["youtube_url"]:
-                youtube_thought, youtube_info = self.process_youtube_video(analysis["youtube_url"])
-                thoughts.append(youtube_thought)
-                if youtube_info:
-                    video_context = f"\nYouTube video information:\nTitle: {youtube_info.get('title')}\n"
-                    
-                    # Add description if not too long
-                    description = youtube_info.get('description', '')
-                    if description and len(description) > 500:
-                        description = description[:500] + "... [truncated]"
-                    video_context += f"Description: {description}\n"
-                    
-                    # Add transcript if available (truncated if too long)
-                    transcript = youtube_info.get('transcript', '')
-                    if transcript and transcript != "Transcript unavailable":
-                        if len(transcript) > 1000:
-                            transcript = transcript[:1000] + "... [truncated]"
-                        video_context += f"Transcript excerpt: {transcript}\n"
-                    
-                    all_context += video_context
-            
-            # Step 6: Handle image/file processing references with appropriate messages
-            if analysis["need_image_processing"]:
-                image_thought = "Note: Referenced image not available for processing."
-                thoughts.append(image_thought)
-                all_context += "\n" + image_thought + "\n"
-                answer_context += "\nIf the question references an image or visual content that is not available, respond with an appropriate message indicating you cannot access the image.\n"
-            
-            if analysis["need_file_processing"]:
-                file_type = analysis["file_type"] or "file"
-                file_thought = f"Note: Referenced {file_type} not available for processing."
-                thoughts.append(file_thought)
-                all_context += "\n" + file_thought + "\n"
-                answer_context += f"\nIf the question references a {file_type} that is not available, respond with an appropriate message indicating you cannot access the {file_type}.\n"
-            
-            # Step 7: Perform reasoning (complex or standard)
-            if analysis["need_complex_reasoning"]:
-                reasoning_output = self.complex_reasoning(question, all_context)
-            else:
-                reasoning_output = self.reasoning(question, all_context)
-                
-            thoughts.append(reasoning_output)
-            
-            # Step 8: Format final answer
-            final_answer = self.format_answer(question, thoughts, answer_context)
-            
-            print(f"Agent returning answer: {final_answer}")
-            return final_answer
-        except Exception as e:
-            print(f"Error running agent workflow: {e}")
-            # Fallback response - return just the error message without the prefix
-            return "Error"
+        fixed_answer = "This is a default answer."
+        print(f"Agent returning fixed answer: {fixed_answer}")
+        return fixed_answer
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):
     """
@@ -519,7 +41,7 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
 
     # 1. Instantiate Agent ( modify this part to create your agent)
     try:
-        agent = BasicAgent()
+        agent = LangGraphAgent()
     except Exception as e:
         print(f"Error instantiating agent: {e}")
         return f"Error initializing agent: {e}", None
@@ -589,7 +111,6 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
             f"Message: {result_data.get('message', 'No message received.')}"
         )
         print("Submission successful.")
-        print(result_data)
         results_df = pd.DataFrame(results_log)
         return final_status, results_df
     except requests.exceptions.HTTPError as e: