Coverage for open_webui/retrieval/loaders/external_web.py: 28%

1import logging 

2from typing import Iterator, List, Union 

3 

4import requests 

5from langchain_core.document_loaders import BaseLoader 

6from langchain_core.documents import Document 

7 

8log = logging.getLogger(__name__) 

9 

10 

11class ExternalWebLoader(BaseLoader): 

12 def __init__( 

13 self, 

14 web_paths: Union[str, List[str]], 

15 external_url: str, 

16 external_api_key: str, 

17 continue_on_failure: bool = True, 

18 **kwargs, 

19 ) -> None: 

20 self.external_url = external_url 

21 self.external_api_key = external_api_key 

22 self.urls = web_paths if isinstance(web_paths, list) else [web_paths] 

23 self.continue_on_failure = continue_on_failure 

24 

25 def lazy_load(self) -> Iterator[Document]: 

26 batch_size = 20 

27 for i in range(0, len(self.urls), batch_size): 

28 urls = self.urls[i : i + batch_size] 

29 try: 

30 response = requests.post( 

31 self.external_url, 

32 headers={ 

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

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

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

36 'User-Agent': 'Open WebUI (https://github.com/open-webui/open-webui) External Web Loader', 

37 'Authorization': f'Bearer {self.external_api_key}', 

38 }, 

39 json={ 

40 'urls': urls, 

41 }, 

42 ) 

43 response.raise_for_status() 

44 results = response.json() 

45 for result in results: 

46 yield Document( 

47 page_content=result.get('page_content', ''), 

48 metadata=result.get('metadata', {}), 

49 ) 

50 except Exception as e: 

51 if self.continue_on_failure: 

52 log.error(f'Error extracting content from batch {urls}: {e}') 

53 else: 

54 raise e