Overview
This pipeline adds a reliable search fallback layer to OpenWebUI deployments. When the primary SearXNG instance returns empty results, times out, or encounters rate limits, the pipeline automatically falls through to Scavio API for structured Google results including AI Overviews. The fallback is transparent to the user and triggers within 3 seconds of a SearXNG failure. Logs track which backend served each query for monitoring SearXNG health over time.
Trigger
On each OpenWebUI search request
Schedule
On-demand per search request
Workflow Steps
Receive search query from OpenWebUI
Intercept the search request from OpenWebUI's search toggle or RAG pipeline.
Query SearXNG primary backend
Forward the query to the local SearXNG instance with an 8-second timeout.
Evaluate SearXNG response quality
Check if SearXNG returned at least 3 results. If fewer, mark as low-quality and trigger fallback.
Fallback to Scavio API
Query Scavio with the same search terms, requesting AI Overview data for richer grounding context.
Format and return results
Normalize the response format and return structured results to OpenWebUI with source attribution.
Python Implementation
import requests
import time
from datetime import datetime
API_KEY = "your_scavio_api_key"
SEARXNG_URL = "http://localhost:8888/search"
MIN_RESULTS = 3
def search_with_fallback(query: str) -> dict:
start = time.time()
source = "searxng"
# Try SearXNG
try:
res = requests.get(
SEARXNG_URL,
params={"q": query, "format": "json"},
timeout=8,
)
res.raise_for_status()
searxng_results = res.json().get("results", [])
if len(searxng_results) >= MIN_RESULTS:
return {
"source": "searxng",
"results": [{"title": r["title"], "link": r["url"], "snippet": r.get("content", "")} for r in searxng_results[:10]],
"latency_ms": int((time.time() - start) * 1000),
}
except Exception:
pass
# Fallback to Scavio
source = "scavio"
res = requests.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": API_KEY},
json={"platform": "google", "query": query, "ai_overview": True},
timeout=15,
)
res.raise_for_status()
data = res.json()
results = [{"title": r.get("title", ""), "link": r.get("link", ""), "snippet": r.get("snippet", "")} for r in data.get("organic", [])[:10]]
return {
"source": source,
"ai_overview": data.get("ai_overview", {}).get("text", ""),
"results": results,
"latency_ms": int((time.time() - start) * 1000),
}
def run():
queries = ["latest openwebui release 2026", "best self-hosted LLM tools", "searxng vs commercial search api"]
for q in queries:
result = search_with_fallback(q)
print(f" [{result['source']}] {q}: {len(result['results'])} results ({result['latency_ms']}ms)")
if __name__ == "__main__":
run()JavaScript Implementation
const API_KEY = "your_scavio_api_key";
async function searchWithFallback(query) {
const start = Date.now();
try {
const res = await fetch(`http://localhost:8888/search?q=${encodeURIComponent(query)}&format=json`, { signal: AbortSignal.timeout(8000) });
const data = await res.json();
if ((data.results ?? []).length >= 3) {
return { source: "searxng", results: data.results.slice(0, 10).map((r) => ({ title: r.title, link: r.url, snippet: r.content ?? "" })), ms: Date.now() - start };
}
} catch {}
const res = await fetch("https://api.scavio.dev/api/v1/search", {
method: "POST",
headers: { "x-api-key": API_KEY, "content-type": "application/json" },
body: JSON.stringify({ platform: "google", query, ai_overview: true }),
});
const data = await res.json();
return { source: "scavio", results: (data.organic ?? []).slice(0, 10).map((r) => ({ title: r.title ?? "", link: r.link ?? "", snippet: r.snippet ?? "" })), ms: Date.now() - start };
}
for (const q of ["latest openwebui release 2026", "best self-hosted LLM tools"]) {
const r = await searchWithFallback(q);
console.log(`[${r.source}] ${q}: ${r.results.length} results (${r.ms}ms)`);
}Platforms Used
Web search with knowledge graph, PAA, and AI overviews