import os import gradio as gr import requests import pandas as pd import traceback import time import mimetypes from tempfile import NamedTemporaryFile # Import smol-agent and tool components from smolagents import CodeAgent, LiteLLMModel, tool from smolagents import DuckDuckGoSearchTool from unstructured.partition.auto import partition # Imports for advanced file processing import speech_recognition as sr from pydub import AudioSegment # --- Constants --- DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" # --- Constants --- DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" # --- Tool Definition (Upgraded for Full Multimodality with pydub) --- @tool def file_reader(file_path: str) -> str: """ Reads and analyzes the content of a file and returns relevant text-based information. Supports: - Text files (PDF, TXT, CSV) - Images (PNG, JPG) with OCR - Audio (MP3, WAV) via speech recognition - Video (MP4, MOV) via speech recognition on audio track Can be used with a local file path or a web URL. Args: file_path (str): The local path or web URL of the file to be read. Returns: str: Extracted or transcribed content as text. """ temp_file_path = None audio_temp_path = None try: # Download the file if it's a URL if file_path.startswith("http://") or file_path.startswith("https://"): temp_file_path = NamedTemporaryFile(delete=False).name response = requests.get(file_path, timeout=20) response.raise_for_status() with open(temp_file_path, "wb") as f: f.write(response.content) local_path = temp_file_path else: local_path = file_path mime_type, _ = mimetypes.guess_type(local_path) recognizer = sr.Recognizer() if mime_type: # Handle audio files if mime_type.startswith("audio/"): with sr.AudioFile(local_path) as source: audio = recognizer.record(source) return recognizer.recognize_whisper(audio) # Handle video files by extracting audio with pydub elif mime_type.startswith("video/"): with NamedTemporaryFile(suffix=".wav", delete=False) as audio_temp: audio_temp_path = audio_temp.name # Extract audio using pydub video_audio = AudioSegment.from_file(local_path, format=mime_type.split('/')[1]) video_audio.export(audio_temp_path, format="wav") with sr.AudioFile(audio_temp_path) as source: audio = recognizer.record(source) return recognizer.recognize_whisper(audio) # Default to handling text and images with OCR if not audio/video elements = partition(local_path) return "\n\n".join([str(el) for el in elements]) except Exception as e: return f"Error reading or processing file '{file_path}': {e}" finally: # Clean up the downloaded file if it exists if temp_file_path and os.path.exists(temp_file_path): os.remove(temp_file_path) # Clean up the temporary audio file if audio_temp_path and os.path.exists(audio_temp_path): os.remove(audio_temp_path) # --- Agent Class (Updated with More Powerful Model and Tools) --- class GaiaSmolAgent: def __init__(self): """ Initializes the optimized agent. Now uses a more powerful model and the agent's native conversation memory. """ print("Initializing Optimized GaiaSmolAgent...") api_key = os.getenv("GEMINI_API_KEY") if not api_key: raise ValueError("API key 'GEMINI_API_KEY' not found in environment secrets.") # Use a more powerful, "clever" model for better reasoning. model = LiteLLMModel( model_id="gemini/gemini-1.5-pro-latest", api_key=api_key, temperature=0.0, timeout=120.0, # Add a timeout to prevent hanging ) # Store the sophisticated system prompt as an instance variable. self.system_prompt = """ You are an expert-level research assistant AI. Your sole purpose is to answer the user's question by breaking it down into logical steps and using the provided tools. You will have access to the conversation history, so use it for context. **Available Tools:** - `duck_duck_go_search(query: str) -> str`: Use this to find information, file URLs, or anything on the web. - `file_reader(file_path: str) -> str`: Use this to read the contents of a file from a local path or a web URL. It can read text, extract text from images (OCR), and transcribe audio from audio/video files. **Your Thought Process:** 1. **Deconstruct the Goal:** Carefully analyze the question to understand what information is needed, considering the previous turns in the conversation. 2. **Formulate a Plan:** Think step-by-step about which tools to use in what order. For example, you might need to search for a URL first, then read the content of that URL. 3. **Execute & Analyze:** Call the necessary tools. Carefully examine the output of each tool to extract the required facts. You can write Python code to process the data returned by the tools. 4. **Synthesize the Answer:** Once you have gathered sufficient information, formulate a final, concise answer to the original question. **CRITICAL INSTRUCTIONS:** - Your final action MUST be a single call to the `final_answer(answer: str)` function. - The `answer` argument must be a string containing only the definitive answer. - All code you write is executed in a restricted Python environment. You can define variables and write logic to process the tool outputs before calling `final_answer`. - Do not ask for clarification. Directly proceed to solve the problem. """ # Initialize the agent with the updated file_reader tool and memory settings. self.agent = CodeAgent( model=model, tools=[file_reader, DuckDuckGoSearchTool()], add_base_tools=True, # Provides the python interpreter and the final_answer function planning_interval=3 # Re-plan every 3 steps, considering memory. ) print("Optimized GaiaSmolAgent initialized successfully with native memory and full multimodal capabilities.") def __call__(self, question: str, reset_memory: bool = False) -> str: """ Directly runs the agent to generate and execute a plan to answer the question. It leverages the agent's built-in memory, controlled by the `reset` parameter. Args: question (str): The user's question. reset_memory (bool): If True, the agent's conversation memory will be cleared before running. Maps to the agent's `reset` parameter. """ print(f"Optimized Agent received question: {question[:100]}...") try: # Combine the system prompt with the current question. The agent will handle the history. full_prompt = f"{self.system_prompt}\n\nCURRENT TASK:\nUser Question: \"{question}\"" # Use the agent's `reset` parameter to control conversation memory. # `reset=False` keeps the memory from previous calls. final_answer = self.agent.run(full_prompt, reset=reset_memory) except Exception as e: print(f"FATAL AGENT ERROR: An exception occurred during agent execution: {e}") print(traceback.format_exc()) # Print full traceback for easier debugging return f"FATAL AGENT ERROR: {e}" print(f"Optimized Agent returning final answer: {final_answer}") return str(final_answer) # --- Main Application Logic (Unchanged) --- def run_and_submit_all( profile: gr.OAuthProfile | None): """ Fetches all questions, runs the BasicAgent on them, submits all answers, and displays the results. """ # --- Determine HF Space Runtime URL and Repo URL --- space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code if profile: username= f"{profile.username}" print(f"User logged in: {username}") else: print("User not logged in.") return "Please Login to Hugging Face with the button.", None api_url = DEFAULT_API_URL questions_url = f"{api_url}/questions" submit_url = f"{api_url}/submit" # 1. Instantiate Agent ( modify this part to create your agent) try: agent = GaiaSmolAgent() except Exception as e: print(f"Error instantiating agent: {e}") return f"Error initializing agent: {e}", None # In the case of an app running as a hugging Face space, this link points toward your codebase ( usefull for others so please keep it public) agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" print(agent_code) # 2. Fetch Questions print(f"Fetching questions from: {questions_url}") try: response = requests.get(questions_url, timeout=15) response.raise_for_status() questions_data = response.json() if not questions_data: 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 Exception as e: print(f"An unexpected error occurred fetching questions: {e}") return f"An unexpected error occurred fetching questions: {e}", None # 3. Run your Agent results_log = [] answers_payload = [] print(f"Running agent on {len(questions_data)} questions...") for item in questions_data: task_id = item.get("task_id") question_text = item.get("question") 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) 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: print(f"Error running agent on task {task_id}: {e}") results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"}) 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) # 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) # 5. Submit print(f"Submitting {len(answers_payload)} answers to: {submit_url}") try: response = requests.post(submit_url, json=submission_data, timeout=60) response.raise_for_status() result_data = response.json() final_status = ( f"Submission Successful!\n" f"User: {result_data.get('username')}\n" f"Overall Score: {result_data.get('score', 'N/A')}% " f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n" f"Message: {result_data.get('message', 'No message received.')}" ) print("Submission successful.") results_df = pd.DataFrame(results_log) return final_status, results_df except requests.exceptions.HTTPError as e: error_detail = f"Server responded with status {e.response.status_code}." try: error_json = e.response.json() error_detail += f" Detail: {error_json.get('detail', e.response.text)}" except requests.exceptions.JSONDecodeError: error_detail += f" Response: {e.response.text[:500]}" status_message = f"Submission Failed: {error_detail}" print(status_message) results_df = pd.DataFrame(results_log) return status_message, results_df except requests.exceptions.Timeout: status_message = "Submission Failed: The request timed out." print(status_message) results_df = pd.DataFrame(results_log) return status_message, results_df except requests.exceptions.RequestException as e: status_message = f"Submission Failed: Network error - {e}" print(status_message) results_df = pd.DataFrame(results_log) return status_message, results_df except Exception as e: status_message = f"An unexpected error occurred during submission: {e}" print(status_message) results_df = pd.DataFrame(results_log) return status_message, results_df # --- Gradio Interface (Updated Instructions) --- with gr.Blocks() as demo: gr.Markdown("# GAIA Agent Evaluation Runner (smol-agent)") gr.Markdown( """ **Instructions:** 1. Ensure you have added your **GEMINI API key** (as `GEMINI_API_KEY`) in the Space's secrets. 2. Log in to your Hugging Face account using the button below. 3. Click 'Run Evaluation & Submit All Answers' to run your agent and see the score. """ ) gr.LoginButton() run_button = gr.Button("Run Evaluation & Submit All Answers") status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False) results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True) run_button.click( fn=run_and_submit_all, outputs=[status_output, results_table] ) if __name__ == "__main__": print("Launching Gradio Interface for GAIA Agent Evaluation...") demo.launch(debug=True, share=False)