File size: 2,950 Bytes
b6654dc 622f2bb aeb4eba b6654dc 2bf962d 622f2bb a9c5bda b6654dc 2200521 622f2bb 5e54175 622f2bb b6654dc |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 |
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(query, max_results=max_results))
out = []
for r in raw:
try:
title = r.get("title", "")
link = r.get("href") or r.get("link", "")
snippet = r.get("body") or r.get("snippet", "")
out.append(f"{title} – {link}\n{snippet}")
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
|