Coverage for open_webui/retrieval/web/external.py: 30%

1import logging 

2from typing import List, Optional 

3 

4import requests 

5from fastapi import Request 

6from open_webui.env import FORWARD_SESSION_INFO_HEADER_CHAT_ID 

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

8from open_webui.utils.headers import include_user_info_headers 

9 

10log = logging.getLogger(__name__) 

11 

12 

13def search_external( 

14 request: Request, 

15 external_url: str, 

16 external_api_key: str, 

17 query: str, 

18 count: int, 

19 filter_list: Optional[List[str]] = None, 

20 user=None, 

21) -> List[SearchResult]: 

22 try: 

23 headers = { 

24 # LICENSE covers this Open WebUI user-agent identifier. 

25 # Do not alter, remove, obscure, or replace it except as LICENSE permits: 

26 # https://docs.openwebui.com/license. 

27 'User-Agent': 'Open WebUI (https://github.com/open-webui/open-webui) RAG Bot', 

28 'Authorization': f'Bearer {external_api_key}', 

29 } 

30 headers = include_user_info_headers(headers, user) 

31 

32 chat_id = getattr(request.state, 'chat_id', None) 

33 if chat_id: 

34 headers[FORWARD_SESSION_INFO_HEADER_CHAT_ID] = str(chat_id) 

35 

36 response = requests.post( 

37 external_url, 

38 headers=headers, 

39 json={ 

40 'query': query, 

41 'count': count, 

42 }, 

43 ) 

44 response.raise_for_status() 

45 results = response.json() 

46 if filter_list: 

47 results = get_filtered_results(results, filter_list) 

48 results = [ 

49 SearchResult( 

50 link=result.get('link'), 

51 title=result.get('title'), 

52 snippet=result.get('snippet'), 

53 ) 

54 for result in results[:count] 

55 ] 

56 log.info('External search results: %s', results) 

57 return results 

58 except Exception as e: 

59 log.error(f'Error in External search: {e}') 

60 return []