Spaces:
Running
Running
import logging | |
from typing import Optional | |
from urllib.parse import urlencode | |
import requests | |
from open_webui.retrieval.web.main import SearchResult, get_filtered_results | |
from open_webui.env import SRC_LOG_LEVELS | |
log = logging.getLogger(__name__) | |
log.setLevel(SRC_LOG_LEVELS["RAG"]) | |
def search_serply( | |
api_key: str, | |
query: str, | |
count: int, | |
hl: str = "us", | |
limit: int = 10, | |
device_type: str = "desktop", | |
proxy_location: str = "US", | |
filter_list: Optional[list[str]] = None, | |
) -> list[SearchResult]: | |
"""Search using serper.dev's API and return the results as a list of SearchResult objects. | |
Args: | |
api_key (str): A serply.io API key | |
query (str): The query to search for | |
hl (str): Host Language code to display results in (reference https://developers.google.com/custom-search/docs/xml_results?hl=en#wsInterfaceLanguages) | |
limit (int): The maximum number of results to return [10-100, defaults to 10] | |
""" | |
log.info("Searching with Serply") | |
url = "https://api.serply.io/v1/search/" | |
query_payload = { | |
"q": query, | |
"language": "en", | |
"num": limit, | |
"gl": proxy_location.upper(), | |
"hl": hl.lower(), | |
} | |
url = f"{url}{urlencode(query_payload)}" | |
headers = { | |
"X-API-KEY": api_key, | |
"X-User-Agent": device_type, | |
"User-Agent": "open-webui", | |
"X-Proxy-Location": proxy_location, | |
} | |
response = requests.request("GET", url, headers=headers) | |
response.raise_for_status() | |
json_response = response.json() | |
log.info(f"results from serply search: {json_response}") | |
results = sorted( | |
json_response.get("results", []), key=lambda x: x.get("realPosition", 0) | |
) | |
if filter_list: | |
results = get_filtered_results(results, filter_list) | |
return [ | |
SearchResult( | |
link=result["link"], | |
title=result.get("title"), | |
snippet=result.get("description"), | |
) | |
for result in results[:count] | |
] | |