app.py

from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool
import datetime
import requests
import pytz
import yaml
from tools.final_answer import FinalAnswerTool

from Gradio_UI import GradioUI
import sqlite3
import os
from smolagents import tool

DB_NAME = "robot_memory.db"
TABLE_NAME = "robot_memories"

def initialize_database():
    """Inizializza il database e crea la tabella della memoria se non esiste già."""
    # Crea (o apre) il database
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    
    # Crea la tabella 'robot_memories' se non esiste già
    cursor.execute(f"""
        CREATE TABLE IF NOT EXISTS {TABLE_NAME} (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            memory_text TEXT NOT NULL,
            tags TEXT
        );
    """)
    conn.commit()
    conn.close()

@tool
def robot_memory_tool(action: str, content: str = "", tags: str = "") -> str:
    """A memory management tool that allows storing, retrieving, and deleting memories 
    from a SQLite database. It supports tagging and keyword search for efficient memory retrieval.
    Args:
        action: The action to perform. Supported actions are:
                      - 'store': Save a new memory with optional tags.
                      - 'retrieve': Search for stored memories based on text content and/or tags.
                      - 'delete': Remove memories based on text content and/or tags.
                      - 'help': Display usage examples.
        content: The text of the memory or search criteria. Defaults to an empty string.
        tags: Tags associated with the memory for categorization or search. Defaults to an empty string.

    Returns:
        str: A message indicating the result of the action:
             - For 'store': Confirmation of successful memory storage.
             - For 'retrieve': A formatted list of matching memories or a message indicating no matches.
             - For 'delete': Confirmation of the number of deleted memories.
             - For 'help': A usage guide with examples.
             - For unsupported actions: An error message.

    Usage Examples:
        >>> robot_memory_tool('store', content='Ho visto un gatto', tags='animale, esterno')
        "[STORE] Ricordo memorizzato con successo: 'Ho visto un gatto' (tags: 'animale, esterno')"

        >>> robot_memory_tool('retrieve', content='gatto')
        "[RETRIEVE] Ricordi trovati:\n- ID: 1, Testo: 'Ho visto un gatto', Tags: 'animale, esterno'"

        >>> robot_memory_tool('delete', tags='animale')
        "[DELETE] Rimossi 1 ricordo/i dal database."

        >>> robot_memory_tool('help')
        "Esempi di utilizzo:\n1) robot_memory_tool('store', content='Ho visto un gatto', tags='animale, esterno')\n2) robot_memory_tool('retrieve', content='gatto')\n3) robot_memory_tool('delete', tags='animale')\n4) robot_memory_tool('help')"
    """
    initialize_database()  # Assicura che il DB e la tabella esistano
    conn = sqlite3.connect(DB_NAME)
    cursor = conn.cursor()
    
    # Normalizza i parametri
    action = action.lower().strip()
    content = content.strip()
    tags = tags.strip()

    if action == 'store':
        # Inserisce un nuovo record nella tabella
        cursor.execute(f"""
            INSERT INTO {TABLE_NAME} (memory_text, tags)
            VALUES (?, ?);
        """, (content, tags))
        conn.commit()
        conn.close()
        return f"[STORE] Ricordo memorizzato con successo: '{content}' (tags: '{tags}')"

    elif action == 'retrieve':
        # Recupera i ricordi in base ai criteri di ricerca
        query = f"SELECT id, memory_text, tags FROM {TABLE_NAME} WHERE 1=1"
        params = []
        
        if content:
            query += " AND memory_text LIKE ?"
            params.append(f"%{content}%")
        if tags:
            query += " AND tags LIKE ?"
            params.append(f"%{tags}%")
        
        cursor.execute(query, tuple(params))
        rows = cursor.fetchall()
        conn.close()

        if not rows:
            return "[RETRIEVE] Nessun ricordo trovato con i criteri specificati."
        
        # Format del risultato
        result = "[RETRIEVE] Ricordi trovati:\n"
        for row in rows:
            rec_id, memory_text, memory_tags = row
            result += f"- ID: {rec_id}, Testo: '{memory_text}', Tags: '{memory_tags}'\n"
        return result

    elif action == 'delete':
        # Elimina i ricordi in base ai criteri di ricerca
        query = f"DELETE FROM {TABLE_NAME} WHERE 1=1"
        params = []
        
        if content:
            query += " AND memory_text LIKE ?"
            params.append(f"%{content}%")
        if tags:
            query += " AND tags LIKE ?"
            params.append(f"%{tags}%")

        cursor.execute(query, tuple(params))
        deleted_count = cursor.rowcount
        conn.commit()
        conn.close()
        
        return f"[DELETE] Rimossi {deleted_count} ricordo/i dal database."

    elif action == 'help':
        conn.close()
        return (
            "Esempi di utilizzo:\n"
            "1) robot_memory_tool('store', content='Ho visto un gatto', tags='animale, esterno')\n"
            "2) robot_memory_tool('retrieve', content='gatto')\n"
            "3) robot_memory_tool('delete', tags='animale')\n"
            "4) robot_memory_tool('help')"
        )

    else:
        conn.close()
        return "[ERROR] Azione non supportata. Usa 'store', 'retrieve', 'delete', oppure 'help'."

@tool
def get_current_time_in_timezone(timezone: str) -> str:
    """A tool that fetches the current local time in a specified timezone.
    Args:
        timezone: A string representing a valid timezone (e.g., 'America/New_York').
    """
    try:
        # Create timezone object
        tz = pytz.timezone(timezone)
        # Get current time in that timezone
        local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
        return f"The current local time in {timezone} is: {local_time}"
    except Exception as e:
        return f"Error fetching time for timezone '{timezone}': {str(e)}"


final_answer = FinalAnswerTool()

# If the agent does not answer, the model is overloaded, please use another model or the following Hugging Face Endpoint that also contains qwen2.5 coder:
# model_id='https://pflgm2locj2t89co.us-east-1.aws.endpoints.huggingface.cloud' 

model = HfApiModel(
max_tokens=2096,
temperature=0.5,
model_id='https://pflgm2locj2t89co.us-east-1.aws.endpoints.huggingface.cloud',# it is possible that this model may be overloaded
custom_role_conversions=None,
)

hf_token_setted = False

@tool
def image_generation_tool(text: str) -> str:
    """Generates an image based on the input text using the 'text-to-image' tool.
    Args:
        text: The input text to generate an image from.
    Returns:
        str: The URL of the generated image.
    """
    #if the user is not autenthicated to hugging face the following code will not work 
    if hf_token_setted == False:
        return final_answer("You need to be authenticated to Hugging Face to use this tool. Provide your 'HF_TOKEN'.")
    
    image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True)
    return image_generation_tool(text)


from huggingface_hub import login
@tool
def login_to_hugging_face(HF_TOKEN: str) -> str:
    """Logs in to the Hugging Face API using the provided API token.
    Args:
        HF_TOKEN: The Hugging Face API token.
    Returns:
        str: A message indicating the result of the login attempt.
    """
    global hf_token_setted
    try:
        login(HF_TOKEN)
    except Exception as e:
        return f"Failed to log in to Hugging Face: {str(e)}"
    hf_token_setted = True
    return "Logged in to Hugging Face successfully."

with open("prompts.yaml", 'r') as stream:
    prompt_templates = yaml.safe_load(stream)
    
agent = CodeAgent(
    model=model,
    tools=[final_answer, robot_memory_tool, image_generation_tool, login_to_hugging_face], ## add your tools here (don't remove final answer)
    max_steps=6,
    verbosity_level=1,
    grammar=None,
    planning_interval=None,
    name=None,
    description=None,
    prompt_templates=prompt_templates
)



GradioUI(agent).launch()