Merge pull request #3029 from Yash-1511/main

feat: add DuckDuckGo search functionality using duckduckgo_search library

backend/apps/rag/main.py

@@ -71,6 +71,7 @@ from apps.rag.search.searxng import search_searxng
 from apps.rag.search.serper import search_serper
 from apps.rag.search.serpstack import search_serpstack
 from apps.rag.search.serply import search_serply
+from apps.rag.search.duckduckgo import search_duckduckgo
 
 from utils.misc import (
     calculate_sha256,
@@ -821,6 +822,8 @@ def search_web(engine: str, query: str) -> list[SearchResult]:
             )
         else:
             raise Exception("No SERPLY_API_KEY found in environment variables")
+    elif engine == "duckduckgo":
+        return search_duckduckgo(query, app.state.config.RAG_WEB_SEARCH_RESULT_COUNT)
     else:
         raise Exception("No search engine API key found in environment variables")
 

backend/apps/rag/search/duckduckgo.py

@@ -0,0 +1,46 @@
+import logging
+
+from apps.rag.search.main import SearchResult
+from duckduckgo_search import DDGS
+from config import SRC_LOG_LEVELS
+
+log = logging.getLogger(__name__)
+log.setLevel(SRC_LOG_LEVELS["RAG"])
+
+
+def search_duckduckgo(query: str, count: int) -> list[SearchResult]:
+    """
+    Search using DuckDuckGo's Search API and return the results as a list of SearchResult objects.
+    Args:
+        query (str): The query to search for
+        count (int): The number of results to return
+
+    Returns:
+        List[SearchResult]: A list of search results
+    """
+    # Use the DDGS context manager to create a DDGS object
+    with DDGS() as ddgs:
+        # Use the ddgs.text() method to perform the search
+        ddgs_gen = ddgs.text(
+            query, safesearch="moderate", max_results=count, backend="api"
+        )
+        # Check if there are search results
+        if ddgs_gen:
+            # Convert the search results into a list
+            search_results = [r for r in ddgs_gen]
+
+    # Create an empty list to store the SearchResult objects
+    results = []
+    # Iterate over each search result
+    for result in search_results:
+        # Create a SearchResult object and append it to the results list
+        results.append(
+            SearchResult(
+                link=result["href"],
+                title=result.get("title"),
+                snippet=result.get("body"),
+            )
+        )
+    print(results)
+    # Return the list of search results
+    return results

backend/requirements.txt

@@ -59,4 +59,5 @@ youtube-transcript-api==0.6.2
 pytube==15.0.0
 
 extract_msg
-pydub
+pydub
+duckduckgo-search~=6.1.5