searchapi.py 1.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748
  1. import logging
  2. from typing import Optional
  3. from urllib.parse import urlencode
  4. import requests
  5. from open_webui.apps.retrieval.web.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_searchapi(
  10. api_key: str,
  11. engine: str,
  12. query: str,
  13. count: int,
  14. filter_list: Optional[list[str]] = None,
  15. ) -> list[SearchResult]:
  16. """Search using searchapi.io's API and return the results as a list of SearchResult objects.
  17. Args:
  18. api_key (str): A searchapi.io API key
  19. query (str): The query to search for
  20. """
  21. url = "https://www.searchapi.io/api/v1/search"
  22. engine = engine or "google"
  23. payload = {"engine": engine, "q": query, "api_key": api_key}
  24. url = f"{url}?{urlencode(payload)}"
  25. response = requests.request("GET", url)
  26. json_response = response.json()
  27. log.info(f"results from searchapi search: {json_response}")
  28. results = sorted(
  29. json_response.get("organic_results", []), key=lambda x: x.get("position", 0)
  30. )
  31. if filter_list:
  32. results = get_filtered_results(results, filter_list)
  33. return [
  34. SearchResult(
  35. link=result["link"], title=result["title"], snippet=result["snippet"]
  36. )
  37. for result in results[:count]
  38. ]