research/search.py

"""DuckDuckGo web search tool for the research agents (free, no API key).

Uses the maintained ``ddgs`` package. Degrades gracefully on rate limits / empty
results so the writing step can still proceed from whatever was gathered.
"""

from __future__ import annotations

import time
from dataclasses import dataclass
from typing import List


@dataclass
class SearchResult:
    title: str
    url: str
    snippet: str

    def as_markdown(self) -> str:
        return f"- [{self.title}]({self.url})\n  {self.snippet}"


def web_search(query: str, max_results: int = 5, *, retries: int = 2) -> List[SearchResult]:
    """Return up to ``max_results`` results for ``query``. Never raises."""
    try:
        from ddgs import DDGS
    except ImportError:  # pragma: no cover - older package name fallback
        try:
            from duckduckgo_search import DDGS  # type: ignore
        except ImportError:
            return []

    last_err: Exception | None = None
    for attempt in range(retries + 1):
        try:
            with DDGS() as ddgs:
                hits = ddgs.text(query, max_results=max_results)
            results = [
                SearchResult(
                    title=h.get("title", "").strip(),
                    url=h.get("href") or h.get("url", ""),
                    snippet=(h.get("body") or h.get("snippet") or "").strip(),
                )
                for h in (hits or [])
            ]
            return [r for r in results if r.url]
        except Exception as exc:  # rate limit / network hiccup
            last_err = exc
            if attempt < retries:
                time.sleep(1.5 * (attempt + 1))
    # Give up quietly; caller handles empty findings.
    print(f"[search] '{query}' failed after {retries + 1} attempts: {last_err}")
    return []