chore: rollback

.gitignore

@@ -1,5 +0,0 @@
-.env
-.python-version
-venv/
-__pycache__/
-*.pyc

FinalAssignmentAgent.py

@@ -1,113 +0,0 @@
-import os
-from smolagents import CodeAgent, DuckDuckGoSearchTool, InferenceClientModel
-from tools import WikipediaSearchTool, CoherenceValidatorTool, MultimodalAnalysisTool, FileDownloaderTool
-from huggingface_hub import login
-
-class FinalAssignmentAgent:
-    def __init__(self):
-        # Modelo HuggingFaceInferenceAPI otimizado
-        self.model = InferenceClientModel(model="Qwen/Qwen2.5-Coder-32B-Instruct", temperature=0)
-        self.wikipedia_tool = WikipediaSearchTool()
-        self.coherence_tool = CoherenceValidatorTool()
-        self.file_downloader = FileDownloaderTool()
-        self.multimodal_tool = MultimodalAnalysisTool()
-        self.validation_cache = {}  # Cache para validações
-
-        self.code_agent = self.CodeSubAgent(self.model, self)
-        self.general_agent = self.GeneralSubAgent(self.model, self)
-        
-    def classify_with_llm(self, question: str) -> str:
-        prompt = ("Classify the following task as 'code' or 'general'. "
-                    "Respond with only one word: 'code' or 'general'.\n"
-                    f"Task: {question}"
-                )
-        messages = [{"role": "user", "content": [{"type": "text", "text": prompt}]}]
-        response = self.model(messages)
-        if hasattr(response, 'content'):
-            text = response.content.strip().lower()
-        else:
-            text = str(response).strip().lower()
-        print(f"[Router] Decisão da LLM: {text}")
-        return "code" if "code" in text else "general"
-
-    def validate_answer(self, question, answer):
-        # Heurística simples antes de chamar o validador
-        if len(answer) < 20 or "não sei" in answer.lower():
-            return "CRITIQUE: Resposta curta ou evasiva."
-        # Cache de validação
-        cache_key = f"{question}|||{answer}"
-        if cache_key in self.validation_cache:
-            return self.validation_cache[cache_key]
-        # Prompt enxuto
-        audit_prompt = (
-            f"Q: {question}\nA: {answer}\n"
-            "Evaluate if A answers Q directly, without errors or inconsistencies. "
-            "Reply only with 'COHERENCE_CHECK_PASSED' or 'CRITIQUE: ...'."
-        )
-        result = self.model(audit_prompt)
-        self.validation_cache[cache_key] = result
-        return result
-
-    def __call__(self, question: str) -> str:
-        agent_type = self.classify_with_llm(question)
-        # Prompt enxuto para o sistema
-        system_prompt = (
-            "You are a multimodal agent specialized in text, images, videos, and code. "
-            "Follow this workflow: 1) Identify the task type. 2) Make a short plan. "
-            "3) Only validate the plan if it involves multiple steps or tools. "
-            "4) Execute. 5) Before responding, validate the final answer. "
-            "If the validator returns CRITIQUE, correct and try again (max 2 attempts)."
-        )
-        
-        if agent_type == "code":
-            return self.code_agent.run(question, system_prompt=system_prompt)
-        else:
-            return self.general_agent.run(question, system_prompt=system_prompt)
-
-    class CodeSubAgent:
-        def __init__(self, model, parent):
-            self.model = model
-            self.parent = parent
-            self.agent = CodeAgent(
-                tools=[],
-                model=model,
-                max_steps=2,
-                add_base_tools=True,    
-            )
-
-        def run(self, question, system_prompt=None):
-            full_query = f"{system_prompt}\n\nTask: {question}" if system_prompt else question
-            answer = self.agent.run(full_query)
-            # Validação final otimizada
-            validation = self.parent.validate_answer(question, answer)
-            if "COHERENCE_CHECK_PASSED" in str(validation):
-                return answer
-            else:
-                # Tenta corrigir uma vez
-                answer2 = self.agent.run(full_query)
-                validation2 = self.parent.validate_answer(question, answer2)
-                return answer2 if "COHERENCE_CHECK_PASSED" in str(validation2) else validation2
-
-    class GeneralSubAgent:
-        def __init__(self, model, parent):
-            self.model = model
-            self.parent = parent
-            self.agent = CodeAgent(
-                tools=[DuckDuckGoSearchTool(), WikipediaSearchTool(), CoherenceValidatorTool(), MultimodalAnalysisTool(), FileDownloaderTool()],
-                add_base_tools=True,
-                max_steps=2,
-                model=model,
-            )
-    
-        def run(self, question, system_prompt=None):
-            full_query = f"{system_prompt}\n\nTask: {question}" if system_prompt else question
-            answer = self.agent.run(full_query)
-            # Validação final otimizada
-            validation = self.parent.validate_answer(question, answer)
-            if "COHERENCE_CHECK_PASSED" in str(validation):
-                return answer
-            else:
-                # Tenta corrigir uma vez
-                answer2 = self.agent.run(full_query)
-                validation2 = self.parent.validate_answer(question, answer2)
-                return answer2 if "COHERENCE_CHECK_PASSED" in str(validation2) else validation2

requirements.txt

@@ -1,5 +1,2 @@
-gradio[oauth]
-requests
-smolagents
-opencv-python
-huggingface_hub
+gradio
+requests

runtime.txt

@@ -1 +0,0 @@
-python-3.10

test.py

@@ -1,22 +0,0 @@
-from FinalAssignmentAgent import FinalAssignmentAgent
-
-# Inicializa o agente
-agent = FinalAssignmentAgent()
-
-# Teste 1: Lógica e Busca
-print("--- Teste de Busca e Lógica ---")
-print("Quem foi Alan Turing e qual a sua principal contribuição?")
-response = agent("Quem foi Alan Turing e qual a sua principal contribuição?")
-print(f"Resposta: {response}")
-
-# Teste 2: Código e Matemática
-print("\n--- Teste de Código ---")
-print("Calcule a raiz quadrada de 144 e multiplique por 5.")
-response = agent("Calcule a raiz quadrada de 144 e multiplique por 5.")
-print(f"Resposta: {response}")
-
-# Teste 3: Task ID (Simulado)
-# Nota: Este teste só funcionará se o servidor da Unit 4 estiver ativo
-print("\n--- Teste de Task ID ---")
-response = agent("What is in the file for task_id 100?")
-print(f"Resposta: {response}")

tools.py

@@ -1,181 +0,0 @@
-from smolagents import Tool, DuckDuckGoSearchTool
-from huggingface_hub import InferenceClient
-import requests
-import os
-import cv2
-
-class WikipediaSearchTool(Tool):
-    name = "wikipedia_search"
-    description = (
-        "Use this tool to find factual information, dates, and descriptions from Wikipedia. "
-        "It prioritizes official Wikipedia pages to ensure accuracy for names and historical events."
-    )
-    inputs = {
-        "query": {
-            "type": "string",
-            "description": "The person, place, event, or object to search for.",
-        },
-        "date_context": {
-            "type": "string",
-            "description": "Optional: A specific year or date to narrow down the search (e.g., '1969').",
-            "nullable": True
-        }
-    }
-    output_type = "string"
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        # Initialize the base search engine as a component
-        self.search_engine = DuckDuckGoSearchTool()
-
-    def forward(self, query: str, date_context: str = None) -> str:
-        # 1. Build a specialized query
-        # Adding 'site:en.wikipedia.org' ensures the top results are from Wikipedia
-        refined_query = f"{query} site:en.wikipedia.org"
-        
-        if date_context:
-            refined_query += f" \"{date_context}\""
-
-        try:
-            # 2. Execute the search
-            results = self.search_engine(refined_query)
-            
-            if not results or "no results" in results.lower():
-                return f"No specific Wikipedia entry found for '{target_name}' with the given context."
-            
-            return results
-        except Exception as e:
-            return f"An error occurred while searching Wikipedia: {str(e)}"
-        
-class CoherenceValidatorTool(Tool):
-    name = "coherence_validator"
-    description = "Checks if a plan or answer is semantically coherent and directly addresses the user request. It validates the relationship between names, ages, logic, and code without re-calculating everything."
-    inputs = {
-        "original_question": {
-            "type": "string",
-            "description": "The user's original query."
-        },
-        "proposed_content": {
-            "type": "string",
-            "description": "The plan, code snippet, or final answer to be audited."
-        }
-    }
-    output_type = "string"
-
-    def forward(self, original_question: str, proposed_content: str) -> str:
-        audit_prompt = f"""
-        Role: Senior Semantic Auditor
-        Task: Evaluate if the 'Proposed Content' is a coherent and valid response to the 'Original Question'.
-
-        Original Question: {original_question}
-        Proposed Content: {proposed_content}
-
-        Evaluation Criteria:
-        1. RELEVANCE: Does the content directly address all parts of the question?
-        2. ENTITY CONSISTENCY: Do names, ages, and dates remain consistent throughout the text?
-        3. CODE/MATH LOGIC: Does the code or mathematical approach 'make sense' for this specific problem (e.g., not calculating temperature when asked for age)?
-        4. ABSURDITY CHECK: Are there any hallucinations or impossible claims?
-
-        Instructions:
-        - If coherent, return: "COHERENCE_CHECK_PASSED"
-        - If NOT coherent, return: "CRITIQUE: [detailed explanation of what doesn't make sense]"
-        """
-        return audit_prompt
-
-class FileDownloaderTool(Tool):
-    name = "file_downloader"
-    description = "Downloads task-related files (images or videos) from the evaluation server using a task_id."
-    inputs = {
-        "task_id": {
-            "type": "string",
-            "description": "The ID of the task to download the file for."
-        }
-    }
-    output_type = "string"
-
-    def forward(self, task_id: str) -> str:
-        api_url = "https://agents-course-unit4-scoring.hf.space/files"
-        try:
-            response = requests.get(f"{api_url}/{task_id}", timeout=15)
-            if response.status_code == 200:
-                # Detect extension
-                ctype = response.headers.get('Content-Type', '')
-                ext = ".mp4" if "video" in ctype else ".jpg"
-                filename = f"file_{task_id}{ext}"
-                
-                with open(filename, "wb") as f:
-                    f.write(response.content)
-                return filename # Returns the local path for other tools to use
-            return f"Error: Server returned status {response.status_code}"
-        except Exception as e:
-            return f"Download failed: {str(e)}"
-
-class MultimodalAnalysisTool(Tool):
-    name = "multimodal_analyzer"
-    description = "Analyzes images and videos. For images, it provides a description. For videos, it extracts frames every 2 seconds to describe the sequence of events."
-    inputs = {
-        "file_path": {
-            "type": "string",
-            "description": "Local path to the image (.jpg, .png) or video (.mp4) file."
-        },
-        "query": {
-            "type": "string",
-            "description": "Optional: Specific question about the image or video content.",
-            "nullable": True
-        }
-    }
-    output_type = "string"
-
-    def forward(self, file_path: str, query: str = None) -> str:
-        client = InferenceClient(model="Salesforce/blip-image-captioning-large")
-        
-        if not os.path.exists(file_path):
-            return f"Error: File {file_path} not found."
-
-        if file_path.lower().endswith(('.mp4', '.avi', '.mov')):
-            descriptions = []
-            video = cv2.VideoCapture(file_path)
-            fps = video.get(cv2.CAP_PROP_FPS)
-            if fps == 0: fps = 24
-            frame_interval = int(fps * 2)
-            
-            count = 0
-            while True:
-                success, frame = video.read()
-                if not success:
-                    break
-                
-                if count % frame_interval == 0:
-                    temp_frame = f"temp_frame_{count}.jpg"
-                    cv2.imwrite(temp_frame, frame)
-                    with open(temp_frame, "rb") as f:
-                        # Analyze the specific frame
-                        desc = client.image_to_text(f.read())
-                        timestamp = count // fps
-                        descriptions.append(f"At {timestamp}s: {desc}")
-                    os.remove(temp_frame)
-                count += 1
-            
-            video.release()
-            return "Video Content Summary: " + " | ".join(descriptions)
-        else:
-            try:
-                with open(file_path, "rb") as f:
-                    image_data = f.read()
-                    description = client.image_to_text(image_data)
-                return f"Image Analysis: {description}"
-            except Exception as e:
-                return f"Error analyzing image: {str(e)}"
-            
-def classify_with_llm(self, question: str) -> str:
-    prompt = (
-        "Classify the following task as 'code' or 'general'. "
-        "Respond with only one word: 'code' or 'general'.\n"
-        f"Task: {question}"
-    )
-    response = self.model(prompt)
-    if "code" in response.lower():
-        return "code"
-    else:
-        return "general"
-