mirror of
https://github.com/open-webui/open-webui.git
synced 2026-10-07 02:58:21 +00:00
9ee810ad09
European deployments that need EU data residency have no hosted web search option out of the box: every plug-and-play provider shipped today is US-based, and the only sovereign alternative is self-hosting SearXNG, which means running and maintaining that infrastructure yourself. Staan (staan.ai) is a European search API with EU data residency, so adding it gives those deployments a drop-in choice. It is configured like any other provider, through the admin UI or STAAN_API_KEY, STAAN_MARKET and STAAN_MAX_SNIPPETS. Market sets the region and language of the results and defaults to en-us. Max snippets asks Staan to fetch each result page and return semantically scored chunks of it, which get merged into that result's snippet so retrieval has more to work with; leaving it at 0 uses the plain search endpoint. Wired the same way as Tavily and Exa, with domain filtering going through the shared get_filtered_results. Requested in #26006.
60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
import requests
|
|
from open_webui.retrieval.web.main import SearchResult, get_filtered_results
|
|
|
|
|
|
def search_staan(
|
|
api_key: str,
|
|
query: str,
|
|
count: int,
|
|
filter_list: list[str] | None = None,
|
|
market: str | None = None,
|
|
max_snippets: int | None = None,
|
|
) -> list[SearchResult]:
|
|
"""Search using Staan's Web Search API and return the results as a list of SearchResult objects.
|
|
|
|
Args:
|
|
api_key (str): A Staan API key
|
|
query (str): The query to search for
|
|
count (int): The maximum number of results to return
|
|
filter_list (list[str] | None): The domains to allow or block
|
|
market (str | None): The market to search in, e.g. 'en-us'
|
|
max_snippets (int | None): The maximum extra snippets to request per result
|
|
|
|
Returns:
|
|
A list of SearchResult objects.
|
|
"""
|
|
url = 'https://api.staan.ai/v2/search/web'
|
|
headers = {
|
|
'Accept': 'application/json',
|
|
'Authorization': f'Bearer {api_key}',
|
|
}
|
|
params = {'q': query, 'market': market, 'count': count}
|
|
|
|
if max_snippets:
|
|
params['extra_snippets'] = 'true'
|
|
params['max_snippets'] = max_snippets
|
|
|
|
response = requests.get(url, headers=headers, params=params)
|
|
response.raise_for_status()
|
|
|
|
results = response.json().get('web', {}).get('results', [])
|
|
if filter_list:
|
|
results = get_filtered_results(results, filter_list)
|
|
|
|
return [
|
|
SearchResult(
|
|
link=result.get('url', ''),
|
|
title=result.get('title'),
|
|
snippet=_build_snippet(result),
|
|
)
|
|
for result in results[:count]
|
|
]
|
|
|
|
|
|
def _build_snippet(result: dict) -> str:
|
|
"""Combine the snippet and the extra snippets list into a single string."""
|
|
parts = [result.get('snippet')]
|
|
parts.extend(extra.get('chunk') for extra in result.get('extra_snippets', []))
|
|
return '\n\n'.join(part for part in parts if part)
|