"""
Search tools for web search capabilities.
"""
import logging
from typing import Any
from urllib.parse import parse_qs, urlparse
import requests
from bs4 import BeautifulSoup
from ddgs import DDGS
from langchain_core.tools import BaseTool, StructuredTool
from langchain_tavily import TavilySearch
from pydantic import BaseModel, Field
from config import config
logger = logging.getLogger(__name__)
DUCKDUCKGO_LITE_URL = "https://lite.duckduckgo.com/lite/"
# DDGS no longer exposes Bing text search; Yahoo still provides that index.
KEYLESS_FALLBACK_BACKENDS = "brave,google,yahoo"
SEARCH_USER_AGENT = (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) Safari/537.36"
)
def _search_error(provider: str, error: Exception) -> list[dict[str, str]]:
"""Return a recoverable tool result instead of crashing the agent graph."""
logger.warning("%s web search failed: %s", provider, error)
return [
{
"error": (
f"{provider} web search is temporarily unavailable. "
"Try again later or refine the query."
)
}
]
def _duckduckgo_result_url(href: str) -> str:
"""Extract the destination from a DuckDuckGo redirect URL."""
absolute_href = f"https:{href}" if href.startswith("//") else href
redirect_target = parse_qs(urlparse(absolute_href).query).get("uddg")
return redirect_target[0] if redirect_target else absolute_href
def _search_duckduckgo_lite(query: str, max_results: int) -> list[dict[str, str]]:
"""Search DuckDuckGo's lightweight endpoint and normalize its HTML results."""
response = requests.get(
DUCKDUCKGO_LITE_URL,
params={"q": query, "kl": config.search_region},
headers={"User-Agent": SEARCH_USER_AGENT},
timeout=config.search_timeout_seconds,
)
response.raise_for_status()
soup = BeautifulSoup(response.text, "html.parser")
results: list[dict[str, str]] = []
seen_urls: set[str] = set()
for link in soup.select("a.result-link"):
href = link.get("href")
if not isinstance(href, str):
continue
url = _duckduckgo_result_url(href)
if not url.startswith(("http://", "https://")) or url in seen_urls:
continue
row = link.find_parent("tr")
snippet_row = row.find_next_sibling("tr") if row else None
snippet = snippet_row.select_one("td.result-snippet") if snippet_row else None
results.append(
{
"title": link.get_text(" ", strip=True),
"url": url,
"snippet": snippet.get_text(" ", strip=True) if snippet else "",
}
)
seen_urls.add(url)
if len(results) >= max_results:
break
if not results:
raise RuntimeError("DuckDuckGo Lite returned no results")
return results
def _search_keyless_fallback(query: str, max_results: int) -> list[dict[str, str]]:
"""Use free search backends if DuckDuckGo temporarily blocks automation."""
results = DDGS(timeout=config.search_timeout_seconds).text(
query,
region=config.search_region,
safesearch="moderate",
max_results=max_results,
backend=KEYLESS_FALLBACK_BACKENDS,
)
if not results:
raise RuntimeError("Keyless fallback returned no results")
return [
{
"title": str(result.get("title", "")),
"url": str(result.get("href", "")),
"snippet": str(result.get("body", "")),
}
for result in results
]
def _create_duckduckgo_tool(max_results: int) -> BaseTool:
"""Create the keyless DuckDuckGo search implementation."""
def search_duckduckgo(query: str) -> list[dict[str, str]]:
if not query.strip():
return [{"error": "A non-empty web-search query is required."}]
duckduckgo_error: Exception | None = None
for attempt in range(2):
try:
return _search_duckduckgo_lite(query, max_results)
except Exception as error: # Search failures must not abort the graph.
duckduckgo_error = error
if attempt == 0:
logger.info("Retrying DuckDuckGo Lite search after: %s", error)
logger.warning(
"DuckDuckGo Lite failed; trying free search fallback: %s",
duckduckgo_error,
)
try:
return _search_keyless_fallback(query, max_results)
except Exception as fallback_error:
return _search_error("DuckDuckGo", fallback_error)
return StructuredTool.from_function(
func=search_duckduckgo,
name="web_search",
description=(
"Search the public web with keyless DuckDuckGo-first search. Returns "
"titles, URLs, and short snippets. Cite the returned URLs in the answer."
),
args_schema=WebSearchInput,
)
def _create_tavily_tool(max_results: int) -> BaseTool:
"""Create the optional Tavily implementation behind the same tool contract."""
if not config.tavily_api_key:
raise ValueError("TAVILY_API_KEY must be set when SEARCH_PROVIDER=tavily")
tavily = TavilySearch(
api_key=config.tavily_api_key,
max_results=max_results,
)
def search_tavily(query: str) -> Any:
if not query.strip():
return [{"error": "A non-empty web-search query is required."}]
try:
return tavily.invoke({"query": query})
except Exception as error: # Search failures must not abort the conversation.
return _search_error("Tavily", error)
return StructuredTool.from_function(
func=search_tavily,
name="web_search",
description=(
"Search the public web with Tavily. Returns sources and content. "
"Cite the returned URLs in the answer."
),
args_schema=WebSearchInput,
)
