final_assignment_agentic_course

Sleeping

final_assignment_agentic_course / tools_langchain_doc.py

Created distincts file for langchain and smolagent tools

3b013ba over 1 year ago

8.44 kB

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)}"

	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)}"