import os import gradio as gr import requests import inspect import pandas as pd from smolagents import ( CodeAgent, LiteLLMModel, DuckDuckGoSearchTool, LogLevel, load_tool, PythonInterpreterTool ) from dotenv import load_dotenv from smolagents import Tool import base64 import anthropic from PIL import Image import io class SimpleExcelTool(Tool): name = "SimpleExcelTool" description = "Load a downloaded Excel file associated with a task ID and perform basic operations like reading data" inputs = { "task_id": { "type": "string", "description": "Task ID for which the Excel file has been downloaded" }, "operation": { "type": "string", "description": "Operation to perform on the Excel file (currently only 'read' is supported)", "nullable": True } } output_type = "string" def forward(self, task_id: str, operation: str = "read") -> str: try: filename = f"{task_id}_downloaded_file" df = pd.read_excel(filename, engine="openpyxl") if operation == "read": return df.head().to_string() else: return f"Unsupported operation: {operation}" except Exception as e: return f"Error reading Excel file: {str(e)}" class ImageAnalysisTool(Tool): name = "ImageAnalysisTool" description = "Analyze a downloaded image file associated with a task ID using Claude Vision. Provide a detailed description of what's in the image." inputs = { "task_id": { "type": "string", "description": "Task ID for which the image file has been downloaded" }, "prompt": { "type": "string", "description": "Optional specific question or aspect to analyze about the image", "nullable": True } } output_type = "string" def __init__(self): super().__init__() self.client = anthropic.Client(api_key="") def forward(self, task_id: str, prompt: str = "Describe what you see in this image in detail.") -> str: try: filename = f"{task_id}_downloaded_file" with open(filename, 'rb') as img_file: img_bytes = img_file.read() img = Image.open(io.BytesIO(img_bytes)) if img.mode != 'RGB': img = img.convert('RGB') img_byte_arr = io.BytesIO() img.save(img_byte_arr, format='JPEG') img_byte_arr = img_byte_arr.getvalue() base64_image = base64.b64encode(img_byte_arr).decode('utf-8') message = self.client.messages.create( model="claude-3-7-sonnet-20250219", max_tokens=1000, messages=[{ "role": "user", "content": [ { "type": "image", "source": { "type": "base64", "media_type": "image/jpeg", "data": base64_image } }, { "type": "text", "text": prompt } ] }] ) return message.content[0].text except Exception as e: return f"Error analyzing image: {str(e)}" # New: TaskFileDownloaderTool class TaskFileDownloaderTool(Tool): name = "TaskFileDownloaderTool" description = "Download a specific file associated with a given task ID and save it locally" inputs = { "task_id": { "type": "string", "description": "Task ID for which to download the associated file" } } output_type = "string" def forward(self, task_id: str) -> str: try: download_url = f"{DEFAULT_API_URL}/files/{task_id}" response = requests.get(download_url) response.raise_for_status() filename = f"{task_id}_downloaded_file" with open(filename, "wb") as f: f.write(response.content) return f"File downloaded successfully and saved as: {filename}" except Exception as e: return f"Error downloading file: {str(e)}" # New: FileOpenerTool class FileOpenerTool(Tool): name = "FileOpenerTool" description = "Open a downloaded file associated with a task ID and read its contents as plain text." inputs = { "task_id": { "type": "string", "description": "Task ID for which the file has been downloaded" }, "num_lines": { "type": "integer", "description": "Number of lines to read from the file", "nullable": True } } output_type = "string" def forward(self, task_id: str, num_lines: int = 10) -> str: try: filename = f"{task_id}_downloaded_file" if not os.path.exists(filename): return f"Error: File {filename} does not exist." with open(filename, "r", encoding="utf-8", errors="ignore") as file: lines = [] for _ in range(num_lines): line = file.readline() if not line: break lines.append(line.strip()) return "\n".join(lines) except Exception as e: return f"Error reading file: {str(e)}" # New: SpeechToTextTool import mlx_whisper class SpeechToTextTool(Tool): name = "SpeechToTextTool" description = "Transcribe a downloaded MP3 audio file associated with a task ID into text." inputs = { "task_id": { "type": "string", "description": "Task ID for which the MP3 audio file has been downloaded" } } output_type = "string" def __init__(self): super().__init__() def forward(self, task_id: str) -> str: try: filename = f"{task_id}_downloaded_file" if not os.path.exists(filename): return f"Error: Audio file {filename} does not exist." result = mlx_whisper.transcribe(filename) return result["text"] except Exception as e: return f"Error transcribing audio file: {str(e)}" import wikipedia class WikipediaSearchTool(Tool): name = "WikipediaSearchTool" description = "Search Wikipedia for a query and return a brief summary." inputs = { "query": { "type": "string", "description": "Query to search on Wikipedia" } } output_type = "string" def __init__(self): super().__init__() wikipedia.set_lang("en") # Ensure English Wikipedia def forward(self, query: str) -> str: try: summary = wikipedia.summary(query, sentences=3000) return summary except wikipedia.exceptions.DisambiguationError as e: return f"Disambiguation error. Possible options: {e.options[:5]}" except wikipedia.exceptions.PageError: return f"Page not found for query: {query}" except Exception as e: return f"Error searching Wikipedia: {str(e)}" def format_transcript(transcript_data): return "\n".join([f"{item['start']}: {item['text']}" for item in transcript_data]) import os from youtube_transcript_api import YouTubeTranscriptApi import yt_dlp import mlx_whisper class YouTubeTranscriptTool(Tool): name = "YouTubeTranscriptTool" description = "Fetches or transcribes the text from a YouTube video ID." inputs = { "video_id": { "type": "string", "description": "YouTube Video ID (the part after 'watch?v=')" } } output_type = "string" def __init__(self): super().__init__() def forward(self, video_id: str) -> str: try: transcript_list = YouTubeTranscriptApi.list_transcripts(video_id) try: # First try manually created transcript transcript = transcript_list.find_manually_created_transcript(['en']) except Exception: # If not found, try auto-generated transcript transcript = transcript_list.find_generated_transcript(['en']) transcript_data = transcript.fetch() # Format nicely text = format_transcript(transcript_data) return text except Exception as e: print(f"No direct transcript found: {e}") print("Trying to download and transcribe audio with Whisper...") # Step 1: Download audio using yt_dlp audio_filename = f"{video_id}.mp3" try: ydl_opts = { 'format': 'bestaudio/best', 'outtmpl': audio_filename, 'postprocessors': [{ 'key': 'FFmpegExtractAudio', 'preferredcodec': 'mp3', 'preferredquality': '192', }], 'quiet': True, } with yt_dlp.YoutubeDL(ydl_opts) as ydl: ydl.download([f"https://www.youtube.com/watch?v={video_id}"]) # Step 2: Transcribe audio using mlx_whisper result = mlx_whisper.transcribe(audio_filename) return result["text"] except Exception as download_error: return f"Error downloading or transcribing YouTube audio: {str(download_error)}" finally: if os.path.exists(audio_filename): os.remove(audio_filename) # Clean up downloaded file # Load environment variables load_dotenv() # (Keep Constants as is) # --- Constants --- DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space" # --- Basic Agent Definition --- class BasicAgent: def __init__(self): print("Initializing Agent with tools...") # Initialize the model using Claude via LiteLLM self.model = LiteLLMModel( model_id="ollama_chat/qwen2:7b", api_base="http://127.0.0.1:11434", temperature=0.7, max_tokens=4096 ) # Initialize tools youtube_transcript_tool = YouTubeTranscriptTool() excel_tool = SimpleExcelTool() image_analysis_tool = ImageAnalysisTool() file_opener_tool = FileOpenerTool() speech_to_text_tool = SpeechToTextTool() task_file_downloader_tool = TaskFileDownloaderTool() wikipedia_search_tool = WikipediaSearchTool() self.tools = [ DuckDuckGoSearchTool(), wikipedia_search_tool, youtube_transcript_tool, PythonInterpreterTool(), excel_tool, image_analysis_tool, file_opener_tool, speech_to_text_tool, task_file_downloader_tool ] # Initialize the agent self.agent = CodeAgent( tools=self.tools, model=self.model, verbosity_level=LogLevel.INFO ) print("Agent initialized successfully") def __call__(self, question: str, task_id: str) -> str: print(f"Agent received question: {question[:100]}...") try: # Step 1: Download the file associated with the task first download_result = self.tools[-1](task_id=task_id) # TaskFileDownloaderTool is the last in self.tools print(download_result) # Step 2: Create a comprehensive prompt for the agent prompt = f"""Please answer the following question. Use the available tools (web search) to gather relevant information before providing a comprehensive answer. Question: {question} Task_id: {task_id} Instructions: 1. Search for relevant information using web search. 2. Look for relevant YouTube content if applicable. 3. If the task requires working with an Excel or image file: - First, download the file associated with the task ID using the file download tool. - Then, perform analysis on the downloaded file. 4. Extract and analyze data from Excel files after downloading. 5. Convert images to text after downloading the image file. 6. Convert attached mp3 to text as seepch to text 7. Make Wikipedia search on facts and for a query and return a brief summary 78. Synthesize all gathered and analyzed information into a clear, well-structured final answer. Answer:""" # Step 3: Get response from the agent response = self.agent.run(prompt) print(f"Agent generated response: {response[:100]}...") return response except Exception as e: error_msg = f"Error generating answer: {str(e)}" print(error_msg) return error_msg 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 try: agent = BasicAgent() except Exception as e: print(f"Error instantiating agent: {e}") return f"Error initializing agent: {e}", None 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() print(questions_data) 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 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, task_id) 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 # --- Build Gradio Interface using Blocks --- with gr.Blocks() as demo: gr.Markdown("# Advanced Agent Evaluation Runner") gr.Markdown( """ **Instructions:** 1. Make sure you have set up your environment variables: - HF_TOKEN: Your Hugging Face API token - YOUTUBE_API_KEY: Your YouTube API key (optional) 2. Log in to your Hugging Face account using the button below 3. Click 'Run Evaluation & Submit All Answers' to process all questions The agent will use: - Web search (DuckDuckGo) - YouTube search (if API key provided) - Mistral-7B-Instruct LLM """ ) 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("\n" + "-"*30 + " App Starting " + "-"*30) # Check for required environment variables hf_token = os.getenv("HF_TOKEN") if not hf_token: print("⚠️ Warning: HF_TOKEN not found in environment variables") youtube_api_key = os.getenv("YOUTUBE_API_KEY") if not youtube_api_key: print("ℹ️ Note: YOUTUBE_API_KEY not found. YouTube search will be disabled") # Check for SPACE_HOST and SPACE_ID space_host_startup = os.getenv("SPACE_HOST") space_id_startup = os.getenv("SPACE_ID") if space_host_startup: print(f"✅ SPACE_HOST found: {space_host_startup}") print(f" Runtime URL should be: https://{space_host_startup}.hf.space") else: print("ℹ️ SPACE_HOST environment variable not found (running locally?).") if space_id_startup: print(f"✅ SPACE_ID found: {space_id_startup}") print(f" Repo URL: https://huggingface.co/spaces/{space_id_startup}") print(f" Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main") else: print("ℹ️ SPACE_ID environment variable not found (running locally?)") print("-"*(60 + len(" App Starting ")) + "\n") print("Launching Gradio Interface for Advanced Agent Evaluation...") demo.launch(debug=True, share=False)