Coverage for open_webui/retrieval/web/exa.py: 32%

1import logging 

2 

3import requests 

4from open_webui.retrieval.web.main import SearchResult 

5 

6log = logging.getLogger(__name__) 

7 

8EXA_API_BASE = 'https://api.exa.ai' 

9 

10 

11def search_exa( 

12 api_key: str, 

13 query: str, 

14 count: int, 

15 filter_list: list[str] | None = None, 

16 max_content_length: int | None = None, 

17) -> list[SearchResult]: 

18 """Search using Exa Search API and return the results as a list of SearchResult objects. 

19 

20 Args: 

21 api_key (str): A Exa Search API key 

22 query (str): The query to search for 

23 count (int): Number of results to return 

24 filter_list (list[str] | None): List of domains to filter results by 

25 max_content_length (int | None): Maximum characters per result; None leaves text unlimited. 

26 """ 

27 log.info('Searching with Exa for query: %s', query) 

28 

29 headers = {'Authorization': f'Bearer {api_key}', 'Content-Type': 'application/json'} 

30 

31 payload = { 

32 'query': query, 

33 'numResults': count or 5, 

34 'includeDomains': filter_list, 

35 'contents': {'text': {'maxCharacters': max_content_length} if max_content_length is not None else True}, 

36 'type': 'auto', # Use the auto search type (keyword or neural) 

37 } 

38 

39 try: 

40 response = requests.post(f'{EXA_API_BASE}/search', headers=headers, json=payload) 

41 response.raise_for_status() 

42 data = response.json() 

43 

44 results = data['results'] 

45 log.info('Found %s results', len(results)) 

46 return [ 

47 SearchResult( 

48 link=result['url'], 

49 title=result['title'], 

50 snippet=(result.get('text') or '')[:max_content_length], 

51 ) 

52 for result in results 

53 ] 

54 except Exception as e: 

55 log.error(f'Error searching Exa: {e}') 

56 return []