import base64 import os import tempfile import uuid from io import BytesIO from pathlib import Path import pandas as pd import requests from langchain_core.messages import HumanMessage from langchain_core.tools import tool from langchain_google_genai import ChatGoogleGenerativeAI from PIL import Image @tool("file-download-tool", parse_docstring=True) def download_file_from_url(url: str, filename: str = None) -> str: """Downloads a file from a URL and saves it to a temporary location. This function downloads a file from a given URL and saves it in a local temporary location as `filename`. If the filename is not provided, a random name is generated. If download succeds, the function returns the local path to the file. If an error occurs, it returns the exception as a string. Args: url (str): The URL of the file to download. filename (str, optional): The name to save the file as. If not set, a random name will be generated for the file. Returns: str: The local path to the file, or the raised exception. """ try: # Generate a filename if not provided if not filename: filename = f"downloaded_{uuid.uuid4()}" # Create temporary file temp_dir = tempfile.gettempdir() filepath = Path(temp_dir) / filename # Download the file response = requests.get(url, stream=True) response.raise_for_status() # Save the file with open(filepath, "wb") as f: for chunk in response.iter_content(chunk_size=8192): f.write(chunk) return filepath except Exception as e: return str(e) @tool("file-csv-read-tool", parse_docstring=True) def read_csv(filepath: str) -> str: """Reads a csv file from filepath. This function reads a csv file from `filepath` and returns its contents as a markdown table. If an error occurs, it returns the exception as a string. Args: filepath (str): The local filepath to the file. Returns: str: The contents of the csv file in markdown table format. """ # Read the csv file df = pd.read_csv(filepath) # Convert to markdown table string text_table = df.to_markdown(index=False) return text_table @tool("file-excel-read-tool", parse_docstring=True) def read_excel(filepath: str) -> str: """Reads an excel file from filepath. This function reads an xlsx file from `filepath` and returns its contents as a markdown table. If an error occurs, it returns the exception as a string. Args: filepath (str): The local filepath to the file. Returns: str: The contents of the excel file in markdown table format. """ # Read the csv file df = pd.read_excel(filepath) # Convert to markdown table string text_table = df.to_markdown(index=False) return text_table @tool("file-code-read-tool", parse_docstring=True) def read_code(filepath: str) -> str: """Reads a code file from filepath. This function reads a code file from `filepath` and returns its contents as a string. If an error occurs, it returns the exception as a string. Args: filepath (str): The local filepath to the file. Returns: str: The contents of the code file. """ try: with open(filepath, "r") as f: code_content = f.read() return code_content except Exception as e: return str(e) @tool("analyze-image-tool", parse_docstring=True) async def analyze_image(img_path: str, query: str) -> str: """Analyzes the image at img_path and responds to the input query. This function reads an image from `img_path`, converts it to its Base64 encoding as a string, and responds to the input query based on the image analysis. If an error occurs, it returns the exception as a string. Args: img_path (str): The local filepath to the image. query (str): The query about the image that needs to be answered. Returns: str: Answer to the query about the image. """ try: # Open the image file with Image.open(img_path) as img: buffered = BytesIO() # Save the image to the buffer in its original format img.save(buffered, format=img.format) img_bytes = buffered.getvalue() b64_image = base64.b64encode(img_bytes).decode('utf-8') # Initialize LLM llm = ChatGoogleGenerativeAI( google_api_key=os.getenv("GOOGLE_API_KEY"), model=os.getenv("GOOGLE_AGENT_MODEL_NAME"), temperature=0.7, max_retries=5 ) # Call the LLM message = HumanMessage( content=[ { "type": "text", "text": query }, { "type": "image", "source_type": "base64", "mime_type": "image/png", "data": b64_image } ], ) response = await llm.ainvoke([message]) return response.content except Exception as e: return str(e) @tool("analyze-video-tool", parse_docstring=True) async def analyze_video(video_url: str, query: str) -> str: """Analyzes the video at video_url and responds to the input query. This function analyzes the video at the provided video_url and responds to the input query based on the video analysis. If an error occurs, it returns the exception as a string. Args: video_url (str): URL where the video is located. query (str): The query about the video that needs to be answered. Returns: str: Answer to the query about the video url. """ try: # Initialize LLM llm = ChatGoogleGenerativeAI( google_api_key=os.getenv("GOOGLE_API_KEY"), model=os.getenv("GOOGLE_AGENT_MODEL_NAME"), temperature=0.7, max_retries=5 ) # Call the LLM message = HumanMessage( content=[ { "type": "text", "text": query }, { "type": "media", "mime_type": "video/mp4", "file_uri": video_url } ], ) response = await llm.ainvoke([message]) return response.content except Exception as e: return str(e) @tool("analyze-audio-tool", parse_docstring=True) async def analyze_audio(audio_path: str, query: str) -> str: """Analyzes the audio at audio_path and responds to the input query. This function reads an audio from `audio_path`, converts it to its Base64 encoding as a string, and responds to the input query based on the audio analysis. If an error occurs, it returns the exception as a string. Args: audio_path: String path where the audio file is located on disk. query: The query about the audio file that needs to be answered. Returns: str: Answer to the query about the audio file. """ audio_mime_type = "audio/mpeg" try: with open(audio_path, "rb") as rb: encoded_audio = base64.b64encode(rb.read()).decode('utf-8') # Initialize LLM llm = ChatGoogleGenerativeAI( google_api_key=os.getenv("GOOGLE_API_KEY"), model=os.getenv("GOOGLE_AGENT_MODEL_NAME"), temperature=0.7, max_retries=5 ) # Call the LLM message = HumanMessage( content=[ { "type": "text", "text": query }, { "type": "media", "data": encoded_audio, "mime_type": audio_mime_type } ], ) response = await llm.ainvoke([message]) return response.content except Exception as e: return str(e)