from langchain_core.tools import tool from langchain_community.document_loaders import ArxivLoader, WikipediaLoader from langchain_community.tools.tavily_search import TavilySearchResults import cmath from PIL import Image, ImageDraw, ImageFont, ImageEnhance, ImageFilter, ImageOps import requests from urllib.parse import urlparse import pytesseract import pandas as pd # The following helper functions should be defined here if image_processing.py is missing. import base64 import io import numpy as np import os import tempfile import uuid from typing import Any, Dict, List, Optional from image_processing import decode_image, encode_image, save_image @tool def add(x: int, y: int) -> int: """Adds two numbers together. Args: x (int): The first number. y (int): The second number. Returns: int: The sum of x and y. Example: add(1, 2) # returns 3 add(10, 20) # returns 30 """ return x + y @tool def subtract(x: int, y: int) -> int: """Subtracts the second number from the first. Args: x (int): The first number. y (int): The second number. Returns: int: The result of x - y. Example: subtract(5, 3) # returns 2 """ return x - y @tool def multiply(x: int, y: int) -> int: """Multiplies two numbers together. Args: x (int): The first number. y (int): The second number. Returns: int: The product of x and y. Example: multiply(3, 4) # returns 12 """ return x * y @tool def divide(x: int, y: int) -> float: """Divides the first number by the second. Args: x (int): The numerator. y (int): The denominator. Returns: float: The result of x / y. Raises: ValueError: If y is zero. Example: divide(10, 2) # returns 5.0 divide(5, 0) # raises ValueError """ if y == 0: raise ValueError("Cannot divide by zero.") return x / y @tool def modulus(x: int, y: int) -> int: """Calculates the modulus of the first number by the second. Args: x (int): The numerator. y (int): The denominator. Returns: int: The remainder of x / y. Raises: ValueError: If y is zero. Example: modulus(10, 2) # returns 0 modulus(5, 0) # raises ValueError """ if y == 0: raise ValueError("Cannot divide by zero.") return x % y @tool def power(a: float, b: float) -> float: """ Get the power of two numbers. Args: a (float): the first number b (float): the second number """ return a**b @tool def square_root(a: float) -> float | complex: """ Get the square root of a number. Args: a (float): the number to get the square root of """ if a >= 0: return a**0.5 return cmath.sqrt(a) @tool 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'\n{doc.page_content}\n' for doc in search_docs ]) return {"wiki_results": formatted_search_docs} @tool def web_search(query: str) -> str: """Search Tavily for a query and return maximum 3 results. Args: query: The search query.""" search_docs = TavilySearchResults(max_results=3).invoke(input=query) formatted_search_docs = "\n\n---\n\n".join( [ f'\n{doc.page_content}\n' for doc in search_docs ]) return {"web_results": formatted_search_docs} @tool def arvix_search(query: str) -> str: """Search Arxiv for a query and return maximum 3 result. Args: query: The search query.""" search_docs = ArxivLoader(query=query, load_max_docs=3).load() formatted_search_docs = "\n\n---\n\n".join( [ f'\n{doc.page_content[:1000]}\n' for doc in search_docs ]) return {"arvix_results": formatted_search_docs} @tool def save_and_read_file(content: str, filename: Optional[str] = None) -> str: """ Save content to a file and return the path. Args: content (str): the content to save to the file filename (str, optional): the name of the file. If not provided, a random name file will be created. """ temp_dir = tempfile.gettempdir() if filename is None: temp_file = tempfile.NamedTemporaryFile(delete=False, dir=temp_dir) filepath = temp_file.name else: filepath = os.path.join(temp_dir, filename) with open(filepath, "w") as f: f.write(content) return f"File saved to {filepath}. You can read this file to process its contents." @tool def download_file_from_url(url: str, filename: Optional[str] = None) -> str: """ Download a file from a URL and save it to a temporary location. Args: url (str): the URL of the file to download. filename (str, optional): the name of the file. If not provided, a random name file will be created. """ try: # Parse URL to get filename if not provided if not filename: path = urlparse(url).path filename = os.path.basename(path) if not filename: filename = f"downloaded_{uuid.uuid4().hex[:8]}" # Create temporary file temp_dir = tempfile.gettempdir() filepath = os.path.join(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 f"File downloaded to {filepath}. You can read this file to process its contents." except Exception as e: return f"Error downloading file: {str(e)}" @tool def extract_text_from_image(image_path: str) -> str: """ Extract text from an image using OCR library pytesseract (if available). Args: image_path (str): the path to the image file. """ try: # Open the image image = Image.open(image_path) # Extract text from the image text = pytesseract.image_to_string(image) return f"Extracted text from image:\n\n{text}" except Exception as e: return f"Error extracting text from image: {str(e)}" @tool def analyze_csv_file(file_path: str, query: str) -> str: """ Analyze a CSV file using pandas and answer a question about it. Args: file_path (str): the path to the CSV file. query (str): Question about the data """ try: # Read the CSV file df = pd.read_csv(file_path) # Run various analyses based on the query result = f"CSV file loaded with {len(df)} rows and {len(df.columns)} columns.\n" result += f"Columns: {', '.join(df.columns)}\n\n" # Add summary statistics result += "Summary statistics:\n" result += str(df.describe()) return result except Exception as e: return f"Error analyzing CSV file: {str(e)}" @tool def analyze_excel_file(file_path: str, query: str) -> str: """ Analyze an Excel file using pandas and answer a question about it. Args: file_path (str): the path to the Excel file. query (str): Question about the data """ try: # Read the Excel file df = pd.read_excel(file_path) # Run various analyses based on the query result = ( f"Excel file loaded with {len(df)} rows and {len(df.columns)} columns.\n" ) result += f"Columns: {', '.join(df.columns)}\n\n" # Add summary statistics result += "Summary statistics:\n" result += str(df.describe()) return result except Exception as e: return f"Error analyzing Excel file: {str(e)}" @tool def analyze_image(image_base64: str) -> Dict[str, Any]: """ Analyze basic properties of an image (size, mode, color analysis, thumbnail preview). Args: image_base64 (str): Base64 encoded image string Returns: Dictionary with analysis result """ try: img = decode_image(image_base64) width, height = img.size mode = img.mode if mode in ("RGB", "RGBA"): arr = np.array(img) avg_colors = arr.mean(axis=(0, 1)) dominant = ["Red", "Green", "Blue"][np.argmax(avg_colors[:3])] brightness = avg_colors.mean() color_analysis = { "average_rgb": avg_colors.tolist(), "brightness": brightness, "dominant_color": dominant, } else: color_analysis = {"note": f"No color analysis for mode {mode}"} thumbnail = img.copy() thumbnail.thumbnail((100, 100)) thumb_path = save_image(thumbnail, "thumbnails") thumbnail_base64 = encode_image(thumb_path) return { "dimensions": (width, height), "mode": mode, "color_analysis": color_analysis, "thumbnail": thumbnail_base64, } except Exception as e: return {"error": str(e)} @tool def transform_image( image_base64: str, operation: str, params: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Apply transformations: resize, rotate, crop, flip, brightness, contrast, blur, sharpen, grayscale. Args: image_base64 (str): Base64 encoded input image operation (str): Transformation operation params (Dict[str, Any], optional): Parameters for the operation Returns: Dictionary with transformed image (base64) """ try: img = decode_image(image_base64) params = params or {} if operation == "resize": img = img.resize( ( params.get("width", img.width // 2), params.get("height", img.height // 2), ) ) elif operation == "rotate": img = img.rotate(params.get("angle", 90), expand=True) elif operation == "crop": img = img.crop( ( params.get("left", 0), params.get("top", 0), params.get("right", img.width), params.get("bottom", img.height), ) ) elif operation == "flip": if params.get("direction", "horizontal") == "horizontal": img = ImageOps.mirror(img) else: img = ImageOps.flip(img) elif operation == "adjust_brightness": img = ImageEnhance.Brightness(img).enhance(params.get("factor", 1.5)) elif operation == "adjust_contrast": img = ImageEnhance.Contrast(img).enhance(params.get("factor", 1.5)) elif operation == "blur": img = img.filter(ImageFilter.GaussianBlur(params.get("radius", 2))) elif operation == "sharpen": img = img.filter(ImageFilter.SHARPEN) elif operation == "grayscale": img = img.convert("L") else: return {"error": f"Unknown operation: {operation}"} result_path = save_image(img) result_base64 = encode_image(result_path) return {"transformed_image": result_base64} except Exception as e: return {"error": str(e)} @tool def draw_on_image( image_base64: str, drawing_type: str, params: Dict[str, Any] ) -> Dict[str, Any]: """ Draw shapes (rectangle, circle, line) or text onto an image. Args: image_base64 (str): Base64 encoded input image drawing_type (str): Drawing type params (Dict[str, Any]): Drawing parameters Returns: Dictionary with result image (base64) """ try: img = decode_image(image_base64) draw = ImageDraw.Draw(img) color = params.get("color", "red") if drawing_type == "rectangle": draw.rectangle( [params["left"], params["top"], params["right"], params["bottom"]], outline=color, width=params.get("width", 2), ) elif drawing_type == "circle": x, y, r = params["x"], params["y"], params["radius"] draw.ellipse( (x - r, y - r, x + r, y + r), outline=color, width=params.get("width", 2), ) elif drawing_type == "line": draw.line( ( params["start_x"], params["start_y"], params["end_x"], params["end_y"], ), fill=color, width=params.get("width", 2), ) elif drawing_type == "text": font_size = params.get("font_size", 20) try: font = ImageFont.truetype("arial.ttf", font_size) except IOError: font = ImageFont.load_default() draw.text( (params["x"], params["y"]), params.get("text", "Text"), fill=color, font=font, ) else: return {"error": f"Unknown drawing type: {drawing_type}"} result_path = save_image(img) result_base64 = encode_image(result_path) return {"result_image": result_base64} except Exception as e: return {"error": str(e)} @tool def generate_simple_image( image_type: str, width: int = 500, height: int = 500, params: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """ Generate a simple image (gradient, noise, pattern, chart). Args: image_type (str): Type of image width (int), height (int) params (Dict[str, Any], optional): Specific parameters Returns: Dictionary with generated image (base64) """ try: params = params or {} if image_type == "gradient": direction = params.get("direction", "horizontal") start_color = params.get("start_color", (255, 0, 0)) end_color = params.get("end_color", (0, 0, 255)) img = Image.new("RGB", (width, height)) draw = ImageDraw.Draw(img) if direction == "horizontal": for x in range(width): r = int( start_color[0] + (end_color[0] - start_color[0]) * x / width ) g = int( start_color[1] + (end_color[1] - start_color[1]) * x / width ) b = int( start_color[2] + (end_color[2] - start_color[2]) * x / width ) draw.line([(x, 0), (x, height)], fill=(r, g, b)) else: for y in range(height): r = int( start_color[0] + (end_color[0] - start_color[0]) * y / height ) g = int( start_color[1] + (end_color[1] - start_color[1]) * y / height ) b = int( start_color[2] + (end_color[2] - start_color[2]) * y / height ) draw.line([(0, y), (width, y)], fill=(r, g, b)) elif image_type == "noise": noise_array = np.random.randint(0, 256, (height, width, 3), dtype=np.uint8) img = Image.fromarray(noise_array, "RGB") else: return {"error": f"Unsupported image_type {image_type}"} result_path = save_image(img) result_base64 = encode_image(result_path) return {"generated_image": result_base64} except Exception as e: return {"error": str(e)} @tool def combine_images( images_base64: List[str], operation: str, params: Optional[Dict[str, Any]] = None ) -> Dict[str, Any]: """ Combine multiple images (collage, stack, blend). Args: images_base64 (List[str]): List of base64 images operation (str): Combination type params (Dict[str, Any], optional) Returns: Dictionary with combined image (base64) """ try: images = [decode_image(b64) for b64 in images_base64] params = params or {} if operation == "stack": direction = params.get("direction", "horizontal") if direction == "horizontal": total_width = sum(img.width for img in images) max_height = max(img.height for img in images) new_img = Image.new("RGB", (total_width, max_height)) x = 0 for img in images: new_img.paste(img, (x, 0)) x += img.width else: max_width = max(img.width for img in images) total_height = sum(img.height for img in images) new_img = Image.new("RGB", (max_width, total_height)) y = 0 for img in images: new_img.paste(img, (0, y)) y += img.height else: return {"error": f"Unsupported combination operation {operation}"} result_path = save_image(new_img) result_base64 = encode_image(result_path) return {"combined_image": result_base64} except Exception as e: return {"error": str(e)}