Update tools/web_search.py

tools/web_search.py

@@ -1,6 +1,8 @@
 from typing import Any, Optional
 from smolagents.tools import Tool
 import duckduckgo_search
+import random
+import time
 
 
 class DuckDuckGoSearchTool(Tool):
@@ -9,10 +11,21 @@ class DuckDuckGoSearchTool(Tool):
     inputs = {'query': {'type': 'string', 'description': 'The search query to perform.'}}
     output_type = "string"
 
-    def __init__(self, max_results=10, backend="auto", **kwargs):
+    def __init__(
+        self,
+        max_results=10,
+        backends=None,
+        retries=2,
+        min_delay=0.2,
+        max_delay=0.8,
+        **kwargs,
+    ):
         super().__init__()
         self.max_results = max_results
-        self.backend = backend
+        self.backends = backends or ["auto", "lite", "html", "api"]
+        self.retries = retries
+        self.min_delay = min_delay
+        self.max_delay = max_delay
         try:
             from duckduckgo_search import DDGS
         except ImportError as e:
@@ -21,23 +34,36 @@ class DuckDuckGoSearchTool(Tool):
             ) from e
         self.ddgs = DDGS(**kwargs)
 
+    def _format_results(self, results) -> str:
+        postprocessed_results = []
+        for result in results:
+            title = result.get("title") or "Untitled"
+            href = result.get("href") or result.get("url") or ""
+            body = result.get("body") or result.get("snippet") or ""
+            postprocessed_results.append(f"[{title}]({href})\n{body}".strip())
+        return "## Search Results\n\n" + "\n\n".join(postprocessed_results)
+
     def forward(self, query: str) -> str:
-        backends_to_try = [self.backend] + [b for b in ("auto", "lite", "html", "api") if b != self.backend]
         last_error = None
-        for backend in backends_to_try:
-            try:
-                results = list(
-                    self.ddgs.text(query, max_results=self.max_results, backend=backend)
-                )
-            except Exception as e:
-                last_error = e
-                continue
-            if results:
-                postprocessed_results = [
-                    f"[{result['title']}]({result['href']})\n{result['body']}"
-                    for result in results
-                ]
-                return "## Search Results\n\n" + "\n\n".join(postprocessed_results)
+        for backend in self.backends:
+            for attempt in range(self.retries + 1):
+                try:
+                    results = list(
+                        self.ddgs.text(query, max_results=self.max_results, backend=backend)
+                    )
+                except Exception as e:
+                    last_error = e
+                    # Jittered backoff to reduce rate-limit issues.
+                    sleep_for = min(
+                        self.max_delay,
+                        self.min_delay * (2 ** attempt) + random.random() * 0.1,
+                    )
+                    time.sleep(sleep_for)
+                    continue
+                if results:
+                    return self._format_results(results)
+                # Empty results can happen; try next backend.
+                break
         if last_error is not None:
             return (
                 "Web search failed. This can happen if outbound network access is blocked or the "
@@ -45,6 +71,6 @@ class DuckDuckGoSearchTool(Tool):
             )
         return (
             "No results found. This may indicate network restrictions or a temporary "
-            f"DuckDuckGo issue. Try a shorter query or retry later. {backend}"
+            "DuckDuckGo issue. Try a shorter query or retry later."
         )