Files
Classic298 52cfb02c72 perf: build debug log messages lazily so disabled debug logs cost nothing (#27834)
GLOBAL_LOG_LEVEL defaults to INFO, so every log.debug(...) in the backend is discarded, but the message is built first: 187 call sites interpolate their payload into an f-string before the logging call runs, so the work happens on every request and the result is thrown away. The worst one sits in process_chat_payload and stringifies the whole request body, full conversation history included, once per chat completion.

That one line with DEBUG disabled, CPython 3.12:

| conversation | payload | before   | after   |
| ------------ | ------- | -------- | ------- |
| 4 messages   | 1.2 kB  | 3.4 us   | 0.07 us |
| 20 messages  | 17 kB   | 24.8 us  | 0.07 us |
| 60 messages  | 123 kB  | 216.6 us | 0.07 us |

The lazy form log.debug('form_data: %s', form_data) hands the payload to record.getMessage(), which the InterceptHandler only reaches once a record has passed the level check. With DEBUG enabled the emitted lines are byte-identical, f'{x=}' sites included: those map to %r. MistralLoader._debug_log callers get the same treatment, since that wrapper already forwards *args.
2026-07-31 19:09:01 -05:00

89 lines
2.8 KiB
Python

import logging
from typing import Optional
import requests
from open_webui.retrieval.web.main import SearchResult, get_filtered_results
from requests.auth import HTTPDigestAuth
log = logging.getLogger(__name__)
def search_yacy(
query_url: str,
username: Optional[str],
password: Optional[str],
query: str,
count: int,
filter_list: Optional[list[str]] = None,
) -> list[SearchResult]:
"""
Search a Yacy instance for a given query and return the results as a list of SearchResult objects.
The function accepts username and password for authenticating to Yacy.
Args:
query_url (str): The base URL of the Yacy server.
username (str): Optional YaCy username.
password (str): Optional YaCy password.
query (str): The search term or question to find in the Yacy database.
count (int): The maximum number of results to retrieve from the search.
Returns:
list[SearchResult]: A list of SearchResults sorted by relevance score in descending order.
Raise:
requests.exceptions.RequestException: If a request error occurs during the search process.
"""
# Use authentication if either username or password is set
yacy_auth = None
if username or password:
yacy_auth = HTTPDigestAuth(username, password)
params = {
'query': query,
'contentdom': 'text',
'resource': 'global',
'maximumRecords': count,
'nav': 'none',
}
# Check if provided a json API URL
if not query_url.endswith('yacysearch.json'):
# Strip all query parameters from the URL
query_url = query_url.rstrip('/') + '/yacysearch.json'
log.debug('searching %s', query_url)
response = requests.get(
query_url,
auth=yacy_auth,
headers={
# LICENSE covers this Open WebUI user-agent identifier.
# Do not alter, remove, obscure, or replace it except as LICENSE permits:
# https://docs.openwebui.com/license.
'User-Agent': 'Open WebUI (https://github.com/open-webui/open-webui) RAG Bot',
'Accept': 'text/html',
'Accept-Encoding': 'gzip, deflate',
'Accept-Language': 'en-US,en;q=0.5',
'Connection': 'keep-alive',
},
params=params,
)
response.raise_for_status() # Raise an exception for HTTP errors.
json_response = response.json()
results = json_response.get('channels', [{}])[0].get('items', [])
sorted_results = sorted(results, key=lambda x: x.get('ranking', 0), reverse=True)
if filter_list:
sorted_results = get_filtered_results(sorted_results, filter_list)
return [
SearchResult(
link=result['link'],
title=result.get('title'),
snippet=result.get('description'),
)
for result in sorted_results[:count]
]