Create agent.py

agent.py

@@ -0,0 +1,795 @@
+import os
+from dotenv import load_dotenv
+from typing import List, Dict, Any, Optional
+import tempfile
+import re
+import json
+import requests
+from urllib.parse import urlparse
+import pytesseract
+from PIL import Image, ImageDraw, ImageFont, ImageEnhance, ImageFilter
+import cmath
+import pandas as pd
+import uuid
+import numpy as np
+import base64
+import time
+from e2b_code_interpreter import Sandbox
+from pydantic.v1 import BaseModel, Field
+
+"""Langraph"""
+from langgraph.graph import StateGraph, MessagesState
+from langchain_community.tools.tavily_search import TavilySearchResults
+from langchain_community.document_loaders import WikipediaLoader
+from langchain_community.document_loaders import ArxivLoader
+from langgraph.prebuilt import ToolNode, tools_condition
+from langchain_google_genai import ChatGoogleGenerativeAI
+from langchain_core.messages import SystemMessage, HumanMessage, ToolMessage
+from langchain_core.tools import tool
+from langchain_core.tools import Tool
+
+load_dotenv()
+
+# Helper functions for image processing
+def encode_image(image_path: str) -> str:
+    """Convert an image file to base64 string."""
+    with open(image_path, "rb") as image_file:
+        return base64.b64encode(image_file.read()).decode("utf-8")
+
+
+def decode_image(base64_string: str) -> Image.Image:
+    """Convert a base64 string to a PIL Image."""
+    image_data = base64.b64decode(base64_string)
+    return Image.open(io.BytesIO(image_data))
+
+
+def save_image(image: Image.Image, directory: str = "image_outputs") -> str:
+    """Save a PIL Image to disk and return the path."""
+    os.makedirs(directory, exist_ok=True)
+    image_id = str(uuid.uuid4())
+    image_path = os.path.join(directory, f"{image_id}.png")
+    image.save(image_path)
+    return image_path
+
+### =============== BROWSER TOOLS =============== ###
+
+
+@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'<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}
+
+
+@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({'query':query})
+    formatted_search_docs = "\n\n---\n\n".join(
+        [
+            f'<Document source="{doc["url"]}" page=""/>\n{doc["content"]}\n</Document>'
+            for doc in search_docs
+        ])
+    return {"web_results": formatted_search_docs}
+
+
+@tool
+def arxiv_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'<Document source="{doc.metadata["source"]}" page="{doc.metadata.get("page", "")}"/>\n{doc.page_content[:1000]}\n</Document>'
+            for doc in search_docs
+        ]
+    )
+    return {"arxiv_results": formatted_search_docs}
+
+
+### =============== CODE INTERPRETER TOOLS =============== ###
+
+
+class RichToolMessage(ToolMessage):
+    raw_output: dict
+
+
+class LangchainCodeInterpreterToolInput(BaseModel):
+    code: str = Field(description="Python code to execute.")
+
+
+class CodeInterpreterFunctionTool:
+    """
+    This class calls arbitrary code against a Python Jupyter notebook.
+    It requires an E2B_API_KEY to create a sandbox.
+    """
+
+    tool_name: str = "code_interpreter"
+
+    def __init__(self):
+        # Instantiate the E2B sandbox - this is a long lived object
+        # that's pinging E2B cloud to keep the sandbox alive.
+        if "E2B_API_KEY" not in os.environ:
+            raise Exception(
+                "Code Interpreter tool called while E2B_API_KEY environment variable is not set. Please get your E2B api key here https://e2b.dev/docs and set the E2B_API_KEY environment variable."
+            )
+        self.code_interpreter = Sandbox()
+
+    def close(self):
+        self.code_interpreter.kill()
+
+    def call(self, parameters: dict, **kwargs: Any):
+        code = parameters.get("code", "")
+        print(f"***Code Interpreting...\n{code}\n====")
+        execution = self.code_interpreter.run_code(code)
+        return {
+            "results": execution.results,
+            "stdout": execution.logs.stdout,
+            "stderr": execution.logs.stderr,
+            "error": execution.error,
+        }
+
+    # langchain does not return a dict as a parameter, only a code string
+    def langchain_call(self, code: str):
+        return self.call({"code": code})
+
+    def to_langchain_tool(self) -> Tool:
+        tool = Tool(
+            name=self.tool_name,
+            description="Execute python code in a Jupyter notebook cell and returns any rich data (eg charts), stdout, stderr, and error.",
+            func=self.langchain_call,
+        )
+        tool.args_schema = LangchainCodeInterpreterToolInput
+        return tool
+
+    @staticmethod
+    def format_to_tool_message(
+        tool_call_id: str,
+        output: dict,
+    ) -> RichToolMessage:
+        """
+        Format the output of the CodeInterpreter tool to be returned as a RichToolMessage.
+        """
+
+        # TODO: Add info about the results for the LLM
+        content = json.dumps(
+            {k: v for k, v in output.items() if k not in ("results")}, indent=2
+        )
+
+        return RichToolMessage(
+            content=content,
+            raw_output=output,
+            tool_call_id=tool_call_id,
+        )
+
+
+### =============== MATHEMATICAL TOOLS =============== ###
+
+
+@tool
+def multiply(a: float, b: float) -> float:
+    """
+    Multiplies two numbers.
+
+    Args:
+        a (float): the first number
+        b (float): the second number
+    """
+    return a * b
+
+
+@tool
+def add(a: float, b: float) -> float:
+    """
+    Adds two numbers.
+
+    Args:
+        a (float): the first number
+        b (float): the second number
+    """
+    return a + b
+
+
+@tool
+def subtract(a: float, b: float) -> int:
+    """
+    Subtracts two numbers.
+
+    Args:
+        a (float): the first number
+        b (float): the second number
+    """
+    return a - b
+
+
+@tool
+def divide(a: float, b: float) -> float:
+    """
+    Divides two numbers.
+
+    Args:
+        a (float): the first float number
+        b (float): the second float number
+    """
+    if b == 0:
+        raise ValueError("Cannot divided by zero.")
+    return a / b
+
+
+@tool
+def modulus(a: int, b: int) -> int:
+    """
+    Get the modulus of two numbers.
+
+    Args:
+        a (int): the first number
+        b (int): the second number
+    """
+    return a % b
+
+
+@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)
+
+
+### =============== DOCUMENT PROCESSING TOOLS =============== ###
+
+
+@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)}"
+
+
+### ============== IMAGE PROCESSING AND GENERATION TOOLS =============== ###
+
+
+@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 = img.transpose(Image.FLIP_LEFT_RIGHT)
+            else:
+                img = img.transpose(Image.FLIP_TOP_BOTTOM)
+        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)}
+
+# System message
+sys_msg = SystemMessage(content=os.getenv("SYSTEM_PROMPT"))
+
+code_interpreter = CodeInterpreterFunctionTool()
+code_interpreter_tool = code_interpreter.to_langchain_tool()
+
+tools = [
+    web_search,
+    wiki_search,
+    arxiv_search,
+    multiply,
+    add,
+    subtract,
+    divide,
+    modulus,
+    power,
+    square_root,
+    save_and_read_file,
+    download_file_from_url,
+    extract_text_from_image,
+    analyze_csv_file,
+    analyze_excel_file,
+    analyze_image,
+    transform_image,
+    draw_on_image,
+    generate_simple_image,
+    combine_images,
+    code_interpreter_tool
+]
+
+
+# Build graph function
+def build_graph(provider: str = "groq"):
+    """Build the graph"""
+    llm = ChatGoogleGenerativeAI(
+        model= "gemini-2.0-flash",
+        temperature=1.0,
+        max_retries=2,
+        google_api_key=os.getenv("GOOGLE_APY_KEY")
+    )
+    # Bind tools to LLM
+    llm_with_tools = llm.bind_tools(tools)
+
+    # Node
+    def assistant(state: MessagesState):
+        """Assistant node"""
+        return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}
+
+    builder = StateGraph(MessagesState)
+    builder.add_node("assistant", assistant)
+    builder.add_node("tools", ToolNode(tools))
+    builder.set_entry_point("assistant")
+    builder.add_conditional_edges(
+        "assistant",
+        tools_condition,
+    )
+    builder.add_edge("tools", "assistant")
+
+    # Compile graph
+    return builder.compile()
+
+
+# test
+if __name__ == "__main__":
+
+    questions_url = f"https://agents-course-unit4-scoring.hf.space/questions"
+
+    # 2. Fetch Questions
+    print(f"Fetching questions from: {questions_url}")
+    try:
+        response = requests.get(questions_url, timeout=15)
+        response.raise_for_status()
+        questions_data = response.json()
+        if not questions_data:
+            print("Fetched questions list is empty.")
+        print(f"Fetched {len(questions_data)} questions.")
+        for question in questions_data:
+            task_id = question.get("task_id")
+            question_text = question.get("question")
+            # print(f"Task ID: {task_id}, Question: {question_text}\n")
+    except requests.exceptions.RequestException as e:
+        print(f"Error fetching questions: {e}")
+    except requests.exceptions.JSONDecodeError as e:
+        print(f"Error decoding JSON response from questions endpoint: {e}")
+        print(f"Response text: {response.text[:500]}")
+    except Exception as e:
+                        print(f"An unexpected error occurred fetching questions: {e}")
+
+    # question = questions_data[0].get('question')
+    for item in questions_data:
+        question = item.get("question")
+        graph = build_graph()
+        messages = [HumanMessage(content=question)]
+        messages = graph.invoke({"messages": messages})
+        for m in messages["messages"]:
+            m.pretty_print()
+        time.sleep(4.1)