serper.py 1.1 KB

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