Add support for changing the search engine and add duckduckgo as another search engine

src/opendeepsearch/ods_agent.py

@@ -1,5 +1,5 @@
 from typing import Optional, Dict, Any
-from opendeepsearch.serp_search.serp_search import SerperAPI
+from opendeepsearch.serp_search.serp_search import SerperAPI, DuckDuckGoAPI
 from opendeepsearch.context_building.process_sources_pro import SourceProcessor
 from opendeepsearch.context_building.build_context import build_context
 from litellm import completion
@@ -20,6 +20,7 @@ def __init__(
         temperature: float = 0.2, # Slight variation while maintaining reliability
         top_p: float = 0.3, # Focus on high-confidence tokens
         reranker: Optional[str] = "None", # Optional reranker identifier
+        search_engine: str = "serper" # Default search engine
     ):
         """
         Initialize an OpenDeepSearch agent that combines web search, content processing, and LLM capabilities.
@@ -44,9 +45,13 @@ def __init__(
                 the output more focused on high-probability tokens.
             reranker (str, optional): Identifier for the reranker to use. If not provided,
                 uses the default reranker from SourceProcessor.
+            search_engine (str, default="serper"): The search engine to use ("serper" or "duckduckgo").
         """
-        # Initialize SerperAPI with optional API key
-        self.serp_search = SerperAPI(api_key=serper_api_key) if serper_api_key else SerperAPI()
+        # Initialize search engine based on the search_engine parameter
+        if search_engine == "duckduckgo":
+            self.serp_search = DuckDuckGoAPI(api_key=serper_api_key)
+        else:
+            self.serp_search = SerperAPI(api_key=serper_api_key) if serper_api_key else SerperAPI()
 
         # Update source_processor_config with reranker if provided
         if source_processor_config is None:

src/opendeepsearch/serp_search/serp_search.py

@@ -37,19 +37,24 @@ def __init__(self, data: Optional[T] = None, error: Optional[str] = None):
     def failed(self) -> bool:
         return not self.success
 
-class SerperAPI:
+class BaseSearchAPI:
+    def __init__(self, api_key: Optional[str] = None):
+        self.api_key = api_key
+
+    @staticmethod
+    def extract_fields(items: List[Dict[str, Any]], fields: List[str]) -> List[Dict[str, Any]]:
+        """Extract specified fields from a list of dictionaries"""
+        return [{key: item.get(key, "") for key in fields if key in item} for item in items]
+
+class SerperAPI(BaseSearchAPI):
     def __init__(self, config: Optional[SerperConfig] = None):
+        super().__init__(api_key=config.api_key if config else None)
         self.config = config or SerperConfig.from_env()
         self.headers = {
             'X-API-KEY': self.config.api_key,
             'Content-Type': 'application/json'
         }
 
-    @staticmethod
-    def extract_fields(items: List[Dict[str, Any]], fields: List[str]) -> List[Dict[str, Any]]:
-        """Extract specified fields from a list of dictionaries"""
-        return [{key: item.get(key, "") for key in fields if key in item} for item in items]
-
     def get_sources(
         self,
         query: str,
@@ -112,4 +117,62 @@ def get_sources(
         except requests.RequestException as e:
             return SearchResult(error=f"API request failed: {str(e)}")
         except Exception as e:
-            return SearchResult(error=f"Unexpected error: {str(e)}")
+            return SearchResult(error=f"Unexpected error: {str(e)}")
+
+class DuckDuckGoAPI(BaseSearchAPI):
+    def __init__(self, api_key: Optional[str] = None):
+        super().__init__(api_key=api_key)
+        self.api_url = "https://api.duckduckgo.com/"
+
+    def get_sources(
+        self,
+        query: str,
+        num_results: int = 8,
+        stored_location: Optional[str] = None
+    ) -> SearchResult[Dict[str, Any]]:
+        """
+        Fetch search results from DuckDuckGo API.
+
+        Args:
+            query: Search query string
+            num_results: Number of results to return (default: 10, max: 20 for pro)
+            stored_location: Optional location string
+
+        Returns:
+            SearchResult containing the search results or error information
+        """
+        if not query.strip():
+            return SearchResult(error="Query cannot be empty")
+
+        try:
+            params = {
+                "q": query,
+                "format": "json",
+                "no_redirect": 1,
+                "no_html": 1,
+                "skip_disambig": 1
+            }
+
+            response = requests.get(
+                self.api_url,
+                params=params,
+                timeout=10
+            )
+            response.raise_for_status()
+            data = response.json()
+
+            results = {
+                'organic': self.extract_fields(
+                    data.get('RelatedTopics', []),
+                    ['Text', 'FirstURL', 'Icon']
+                ),
+                'answerBox': data.get('AbstractText'),
+                'relatedSearches': data.get('RelatedTopics')
+            }
+
+            return SearchResult(data=results)
+
+        except requests.RequestException as e:
+            return SearchResult(error=f"API request failed: {str(e)}")
+        except Exception as e:
+            return SearchResult(error=f"Unexpected error: {str(e)}")