tools.py

from typing import List, Callable
from duckduckgo_search import DDGS   # pip install -U duckduckgo-search
import re
import time

# -------- helper to shorten very long GAIA questions (optional but helpful)
def tighten(q: str) -> str:
    quoted = re.findall(r'"([^"]+)"', q)
    caps   = re.findall(r'\b([A-Z0-9][\w-]{2,})', q)
    short  = " ".join(quoted + caps)
    return short or q

def _raw_search(query: str, max_results: int = 5) -> List[str]:
    """Internal function that performs the actual DuckDuckGo search."""
    with DDGS() as ddgs:
        raw = list(ddgs.text(keywords=query, max_results=max_results))
    out = []
    for r in raw:
        try:
            title = r.get("title", "")
            link  = r.get("href") or r.get("link", "")
            out.append(f"{title} – {link}")
        except Exception:
            pass
    return out

def retry_ddg(
    query: str,
    max_results: int = 5,
    attempts: int = 4,
    delay_sec: int = 10,
    search_fn: Callable[[str, int], List[str]] = _raw_search,
) -> List[str]:
    """
    Retry DuckDuckGo search up to *attempts* times, waiting *delay_sec* seconds
    between attempts if no results were returned or an exception was raised.

    Parameters
    ----------
    query : str
        Search query.
    max_results : int, default 5
        Number of results to return.
    attempts : int, default 4
        Maximum number of attempts before giving up.
    delay_sec : int, default 10
        Seconds to sleep between attempts.
    search_fn : Callable
        A function with signature (query: str, max_results: int) -> List[str].
        Defaults to _raw_search.

    Returns
    -------
    List[str]
        List of result strings; may be empty if every attempt failed.
    """
    last_err = None
    for i in range(1, attempts + 1):
        try:
            results = search_fn(query, max_results)
            if results:                       # Success
                return results
            print(f"Attempt {i}/{attempts}: no results, retrying in {delay_sec}s…")
        except Exception as e:
            last_err = e                      # Keep last error for optional logging
            print(f"Attempt {i}/{attempts} failed: {e}. Retrying in {delay_sec}s…")

        if i < attempts:
            time.sleep(delay_sec)

    # All attempts failed or returned empty
    if last_err:
        print(f"All {attempts} attempts failed. Last exception: {last_err}")
    else:
        print(f"All {attempts} attempts returned empty results.")
    return []

# -------- the only search function your agent will call
def simple_search(query: str, max_results: int = 5) -> List[str]:
    """
    Perform a DuckDuckGo search and return 'title – url' snippets.
    """
    query = tighten(query)           # optional heuristic cleaner
    return retry_ddg(query, max_results)  # retry on failure