Initial version of the agent WIP

.gitignore

@@ -0,0 +1,28 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+build/
+dist/
+*.egg-info/
+
+# Virtual environments
+.env
+.venv
+env/
+venv/
+
+# VSCode
+.vscode/
+*.code-workspace
+
+# OS files
+.DS_Store
+Thumbs.db
+
+# Logs
+*.log
+
+# Sandbox folders
+/playground

agent.py

@@ -0,0 +1,135 @@
+import json
+from typing import TypedDict, Annotated
+from dotenv import load_dotenv
+import base64
+import mimetypes
+from pathlib import Path
+
+from langgraph.graph.message import add_messages
+from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
+from langgraph.graph import START, StateGraph
+from langgraph.prebuilt import ToolNode
+from langgraph.prebuilt import tools_condition
+from langchain_community.tools import DuckDuckGoSearchRun
+import requests
+
+from agent_state import AgentState
+from agent_tools import download_file, query_spreadsheet, read_media_file
+from langchain_google_genai import ChatGoogleGenerativeAI
+
+load_dotenv()
+
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+questions_url = f"{DEFAULT_API_URL}/random-question"
+
+llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash")
+search_tool = DuckDuckGoSearchRun()
+tools = [search_tool, download_file, query_spreadsheet, read_media_file]
+llm_with_tools = llm.bind_tools(tools, parallel_tool_calls=False)
+
+TOOLS_DESCRIPTION = """
+tool name: search_tool
+description: A tool to search the web for information.
+input: A string with the query to search for.
+output: A string with the search results.
+
+tool name: download_file
+description: A tool to download a file from a URL.
+input: A string with the URL of the file to download.
+output: A dictionary with the path to the downloaded file and a dictionary with metadata about the file.
+
+tool name: query_spreadsheet
+description: A tool to query a spreadsheet file.
+input: A string with the path to the spreadsheet file and a string with the pandas query to execute, the dataframe is called "df". For example: "df['Burguers'].sum()"
+output: A string with the result of the pandas query.
+
+tool name: read_media_file
+description: A tool to read a media file (image or audio).
+input: A string with the path to the media file.
+output: The file content in the format to be used.
+"""
+
+ASSISTANT_PROMPT = f"""
+You are a precise, expert-level answering agent designed to solve a wide range of factual, perceptual, and reasoning-based questions. These questions may involve:
+
+* **Information retrieval** from online sources (e.g., Wikipedia)
+* **Analysis of structured data**, such as tables or spreadsheets
+* **Understanding and interpreting images, videos, and audio**
+* **Logical reasoning**, including math, code, and puzzles
+* **Linguistic manipulation**, such as reversing, translating, or categorizing
+
+You have access to the following tools to help you solve the questions:
+
+{TOOLS_DESCRIPTION}
+
+Read the question carefully, think step by step and formulate a plan to answer the question.
+Then follow the plan to answer the question.
+If necessary, use the tools to answer the question.
+
+The questions come in the following format:
+
+question: question text
+question_file_url (optional): path to the url to download the file required to answer the question
+
+You must follow these strict response rules:
+
+1. **Only provide the final answer**, exactly in the format requested by the question.
+2. **Do not explain, comment, or add phrases** like “The answer is” or “Here’s what I found.”
+3. If a question requires a list, a single word, a number, or a specific notation (e.g., algebraic chess move), **match the format exactly**.
+4. **Always prioritize accuracy, formatting, and precision**, since your answers will be evaluated automatically.
+5. Use your tools only when required to complete the task with confidence.
+"""
+
+
+def assistant(state: AgentState) -> AgentState:
+    sys_msg = SystemMessage(content=ASSISTANT_PROMPT)
+    return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}
+
+
+graph = StateGraph(AgentState)
+
+graph.add_node("assistant", assistant)
+graph.add_node("tools", ToolNode(tools))
+
+graph.add_edge(START, "assistant")
+graph.add_conditional_edges(
+    "assistant",
+    # If the latest message requires a tool, route to tools, otherwise, provide a direct response
+    tools_condition,
+)
+graph.add_edge("tools", "assistant")
+neighborhood_assistant = graph.compile()
+
+neighborhood_assistant.get_graph().draw_mermaid_png(
+    output_file_path="neighborhood_agent_graph.png"
+)
+
+# ===== EXAMPLES OF USAGE =====
+
+# Get a random question
+# response = requests.get(questions_url, timeout=5)
+# response.raise_for_status()
+# raw_question_data = response.json()
+
+# question_data = {
+#     "question": raw_question_data["question"],
+#     "question_file_url": f"{DEFAULT_API_URL}/files/{raw_question_data['task_id']}"
+# }
+
+# Test question
+question_data = {
+    "question": "Hi, I was out sick from my classes on Friday, so I'm trying to figure out what I need to study for my Calculus mid-term next week. My friend from class sent me an audio recording of Professor Willowbrook giving out the recommended reading for the test, but my headphones are broken :(\n\nCould you please listen to the recording for me and tell me the page numbers I'm supposed to go over? I've attached a file called Homework.mp3 that has the recording. Please provide just the page numbers as a comma-delimited list. And please provide the list in ascending order.",
+    "question_file_url": f"{DEFAULT_API_URL}/files/1f975693-876d-457b-a649-393859e79bf3"
+}
+
+# Process the question with the agent
+
+user_input = [HumanMessage(json.dumps(question_data))]
+response = neighborhood_assistant.invoke({"messages": user_input})
+final_answer = response['messages'][-1].content
+
+print(response)
+print("🤖 Question:")
+print(question_data["question"])
+print("🤖 Agent's Response:")
+print(final_answer)

agent_state.py

@@ -0,0 +1,7 @@
+from typing import Annotated, TypedDict
+from langgraph.graph.message import add_messages
+from langchain_core.messages import AnyMessage
+
+
+class AgentState(TypedDict):
+    messages: Annotated[list[AnyMessage], add_messages]

agent_tools.py

@@ -0,0 +1,196 @@
+import base64
+import mimetypes
+import os
+from pathlib import Path
+from urllib.parse import urlparse
+import re
+
+import pandas as pd
+import requests
+from langchain_core.tools import tool
+
+from logging_config import get_logger
+
+
+
+logger = get_logger(__name__)
+
+
+@tool
+def download_file(url: str) -> dict:
+    """Download a file from a URL.
+        
+    Args:
+        url: The URL of the file to download.
+
+    Returns:
+        dict: A dictionary containing the path to the downloaded file and a dictionary with metadata about the file.
+    """
+    logger.info(f"Downloading file from {url}")
+    response = requests.get(url, stream=True)
+    response.raise_for_status()
+
+    parsed = urlparse(url)
+    base = os.path.basename(parsed.path)
+    file_name, file_extension = os.path.splitext(base)
+    file_extension = file_extension.lower()
+
+    # Si no hay extensión en URL, intentar con Content-Disposition
+    if not file_extension:
+        cd = response.headers.get('content-disposition', '')
+        if cd:
+            match = re.search(r'filename\*?=(?:UTF-8\'\')?"?([^\";]+)"?', cd)
+            if match:
+                fname = os.path.basename(match.group(1))
+                name2, ext2 = os.path.splitext(fname)
+                if ext2:
+                    file_name, file_extension = name2, ext2.lower()
+
+    # Si aún sin extensión, dejar ext vacía
+    if not file_name:
+        file_name = "downloaded_file"
+
+
+    filename = f"{file_name}{file_extension}"
+    full_path = os.path.join("agent_files", filename)
+
+    size = 0
+    os.makedirs("agent_files", exist_ok=True)
+    with open(full_path, "wb") as f:
+        for chunk in response.iter_content(chunk_size=8192):
+            if chunk:
+                size += len(chunk)
+                f.write(chunk)
+
+    metadata = {"file_name": file_name,
+            "file_extension": file_extension,
+            "size_bytes": size,
+            "file_path": os.path.abspath(full_path)}
+
+    if file_extension in (".csv", ".xls", ".xlsx", ".xlsm", ".xlsb", ".ods"):
+        try:
+            df = pd.read_excel(full_path) if file_extension != ".csv" else pd.read_csv(full_path)
+            metadata["type"] = "table"
+            metadata["num_rows"], metadata["num_columns"] = df.shape
+            metadata["columns"] = [
+                {"name": col, "dtype": str(df[col].dtype)} for col in df.columns
+            ]
+            if df.shape[0] >= 1:
+                first_row = df.iloc[0].to_dict()
+                metadata["first_row"] = first_row
+        except Exception:
+            pass
+
+    logger.info(f"File downloaded: {os.path.abspath(full_path)}")
+    return {"file_path": os.path.abspath(full_path), "metadata": metadata}
+
+
+@tool
+def query_spreadsheet(file_path: str, pandas_query: str) -> str:
+    """Execute a pandas query on a spreadsheet file.
+
+    Args:
+        file_path: The path to the spreadsheet file.
+        pandas_code: The pandas code to execute.
+
+    Returns:
+        str: The result of the pandas code execution.
+    """
+    logger.info(f"Querying spreadsheet: {file_path} with code: {pandas_query}")
+    if file_path.endswith(".csv"):
+        df = pd.read_csv(file_path)
+    elif file_path.endswith((".xls", ".xlsx")):
+        df = pd.read_excel(file_path)
+    else:
+        raise ValueError("Formato no soportado")
+
+    # Ejecutar el código generado por el LLM
+    local_vars = {"df": df}
+    try:
+        exec("result = " + pandas_query, {}, local_vars)
+        result = local_vars["result"]
+        logger.info(f"Spreadsheet query result: {result}")
+        return result.to_string(index=False) if hasattr(result, "to_string") else str(result)
+    except Exception as e:
+        logger.error(f"Error executing pandas query: {e}")
+        return f"Error executing pandas query: {e}"
+
+
+@tool
+def read_media_file(file_path: str) -> list[dict]:
+    """Read a media file (image or audio).
+    
+    Args:
+        file_path: Path to the image or audio file
+        
+    Returns:
+        list[dict]: A list of dictionaries with multimodal content.
+    """
+    logger.info(f"Reading media file: {file_path}")
+    if not os.path.exists(file_path):
+        raise FileNotFoundError(f"File not found: {file_path}")
+    
+    # Get MIME type to determine if it's image or audio
+    mime_type, _ = mimetypes.guess_type(file_path)
+
+    if mime_type and mime_type.startswith('image/'):
+        return [_encode_image(file_path)]
+    elif mime_type and mime_type.startswith('audio/'):
+        return [_encode_audio(file_path)]
+    else:
+        raise ValueError(f"Unsupported file type: {mime_type}")
+
+def _encode_image(image_path: str) -> dict:
+    """Encode an image file to base64 format for Gemini model. Supports: PNG, JPEG, WEBP, HEIC, HEIF"""
+    logger.info(f"Encoding image file: {image_path}")
+    path = Path(image_path)
+    if not path.exists():
+        raise FileNotFoundError(f"Image file not found: {image_path}")
+    
+    # Get MIME type
+    mime_type, _ = mimetypes.guess_type(image_path)
+    if not mime_type or not mime_type.startswith('image/'):
+        raise ValueError(f"Unsupported image format: {mime_type}")
+    
+    # Read and encode image
+    with open(image_path, "rb") as image_file:
+        encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
+    
+    logger.info(f"Image encoded: {image_path}")
+    return {
+        "type": "image_url",
+        "image_url": {
+            "url": f"data:{mime_type};base64,{encoded_string}"
+        }
+    }
+
+def _encode_audio(audio_path: str) -> dict:
+    """Encode an audio file to base64 format for Gemini model. Supports: MP3, MPEG, MP4, MPG, AVI, WMV, MPEGPS, FLV"""
+    logger.info(f"Encoding audio file: {audio_path}")
+    path = Path(audio_path)
+    if not path.exists():
+        raise FileNotFoundError(f"Audio file not found: {audio_path}")
+    
+    # Get MIME type
+    mime_type, _ = mimetypes.guess_type(audio_path)
+    if not mime_type or not mime_type.startswith('audio/'):
+        # Handle common audio formats that might not be detected
+        if audio_path.lower().endswith('.mp3'):
+            mime_type = 'audio/mpeg'
+        elif audio_path.lower().endswith('.wav'):
+            mime_type = 'audio/wav'
+        elif audio_path.lower().endswith('.m4a'):
+            mime_type = 'audio/mp4'
+        else:
+            raise ValueError(f"Unsupported audio format: {audio_path}")
+    
+    # Read and encode audio
+    with open(audio_path, "rb") as audio_file:
+        encoded_string = base64.b64encode(audio_file.read()).decode('utf-8')
+    
+    logger.info(f"Audio encoded: {audio_path}")
+    return {
+        "type": "media",
+        "media_type": mime_type,
+        "data": encoded_string
+    }

app.py

@@ -1,8 +1,8 @@
 import os
 import gradio as gr
 import requests
-import inspect
 import pandas as pd
+from agent import BasicAgent
 
 # (Keep Constants as is)
 # --- Constants ---
@@ -10,19 +10,11 @@ DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
 # --- Basic Agent Definition ---
 # ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
-class BasicAgent:
-    def __init__(self):
-        print("BasicAgent initialized.")
-    def __call__(self, question: str) -> str:
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
-        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):
-    """
-    Fetches all questions, runs the BasicAgent on them, submits all answers,
-    and displays the results.
+    """Fetches all questions, runs the BasicAgent on them, submits all answers, and displays the results.
+    
+    Args:
+        profile: The profile of the user who is logged in.
     """
     # --- Determine HF Space Runtime URL and Repo URL ---
     space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code
@@ -193,4 +185,4 @@ if __name__ == "__main__":
     print("-"*(60 + len(" App Starting ")) + "\n")
 
     print("Launching Gradio Interface for Basic Agent Evaluation...")
-    demo.launch(debug=True, share=False)
+    demo.launch(debug=True, share=False)

check_models.py

@@ -0,0 +1,16 @@
+from dotenv import load_dotenv
+import google.generativeai as genai
+import os
+
+load_dotenv()
+
+# Configure the API key
+genai.configure(api_key=os.getenv("GOOGLE_API_KEY"))
+
+print("Available Gemini models:")
+for model in genai.list_models():
+    if 'gemini' in model.name.lower():
+        print(f"- {model.name}")
+        print(f"  Display name: {model.display_name}")
+        print(f"  Description: {model.description}")
+        print() 

file_download_tool.py

@@ -0,0 +1,87 @@
+import requests
+import os
+from urllib.parse import urlparse
+from langchain.tools import Tool
+import tempfile
+from pathlib import Path
+
+
+def download_file(url: str) -> str:
+    """Download a file from a URL and save it locally.
+
+    Args:
+        url: The URL of the file to download.
+
+    Returns:
+        str: The local file path where the file was saved, or an error message.
+    """
+    try:
+        # Validate URL
+        parsed_url = urlparse(url)
+        if not parsed_url.scheme or not parsed_url.netloc:
+            return f"Error: Invalid URL format: {url}"
+        
+        # Send GET request to download the file
+        response = requests.get(url, stream=True, timeout=30)
+        response.raise_for_status()
+        
+        # Get filename from URL or Content-Disposition header
+        filename = None
+        if 'Content-Disposition' in response.headers:
+            content_disposition = response.headers['Content-Disposition']
+            if 'filename=' in content_disposition:
+                filename = content_disposition.split('filename=')[1].strip('"')
+        
+        if not filename:
+            # Extract filename from URL
+            filename = os.path.basename(parsed_url.path)
+            if not filename or '.' not in filename:
+                # If no filename, create one based on content type
+                content_type = response.headers.get('content-type', '')
+                if 'json' in content_type:
+                    filename = 'downloaded_file.json'
+                elif 'csv' in content_type:
+                    filename = 'downloaded_file.csv'
+                elif 'xml' in content_type:
+                    filename = 'downloaded_file.xml'
+                elif 'pdf' in content_type:
+                    filename = 'downloaded_file.pdf'
+                else:
+                    filename = 'downloaded_file.txt'
+        
+        # Create downloads directory if it doesn't exist
+        downloads_dir = Path("downloads")
+        downloads_dir.mkdir(exist_ok=True)
+        
+        # Save file to downloads directory
+        file_path = downloads_dir / filename
+        
+        # Handle duplicate filenames
+        counter = 1
+        original_path = file_path
+        while file_path.exists():
+            stem = original_path.stem
+            suffix = original_path.suffix
+            file_path = downloads_dir / f"{stem}_{counter}{suffix}"
+            counter += 1
+        
+        # Write file content
+        with open(file_path, 'wb') as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                f.write(chunk)
+        
+        file_size = file_path.stat().st_size
+        return f"File successfully downloaded to: {file_path}\nFile size: {file_size} bytes\nContent type: {response.headers.get('content-type', 'unknown')}"
+        
+    except requests.exceptions.RequestException as e:
+        return f"Error downloading file: {str(e)}"
+    except Exception as e:
+        return f"Unexpected error: {str(e)}"
+
+
+# Create the tool
+file_download_tool = Tool(
+    name="file_download_tool",
+    description="Use this tool to download files from URLs. Provide the URL of the file you want to download.",
+    func=download_file
+) 

file_processor_tool.py

@@ -0,0 +1,185 @@
+import pandas as pd
+import json
+import csv
+from pathlib import Path
+from langchain.tools import Tool
+
+
+def process_file(file_path: str, query: str = None) -> str:
+    """Process and analyze a file, optionally with a specific query.
+
+    Args:
+        file_path: The path to the file to process.
+        query: Optional query to filter or search within the file content.
+
+    Returns:
+        str: Analysis results or file content based on the query.
+    """
+    try:
+        file_path = Path(file_path)
+        
+        if not file_path.exists():
+            return f"Error: File not found at {file_path}"
+        
+        file_extension = file_path.suffix.lower()
+        
+        # Handle different file types
+        if file_extension in ['.csv']:
+            return _process_csv(file_path, query)
+        elif file_extension in ['.json']:
+            return _process_json(file_path, query)
+        elif file_extension in ['.xlsx', '.xls']:
+            return _process_excel(file_path, query)
+        elif file_extension in ['.txt', '.md']:
+            return _process_text(file_path, query)
+        else:
+            # Try to process as text for unknown formats
+            return _process_text(file_path, query)
+            
+    except Exception as e:
+        return f"Error processing file: {str(e)}"
+
+
+def _process_csv(file_path: Path, query: str = None) -> str:
+    """Process CSV files."""
+    try:
+        df = pd.read_csv(file_path)
+        
+        result = f"CSV File Analysis for {file_path.name}:\n"
+        result += f"- Rows: {len(df)}\n"
+        result += f"- Columns: {len(df.columns)}\n"
+        result += f"- Column names: {', '.join(df.columns.tolist())}\n\n"
+        
+        if query:
+            # Search for query in the dataframe
+            query_lower = query.lower()
+            matching_rows = df[df.astype(str).apply(lambda x: x.str.lower().str.contains(query_lower, na=False)).any(axis=1)]
+            
+            if not matching_rows.empty:
+                result += f"Rows matching '{query}':\n"
+                result += matching_rows.to_string(index=False)
+            else:
+                result += f"No rows found matching '{query}'"
+        else:
+            # Show first few rows and basic statistics
+            result += "First 5 rows:\n"
+            result += df.head().to_string(index=False)
+            
+            if len(df) > 5:
+                result += f"\n\n... and {len(df) - 5} more rows"
+                
+        return result
+        
+    except Exception as e:
+        return f"Error processing CSV: {str(e)}"
+
+
+def _process_json(file_path: Path, query: str = None) -> str:
+    """Process JSON files."""
+    try:
+        with open(file_path, 'r', encoding='utf-8') as f:
+            data = json.load(f)
+        
+        result = f"JSON File Analysis for {file_path.name}:\n"
+        
+        if isinstance(data, dict):
+            result += f"- Type: Dictionary with {len(data)} keys\n"
+            result += f"- Keys: {', '.join(data.keys())}\n\n"
+        elif isinstance(data, list):
+            result += f"- Type: List with {len(data)} items\n\n"
+        
+        if query:
+            # Search for query in JSON content
+            json_str = json.dumps(data, indent=2).lower()
+            if query.lower() in json_str:
+                result += f"Found '{query}' in the JSON content.\n"
+                # Show relevant parts (simplified)
+                result += f"JSON content preview:\n{json.dumps(data, indent=2)[:1000]}..."
+            else:
+                result += f"'{query}' not found in JSON content."
+        else:
+            # Show JSON structure
+            result += f"JSON content preview:\n{json.dumps(data, indent=2)[:1000]}"
+            if len(json.dumps(data)) > 1000:
+                result += "..."
+                
+        return result
+        
+    except Exception as e:
+        return f"Error processing JSON: {str(e)}"
+
+
+def _process_excel(file_path: Path, query: str = None) -> str:
+    """Process Excel files."""
+    try:
+        # Read the first sheet
+        df = pd.read_excel(file_path)
+        
+        result = f"Excel File Analysis for {file_path.name}:\n"
+        result += f"- Rows: {len(df)}\n"
+        result += f"- Columns: {len(df.columns)}\n"
+        result += f"- Column names: {', '.join(df.columns.tolist())}\n\n"
+        
+        if query:
+            # Search for query in the dataframe
+            query_lower = query.lower()
+            matching_rows = df[df.astype(str).apply(lambda x: x.str.lower().str.contains(query_lower, na=False)).any(axis=1)]
+            
+            if not matching_rows.empty:
+                result += f"Rows matching '{query}':\n"
+                result += matching_rows.to_string(index=False)
+            else:
+                result += f"No rows found matching '{query}'"
+        else:
+            # Show first few rows
+            result += "First 5 rows:\n"
+            result += df.head().to_string(index=False)
+            
+            if len(df) > 5:
+                result += f"\n\n... and {len(df) - 5} more rows"
+                
+        return result
+        
+    except Exception as e:
+        return f"Error processing Excel: {str(e)}"
+
+
+def _process_text(file_path: Path, query: str = None) -> str:
+    """Process text files."""
+    try:
+        with open(file_path, 'r', encoding='utf-8') as f:
+            content = f.read()
+        
+        result = f"Text File Analysis for {file_path.name}:\n"
+        result += f"- File size: {len(content)} characters\n"
+        result += f"- Lines: {len(content.splitlines())}\n\n"
+        
+        if query:
+            lines_with_query = [line for line in content.splitlines() if query.lower() in line.lower()]
+            if lines_with_query:
+                result += f"Lines containing '{query}':\n"
+                for i, line in enumerate(lines_with_query[:10]):  # Limit to first 10 matches
+                    result += f"{i+1}: {line}\n"
+                if len(lines_with_query) > 10:
+                    result += f"... and {len(lines_with_query) - 10} more matches"
+            else:
+                result += f"No lines found containing '{query}'"
+        else:
+            # Show first 1000 characters
+            result += "Content preview:\n"
+            result += content[:1000]
+            if len(content) > 1000:
+                result += "..."
+                
+        return result
+        
+    except Exception as e:
+        return f"Error processing text file: {str(e)}"
+
+
+# Create the tool
+file_processor_tool = Tool(
+    name="file_processor_tool",
+    description="Use this tool to process and analyze downloaded files. Supports CSV, JSON, Excel, and text files. You can provide a query to search within the file content.",
+    func=lambda file_path_and_query: process_file(*file_path_and_query.split(' | ')) if ' | ' in file_path_and_query else process_file(file_path_and_query)
+) 

logging_config.py

@@ -0,0 +1,100 @@
+import logging
+import logging.config
+from pathlib import Path
+from typing import Optional
+
+
+def setup_logging(
+    log_level: str = "INFO",
+    log_file: Optional[str] = None,
+    log_format: Optional[str] = None
+) -> None:
+    """
+    Set up logging configuration for the Pokemon agent project.
+    
+    Args:
+        log_level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
+        log_file: Optional file path to write logs to
+        log_format: Optional custom log format
+    """
+    if log_format is None:
+        log_format = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+    
+    # Create logs directory if logging to file
+    if log_file:
+        log_path = Path(log_file)
+        log_path.parent.mkdir(parents=True, exist_ok=True)
+    
+    # Configure logging
+    config = {
+        "version": 1,
+        "disable_existing_loggers": False,
+        "formatters": {
+            "standard": {
+                "format": log_format,
+                "datefmt": "%Y-%m-%d %H:%M:%S"
+            },
+            "detailed": {
+                "format": "%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s",
+                "datefmt": "%Y-%m-%d %H:%M:%S"
+            }
+        },
+        "handlers": {
+            "console": {
+                "class": "logging.StreamHandler",
+                "level": log_level,
+                "formatter": "standard",
+                "stream": "ext://sys.stdout"
+            }
+        },
+        "loggers": {
+            "pokemon_agent": {
+                "level": log_level,
+                "handlers": ["console"],
+                "propagate": False
+            },
+            "langgraph": {
+                "level": "WARNING",
+                "handlers": ["console"],
+                "propagate": False
+            },
+            "langchain": {
+                "level": "WARNING", 
+                "handlers": ["console"],
+                "propagate": False
+            }
+        },
+        "root": {
+            "level": log_level,
+            "handlers": ["console"]
+        }
+    }
+    
+    # Add file handler if log_file is specified
+    if log_file:
+        config["handlers"]["file"] = {
+            "class": "logging.handlers.RotatingFileHandler",
+            "level": log_level,
+            "formatter": "detailed",
+            "filename": log_file,
+            "maxBytes": 10485760,  # 10MB
+            "backupCount": 5
+        }
+        # Add file handler to all loggers
+        config["loggers"]["pokemon_agent"]["handlers"].append("file")
+        config["root"]["handlers"].append("file")
+    
+    logging.config.dictConfig(config)
+
+
+def get_logger(name: str) -> logging.Logger:
+    """
+    Get a logger instance for a specific module.
+    
+    Args:
+        name: Logger name (typically __name__)
+        
+    Returns:
+        Configured logger instance
+    """
+    return logging.getLogger(f"pokemon_agent.{name}") 

multimodal_demo.py

@@ -0,0 +1,53 @@
+"""
+Demonstration of the multimodal download capabilities.
+
+This shows how the agent can download and analyze images/audio files from URLs.
+"""
+
+from agent import BasicAgent
+from langchain_core.messages import HumanMessage
+
+def test_multimodal_download():
+    """Test the multimodal download functionality."""
+    
+    # Initialize the agent
+    agent = BasicAgent()
+    
+    print("🤖 Testing Multimodal Download Agent")
+    print("=" * 50)
+    
+    # Example 1: Image analysis
+    print("\n📸 Example 1: Analyzing an image from URL")
+    print("-" * 30)
+    
+    # Example with a sample image URL
+    image_question = """
+    Can you download and analyze this image: https://upload.wikimedia.org/wikipedia/commons/thumb/5/50/Vd-Orig.svg/256px-Vd-Orig.svg.png
+    What do you see in this image?
+    """
+    
+    try:
+        response = agent(image_question)
+        print(f"Agent Response: {response}")
+    except Exception as e:
+        print(f"Error: {e}")
+    
+    print("\n" + "=" * 50)
+    
+    # Example 2: Different type of question
+    print("\n🏠 Example 2: Neighborhood query")
+    print("-" * 30)
+    
+    neighborhood_question = "Who lives in apartment 302?"
+    
+    try:
+        response = agent(neighborhood_question)
+        print(f"Agent Response: {response}")
+    except Exception as e:
+        print(f"Error: {e}")
+    
+    print("\n" + "=" * 50)
+    print("✅ Demo completed!")
+
+if __name__ == "__main__":
+    test_multimodal_download() 

multimodal_download_tool.py

@@ -0,0 +1,151 @@
+import requests
+import os
+from urllib.parse import urlparse
+from langchain.tools import Tool
+from pathlib import Path
+import base64
+import mimetypes
+
+
+def download_and_process_media(url: str, media_type: str = "auto") -> str:
+    """Download media file (image/audio) from URL and prepare it for multimodal processing.
+
+    Args:
+        url: The URL of the media file to download.
+        media_type: Type of media - "image", "audio", or "auto" to detect automatically.
+
+    Returns:
+        str: Information about the downloaded file and instructions for the agent.
+    """
+    try:
+        # Validate URL
+        parsed_url = urlparse(url)
+        if not parsed_url.scheme or not parsed_url.netloc:
+            return f"Error: Invalid URL format: {url}"
+        
+        # Send GET request to download the file
+        response = requests.get(url, stream=True, timeout=30)
+        response.raise_for_status()
+        
+        # Get content type and determine file type
+        content_type = response.headers.get('content-type', '').lower()
+        
+        # Determine if it's image or audio
+        is_image = any(img_type in content_type for img_type in ['image/jpeg', 'image/png', 'image/webp', 'image/heic', 'image/heif'])
+        is_audio = any(audio_type in content_type for audio_type in ['audio/mpeg', 'audio/mp3', 'audio/wav', 'audio/mp4', 'audio/m4a'])
+        
+        if media_type == "auto":
+            if is_image:
+                media_type = "image"
+            elif is_audio:
+                media_type = "audio"
+            else:
+                return f"Error: Unable to determine media type from content-type: {content_type}. Please specify 'image' or 'audio'."
+        
+        # Get filename from URL or create one based on content type
+        filename = os.path.basename(parsed_url.path)
+        if not filename or '.' not in filename:
+            if media_type == "image":
+                if 'jpeg' in content_type or 'jpg' in content_type:
+                    filename = 'downloaded_image.jpg'
+                elif 'png' in content_type:
+                    filename = 'downloaded_image.png'
+                elif 'webp' in content_type:
+                    filename = 'downloaded_image.webp'
+                else:
+                    filename = 'downloaded_image.jpg'
+            elif media_type == "audio":
+                if 'mpeg' in content_type or 'mp3' in content_type:
+                    filename = 'downloaded_audio.mp3'
+                elif 'wav' in content_type:
+                    filename = 'downloaded_audio.wav'
+                elif 'mp4' in content_type:
+                    filename = 'downloaded_audio.mp4'
+                else:
+                    filename = 'downloaded_audio.mp3'
+        
+        # Create downloads directory if it doesn't exist
+        downloads_dir = Path("downloads")
+        downloads_dir.mkdir(exist_ok=True)
+        
+        # Save file to downloads directory
+        file_path = downloads_dir / filename
+        
+        # Handle duplicate filenames
+        counter = 1
+        original_path = file_path
+        while file_path.exists():
+            stem = original_path.stem
+            suffix = original_path.suffix
+            file_path = downloads_dir / f"{stem}_{counter}{suffix}"
+            counter += 1
+        
+        # Write file content
+        with open(file_path, 'wb') as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                f.write(chunk)
+        
+        file_size = file_path.stat().st_size
+        
+        # Prepare the response with file information
+        result = f"Successfully downloaded {media_type} file:\n"
+        result += f"- File path: {file_path}\n"
+        result += f"- File size: {file_size} bytes\n"
+        result += f"- Content type: {content_type}\n"
+        result += f"- Media type: {media_type}\n\n"
+        
+        if media_type == "image":
+            result += f"DOWNLOADED_IMAGE_PATH:{file_path}"
+        elif media_type == "audio":
+            result += f"DOWNLOADED_AUDIO_PATH:{file_path}"
+            
+        return result
+        
+    except requests.exceptions.RequestException as e:
+        return f"Error downloading media file: {str(e)}"
+    except Exception as e:
+        return f"Unexpected error: {str(e)}"
+
+
+def analyze_downloaded_media(file_path: str, query: str = "") -> str:
+    """Analyze a downloaded media file using multimodal capabilities.
+
+    Args:
+        file_path: Path to the downloaded media file.
+        query: Optional query about the media content.
+
+    Returns:
+        str: Special marker for the agent to process this with multimodal capabilities.
+    """
+    try:
+        file_path = Path(file_path)
+        
+        if not file_path.exists():
+            return f"Error: File not found at {file_path}"
+        
+        # Get MIME type to determine if it's image or audio
+        mime_type, _ = mimetypes.guess_type(str(file_path))
+        
+        if mime_type and mime_type.startswith('image/'):
+            return f"ANALYZE_IMAGE:{file_path}|{query}"
+        elif mime_type and mime_type.startswith('audio/'):
+            return f"ANALYZE_AUDIO:{file_path}|{query}"
+        else:
+            return f"Error: Unsupported media type for file: {file_path}"
+            
+    except Exception as e:
+        return f"Error preparing media analysis: {str(e)}"
+
+
+# Create the tools
+multimodal_download_tool = Tool(
+    name="multimodal_download_tool",
+    description="Download image or audio files from URLs for multimodal analysis. Specify media_type as 'image', 'audio', or 'auto' to detect automatically. Use format: 'URL|media_type' or just 'URL' for auto-detection.",
+    func=lambda url_and_type: download_and_process_media(*url_and_type.split('|')) if '|' in url_and_type else download_and_process_media(url_and_type)
+)
+
+media_analysis_tool = Tool(
+    name="media_analysis_tool", 
+    description="Analyze downloaded image or audio files with optional query. Use format: 'file_path|query' or just 'file_path'.",
+    func=lambda path_and_query: analyze_downloaded_media(*path_and_query.split('|', 1)) if '|' in path_and_query else analyze_downloaded_media(path_and_query)
+) 

pyproject.toml

@@ -0,0 +1,32 @@
+[project]
+name = "final-assignment-template"
+version = "0.1.0"
+description = "My attempt to the final assignment of hugging face's agents course"
+authors = [
+    {name = "Angelo Mosquera",email = "anggelomos@outlook.com"}
+]
+readme = "README.md"
+requires-python = ">=3.10,<4.0"
+dependencies = [
+    "gradio (>=5.38.2,<6.0.0)",
+    "requests (>=2.32.4,<3.0.0)",
+    "python-dotenv (>=1.1.1,<2.0.0)",
+    "langchain-community (>=0.3.27,<0.4.0)",
+    "langgraph (>=0.5.4,<0.6.0)",
+    "rank-bm25 (>=0.2.2,<0.3.0)",
+    "langchain-google-genai (>=2.1.8,<3.0.0)",
+    "pandas (>=2.0.0,<3.0.0)",
+    "openpyxl (>=3.0.0,<4.0.0)",
+    "duckduckgo-search (>=8.1.1,<9.0.0)",
+]
+
+[tool.poetry]
+package-mode = false
+
+[tool.poetry.group.dev.dependencies]
+flake8 = "^7.3.0"
+ruff = "^0.12.5"
+
+[build-system]
+requires = ["poetry-core>=2.0.0,<3.0.0"]
+build-backend = "poetry.core.masonry.api"

requirements.txt

@@ -1,2 +0,0 @@
-gradio
-requests