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