final_assignment_agentic_course

Sleeping

File size: 8,440 Bytes

import os
import pandas as pd
import re
import requests
import tempfile
import uuid

from langchain_core.tools import tool

from google import genai
from google.genai import types

# from smolagents import tool
from typing import Optional, Dict, Union


@tool
def save_and_read_file(content: str, filename: Optional[str] = None) -> str:
    """
    Save content to a temporary file and return the path.
    Useful for processing files from the GAIA API.

    Args:
        content: The content to save to the file
        filename: Optional filename, will generate a random name if not provided

    Returns:
        Path to the saved file
    """
    temp_dir = tempfile.gettempdir()
    if filename is None:
        temp_file = tempfile.NamedTemporaryFile(delete=False)
        filepath = temp_file.name
    else:
        filepath = os.path.join(temp_dir, filename)

    # Write content to the file
    with open(filepath, "w") as f:
        f.write(content)

    return f"File saved to {filepath}. You can read this file to process its contents."


# File Download Tool
@tool
def download_file_from_url(
    url: str, directory: str
) -> Dict[str, Union[str, None]]:
    """Downloads a file from a URL and saves it to a directory.
    Args:
        url (str): the URL to download the file from.
        directory (str): the directory to save the file to.
    Returns:
        Dict[str, Union[str, None]]: A dictionary containing the file type and path.
    """

    try:
        response = requests.get(url, stream=True, timeout=10)
        response.raise_for_status()

        content_type = response.headers.get("content-type", "").lower()

        # Try to get filename from headers
        filename = None
        cd = response.headers.get("content-disposition", "")
        match = re.search(r"filename\*=UTF-8\'\'(.+)", cd) or re.search(
            r'filename="?([^"]+)"?', cd
        )
        if match:
            filename = match.group(1)

        # If not in headers, try URL
        if not filename:
            filename = os.path.basename(url.split("?")[0])

        # Fallback to generated filename
        if not filename:
            extension = {
                "image/jpeg": ".jpg",
                "image/png": ".png",
                "image/gif": ".gif",
                "audio/wav": ".wav",
                "audio/mpeg": ".mp3",
                "video/mp4": ".mp4",
                "text/plain": ".txt",
                "text/csv": ".csv",
                "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
                "application/vnd.ms-excel": ".xls",
                "application/octet-stream": ".bin",
            }.get(content_type, ".bin")
            filename = f"downloaded_{uuid.uuid4().hex[:8]}{extension}"

        os.makedirs(directory, exist_ok=True)
        file_path = os.path.join(directory, filename)

        with open(file_path, "wb") as f:
            for chunk in response.iter_content(chunk_size=8192):
                f.write(chunk)

        # shutil.copy(file_path, os.getcwd())

        if os.path.exists(file_path) and os.path.getsize(file_path) > 0:
            return {"type": content_type, "path": file_path}
        else:
            return {
                "type": "error",
                "path": None,
                "error": "Failed to save file",
            }

    except Exception as e:
        return {
            "type": "error",
            "path": None,
            "error": f"Error downloading file: {str(e)}",
        }


@tool
def extract_text_from_image(image_path: str) -> str:
    """
    Extract text from an image using pytesseract (if available).

    Args:
        image_path: Path to the image file

    Returns:
        Extracted text or error message
    """
    try:
        # Try to import pytesseract
        import pytesseract
        from PIL import Image

        # Open the image
        image = Image.open(image_path)

        # Extract text
        text = pytesseract.image_to_string(image)

        return f"Extracted text from image:\n\n{text}"
    except ImportError:
        return "Error: pytesseract is not installed. Please install it with 'pip install pytesseract' and ensure Tesseract OCR is installed on your system."
    except Exception as e:
        return f"Error extracting text from image: {str(e)}"


# CSV Analysis Tool
@tool
def analyze_csv_file(file_path: str, query: str) -> str:
    """Analyzes a CSV file and answers questions about its contents using Gemini.
    Args:
        file_path (str): the path to the CSV file to analyze.
        query (str): the question to answer about the CSV file.
    Returns:
        str: The result of the analysis.
    """
    try:
        # Read the CSV file
        df = pd.read_csv(file_path)

        # Initialize Gemini
        client = genai.Client(api_key=os.getenv("GEMINI_KEY"))
        model = "models/gemini-1.5-flash-8b"

        # Convert DataFrame to a string representation
        df_str = df.to_string()

        # Create a prompt for Gemini
        prompt = f"""Analyze this CSV data and provide insights:
Dimensions: {len(df)} rows × {len(df.columns)} columns
Data:
{df_str}
Please provide:
1. A summary of the data structure and content
2. Key patterns and insights
3. Potential data quality issues
4. Suggestions for analysis
User Query: {query}
Please format your response in a clear, structured way with sections and bullet points."""

        # Get analysis from Gemini
        response = client.models.generate_content(
            model=model,
            contents=types.Content(
                parts=[
                    types.Part(text=df_str),
                    types.Part(text=prompt),
                ]
            ),
        )

        result = f"CSV file loaded with {len(df)} rows and {len(df.columns)} columns.\n\n"
        result += response.text

        return result
    except Exception as e:
        return f"Error analyzing CSV file: {str(e)}"


# Excel Analysis Tool
@tool
def analyze_excel_file(file_path: str, query: str) -> str:
    """Analyzes an Excel file and answers questions about its contents using Gemini.
    Args:
        file_path (str): the path to the Excel file to analyze.
        query (str): the question to answer about the Excel file.
    Returns:
        str: The result of the analysis.
    """
    try:
        # Read all sheets from the Excel file
        excel_file = pd.ExcelFile(file_path)
        sheet_names = excel_file.sheet_names

        # Initialize Gemini
        client = genai.Client(api_key=os.getenv("GEMINI_KEY"))
        model = "models/gemini-1.5-flash-8b"

        result = f"Excel file loaded with {len(sheet_names)} sheets: {', '.join(sheet_names)}\n\n"

        # Analyze each sheet
        for sheet_name in sheet_names:
            df = pd.read_excel(file_path, sheet_name=sheet_name)

            # Convert DataFrame to a string representation
            df_str = df.to_string()

            # Create a prompt for Gemini
            prompt = f"""Analyze this Excel sheet data and provide insights:
Sheet Name: {sheet_name}
Dimensions: {len(df)} rows × {len(df.columns)} columns
Data:
{df_str}
Please provide:
1. A summary of the data structure and content
2. Key patterns and insights
3. Potential data quality issues
4. Suggestions for analysis
User Query: {query}
Please format your response in a clear, structured way with sections and bullet points."""

            # Get analysis from Gemini
            response = client.models.generate_content(
                model=model,
                contents=types.Content(
                    parts=[types.Part(text=df_str), types.Part(text=prompt)]
                ),
            )

            result += f"=== Sheet: {sheet_name} ===\n"
            result += response.text + "\n"
            result += "=" * 50 + "\n\n"

        return result
    except Exception as e:
        return f"Error analyzing Excel file: {str(e)}"


@tool
def read_file(filepath: str) -> str:
    """Reads the content of a text file.
    Args:
        filepath (str): the path to the file to read.
    Returns:
        str: The content of the file.
    """
    try:
        with open(filepath, "r", encoding="utf-8") as file:
            content = file.read()
        return content
    except FileNotFoundError:
        return f"File not found: {filepath}"
    except IOError as e:
        return f"Error reading file: {str(e)}"