kagi.py 1.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748
  1. import logging
  2. from typing import Optional
  3. import requests
  4. from open_webui.retrieval.web.main import SearchResult, get_filtered_results
  5. from open_webui.env import SRC_LOG_LEVELS
  6. log = logging.getLogger(__name__)
  7. log.setLevel(SRC_LOG_LEVELS["RAG"])
  8. def search_kagi(
  9. api_key: str, query: str, count: int, filter_list: Optional[list[str]] = None
  10. ) -> list[SearchResult]:
  11. """Search using Kagi's Search API and return the results as a list of SearchResult objects.
  12. The Search API will inherit the settings in your account, including results personalization and snippet length.
  13. Args:
  14. api_key (str): A Kagi Search API key
  15. query (str): The query to search for
  16. count (int): The number of results to return
  17. """
  18. url = "https://kagi.com/api/v0/search"
  19. headers = {
  20. "Authorization": f"Bot {api_key}",
  21. }
  22. params = {"q": query, "limit": count}
  23. response = requests.get(url, headers=headers, params=params)
  24. response.raise_for_status()
  25. json_response = response.json()
  26. search_results = json_response.get("data", [])
  27. results = [
  28. SearchResult(
  29. link=result["url"], title=result["title"], snippet=result.get("snippet")
  30. )
  31. for result in search_results
  32. if result["t"] == 0
  33. ]
  34. print(results)
  35. if filter_list:
  36. results = get_filtered_results(results, filter_list)
  37. return results