Spaces:
Sleeping
Sleeping
| 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 | |
| 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) | |
| 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 | |
| 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 | |
| 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) | |
| 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) | |
| 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) | |
| 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) | |