tools/search.py

"""웹 검색 툴. 백엔드 우선순위: SearXNG → Tavily → Brave → DuckDuckGo.

SearXNG 는 키 없이 동작하는 메타검색(공개 인스턴스 폴백). Tavily/Brave 는 환경변수로
API 키가 설정된 경우에만 사용. 둘 다 무료 티어가 있다 (Tavily 1k/월, Brave 2k/월
가량). DDG는 마지막 안전망이지만 종종 차단/디코딩 에러가 나므로 우선 백엔드들을
앞에 두는 게 안정적이다.

각 백엔드는 결과가 있으면 포맷된 문자열, 없으면 None 을 반환. 호출자(web_search)는
None 을 만나면 다음 백엔드로 폴백한다. DDG 는 마지막 폴백이라 None 대신 항상
문자열(에러 메시지 또는 "No results found.")을 반환한다.

환경변수:
    TAVILY_API_KEY  Tavily Search API 키 (옵션)
    BRAVE_API_KEY   Brave Search API 키 (옵션)
"""
import os
import random
import requests
from smolagents import tool

_TAVILY_URL = "https://api.tavily.com/search"
_BRAVE_URL = "https://api.search.brave.com/res/v1/web/search"

# SearXNG 공개 인스턴스 풀. 키 불필요. 호출마다 일부만 무작위로 골라 시도해서
# (a) 한 인스턴스가 IP 차단 가속되는 걸 분산하고 (b) 누적 timeout 상한을 통제한다.
# searx.space 가용 목록을 주기적으로 갱신할 것.
_SEARXNG_INSTANCES = (
    "https://searx.be",
    "https://searx.tiekoetter.com",
    "https://search.inetol.net",
    "https://searxng.online",
    "https://priv.au",
)
_SEARXNG_TRY_COUNT = 3        # 호출당 시도할 인스턴스 수
_SEARXNG_TIMEOUT = 5          # 인스턴스당 타임아웃(초) — 누적 상한 ~15s


def _format_results(items) -> str:
    """공통 포매터: (title, url, snippet) 튜플 리스트를 LLM-friendly 텍스트로."""
    lines = [f"- {t}\n  {u}\n  {b}" for t, u, b in items if (t or u or b)]
    return "\n".join(lines) if lines else ""


def _search_searxng(query: str) -> str | None:
    """SearXNG 메타검색. Google/Bing/DDG 등 70+ 엔진을 묶어 반환. 키 불필요.
    공개 인스턴스 폴백 — 한 곳 죽으면 다음으로. 모두 실패하면 None 반환해
    호출자가 다음 백엔드(Tavily/Brave/DDG)로 폴백하게 한다.

    일부 인스턴스는 빈 UA 또는 봇처럼 보이는 요청을 차단하므로 브라우저 UA를 명시.
    """
    headers = {
        "User-Agent": (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
            "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
        ),
        "Accept": "application/json",
    }
    # 호출마다 무작위 부분집합 → 부하 분산 + 누적 timeout 통제.
    candidates = random.sample(_SEARXNG_INSTANCES, _SEARXNG_TRY_COUNT)
    for base in candidates:
        try:
            r = requests.get(
                f"{base}/search",
                params={"q": query, "format": "json", "language": "en"},
                headers=headers,
                timeout=_SEARXNG_TIMEOUT,
            )
            if r.status_code != 200:
                continue
            results = r.json().get("results", [])
            if not results:
                continue
            items = [
                (x.get("title", ""), x.get("url", ""), x.get("content", ""))
                for x in results[:8]  # 토큰 제어, DDG와 동일한 max_results=8
            ]
            formatted = _format_results(items)
            if formatted:
                return formatted
        except Exception as e:
            print(f"SearXNG ({base}) failed: {e}")
            continue
    return None


def _search_tavily(query: str) -> str | None:
    """Tavily Search API. TAVILY_API_KEY 가 있어야 호출."""
    api_key = os.getenv("TAVILY_API_KEY")
    if not api_key:
        return None
    try:
        r = requests.post(
            _TAVILY_URL,
            json={"api_key": api_key, "query": query, "max_results": 8},
            timeout=15,
        )
        r.raise_for_status()
        results = r.json().get("results", [])
        if not results:
            return None
        items = [
            (x.get("title", ""), x.get("url", ""), x.get("content", ""))
            for x in results
        ]
        formatted = _format_results(items)
        return formatted or None
    except Exception as e:
        print(f"Tavily search failed (falling back): {e}")
        return None


def _search_brave(query: str) -> str | None:
    """Brave Search API. BRAVE_API_KEY 가 있어야 호출."""
    api_key = os.getenv("BRAVE_API_KEY")
    if not api_key:
        return None
    try:
        r = requests.get(
            _BRAVE_URL,
            params={"q": query, "count": 8},
            headers={
                "X-Subscription-Token": api_key,
                "Accept": "application/json",
            },
            timeout=15,
        )
        r.raise_for_status()
        results = r.json().get("web", {}).get("results", [])
        if not results:
            return None
        items = [
            (x.get("title", ""), x.get("url", ""), x.get("description", ""))
            for x in results
        ]
        formatted = _format_results(items)
        return formatted or None
    except Exception as e:
        print(f"Brave search failed (falling back): {e}")
        return None


def _search_ddg(query: str) -> str:
    """DuckDuckGo. ddgs 패키지 우선, 실패 시 구 duckduckgo_search 폴백.
    마지막 폴백이라 None 대신 항상 문자열을 반환한다(에러 메시지 또는 "No results found.")."""
    # DDG 클라이언트 패키지 이름이 `duckduckgo_search` → `ddgs`로 바뀌었고
    # 구 패키지에서는 "Body collection error: ..." 같은 디코딩 에러가 빈번했다.
    last_err = None
    for module_name in ("ddgs", "duckduckgo_search"):
        try:
            mod = __import__(module_name, fromlist=["DDGS"])
            DDGS = getattr(mod, "DDGS")
            with DDGS() as ddgs:
                # max_results=8: 너무 적으면 정답 사이트 누락, 너무 많으면 컨텍스트 낭비.
                results = list(ddgs.text(query, max_results=8))
            if not results:
                continue
            # 두 패키지가 키 이름이 미묘하게 다르므로 양쪽 모두 처리.
            items = [
                (
                    r.get("title", ""),
                    r.get("href", "") or r.get("url", ""),
                    r.get("body", "") or r.get("snippet", ""),
                )
                for r in results
            ]
            formatted = _format_results(items)
            if formatted:
                return formatted
        except Exception as e:
            last_err = e
            continue
    if last_err:
        return f"web_search error: {last_err}"
    return "No results found."


@tool
def web_search(query: str) -> str:
    """Search the web and return a list of titles, URLs, and snippets.
    Backend priority: SearXNG public instances (no key) -> Tavily/Brave (only if their
    API keys are set in environment variables TAVILY_API_KEY, BRAVE_API_KEY) ->
    DuckDuckGo fallback.

    Args:
        query: The search query string.
    """
    # SearXNG가 1순위: 키 없이 가장 양질의 결과를 주는 백엔드.
    # Tavily/Brave는 키가 환경변수에 있을 때만 시도(없으면 None 반환하고 통과).
    # DDG는 마지막 안전망.
    out = _search_searxng(query)
    if out:
        return out
    for backend in (_search_tavily, _search_brave):
        out = backend(query)
        if out:
            return out
    return _search_ddg(query)