Test_Run_with_CodeAgent

answering.py

@@ -1,17 +0,0 @@
-import os
-
-from smolagents import CodeAgent, InferenceClientModel, FinalAnswerTool, DuckDuckGoSearchTool, WikipediaSearchTool, PythonInterpreterTool
-
-from dotenv import load_dotenv
-load_dotenv()
-
-
-def gen_question_answer(question: str) -> str:
-    duck_duck_go_search_tool = DuckDuckGoSearchTool()
-    wikipedia_search_tool = WikipediaSearchTool()
-    final_answer_tool = FinalAnswerTool()
-    python_interpreter_tool = PythonInterpreterTool()    
-    agent = CodeAgent(tools=[duck_duck_go_search_tool, wikipedia_search_tool, python_interpreter_tool, final_answer_tool], model=InferenceClientModel(), add_base_tools=True, additional_authorized_imports=["pandas"]) 
-    
-    response = agent.run(question)  # Use run() instead of query()
-    return str(response)

app.py

@@ -3,11 +3,6 @@ import gradio as gr
 import requests
 import inspect
 import pandas as pd
-from workflow import GAIAAnsweringWorkflow
-
-
-from dotenv import load_dotenv
-load_dotenv()
 
 # (Keep Constants as is)
 # --- Constants ---
@@ -15,16 +10,14 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
 # --- Basic Agent Definition ---
 # ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
-class JeSaisToutAgent(GAIAAnsweringWorkflow):
+class BasicAgent:
     def __init__(self):
-        super().__init__()
-        print("Agentics Workflow initialized.")
-    def __call__(self, question: str, context) -> str:
+        print("BasicAgent initialized.")
+    def __call__(self, question: str) -> str:
         print(f"Agent received question (first 50 chars): {question[:50]}...")
-        answer = "This is a default answer."
-        answer=super().__call__(question, context)
-        print(f"Agent returning answer: {answer}")
-        return answer
+        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):
     """
@@ -47,7 +40,7 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
 
     # 1. Instantiate Agent ( modify this part to create your agent)
     try:
-        agent = JeSaisToutAgent()
+        agent = BasicAgent()
     except Exception as e:
         print(f"Error instantiating agent: {e}")
         return f"Error initializing agent: {e}", None
@@ -65,13 +58,13 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
              print("Fetched questions list is empty.")
              return "Fetched questions list is empty or invalid format.", None
         print(f"Fetched {len(questions_data)} questions.")
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching questions: {e}")
+        return f"Error fetching questions: {e}", None
     except requests.exceptions.JSONDecodeError as e:
          print(f"Error decoding JSON response from questions endpoint: {e}")
          print(f"Response text: {response.text[:500]}")
          return f"Error decoding server response for questions: {e}", None
-    except requests.exceptions.RequestException as e:
-        print(f"Error fetching questions: {e}")
-        return f"Error fetching questions: {e}", None
     except Exception as e:
         print(f"An unexpected error occurred fetching questions: {e}")
         return f"An unexpected error occurred fetching questions: {e}", None
@@ -83,15 +76,11 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
     for item in questions_data:
         task_id = item.get("task_id")
         question_text = item.get("question")
-        question_context = item
-        print(f"=== item[{task_id}]===")
-        print(item)
-        print("\n")
         if not task_id or question_text is None:
             print(f"Skipping item with missing task_id or question: {item}")
             continue
         try:
-            submitted_answer = agent(question_text, question_context)
+            submitted_answer = agent(question_text)
             answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
             results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})
         except Exception as e:
@@ -101,19 +90,12 @@ def run_and_submit_all( profile: gr.OAuthProfile | None):
     if not answers_payload:
         print("Agent did not produce any answers to submit.")
         return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)
-    if os.getenv("TESTRUN"):
-        print("Skipping submission in test environment.")
-        return "Skipping submission in test environment.", pd.DataFrame(results_log)
-    
+
     # 4. Prepare Submission 
     submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
     status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
     print(status_update)
-    
-    if os.getenv("TESTRUN"):
-        print(f"Skipping submission in test environment. \n{submission_data}")
-        return "Skipping submission in test environment.", pd.DataFrame(submission_data)
-    
+
     # 5. Submit
     print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
     try:

requirements.txt

@@ -1,17 +1,2 @@
 gradio
-requests
-pandas
-langgraph
-smolagents
-transformers
-torch
-huggingface_hub
-python-dotenv
-pillow
-duckduckgo-search
-wikipedia-api
-markdownify 
-langchain_huggingface
-langchain-community
-llama-index
-sseclient-py
+requests

workflow.py

@@ -1,316 +0,0 @@
-from typing import List, Dict, TypedDict, Annotated, Callable, Optional, Any
-from huggingface_hub.inference._generated.types import question_answering
-from langgraph.graph import StateGraph, END
-from langchain_core.messages import HumanMessage
-#from langchain_community.chat_models import ChatHuggingFace
-from langchain_huggingface import ChatHuggingFace, HuggingFaceEndpoint
-from answering import gen_question_answer
-
-import requests
-import os
-from dotenv import load_dotenv
-load_dotenv()
-
-
-class AgentState(TypedDict):
-    context: Dict[str, str]
-    question: str
-    answer: Annotated[str, lambda x, y: y]  # Overwrite with new value
-    formatted_answer: Annotated[str, lambda x, y: y]
-    format_requirement: str    
-    uris: List[str]
-    reasoning: List[str]
-
-class GAIAAnsweringWorkflow:
-    def __init__(
-        self,
-        qa_function: Optional[Callable[[str], str]] = None,
-        formatter: Optional[Callable[[str], str]] = None
-    ):
-        """
-        Initialize the GAIA agent workflow
-        
-        Args: 
-            qa_function: Core question answering function (gen_question_answer)
-            formatter: Answer formatting function (default: GAIA boxed format)
-        """
-        self.qa_function = gen_question_answer #qa_function or self.default_qa_function
-        self.formatter = formatter or self.default_formatter
-        self.workflow = self.build_workflow()
-        # Initialize model with HF Inference API
-        llm_endpoint = HuggingFaceEndpoint(
-            model="deepseek-ai/DeepSeek-R1",#endpoint_url="https://api-inference.huggingface.co/models/cortexso/deepseek-r1:7b",
-            huggingfacehub_api_token=os.getenv("HF_TOKEN"),
-            task="text-generation",
-            #max_tokens=1024
-        )
-        self.reasoning_llm  = ChatHuggingFace(llm=llm_endpoint)
-
-        self.llm = ChatHuggingFace(
-            llm=HuggingFaceEndpoint(
-                model="mistralai/Mistral-7B-Instruct-v0.3",
-                huggingfacehub_api_token=os.getenv("HF_TOKEN"),
-                task="text-generation",                
-            )
-
-        )
-
-
-
-    def ask_llm(self, question: str, do_reasoning=False)->str:
-        prompt = question
-        messages = [HumanMessage(content=prompt)]
-        response = self.llm.invoke(messages) 
-        answer = str(response.content)      
-        return answer
-
-
-    def extract_noted_urls_with_llm(self, question: str) -> List[str]:
-        """Use LLM to extract URLs specifically noted in the question"""
-        prompt = f"""
-        Analyze the following question and extract ONLY URLs that are explicitly noted or referenced.
-        Return each URL on a separate line. If no URLs are noted, return an empty string.
-        
-        QUESTION: {question}
-        
-        Respond ONLY with the URLs, one per line, with no additional text or formatting.
-        """
-        
-        try:
-            # Use your LLM to generate the response
-            response = self.ask_llm(prompt)
-            
-            # Parse the response to extract URLs
-            urls = []
-            for line in response.split('\n'):
-                line = line.strip()
-                #if line.startswith(('http://', 'https://', 'www.')):
-                urls.append(line)
-            
-            return urls
-        except Exception as e:
-            print(f"LLM-based URL extraction failed: {str(e)}")
-            return []
-
-    def download_file(self, task_id: str, file_name: str) -> str:
-        """Download file from API and return local path"""
-        try:
-            #os.makedirs("files", exist_ok=True)
-            file_path = f"./{file_name}"#files/{file_id}"
-            api_base_url: str = "https://agents-course-unit4-scoring.hf.space"
-            api_endpoint = f"{api_base_url}/files/{task_id}"
-            response = requests.get(api_endpoint)
-            response.raise_for_status()
-            
-            with open(file_path, "wb") as f:
-                f.write(response.content)
-                
-            print(f"File saved: {file_path}")
-            return file_path
-        except Exception as e:
-            print(f"File download failed: {str(e)}")
-            return ""
-
-    
-    def check_context_independent(self, state: AgentState)->bool:
-        if ctx := state.get("context"):
-            if ctx.get("filename"): 
-                return False
-        prompt = f"""
-        
-        I have a CodeAgent based on the text-to-text model that can use Internet search and parse the information found. 
-        If this approach is enough to successfully cope with the task, then we will call such a task an "easy question"
-
-        AS AN ERUDITE PERSON  YOU must analyze how difficult it will be to solve the next question 
-        <<{state["question"]}>>
-        
-        If you think that the question is easy, then return an empty string. Important! You should NOT add any symbols to the output in this case!
-        If the question concerns the use of additional resources such as complex analysis of downloaded files or resources on the Internet, then return an action plan
-
-        """
-        reply = self.ask_llm(prompt, True)
-        prompt = f""" The reasonings from other LLM is provided: <<{reply}>>
-        You have to Summarize: 
-        output either empty string ('') for easy question 
-        or extract action plan for non-easy question
-        """
-        reply = self.ask_llm(prompt, False)
-        if reply:
-            state["reasoning"].append(reply)
-            return False
-        return True         
-
-    def preparations_node(self, state: AgentState) -> dict:        
-        if not state["context"]:
-            return {}
-        """Node to prepare resources"""
-        context = state["context"]
-        question = state["question"]
-        uris = state["uris"]
-        
-        # 1. Handle file_id in context
-        if file_name:= context.get("file_name"):
-            file_path = self.download_file(context["task_id"], file_name)
-            if file_path:
-                uris.append(file_path)
-        
-        # 2. Extract URLs from question
-        found_urls = self.extract_noted_urls_with_llm(question)
-        if found_urls:
-            uris.extend(found_urls)
-            print(f"Added {len(found_urls)} URL(s) from question")
-        
-        return {"uris": uris}
-        
-
-    def triage_node(self, state: AgentState) -> dict:
-        return {}
-    
-    def deep_processing_node(self, state: AgentState) -> dict:
-        question = f"""
-        question: \n <<{state["question"]}>> \n 
-        resources: {str(state["uris"])} \n 
-        reasoning: <<{state["reasoning"]}>>
-        """
-        answer = gen_question_answer(question)
-        return {state["answer"]: answer}    
-
-    def generate_answer_node(self, state: AgentState) -> dict:
-        """Node that executes the question answering tool"""
-        try:
-            answer = self.qa_function(state["question"])
-            return {"answer": answer}
-        except Exception as e:
-            print(str(e))
-            return {"answer": f"Error: {str(e)}"}
-    
-    def format_output_node(self, state: AgentState) -> dict:
-        """Node that formats the answer for GAIA benchmark"""
-        prompt = f"""
-        
-        As a very smart person, you should formulate what should be the output format of the answer to the question:
-        <<{state["question"]}>>
-
-        You must formulate it very briefly and clearly!
-        The common requirements is:
-        << 
-    OUR FINAL ANSWER should be a number OR as few words as possible OR a comma separated list of numbers and/or strings.
-
-    If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise, and don't include additional text. 
-    If the answer is a number, represent it with digits.
-    If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. 
-    If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string.
-    >>
-        But you have to figure out how the answer should looks like in the given case and reformulate requirement according to the specified question
-
-        """
-        format_requirement = self.ask_llm(prompt, True)
-
-        prompt = f"""
-Your attentiveness and responsibility are very much needed! We are solving a strict test that is automatically checked, so we must formulate the answer in strict accordance with the task and the required format! Even one incorrect symbol in the answer can fail the task! Pull yourself together!
-
-You will be required to produce an output of the answer, but formatted in accordance with the task
-
-Received answer: <<{state['answer']}>>
-
-Format requirements: <<{format_requirement}>>   
-
-Do NOT include << >> in your answer! Don't use full answer formulations! If you are asked about a number it MUST be just a number, nothing more! Each time it should be a clear answer (checked automatically) 
-        """ 
-        try:
-            formatted = self.ask_llm(prompt)
-            return {"formatted_answer": formatted.strip()}
-        except Exception as e:
-            return {"formatted_answer": f"\\boxed{{\\text{{Formatting error: {str(e)}}}}}"}
-    
-    def build_workflow(self) -> Any:
-        """Construct and compile the LangGraph workflow"""
-        # Create graph
-        workflow = StateGraph(AgentState)
-        
-        # Add nodes
-        workflow.add_node("preparations", self.preparations_node)
-        workflow.add_node("triage", self.triage_node)
-        workflow.add_node("deep_processing", self.deep_processing_node)
-        workflow.add_node("generate_answer", self.generate_answer_node)
-        workflow.add_node("format_output", self.format_output_node)
-        
-        # Define edges
-        workflow.set_entry_point("preparations")
-        workflow.add_edge("preparations", "triage")
-        workflow.add_conditional_edges("triage"
-                                        , self.check_context_independent
-                                        , {
-                                            True: "generate_answer",
-                                            False: "deep_processing"
-                                        })
-        workflow.add_edge("deep_processing", "format_output")
-
-        workflow.add_edge("generate_answer", "format_output")
-        workflow.add_edge("format_output", END)
-        
-        return workflow.compile()
-
-
-    
-    def __call__(self, question: str, context: Dict|None=None) -> str:
-        """
-        Execute the agent workflow for a given question
-        
-        Args:
-            question: Input question string
-            
-        Returns:
-            Formatted GAIA answer
-        """
-        # Initialize state
-        initial_state = {
-            "context": context,
-            "question": question,
-            "answer": "",
-            "formatted_answer": "",
-            "format_requirement": "",    
-            "uris": [],
-            "reasoning": []            
-        }
-        
-        # Execute workflow
-        result = self.workflow.invoke(initial_state)
-        return result["answer"]#["formatted_answer"]
-
-    @staticmethod
-    def default_qa_function(question: str) -> str:
-        """Placeholder QA function (override with your CodeAgent)"""
-        return "42"
-    
-    @staticmethod
-    def default_formatter(answer: str) -> str:
-        """Default GAIA formatting"""
-        return answer #f"\\boxed{{{answer}}}"
-
-# Example usage with custom QA function
-if __name__ == "__main__":
-    # Custom QA function (replace with your CodeAgent integration)
-    def custom_qa(question: str) -> str:
-        if "life" in question:
-            return "42"
-        elif "prime" in question:
-            return "101"
-        return "unknown"
-    
-    # Create agent instance
-    agent = GAIAAnsweringWorkflow(
-        qa_function=gen_question_answer,
-        formatter=lambda ans: ans #f"ANSWER: \\boxed{{{ans}}}"  # Custom formatting
-    )
-    
-    # Test cases
-    questions = [
-        "What is the answer to life, the universe, and everything?",
-        "What is the smallest 3-digit prime number?",
-        "Unknown question type?"
-    ]
-    
-    for q in questions:
-        result = agent(q)
-        print(f"Question: {q}\nAnswer: {result}\n")