serper.py 1.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243
  1. import json
  2. import logging
  3. from typing import Optional
  4. import requests
  5. from open_webui.apps.rag.search.main import SearchResult, get_filtered_results
  6. from open_webui.env import SRC_LOG_LEVELS
  7. log = logging.getLogger(__name__)
  8. log.setLevel(SRC_LOG_LEVELS["RAG"])
  9. def search_serper(
  10. api_key: str, query: str, count: int, filter_list: Optional[list[str]] = None
  11. ) -> list[SearchResult]:
  12. """Search using serper.dev's API and return the results as a list of SearchResult objects.
  13. Args:
  14. api_key (str): A serper.dev API key
  15. query (str): The query to search for
  16. """
  17. url = "https://google.serper.dev/search"
  18. payload = json.dumps({"q": query})
  19. headers = {"X-API-KEY": api_key, "Content-Type": "application/json"}
  20. response = requests.request("POST", url, headers=headers, data=payload)
  21. response.raise_for_status()
  22. json_response = response.json()
  23. results = sorted(
  24. json_response.get("organic", []), key=lambda x: x.get("position", 0)
  25. )
  26. if filter_list:
  27. results = get_filtered_results(results, filter_list)
  28. return [
  29. SearchResult(
  30. link=result["link"],
  31. title=result.get("title"),
  32. snippet=result.get("description"),
  33. )
  34. for result in results[:count]
  35. ]