diff --git a/backend/open_webui/retrieval/loaders/local.py b/backend/open_webui/retrieval/loaders/local.py new file mode 100644 index 0000000000..fad7a930be --- /dev/null +++ b/backend/open_webui/retrieval/loaders/local.py @@ -0,0 +1,121 @@ +from importlib import import_module +from pathlib import Path + +from bs4 import BeautifulSoup +from langchain_core.documents import Document + + +class TextLoader: + def __init__(self, file_path, encoding=None): + self.file_path = str(file_path) + self.encoding = encoding + + def load(self) -> list[Document]: + try: + text = Path(self.file_path).read_text(encoding=self.encoding) + except Exception as e: + raise RuntimeError(f'Error loading {self.file_path}') from e + return [ + Document( + page_content=text, + metadata={'source': self.file_path}, + ) + ] + + +class HTMLLoader(TextLoader): + def load(self) -> list[Document]: + with open(self.file_path, encoding=self.encoding) as file: + soup = BeautifulSoup(file, 'lxml') + return [ + Document( + page_content=soup.get_text(), + metadata={'source': self.file_path, 'title': str(soup.title.string) if soup.title else ''}, + ) + ] + + +class DocxLoader(TextLoader): + def load(self) -> list[Document]: + import docx2txt + + return [ + Document( + page_content=docx2txt.process(Path(self.file_path).expanduser()), + metadata={'source': self.file_path}, + ) + ] + + +class UnstructuredLoader: + def __init__(self, file_path, file_format, mode='single', **kwargs): + # Match the optional-package check; format dependencies are loaded when parsing. + import_module('unstructured') + self.file_path = file_path + self.file_format = file_format + self.mode = mode + self.kwargs = kwargs + + def load(self) -> list[Document]: + file_format = self.file_format + if file_format in ('doc', 'ppt', 'pptx'): + from unstructured.file_utils.filetype import detect_filetype + + legacy_format = 'doc' if file_format == 'doc' else 'ppt' + try: + import_module('magic') + except ImportError: + is_legacy = Path(self.file_path).suffix == f'.{legacy_format}' + else: + is_legacy = detect_filetype(self.file_path).name.lower() == legacy_format + file_format = legacy_format if is_legacy else legacy_format + 'x' + elif file_format == 'msg': + from unstructured.file_utils.filetype import detect_filetype + + detected = detect_filetype(self.file_path).name + if detected not in ('EML', 'MSG'): + raise ValueError(f'Unsupported email file type: {detected}') + file_format = 'email' if detected == 'EML' else 'msg' + + module = import_module(f'unstructured.partition.{file_format}') + elements = getattr(module, f'partition_{file_format}')(filename=self.file_path, **self.kwargs) + metadata = {'source': str(self.file_path)} + if self.mode == 'elements': + return [ + Document( + page_content=str(element), + metadata={ + **metadata, + **element.metadata.to_dict(), + 'category': element.category, + 'element_id': element.id, + }, + ) + for element in elements + ] + return [Document(page_content='\n\n'.join(map(str, elements)), metadata=metadata)] + + +class DocumentIntelligenceLoader: + def __init__(self, file_path, api_endpoint, api_key=None, azure_credential=None, api_model='prebuilt-layout'): + if (api_key is None) == (azure_credential is None): + raise ValueError('Provide exactly one of api_key or azure_credential.') + self.file_path = file_path + self.api_endpoint = api_endpoint + self.api_key = api_key + self.azure_credential = azure_credential + self.api_model = api_model + + def load(self) -> list[Document]: + from azure.ai.documentintelligence import DocumentIntelligenceClient + from azure.core.credentials import AzureKeyCredential + + credential = self.azure_credential if self.azure_credential is not None else AzureKeyCredential(self.api_key) + with DocumentIntelligenceClient(self.api_endpoint, credential) as client, open(self.file_path, 'rb') as file: + result = client.begin_analyze_document( + self.api_model, + body=file, + content_type='application/octet-stream', + output_content_format='markdown', + ).result() + return [Document(page_content=result.content, metadata=result.as_dict())] diff --git a/backend/open_webui/retrieval/loaders/main.py b/backend/open_webui/retrieval/loaders/main.py index 7a34edb2cb..9188c723c1 100644 --- a/backend/open_webui/retrieval/loaders/main.py +++ b/backend/open_webui/retrieval/loaders/main.py @@ -9,14 +9,6 @@ import ftfy import requests from fastapi import HTTPException from azure.identity import DefaultAzureCredential -from langchain_community.document_loaders import ( - AzureAIDocumentIntelligenceLoader, - BSHTMLLoader, - CSVLoader, - Docx2txtLoader, - PyPDFLoader, - TextLoader, -) from langchain_core.documents import Document from open_webui.env import ( AIOHTTP_CLIENT_SESSION_SSL, @@ -27,9 +19,17 @@ from open_webui.env import ( ) from open_webui.retrieval.loaders.datalab_marker import DatalabMarkerLoader from open_webui.retrieval.loaders.external_document import ExternalDocumentLoader +from open_webui.retrieval.loaders.local import ( + DocumentIntelligenceLoader, + DocxLoader, + HTMLLoader, + TextLoader, + UnstructuredLoader, +) from open_webui.retrieval.loaders.mineru import MinerULoader from open_webui.retrieval.loaders.mistral import MistralLoader from open_webui.retrieval.loaders.paddleocr_vl import PADDLEOCR_VL_SUPPORTED_EXTENSIONS, PaddleOCRVLLoader +from open_webui.retrieval.loaders.pdf import PDFLoader from open_webui.utils.headers import get_user_groups_for_custom_headers from open_webui.utils.json_codec import JSONCodec @@ -163,7 +163,21 @@ class CSVLoaderWithSummary: self.encoding = encoding def load(self) -> list[Document]: - docs = CSVLoader(self.file_path, encoding=self.encoding).load() + docs = [] + try: + with open(self.file_path, newline='', encoding=self.encoding) as file: + for index, row in enumerate(csv.DictReader(file)): + fields = [] + for key, value in row.items(): + if isinstance(value, str): + value = value.strip() + elif isinstance(value, list): + value = ','.join(v.strip() for v in value) + fields.append(f'{key.strip() if key is not None else key}: {value}') + content = '\n'.join(fields) + docs.append(Document(page_content=content, metadata={'source': self.file_path, 'row': index})) + except Exception as e: + raise RuntimeError(f'Error loading {self.file_path}') from e if os.getenv('ENABLE_RAG_CSV_SUMMARY', 'False').lower() == 'true': summary = get_csv_summary(self.filename, self.file_path, self.encoding) if summary: @@ -620,14 +634,14 @@ class Loader: ) ): if self.kwargs.get('DOCUMENT_INTELLIGENCE_KEY') != '': - loader = AzureAIDocumentIntelligenceLoader( + loader = DocumentIntelligenceLoader( file_path=file_path, api_endpoint=self.kwargs.get('DOCUMENT_INTELLIGENCE_ENDPOINT'), api_key=self.kwargs.get('DOCUMENT_INTELLIGENCE_KEY'), api_model=self.kwargs.get('DOCUMENT_INTELLIGENCE_MODEL'), ) else: - loader = AzureAIDocumentIntelligenceLoader( + loader = DocumentIntelligenceLoader( file_path=file_path, api_endpoint=self.kwargs.get('DOCUMENT_INTELLIGENCE_ENDPOINT'), azure_credential=DefaultAzureCredential(), @@ -677,7 +691,7 @@ class Loader: if file_ext == 'csv': return CSVLoaderWithSummary(file_path, filename, self._detect_text_encoding(file_path)) if file_ext in ['htm', 'html']: - return BSHTMLLoader(file_path, open_encoding=self._detect_text_encoding(file_path)) + return HTMLLoader(file_path, encoding=self._detect_text_encoding(file_path)) if file_ext in ['txt', 'md', 'markdown', 'rst', 'xml'] or self._is_text_file( file_ext, file_content_type ): @@ -687,7 +701,7 @@ class Loader: 'This file type requires an external document extractor in slim. Configure one that supports it.', ) if file_ext == 'pdf': - loader = PyPDFLoader( + loader = PDFLoader( file_path, extract_images=self.kwargs.get('PDF_EXTRACT_IMAGES'), mode=self.kwargs.get('PDF_LOADER_MODE', 'page'), @@ -700,9 +714,7 @@ class Loader: ) elif file_ext == 'rst': try: - from langchain_community.document_loaders import UnstructuredRSTLoader - - loader = UnstructuredRSTLoader(file_path, mode='elements') + loader = UnstructuredLoader(file_path, 'rst', mode='elements') except ImportError: log.warning( "The 'unstructured' package is not installed. " @@ -712,9 +724,7 @@ class Loader: loader = TextLoader(file_path, encoding=self._detect_text_encoding(file_path)) elif file_ext == 'xml': try: - from langchain_community.document_loaders import UnstructuredXMLLoader - - loader = UnstructuredXMLLoader(file_path) + loader = UnstructuredLoader(file_path, 'xml') except ImportError: log.warning( "The 'unstructured' package is not installed. " @@ -723,14 +733,12 @@ class Loader: ) loader = TextLoader(file_path, encoding=self._detect_text_encoding(file_path)) elif file_ext in ['htm', 'html']: - loader = BSHTMLLoader(file_path, open_encoding='unicode_escape') + loader = HTMLLoader(file_path, encoding='unicode_escape') elif file_ext == 'md': loader = TextLoader(file_path, encoding=self._detect_text_encoding(file_path)) elif file_content_type == 'application/epub+zip': try: - from langchain_community.document_loaders import UnstructuredEPubLoader - - loader = UnstructuredEPubLoader(file_path) + loader = UnstructuredLoader(file_path, 'epub') except ImportError: raise ValueError( "Processing .epub files requires the 'unstructured' package. " @@ -740,12 +748,10 @@ class Loader: file_content_type == 'application/vnd.openxmlformats-officedocument.wordprocessingml.document' or file_ext == 'docx' ): - loader = Docx2txtLoader(file_path) + loader = DocxLoader(file_path) elif file_ext == 'doc' or file_content_type == 'application/msword': try: - from langchain_community.document_loaders import UnstructuredWordDocumentLoader - - loader = UnstructuredWordDocumentLoader(file_path) + loader = UnstructuredLoader(file_path, 'doc') except ImportError: raise ValueError( "Processing .doc files requires the 'unstructured' package. " @@ -756,9 +762,7 @@ class Loader: 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', ] or file_ext in ['xls', 'xlsx']: try: - from langchain_community.document_loaders import UnstructuredExcelLoader - - loader = UnstructuredExcelLoader(file_path) + loader = UnstructuredLoader(file_path, 'xlsx') except ImportError: log.warning( "The 'unstructured' package is not installed. " @@ -771,9 +775,7 @@ class Loader: 'application/vnd.openxmlformats-officedocument.presentationml.presentation', ] or file_ext in ['ppt', 'pptx']: try: - from langchain_community.document_loaders import UnstructuredPowerPointLoader - - loader = UnstructuredPowerPointLoader(file_path) + loader = UnstructuredLoader(file_path, 'ppt' if file_ext == 'ppt' else 'pptx') except ImportError: log.warning( "The 'unstructured' package is not installed. " @@ -783,12 +785,8 @@ class Loader: loader = PptxLoader(file_path) elif file_ext == 'msg': try: - from langchain_community.document_loaders import ( - UnstructuredEmailLoader, - ) - # unstructured parses .msg via python-oxmsg; avoids extract_msg's beautifulsoup4<4.14 conflict - loader = UnstructuredEmailLoader(file_path, process_attachments=False) + loader = UnstructuredLoader(file_path, 'msg', process_attachments=False) except ImportError: raise ValueError( "Processing .msg files requires the 'unstructured' package. " @@ -796,9 +794,7 @@ class Loader: ) elif file_ext == 'odt': try: - from langchain_community.document_loaders import UnstructuredODTLoader - - loader = UnstructuredODTLoader(file_path) + loader = UnstructuredLoader(file_path, 'odt') except ImportError: raise ValueError( "Processing .odt files requires the 'unstructured' package. " diff --git a/backend/open_webui/retrieval/loaders/pdf.py b/backend/open_webui/retrieval/loaders/pdf.py new file mode 100644 index 0000000000..38ecd3b1a9 --- /dev/null +++ b/backend/open_webui/retrieval/loaders/pdf.py @@ -0,0 +1,101 @@ +import datetime as dt +import io +import logging +from pathlib import Path + +from langchain_core.document_loaders import BaseLoader +from langchain_core.documents import Document + +log = logging.getLogger(__name__) + + +class PDFLoader(BaseLoader): + def __init__(self, file_path, *, extract_images=False, mode='page'): + if mode not in ('single', 'page'): + raise ValueError("PDF mode must be 'single' or 'page'") + self.file_path = str(Path(file_path).expanduser()) + self.extract_images = extract_images + self.mode = mode + self.ocr = None + + def lazy_load(self): + from pypdf import PdfReader + + with open(self.file_path, 'rb') as file: + reader = PdfReader(file) + metadata = {'producer': 'PyPDF', 'creator': 'PyPDF', 'creationdate': ''} + for key, value in (reader.metadata or {}).items(): + key = key.removeprefix('/').lower() + value = value if type(value) in (str, int) else str(value) + if key in ('creationdate', 'moddate') and isinstance(value, str): + try: + value = dt.datetime.strptime(value.replace("'", ''), 'D:%Y%m%d%H%M%S%z').isoformat() + except ValueError: + pass + metadata[key] = ( + value.strip() + if isinstance(value, str) and key not in ('creationdate', 'moddate', 'page_count', 'file_path') + else value + ) + metadata.update(source=self.file_path, total_pages=len(reader.pages)) + labels = reader.page_labels if self.mode == 'page' else None + texts = [] + for index, page in enumerate(reader.pages): + text = page.extract_text() + if self.extract_images: + image_text = self._extract_images(page) + if image_text: + text = self._merge_image_text(text, image_text) + text = text.strip() + if self.mode == 'page': + yield Document(page_content=text, metadata={**metadata, 'page': index, 'page_label': labels[index]}) + else: + texts.append(text) + if self.mode == 'single': + yield Document(page_content='\n\f'.join(texts), metadata=metadata) + + @staticmethod + def _merge_image_text(text, image_text): + # Insert before the final paragraphs/footer where possible, matching existing chunks. + position, separator = len(text), '\n\n' + for _ in range(2): + for delimiter in ('\n\n\n', '\n\n'): + found = text.rfind(delimiter, 0, position) + if found >= 0: + position, separator = found, delimiter + break + else: + break + return text[:position] + separator + image_text + text[position:] + + def _extract_images(self, page): + import numpy as np + from PIL import Image, UnidentifiedImageError + + if '/Resources' not in page or '/XObject' not in page['/Resources']: + return '' + texts = [] + xobjects = page['/Resources']['/XObject'] + for name in xobjects: + try: + stream = xobjects[name] + if stream.get('/Subtype') != '/Image': + continue + try: + # Encoded images, including CMYK JPEGs, can go straight to Pillow. + image = Image.open(io.BytesIO(stream.get_data())) + except UnidentifiedImageError: + image = stream.decode_as_image() + pixels = np.array(image.convert('RGB')) + except Exception as e: + log.warning('Skipping unreadable PDF image %s: %s', name, e) + continue + + if self.ocr is None: + from rapidocr import RapidOCR + + self.ocr = RapidOCR() + result = self.ocr(pixels) + if result and result.txts: + texts.append('\n'.join(result.txts).strip()) + return '\n\n' + '\n'.join(filter(None, texts)) + '\n\n' if any(texts) else '' diff --git a/backend/open_webui/retrieval/utils.py b/backend/open_webui/retrieval/utils.py index d064e8575e..1a6c927c5f 100644 --- a/backend/open_webui/retrieval/utils.py +++ b/backend/open_webui/retrieval/utils.py @@ -17,7 +17,6 @@ from langchain_classic.retrievers import ( ContextualCompressionRetriever, EnsembleRetriever, ) -from langchain_community.retrievers import BM25Retriever from langchain_core.documents import Document from open_webui.config import ( RAG_EMBEDDING_CONTENT_PREFIX, @@ -65,6 +64,15 @@ from langchain_core.callbacks import CallbackManagerForRetrieverRun from langchain_core.retrievers import BaseRetriever +class BM25Retriever(BaseRetriever): + docs: list[Document] + vectorizer: Any + k: int + + def _get_relevant_documents(self, query: str, *, run_manager: CallbackManagerForRetrieverRun) -> list[Document]: + return self.vectorizer.get_top_n(query.split(), self.docs, n=self.k) + + def is_youtube_url(url: str) -> bool: youtube_regex = r'^(https?://)?(www\.)?(youtube\.com|youtu\.be)/.+$' return re.match(youtube_regex, url) is not None @@ -544,11 +552,13 @@ async def query_doc_with_hybrid_search( bm25_texts = get_enriched_texts(collection_result) if enable_enriched_texts else original_texts - bm25_retriever = BM25Retriever.from_texts( - texts=bm25_texts, - metadatas=bm25_metadatas, + from rank_bm25 import BM25Okapi + + bm25_retriever = BM25Retriever( + docs=[Document(page_content=text, metadata=meta) for text, meta in zip(bm25_texts, bm25_metadatas)], + vectorizer=BM25Okapi([text.split() for text in bm25_texts]), + k=k, ) - bm25_retriever.k = k vector_search_retriever = VectorSearchRetriever( collection_name=collection_name, diff --git a/backend/open_webui/retrieval/web/utils.py b/backend/open_webui/retrieval/web/utils.py index c9509edc87..f16e7ff626 100644 --- a/backend/open_webui/retrieval/web/utils.py +++ b/backend/open_webui/retrieval/web/utils.py @@ -8,6 +8,7 @@ import time import urllib.parse import urllib.request from datetime import datetime, timedelta +from importlib import import_module from typing import ( Any, AsyncIterator, @@ -31,8 +32,7 @@ import validators from requests.adapters import HTTPAdapter from fastapi import HTTPException from fastapi.concurrency import run_in_threadpool -from langchain_community.document_loaders import PlaywrightURLLoader, WebBaseLoader -from langchain_community.document_loaders.base import BaseLoader +from langchain_core.document_loaders import BaseLoader from langchain_core.documents import Document from open_webui.config import ( ENABLE_LOCAL_WEB_FETCH, @@ -648,7 +648,7 @@ class SafeMicrosoftWebIQLoader(BaseLoader, RateLimitMixin, URLProcessingMixin): raise e -class SafePlaywrightURLLoader(PlaywrightURLLoader, RateLimitMixin, URLProcessingMixin): +class SafePlaywrightURLLoader(BaseLoader, RateLimitMixin, URLProcessingMixin): """Load HTML pages safely with Playwright, supporting SSL verification, rate limiting, and remote browser connection. Attributes: @@ -684,6 +684,9 @@ class SafePlaywrightURLLoader(PlaywrightURLLoader, RateLimitMixin, URLProcessing 503, 'Playwright is unavailable in slim. Use basic HTTP fetching or an external web loader.' ) + for package in ('playwright', 'unstructured'): + import_module(package) + proxy_server = proxy.get('server') if proxy else None if trust_env and not proxy_server: env_proxies = urllib.request.getproxies() @@ -694,14 +697,11 @@ class SafePlaywrightURLLoader(PlaywrightURLLoader, RateLimitMixin, URLProcessing else: proxy = {'server': env_proxy_server} - # We'll set headless to False if using playwright_ws_url since it's handled by the remote browser - super().__init__( - urls=web_paths, - continue_on_failure=continue_on_failure, - headless=headless if playwright_ws_url is None else False, - remove_selectors=remove_selectors, - proxy=proxy, - ) + self.urls = web_paths + self.continue_on_failure = continue_on_failure + self.headless = headless if playwright_ws_url is None else False + self.remove_selectors = remove_selectors or [] + self.proxy = proxy self.verify_ssl = verify_ssl self.requests_per_second = requests_per_second self.last_request_time = None @@ -709,6 +709,12 @@ class SafePlaywrightURLLoader(PlaywrightURLLoader, RateLimitMixin, URLProcessing self.trust_env = trust_env self.playwright_timeout = playwright_timeout + @staticmethod + def _extract_html(html): + from unstructured.partition.html import partition_html + + return '\n\n'.join(str(element) for element in partition_html(text=html)) + def _request_timeout(self) -> float: # per-hop budget, since page.goto's timeout cannot reach into our own fetch and 0 disables # it. aiohttp treats it as a total where requests only caps each read, so sync runs looser. @@ -867,7 +873,11 @@ class SafePlaywrightURLLoader(PlaywrightURLLoader, RateLimitMixin, URLProcessing if response is None: raise ValueError(f'page.goto() returned None for url {url}') - text = self.evaluator.evaluate(page, browser, response) + for selector in self.remove_selectors: + for element in page.locator(selector).all(): + if element.is_visible(): + element.evaluate('element => element.remove()') + text = self._extract_html(page.content()) page.unroute_all(behavior='ignoreErrors') metadata = {'source': url} yield Document(page_content=text, metadata=metadata) @@ -903,7 +913,11 @@ class SafePlaywrightURLLoader(PlaywrightURLLoader, RateLimitMixin, URLProcessing if response is None: raise ValueError(f'page.goto() returned None for url {url}') - text = await self.evaluator.evaluate_async(page, browser, response) + for selector in self.remove_selectors: + for element in await page.locator(selector).all(): + if await element.is_visible(): + await element.evaluate('element => element.remove()') + text = await asyncio.to_thread(self._extract_html, await page.content()) await page.unroute_all(behavior='ignoreErrors') metadata = {'source': url} yield Document(page_content=text, metadata=metadata) @@ -914,42 +928,58 @@ class SafePlaywrightURLLoader(PlaywrightURLLoader, RateLimitMixin, URLProcessing raise e -class SafeWebBaseLoader(WebBaseLoader): - """WebBaseLoader with enhanced error handling for URLs.""" +class SafeWebBaseLoader(BaseLoader): + """Fetch pages with connect-time address checks and bounded concurrency.""" - def __init__(self, trust_env: bool = False, *args, **kwargs): - """Initialize SafeWebBaseLoader - Args: - trust_env (bool, optional): set to True if using proxy to make web requests, for example - using http(s)_proxy environment variables. Defaults to False. - """ - # lxml parses scraped pages far faster than the html.parser default - kwargs.setdefault('default_parser', 'lxml') - super().__init__(*args, **kwargs) + def __init__( + self, + web_paths, + verify_ssl=True, + trust_env=False, + requests_per_second=2, + continue_on_failure=False, + requests_kwargs=None, + raise_for_status=False, + default_parser='lxml', + bs_kwargs=None, + bs_get_text_kwargs=None, + ): + self.web_paths = list(web_paths) self.trust_env = trust_env - - # Propagate USER_AGENT env var so that both the sync _scrape() and - # async _fetch() paths present a real UA instead of python-requests/2.x - # which gets blocked by Cloudflare, Wikipedia, and similar bot-detection. - # _fetch() forwards self.session.headers to the aiohttp session, so - # setting it here covers both code-paths. - if USER_AGENT: - self.session.headers['User-Agent'] = USER_AGENT - - # Prevent redirect-based SSRF on the synchronous _scrape() path. - # validate_url() is called once on the originally-submitted URL, but the - # parent WebBaseLoader's _scrape() invokes self.session.get(url, **self.requests_kwargs) - # which by default follows redirects. Without the override below, an attacker - # can submit a public URL that 302-redirects to an internal address (RFC1918, - # 127.0.0.1, 169.254.169.254, etc.) and the redirected target is fetched without - # re-validation. Matches the policy enforced on the async _fetch() path below. - self.requests_kwargs = { - **(self.requests_kwargs or {}), - 'allow_redirects': AIOHTTP_CLIENT_ALLOW_REDIRECTS, + self.requests_per_second = requests_per_second + self.continue_on_failure = continue_on_failure + self.requests_kwargs = {**(requests_kwargs or {}), 'allow_redirects': AIOHTTP_CLIENT_ALLOW_REDIRECTS} + self.raise_for_status = raise_for_status + self.default_parser = default_parser + self.bs_kwargs = bs_kwargs or {} + self.bs_get_text_kwargs = bs_get_text_kwargs or {} + # Preserve the synchronous loader's environment-proxy behavior. + self.session = get_ssrf_safe_requests_session() + self.session.verify = verify_ssl + self.session.headers = { + 'User-Agent': USER_AGENT or 'DefaultLangchainUserAgent', + 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', + 'Accept-Language': 'en-US,en;q=0.5', + 'Referer': 'https://www.google.com/', + 'DNT': '1', + 'Connection': 'keep-alive', + 'Upgrade-Insecure-Requests': '1', } - self.session.mount('http://', _SSRFSafeAdapter()) - self.session.mount('https://', _SSRFSafeAdapter()) + async def fetch_all(self, urls): + semaphore = asyncio.Semaphore(self.requests_per_second) + + async def fetch(url): + async with semaphore: + try: + return await self._fetch(url) + except Exception as e: + if not self.continue_on_failure: + raise + log.warning('Error fetching %s: %s', url, e) + return '' + + return await asyncio.gather(*(fetch(url) for url in urls)) async def _fetch(self, url: str, retries: int = 3, cooldown: int = 2, backoff: float = 1.5) -> str: connector = _SSRFSafeConnector() @@ -965,10 +995,10 @@ class SafeWebBaseLoader(WebBaseLoader): else: kwargs['ssl'] = AIOHTTP_CLIENT_SESSION_SSL - async with session.get( - url, - **(self.requests_kwargs | kwargs), - ) as response: + options = self.requests_kwargs | kwargs + if isinstance(options.get('timeout'), (int, float)): + options['timeout'] = aiohttp.ClientTimeout(total=options['timeout']) + async with session.get(url, **options) as response: if self.raise_for_status: response.raise_for_status() return await response.text() @@ -980,38 +1010,24 @@ class SafeWebBaseLoader(WebBaseLoader): await asyncio.sleep(cooldown * backoff**i) raise ValueError('retry count exceeded') - def _unpack_fetch_results(self, results: Any, urls: List[str], parser: Union[str, None] = None) -> List[Any]: - """Unpack fetch results into BeautifulSoup objects.""" - from bs4 import BeautifulSoup - - final_results = [] - for i, result in enumerate(results): - url = urls[i] - url_parser = parser - if url_parser is None: - url_parser = 'xml' if url.endswith('.xml') else self.default_parser - self._check_parser(url_parser) - final_results.append(BeautifulSoup(result, url_parser, **self.bs_kwargs)) - return final_results - def lazy_load(self) -> Iterator[Document]: """Lazy load text from the url(s) in web_path with error handling.""" for path in self.web_paths: try: - soup = self._scrape(path, bs_kwargs=self.bs_kwargs) - text = soup.get_text(**self.bs_get_text_kwargs) - - # Build metadata - metadata = extract_metadata(soup, path) - - yield Document(page_content=text, metadata=metadata) + with self.session.get(path, **self.requests_kwargs) as response: + if self.raise_for_status: + response.raise_for_status() + response.encoding = response.apparent_encoding + yield self._document_from_html(response.text, path) except Exception as e: - # Log the error and continue with the next URL log.exception(f'Error loading {path}: {e}') def _document_from_html(self, html: str, url: str) -> Document: """Build one Document.""" - soup = self._unpack_fetch_results([html], [url])[0] + from bs4 import BeautifulSoup + + parser = 'xml' if url.endswith('.xml') else self.default_parser + soup = BeautifulSoup(html, parser, **self.bs_kwargs) return Document( page_content=soup.get_text(**self.bs_get_text_kwargs), metadata=extract_metadata(soup, url), diff --git a/backend/requirements-slim.txt b/backend/requirements-slim.txt index 37f1213fa9..8171a6f9e6 100644 --- a/backend/requirements-slim.txt +++ b/backend/requirements-slim.txt @@ -51,7 +51,7 @@ mcp==1.27.2 openai==2.29.0 -langchain-community==0.4.2 +langchain-core==1.4.8 langchain-classic==1.0.7 langchain-text-splitters==1.1.2 diff --git a/backend/requirements.txt b/backend/requirements.txt index 9e7f798360..e929beafb6 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -47,7 +47,7 @@ mcp==1.27.2 openai==2.29.0 anthropic==0.86.0 -langchain-community==0.4.2 +langchain-core==1.4.8 langchain-classic==1.0.7 langchain-text-splitters==1.1.2 diff --git a/pyproject.toml b/pyproject.toml index 02f72633af..2fc362ae97 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -55,7 +55,7 @@ dependencies = [ "openai==2.29.0", "anthropic==0.86.0", - "langchain-community==0.4.2", + "langchain-core==1.4.8", "langchain-classic==1.0.7", "langchain-text-splitters==1.1.2", diff --git a/uv.lock b/uv.lock index dde37e4f73..701157a9eb 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 3 +revision = 2 requires-python = ">=3.11, <3.13.0" resolution-markers = [ "python_full_version >= '3.12' and sys_platform == 'win32'", @@ -1957,28 +1957,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f5/78/2d9980d028ff0523eea503a77c200e2ff252a3a75eb6e7842bcf5f9c979b/langchain_classic-1.0.7-py3-none-any.whl", hash = "sha256:d9d9be38f7aa534ed0259c2410432e34a1f80b1d491e686749bb55af56479be3", size = 1041386, upload-time = "2026-05-07T15:46:54.845Z" }, ] -[[package]] -name = "langchain-community" -version = "0.4.2" -source = { registry = "https://pypi.org/simple" } -dependencies = [ - { name = "aiohttp" }, - { name = "httpx-sse" }, - { name = "langchain-classic" }, - { name = "langchain-core" }, - { name = "langsmith" }, - { name = "numpy" }, - { name = "pydantic-settings" }, - { name = "pyyaml" }, - { name = "requests" }, - { name = "sqlalchemy" }, - { name = "tenacity" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/ea/0c/e3aca1f2b1c5b95f8b87cb2b6e81a6f20d538c07a128419dc01cef0617b6/langchain_community-0.4.2.tar.gz", hash = "sha256:a99308160d53d7e9b5965ee665e5173709914338210089fd5788ad724432c21e", size = 33268708, upload-time = "2026-05-22T19:42:59.374Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/8f/39/5d97e42a3e95dc2a6d71b2f902a3fae71786131e11d01bddb604accb0ebe/langchain_community-0.4.2-py3-none-any.whl", hash = "sha256:84dd8c5122532394d5b6849a5fc9995ef28e4f77227daeb09f24b3d942e9e466", size = 2364406, upload-time = "2026-05-22T19:42:57.103Z" }, -] - [[package]] name = "langchain-core" version = "1.4.8" @@ -2784,7 +2762,7 @@ dependencies = [ { name = "itsdangerous" }, { name = "joserfc" }, { name = "langchain-classic" }, - { name = "langchain-community" }, + { name = "langchain-core" }, { name = "langchain-text-splitters" }, { name = "ldap3" }, { name = "loguru" }, @@ -2910,7 +2888,7 @@ requires-dist = [ { name = "itsdangerous", specifier = "==2.2.0" }, { name = "joserfc", specifier = "==1.7.4" }, { name = "langchain-classic", specifier = "==1.0.7" }, - { name = "langchain-community", specifier = "==0.4.2" }, + { name = "langchain-core", specifier = "==1.4.8" }, { name = "langchain-text-splitters", specifier = "==1.1.2" }, { name = "ldap3", specifier = "==2.9.1" }, { name = "loguru", specifier = "==0.7.3" },