tools.py

import os
import time
import json
import requests
from duckduckgo_search import DDGS
from smolagents import HfApiModel, tool

MEMORY_FILE = "memory.json"

def load_memory():
    if os.path.exists(MEMORY_FILE):
        try:
            with open(MEMORY_FILE, "r", encoding="utf-8") as f:
                content = f.read().strip()
                if not content:
                    return {}
                return json.loads(content)
        except json.JSONDecodeError:
            return {}
    return {}

def save_memory(memory):
    with open(MEMORY_FILE, "w", encoding="utf-8") as f:
        json.dump(memory, f, ensure_ascii=False, indent=2)

# Modello di sintesi testo-based
summarizer_model = HfApiModel(
    model_id="google/flan-t5-base",
    token=os.getenv("HUGGINGFACEHUB_API_TOKEN")
)

def summarize_text(text: str, max_length: int = 150) -> str:
    prompt = f"Riassumi brevemente il seguente testo:\n\n{text}\n\nRiassunto:"
    summary = summarizer_model.generate(prompt=prompt, max_new_tokens=max_length)
    return summary.strip()

@tool
def duckduckgo_search(query: str, max_results: int = 3) -> str:
    """
    Performs a search on DuckDuckGo and returns the results.

    Args:
        query (str): The search query.
        max_results (int, optional): Maximum number of results to return. Defaults to 3.

    Returns:
        str: A formatted string containing the search results.

    Raises:
        [Insert any exceptions the function might raise, if applicable]
    """
          
    results_texts = []
    retry_attempts = 2
    delay_seconds = 5
    
    for attempt in range(retry_attempts):
        try:
            with DDGS() as ddgs:
                results_gen = ddgs.text(query, max_results=max_results)
                for i, result in enumerate(results_gen):
                    if i >= max_results:
                        break
                    snippet = result.get('body') or result.get('text') or ''
                    results_texts.append(snippet)
            if results_texts:
                return "\n".join(results_texts)
            else:
                return "Nessun risultato DuckDuckGo."
        except Exception as e:
            if "Ratelimit" in str(e) or "429" in str(e):
                if attempt < retry_attempts - 1:
                    time.sleep(delay_seconds)
                    delay_seconds *= 2  # Backoff esponenziale
                    continue
                else:
                    return "Errore: superato limite richieste DuckDuckGo. Riprova tra qualche minuto."
            else:
                return f"Errore durante la ricerca DuckDuckGo: {str(e)}"
    return "Errore sconosciuto durante la ricerca DuckDuckGo."

@tool
def wikipedia_search(query: str) -> str:
    """
    Performs a search on Wikipedia  and returns the results.

    Args:
        query (str): The search query.
        max_results (int, optional): Maximum number of results to return. Defaults to 3.

    Returns:
        str: A formatted string containing the search results.

    Raises:
        [Insert any exceptions the function might raise, if applicable]
    """
    
    url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{query.replace(' ', '_')}"
    headers = {"User-Agent": "MyAgent/1.0 ([email protected])"}  # Sostituisci con il tuo indirizzo email
    try:
        resp = requests.get(url, headers=headers, timeout=10)
        if resp.status_code == 200:
            data = resp.json()
            return data.get("extract", "Nessun sommario disponibile.")
        elif resp.status_code == 404:
            return "Pagina Wikipedia non trovata."
        else:
            return f"Errore nella richiesta Wikipedia: {resp.status_code}"
    except requests.RequestException as e:
        return f"Errore nella richiesta Wikipedia: {str(e)}"