Final_Assignment_Template

Runtime error

05a60d3 verified over 1 year ago

20.5 kB

from langchain_community.tools import DuckDuckGoSearchRun
import base64
from langchain.tools import Tool
import random
from langchain.tools import Tool
from huggingface_hub import list_models
from langchain_community.document_loaders import WikipediaLoader
from langchain_community.document_loaders import ArxivLoader
from langchain_community.document_loaders.csv_loader import CSVLoader
import math
from langgraph.prebuilt import tools_condition
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
import os
from typing import Optional, Tuple
from huggingface_hub import list_models
from pathlib import Path
import json 
from langchain_community.document_loaders import UnstructuredExcelLoader
from langchain_tavily import TavilySearch
from langchain.schema import Document
from langchain_community.document_loaders import YoutubeLoader
from langchain_community.document_loaders import AssemblyAIAudioTranscriptLoader
from langchain_community.document_loaders.assemblyai import TranscriptFormat
import assemblyai as aai
import tempfile
from typing import Dict, List
from langchain_community.document_loaders import GoogleApiClient, GoogleApiYoutubeLoader
from langchain_openai import ChatOpenAI

def create_credentials_file(api_key: str) -> str:
    """
    Create a temporary credentials file for Google API access.
    
    This function creates the necessary credentials structure that Google's API expects.
    In a production environment, you would typically have a proper service account
    JSON file, but for this educational example, we'll create a minimal structure.
    
    Args:
        api_key: "The API Key"
        
    Returns:
        Path to the temporary credentials file
    """
    # Create a minimal credentials structure
    # Note: In production, you would use proper service account credentials
    credentials_data = {
        "type": "service_account",
        "project_id": "your-project-id",
        "private_key_id": "dummy-key-id",
        "private_key": "dummy-private-key",
        "client_email": "tu101@pnw.edu",
        "client_id": "dummy-client-id",
        "auth_uri": "https://accounts.google.com/o/oauth2/auth",
        "token_uri": "https://oauth2.googleapis.com/token",
        "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
        "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/dummy%40your-project-id.iam.gserviceaccount.com"
    }
    
    # Create a temporary file to store credentials
    temp_file = tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False)
    json.dump(credentials_data, temp_file)
    temp_file.close()
    
    return temp_file.name

def extract_video_id_from_url(url: str) -> Optional[str]:
    """
    Extract the video ID from a YouTube URL.
    
    This function handles different YouTube URL formats and extracts the unique
    video identifier that the API needs to fetch the video content.
    
    Args:
        url: YouTube URL in various formats
        
    Returns:
        Video ID string or None if URL is invalid
    """
    import re
    
    # Handle different YouTube URL formats
    patterns = [
        r'(?:https?://)?(?:www\.)?youtube\.com/watch\?v=([a-zA-Z0-9_-]+)',
        r'(?:https?://)?(?:www\.)?youtu\.be/([a-zA-Z0-9_-]+)',
        r'(?:https?://)?(?:www\.)?youtube\.com/embed/([a-zA-Z0-9_-]+)',
    ]
    
    for pattern in patterns:
        match = re.search(pattern, url)
        if match:
            return match.group(1)
    
    return None

def youtube_search_google_api(link: str) -> Dict:
    """
    Search YouTube using Google's official API and extract transcript/captions.
    
    This function represents a more robust approach to YouTube content access.
    Instead of web scraping, we use Google's official API endpoints, which
    provide reliable access to video metadata and transcript data.
    
    Args:
        link: YouTube URL to process
        
    Returns:
        Dictionary containing video documents or error information
    """
    
    # Your Google Cloud API key
    api_key = "AIzaSyA-1ZuYz-PYnnE9PQsWA7vOuJeRz-E2sbU"
    
    try:
        # First, let's validate and extract the video ID
        video_id = extract_video_id_from_url(link)
        if not video_id:
            return {
                "YT_docs": [{
                    "text": "Invalid YouTube URL format. Please provide a valid YouTube link.",
                    "metadata": {"error": True, "url": link}
                }]
            }
        
        # For this implementation, we'll use a simpler approach with the API key
        # In a full production system, you would use proper service account credentials
        
        # Alternative approach using youtube-transcript-api for transcript data
        # This is often more reliable for transcript access
        try:
            from youtube_transcript_api import YouTubeTranscriptApi
            
            # Get the transcript for the video
            transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
            
            # Combine all transcript segments into a single text
            full_transcript = ""
            for segment in transcript_list:
                full_transcript += segment['text'] + " "
            
            # Create a document-like structure similar to other loaders
            documents = [{
                "text": full_transcript.strip(),
                "metadata": {
                    "video_id": video_id,
                    "source": link,
                    "title": f"YouTube Video {video_id}",
                    "segments_count": len(transcript_list)
                }
            }]
            
            return {"YT_docs": documents}
            
        except Exception as transcript_error:
            # If transcript API fails, provide helpful error information
            error_msg = f"Could not retrieve transcript for video {video_id}. "
            error_msg += "This video may not have captions available, may be private, "
            error_msg += "or may have restricted access. "
            error_msg += f"Original error: {str(transcript_error)}"
            
            return {
                "YT_docs": [{
                    "text": error_msg,
                    "metadata": {"error": True, "video_id": video_id}
                }]
            }
    
    except Exception as e:
        # Comprehensive error handling with debugging information
        error_msg = f"Error processing YouTube video: {str(e)}"
        return {
            "YT_docs": [{
                "text": error_msg,
                "metadata": {"error": True, "url": link}
            }]
        }

YT_search_tool = Tool(
    name="YT_search",
    func=youtube_search_google_api,
    description="Searches YouTube using Google's official services to extract video transcripts and captions. Handles various YouTube URL formats and provides detailed error information when content is not available."
)


def audio_search(path: str) -> str:
    """Read an audio mp3 file and extract the text to understand it. 
    
    Args:
        path: The path to the file. If the file attached is just {name}.mp3, then it is local, and just use {name}.mp3. """
    aai.settings.api_key = "945f7d6c00f142158322cd0c13901024"
    possible_paths = [
        path,  # Try the path as-is first
        os.path.join(".", path),  # Try current directory
        os.path.join("/tmp", path),  # Try common temp directory
        os.path.join("uploads", path),  # Try uploads directory
    ]
    
    # Find the actual file location
    actual_path = None
    for potential_path in possible_paths:
        if os.path.exists(potential_path):
            actual_path = potential_path
            break
    
    if actual_path is None:
        # If we can't find the file, let's provide debugging information
        current_dir_files = os.listdir(".")
        error_msg = f"Could not find audio file. Searched in: {possible_paths}. Current directory contains: {current_dir_files[:10]}"  # Limit output for readability
        return {"audio_docs": [{"text": error_msg, "metadata": {"error": True}}]}
    
    try:
        # Use the found path
        loader = AssemblyAIAudioTranscriptLoader(
            file_path=actual_path, 
            transcript_format=TranscriptFormat.PARAGRAPHS
        )
        
        raw_docs = loader.load()
        
        if not raw_docs:
            return {"audio_docs": [{"text": "No audio content could be transcribed from this file.", "metadata": {"warning": True}}]}
        
        # Convert each Document into a serializable dict
        serialized_docs = [
            {"text": doc.page_content, "metadata": doc.metadata}
            for doc in raw_docs
        ]
        
        return {"audio_docs": serialized_docs}
        
    except Exception as e:
        error_msg = f"Error processing audio file at {actual_path}: {str(e)}"
        return {"audio_docs": [{"text": error_msg, "metadata": {"error": True}}]}

        


audio_search_tool = Tool(
    name="audio_search",
    func=audio_search,
    description="Uses an mp3 audio file path and extract it into paragraphs of text that are searchable."
)


def excel_search(excel_path: str ) -> str:
    """Given the path to an excel file, return the excel file contents as raw text.
    
    Args:
        excel_path: The path to where the excel file is stored."""
    loader = UnstructuredExcelLoader(excel_path, mode="elements")
    raw_excel_docs = loader.load()
    formatted_search_docs = "\n\n---\n\n".join(
        [
            f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
            for doc in raw_excel_docs
        ])
    return {"excel_results": formatted_search_docs}

excel_search_tool = Tool(
    name="excel_search",
    func=excel_search,
    description="Searches for and retrieves information in an Excel Document using the given path."
)
def wiki_search(query: str) -> str:
    """Search Wikipedia for a query and return maximum 2 results.
    
    Args:
        query: The search query."""
    search_docs = WikipediaLoader(query=query, load_max_docs=2).load()
    formatted_search_docs = "\n\n---\n\n".join(
        [
            f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
            for doc in search_docs
        ])
    return {"wiki_results": formatted_search_docs}

wiki_search_tool = Tool(
    name="wiki_search",
    func=wiki_search,
    description="Searches for and retrieves information on Wikipedia using an inputted query."
)

def web_search(query: str) -> str:
    """Search Tavily for a query and return maximum 5 results. Make sure to format the args as query and have the query sound like a sentence, hopefully matching part of the question the search is used for.
    Args:
    query: The search query. The query fulfils the argument that is input."""
    tavily = TavilySearch(max_results=5, topic="general", include_answer=True, include_image_descriptions=True)
    raw_results = tavily.invoke(query)
    
    # Extract the actual search results from the response dictionary
    # The 'results' key contains the list of search results we want
    search_results = raw_results.get('results', [])
    
    docs = []
    for idx, result in enumerate(search_results):
        # Each result is a dictionary with keys like 'title', 'content', 'url'
        # We'll use the content as the main text and include other metadata
        content = result.get('content', result.get('title', 'No content available'))
        source_url = result.get('url', f'Tavily result #{idx+1}')
        
        docs.append(Document(
            page_content=content,
            metadata={"source": source_url, "title": result.get('title', 'No title')}
        ))
    
    formatted_docs = "\n\n---\n\n".join(
        f'<Document source="{doc.metadata["source"]}">\n'
        f'{doc.page_content}\n'
        f'</Document>'
        for doc in docs
    )
    
    # Fix the return type - should return string, not dictionary
    return formatted_docs

web_search_tool = Tool(
    name="web_search",
    func=web_search,
    description="Searches for and retrieves information online using an inputted query using Tavily. Please use a query that sounds like a sentence and not just keywords."
)

def arvix_search(query: str) -> str:
    """Search Arxiv for a query and return maximum 3 results.
    
    Args:
        query: The search query.
    
    Returns:
        Formatted string containing search results from Arxiv papers.
    """
    try:
        # Load documents from Arxiv
        search_docs = ArxivLoader(query=query, load_max_docs=3).load()
        
        # Check if we got any results
        if not search_docs:
            return "No results found for the query: " + query
        
        # Process each document with safe metadata access
        formatted_docs = []
        for doc in search_docs:
            # Use .get() method for safe dictionary access with fallback values
            source = doc.metadata.get("source", "Unknown source")
            page = doc.metadata.get("page", "N/A")
            
            # Create the formatted document string
            formatted_doc = f'<Document source="{source}" page="{page}"/>\n{doc.page_content[:1000]}\n</Document>'
            formatted_docs.append(formatted_doc)
        
        # Join all documents with separators
        formatted_search_docs = "\n\n---\n\n".join(formatted_docs)
        
        # Return the string directly (not wrapped in a dictionary)
        return formatted_search_docs
        
    except Exception as e:
        # Handle any unexpected errors gracefully
        return f"Error searching Arxiv: {str(e)}"
        
arvix_search_tool = Tool(
    name="arvix_search",
    func=arvix_search,
    description="Searches for and retrieves information from Arxiv, a database storing scientific papers. Returns formatted text containing paper abstracts and metadata."
)

def extract_image_text(img_path: str) -> str:
    """
    Extract text from an image file using a multimodal model.
    Args:
        img_path: A local image file path (strings).
    Returns:
        A single string containing the concatenated text extracted from each image.
    """
    vision_llm = ChatOpenAI(model="gpt-4o")
    all_text = ""
    try:

        # Read image and encode as base64
        with open(img_path, "rb") as image_file:
            image_bytes = image_file.read()

        image_base64 = base64.b64encode(image_bytes).decode("utf-8")

        # Prepare the prompt including the base64 image data
        message = [
            HumanMessage(
                content=[
                    {
                        "type": "text",
                        "text": (
                            "Extract all the text from this image. "
                            "Return only the extracted text, no explanations."
                        ),
                    },
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/png;base64,{image_base64}"
                        },
                    },
                ]
            )
        ]

        # Call the vision-capable model
        response = vision_llm.invoke(message)

        # Append extracted text
        all_text += response.content + "\n\n"

        return all_text.strip()
    except Exception as e:
        # You can choose whether to raise or just return an empty string / error message
        error_msg = f"Error extracting text: {str(e)}"
        print(error_msg)
        return ""
    
extract_image_text = Tool(
    name="extract_image_text",
    func=extract_image_text,
    description="Extracts text that might be visible on an image by reading the image. Works with path/url to image."
)


def multiply(a: int, b: int) -> int:
    """Multiply two numbers.
    Args:
        a: first int
        b: second int
    """
    return a * b

multiply = Tool(
    name="multiply",
    func=multiply,
    description="Multiplies two integer numbers to get the product."
)

def add(a: int, b: int) -> int:
    """Add two numbers.
    
    Args:
        a: first int
        b: second int
    """
    return a + b
    
add = Tool(
    name="add",
    func=add,
    description="Adds two integer numbers to get the result."
)

def subtract(a: int, b: int) -> int:
    """Subtract two numbers.
    
    Args:
        a: first int
        b: second int
    """
    return a - b
subtract = Tool(
    name="subtract",
    func=subtract,
    description="Subtracts two integer numbers to get the result."
)

def divide(a: int, b: int) -> int:
    """Divide two numbers.
    
    Args:
        a: first int
        b: second int
    """
    if b == 0:
        raise ValueError("Cannot divide by zero.")
    return a / b
divide = Tool(
    name="divide",
    func=divide,
    description="Divides two integer numbers to get the result."
)

def modulus(a: int, b: int) -> int:
    """Get the modulus of two numbers.
    
    Args:
        a: first int
        b: second int
    """
    return a % b
    
modulus = Tool(
    name="modulus",
    func=modulus,
    description="Calculates the modulus of two integer numbers, a mod b, to get the result."
)

def calculate_cargo_travel_time(
    origin_coords: Tuple[float, float],
    destination_coords: Tuple[float, float],
    cruising_speed_kmh: Optional[float] = 750.0,  # Average speed for cargo planes
) -> float:
    """
    Calculate the travel time for an object, usually a cargo plane, between two points on Earth using great-circle distance.
    Args:
        origin_coords: Tuple of (latitude, longitude) for the starting point
        destination_coords: Tuple of (latitude, longitude) for the destination
        cruising_speed_kmh: Optional cruising speed in km/h (defaults to 750 km/h for typical cargo planes). Adjustable depending on object.
    Returns:
        float: The estimated travel time in hours
    Example:
        >>> # Chicago (41.8781° N, 87.6298° W) to Sydney (33.8688° S, 151.2093° E)
        >>> result = calculate_cargo_travel_time((41.8781, -87.6298), (-33.8688, 151.2093))
    """

    def to_radians(degrees: float) -> float:
        return degrees * (math.pi / 180)

    # Extract coordinates
    lat1, lon1 = map(to_radians, origin_coords)
    lat2, lon2 = map(to_radians, destination_coords)

    # Earth's radius in kilometers
    EARTH_RADIUS_KM = 6371.0

    # Calculate great-circle distance using the haversine formula
    dlon = lon2 - lon1
    dlat = lat2 - lat1

    a = (
        math.sin(dlat / 2) ** 2
        + math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2) ** 2
    )
    c = 2 * math.asin(math.sqrt(a))
    distance = EARTH_RADIUS_KM * c

    # Add 10% to account for non-direct routes and air traffic controls
    actual_distance = distance * 1.1

    # Calculate flight time
    # Add 1 hour for takeoff and landing procedures
    flight_time = (actual_distance / cruising_speed_kmh) + 1.0

    # Format the results

    return round(flight_time, 2)

calculate_cargo_travel_time = Tool(
    name="calculate_cargo_travel_time",
    func=calculate_cargo_travel_time,
    description="Calculates the travel time between two points on the earth based on an object with optionally set speed, usually a cargo plane."
)


def get_hub_stats(author: str) -> str:
    """Fetches the most downloaded model from a specific author on the Hugging Face Hub."""
    try:
        # List models from the specified author, sorted by downloads
        models = list(list_models(author=author, sort="downloads", direction=-1, limit=1))

        if models:
            model = models[0]
            return f"The most downloaded model by {author} is {model.id} with {model.downloads:,} downloads."
        else:
            return f"No models found for author {author}."
    except Exception as e:
        return f"Error fetching models for {author}: {str(e)}"

# Initialize the tool
hub_stats_tool = Tool(
    name="get_hub_stats",
    func=get_hub_stats,
    description="Fetches the most downloaded model from a specific author on the Hugging Face Hub."
)

	from langchain_community.tools import DuckDuckGoSearchRun
	import base64
	from langchain.tools import Tool
	import random
	from langchain.tools import Tool
	from huggingface_hub import list_models
	from langchain_community.document_loaders import WikipediaLoader
	from langchain_community.document_loaders import ArxivLoader
	from langchain_community.document_loaders.csv_loader import CSVLoader
	import math
	from langgraph.prebuilt import tools_condition
	from langgraph.prebuilt import ToolNode
	from langchain_core.tools import tool
	from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
	import os
	from typing import Optional, Tuple
	from huggingface_hub import list_models
	from pathlib import Path
	import json
	from langchain_community.document_loaders import UnstructuredExcelLoader
	from langchain_tavily import TavilySearch
	from langchain.schema import Document
	from langchain_community.document_loaders import YoutubeLoader
	from langchain_community.document_loaders import AssemblyAIAudioTranscriptLoader
	from langchain_community.document_loaders.assemblyai import TranscriptFormat
	import assemblyai as aai
	import tempfile
	from typing import Dict, List
	from langchain_community.document_loaders import GoogleApiClient, GoogleApiYoutubeLoader
	from langchain_openai import ChatOpenAI

	def create_credentials_file(api_key: str) -> str:
	"""
	Create a temporary credentials file for Google API access.

	This function creates the necessary credentials structure that Google's API expects.
	In a production environment, you would typically have a proper service account
	JSON file, but for this educational example, we'll create a minimal structure.

	Args:
	api_key: "The API Key"

	Returns:
	Path to the temporary credentials file
	"""
	# Create a minimal credentials structure
	# Note: In production, you would use proper service account credentials
	credentials_data = {
	"type": "service_account",
	"project_id": "your-project-id",
	"private_key_id": "dummy-key-id",
	"private_key": "dummy-private-key",
	"client_email": "tu101@pnw.edu",
	"client_id": "dummy-client-id",
	"auth_uri": "https://accounts.google.com/o/oauth2/auth",
	"token_uri": "https://oauth2.googleapis.com/token",
	"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
	"client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/dummy%40your-project-id.iam.gserviceaccount.com"
	}

	# Create a temporary file to store credentials
	temp_file = tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False)
	json.dump(credentials_data, temp_file)
	temp_file.close()

	return temp_file.name

	def extract_video_id_from_url(url: str) -> Optional[str]:
	"""
	Extract the video ID from a YouTube URL.

	This function handles different YouTube URL formats and extracts the unique
	video identifier that the API needs to fetch the video content.

	Args:
	url: YouTube URL in various formats

	Returns:
	Video ID string or None if URL is invalid
	"""
	import re

	# Handle different YouTube URL formats
	patterns = [
	r'(?:https?://)?(?:www\.)?youtube\.com/watch\?v=([a-zA-Z0-9_-]+)',
	r'(?:https?://)?(?:www\.)?youtu\.be/([a-zA-Z0-9_-]+)',
	r'(?:https?://)?(?:www\.)?youtube\.com/embed/([a-zA-Z0-9_-]+)',
	]

	for pattern in patterns:
	match = re.search(pattern, url)
	if match:
	return match.group(1)

	return None

	def youtube_search_google_api(link: str) -> Dict:
	"""
	Search YouTube using Google's official API and extract transcript/captions.

	This function represents a more robust approach to YouTube content access.
	Instead of web scraping, we use Google's official API endpoints, which
	provide reliable access to video metadata and transcript data.

	Args:
	link: YouTube URL to process

	Returns:
	Dictionary containing video documents or error information
	"""

	# Your Google Cloud API key
	api_key = "AIzaSyA-1ZuYz-PYnnE9PQsWA7vOuJeRz-E2sbU"

	try:
	# First, let's validate and extract the video ID
	video_id = extract_video_id_from_url(link)
	if not video_id:
	return {
	"YT_docs": [{
	"text": "Invalid YouTube URL format. Please provide a valid YouTube link.",
	"metadata": {"error": True, "url": link}
	}]
	}

	# For this implementation, we'll use a simpler approach with the API key
	# In a full production system, you would use proper service account credentials

	# Alternative approach using youtube-transcript-api for transcript data
	# This is often more reliable for transcript access
	try:
	from youtube_transcript_api import YouTubeTranscriptApi

	# Get the transcript for the video
	transcript_list = YouTubeTranscriptApi.get_transcript(video_id)

	# Combine all transcript segments into a single text
	full_transcript = ""
	for segment in transcript_list:
	full_transcript += segment['text'] + " "

	# Create a document-like structure similar to other loaders
	documents = [{
	"text": full_transcript.strip(),
	"metadata": {
	"video_id": video_id,
	"source": link,
	"title": f"YouTube Video {video_id}",
	"segments_count": len(transcript_list)
	}
	}]

	return {"YT_docs": documents}

	except Exception as transcript_error:
	# If transcript API fails, provide helpful error information
	error_msg = f"Could not retrieve transcript for video {video_id}. "
	error_msg += "This video may not have captions available, may be private, "
	error_msg += "or may have restricted access. "
	error_msg += f"Original error: {str(transcript_error)}"

	return {
	"YT_docs": [{
	"text": error_msg,
	"metadata": {"error": True, "video_id": video_id}
	}]
	}

	except Exception as e:
	# Comprehensive error handling with debugging information
	error_msg = f"Error processing YouTube video: {str(e)}"
	return {
	"YT_docs": [{
	"text": error_msg,
	"metadata": {"error": True, "url": link}
	}]
	}

	YT_search_tool = Tool(
	name="YT_search",
	func=youtube_search_google_api,
	description="Searches YouTube using Google's official services to extract video transcripts and captions. Handles various YouTube URL formats and provides detailed error information when content is not available."
	)


	def audio_search(path: str) -> str:
	"""Read an audio mp3 file and extract the text to understand it.

	Args:
	path: The path to the file. If the file attached is just {name}.mp3, then it is local, and just use {name}.mp3. """
	aai.settings.api_key = "945f7d6c00f142158322cd0c13901024"
	possible_paths = [
	path, # Try the path as-is first
	os.path.join(".", path), # Try current directory
	os.path.join("/tmp", path), # Try common temp directory
	os.path.join("uploads", path), # Try uploads directory
	]

	# Find the actual file location
	actual_path = None
	for potential_path in possible_paths:
	if os.path.exists(potential_path):
	actual_path = potential_path
	break

	if actual_path is None:
	# If we can't find the file, let's provide debugging information
	current_dir_files = os.listdir(".")
	error_msg = f"Could not find audio file. Searched in: {possible_paths}. Current directory contains: {current_dir_files[:10]}" # Limit output for readability
	return {"audio_docs": [{"text": error_msg, "metadata": {"error": True}}]}

	try:
	# Use the found path
	loader = AssemblyAIAudioTranscriptLoader(
	file_path=actual_path,
	transcript_format=TranscriptFormat.PARAGRAPHS
	)

	raw_docs = loader.load()

	if not raw_docs:
	return {"audio_docs": [{"text": "No audio content could be transcribed from this file.", "metadata": {"warning": True}}]}

	# Convert each Document into a serializable dict
	serialized_docs = [
	{"text": doc.page_content, "metadata": doc.metadata}
	for doc in raw_docs
	]

	return {"audio_docs": serialized_docs}

	except Exception as e:
	error_msg = f"Error processing audio file at {actual_path}: {str(e)}"
	return {"audio_docs": [{"text": error_msg, "metadata": {"error": True}}]}




	audio_search_tool = Tool(
	name="audio_search",
	func=audio_search,
	description="Uses an mp3 audio file path and extract it into paragraphs of text that are searchable."
	)


	def excel_search(excel_path: str ) -> str:
	"""Given the path to an excel file, return the excel file contents as raw text.

	Args:
	excel_path: The path to where the excel file is stored."""
	loader = UnstructuredExcelLoader(excel_path, mode="elements")
	raw_excel_docs = loader.load()
	formatted_search_docs = "\n\n---\n\n".join(
	[
	f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
	for doc in raw_excel_docs
	])
	return {"excel_results": formatted_search_docs}

	excel_search_tool = Tool(
	name="excel_search",
	func=excel_search,
	description="Searches for and retrieves information in an Excel Document using the given path."
	)
	def wiki_search(query: str) -> str:
	"""Search Wikipedia for a query and return maximum 2 results.

	Args:
	query: The search query."""
	search_docs = WikipediaLoader(query=query, load_max_docs=2).load()
	formatted_search_docs = "\n\n---\n\n".join(
	[
	f'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content}\n</Document>'
	for doc in search_docs
	])
	return {"wiki_results": formatted_search_docs}

	wiki_search_tool = Tool(
	name="wiki_search",
	func=wiki_search,
	description="Searches for and retrieves information on Wikipedia using an inputted query."
	)

	def web_search(query: str) -> str:
	"""Search Tavily for a query and return maximum 5 results. Make sure to format the args as query and have the query sound like a sentence, hopefully matching part of the question the search is used for.
	Args:
	query: The search query. The query fulfils the argument that is input."""
	tavily = TavilySearch(max_results=5, topic="general", include_answer=True, include_image_descriptions=True)
	raw_results = tavily.invoke(query)

	# Extract the actual search results from the response dictionary
	# The 'results' key contains the list of search results we want
	search_results = raw_results.get('results', [])

	docs = []
	for idx, result in enumerate(search_results):
	# Each result is a dictionary with keys like 'title', 'content', 'url'
	# We'll use the content as the main text and include other metadata
	content = result.get('content', result.get('title', 'No content available'))
	source_url = result.get('url', f'Tavily result #{idx+1}')

	docs.append(Document(
	page_content=content,
	metadata={"source": source_url, "title": result.get('title', 'No title')}
	))

	formatted_docs = "\n\n---\n\n".join(
	f'<Document source="{doc.metadata["source"]}">\n'
	f'{doc.page_content}\n'
	f'</Document>'
	for doc in docs
	)

	# Fix the return type - should return string, not dictionary
	return formatted_docs

	web_search_tool = Tool(
	name="web_search",
	func=web_search,
	description="Searches for and retrieves information online using an inputted query using Tavily. Please use a query that sounds like a sentence and not just keywords."
	)

	def arvix_search(query: str) -> str:
	"""Search Arxiv for a query and return maximum 3 results.

	Args:
	query: The search query.

	Returns:
	Formatted string containing search results from Arxiv papers.
	"""
	try:
	# Load documents from Arxiv
	search_docs = ArxivLoader(query=query, load_max_docs=3).load()

	# Check if we got any results
	if not search_docs:
	return "No results found for the query: " + query

	# Process each document with safe metadata access
	formatted_docs = []
	for doc in search_docs:
	# Use .get() method for safe dictionary access with fallback values
	source = doc.metadata.get("source", "Unknown source")
	page = doc.metadata.get("page", "N/A")

	# Create the formatted document string
	formatted_doc = f'<Document source="{source}" page="{page}"/>\n{doc.page_content[:1000]}\n</Document>'
	formatted_docs.append(formatted_doc)

	# Join all documents with separators
	formatted_search_docs = "\n\n---\n\n".join(formatted_docs)

	# Return the string directly (not wrapped in a dictionary)
	return formatted_search_docs

	except Exception as e:
	# Handle any unexpected errors gracefully
	return f"Error searching Arxiv: {str(e)}"

	arvix_search_tool = Tool(
	name="arvix_search",
	func=arvix_search,
	description="Searches for and retrieves information from Arxiv, a database storing scientific papers. Returns formatted text containing paper abstracts and metadata."
	)

	def extract_image_text(img_path: str) -> str:
	"""
	Extract text from an image file using a multimodal model.

	Args:
	img_path: A local image file path (strings).

	Returns:
	A single string containing the concatenated text extracted from each image.
	"""
	vision_llm = ChatOpenAI(model="gpt-4o")
	all_text = ""
	try:

	# Read image and encode as base64
	with open(img_path, "rb") as image_file:
	image_bytes = image_file.read()

	image_base64 = base64.b64encode(image_bytes).decode("utf-8")

	# Prepare the prompt including the base64 image data
	message = [
	HumanMessage(
	content=[
	{
	"type": "text",
	"text": (
	"Extract all the text from this image. "
	"Return only the extracted text, no explanations."
	),
	},
	{
	"type": "image_url",
	"image_url": {
	"url": f"data:image/png;base64,{image_base64}"
	},
	},
	]
	)
	]

	# Call the vision-capable model
	response = vision_llm.invoke(message)

	# Append extracted text
	all_text += response.content + "\n\n"

	return all_text.strip()
	except Exception as e:
	# You can choose whether to raise or just return an empty string / error message
	error_msg = f"Error extracting text: {str(e)}"
	print(error_msg)
	return ""

	extract_image_text = Tool(
	name="extract_image_text",
	func=extract_image_text,
	description="Extracts text that might be visible on an image by reading the image. Works with path/url to image."
	)


	def multiply(a: int, b: int) -> int:
	"""Multiply two numbers.
	Args:
	a: first int
	b: second int
	"""
	return a * b

	multiply = Tool(
	name="multiply",
	func=multiply,
	description="Multiplies two integer numbers to get the product."
	)

	def add(a: int, b: int) -> int:
	"""Add two numbers.

	Args:
	a: first int
	b: second int
	"""
	return a + b

	add = Tool(
	name="add",
	func=add,
	description="Adds two integer numbers to get the result."
	)

	def subtract(a: int, b: int) -> int:
	"""Subtract two numbers.

	Args:
	a: first int
	b: second int
	"""
	return a - b
	subtract = Tool(
	name="subtract",
	func=subtract,
	description="Subtracts two integer numbers to get the result."
	)

	def divide(a: int, b: int) -> int:
	"""Divide two numbers.

	Args:
	a: first int
	b: second int
	"""
	if b == 0:
	raise ValueError("Cannot divide by zero.")
	return a / b
	divide = Tool(
	name="divide",
	func=divide,
	description="Divides two integer numbers to get the result."
	)

	def modulus(a: int, b: int) -> int:
	"""Get the modulus of two numbers.

	Args:
	a: first int
	b: second int
	"""
	return a % b

	modulus = Tool(
	name="modulus",
	func=modulus,
	description="Calculates the modulus of two integer numbers, a mod b, to get the result."
	)

	def calculate_cargo_travel_time(
	origin_coords: Tuple[float, float],
	destination_coords: Tuple[float, float],
	cruising_speed_kmh: Optional[float] = 750.0, # Average speed for cargo planes
	) -> float:
	"""
	Calculate the travel time for an object, usually a cargo plane, between two points on Earth using great-circle distance.

	Args:
	origin_coords: Tuple of (latitude, longitude) for the starting point
	destination_coords: Tuple of (latitude, longitude) for the destination
	cruising_speed_kmh: Optional cruising speed in km/h (defaults to 750 km/h for typical cargo planes). Adjustable depending on object.

	Returns:
	float: The estimated travel time in hours

	Example:
	>>> # Chicago (41.8781° N, 87.6298° W) to Sydney (33.8688° S, 151.2093° E)
	>>> result = calculate_cargo_travel_time((41.8781, -87.6298), (-33.8688, 151.2093))
	"""

	def to_radians(degrees: float) -> float:
	return degrees * (math.pi / 180)

	# Extract coordinates
	lat1, lon1 = map(to_radians, origin_coords)
	lat2, lon2 = map(to_radians, destination_coords)

	# Earth's radius in kilometers
	EARTH_RADIUS_KM = 6371.0

	# Calculate great-circle distance using the haversine formula
	dlon = lon2 - lon1
	dlat = lat2 - lat1

	a = (
	math.sin(dlat / 2) ** 2
	+ math.cos(lat1) * math.cos(lat2) * math.sin(dlon / 2) ** 2
	)
	c = 2 * math.asin(math.sqrt(a))
	distance = EARTH_RADIUS_KM * c

	# Add 10% to account for non-direct routes and air traffic controls
	actual_distance = distance * 1.1

	# Calculate flight time
	# Add 1 hour for takeoff and landing procedures
	flight_time = (actual_distance / cruising_speed_kmh) + 1.0

	# Format the results

	return round(flight_time, 2)

	calculate_cargo_travel_time = Tool(
	name="calculate_cargo_travel_time",
	func=calculate_cargo_travel_time,
	description="Calculates the travel time between two points on the earth based on an object with optionally set speed, usually a cargo plane."
	)


	def get_hub_stats(author: str) -> str:
	"""Fetches the most downloaded model from a specific author on the Hugging Face Hub."""
	try:
	# List models from the specified author, sorted by downloads
	models = list(list_models(author=author, sort="downloads", direction=-1, limit=1))

	if models:
	model = models[0]
	return f"The most downloaded model by {author} is {model.id} with {model.downloads:,} downloads."
	else:
	return f"No models found for author {author}."
	except Exception as e:
	return f"Error fetching models for {author}: {str(e)}"

	# Initialize the tool
	hub_stats_tool = Tool(
	name="get_hub_stats",
	func=get_hub_stats,
	description="Fetches the most downloaded model from a specific author on the Hugging Face Hub."
	)