Coverage for open_webui/retrieval/web/linkup.py: 23%

1import logging 

2from typing import Optional 

3 

4import requests 

5 

6from open_webui.retrieval.web.main import SearchResult, get_filtered_results 

7 

8log = logging.getLogger(__name__) 

9 

10DEFAULT_LINKUP_PARAMS = { 

11 'url': 'https://api.linkup.so/v1/search', 

12 'depth': 'standard', 

13 'outputType': 'sourcedAnswer', 

14} 

15 

16 

17def search_linkup( 

18 api_key: str, 

19 query: str, 

20 count: int, 

21 filter_list: Optional[list[str]] = None, 

22 params: Optional[dict] = None, 

23) -> list[SearchResult]: 

24 """Search using the Linkup Search API. 

25 

26 ``params`` is forwarded almost verbatim as the JSON body; only ``q`` 

27 and ``maxResults`` are injected automatically. The special key 

28 ``url`` (default ``https://api.linkup.so/v1/search``) is popped and 

29 used as the endpoint. 

30 """ 

31 if hasattr(api_key, '__str__'): 

32 api_key = str(api_key) 

33 

34 merged = {**DEFAULT_LINKUP_PARAMS, **(params or {})} 

35 api_url = str(merged.pop('url', DEFAULT_LINKUP_PARAMS['url'])) 

36 

37 payload = {**merged, 'q': query, 'maxResults': count} 

38 

39 try: 

40 response = requests.post( 

41 api_url, 

42 headers={ 

43 'Authorization': f'Bearer {api_key}', 

44 'Content-Type': 'application/json', 

45 }, 

46 json=payload, 

47 timeout=30, 

48 ) 

49 response.raise_for_status() 

50 json_response = response.json() 

51 

52 output_type = merged.get('outputType', 'sourcedAnswer') 

53 search_results = ( 

54 json_response.get('sources', []) if output_type == 'sourcedAnswer' else json_response.get('results', []) 

55 ) 

56 

57 if filter_list: 

58 search_results = get_filtered_results(search_results, filter_list) 

59 

60 return [ 

61 SearchResult( 

62 link=r.get('url', ''), 

63 title=r.get('name') or r.get('title'), 

64 snippet=r.get('content') or r.get('text') or r.get('snippet'), 

65 ) 

66 for r in search_results 

67 ][:count] 

68 

69 except requests.exceptions.RequestException as e: 

70 log.error(f'Linkup API request failed: {e}') 

71 raise Exception(f'Linkup search failed: {str(e)}') 

72 except Exception as e: 

73 log.error(f'Error searching Linkup: {e}') 

74 raise Exception(f'Linkup search error: {str(e)}')