RAG systems that only search a local index cannot answer questions outside their corpus. This tutorial builds a hybrid retrieval system: local document search for known-domain questions (fast, free, private) with automatic fallback to Scavio's search API for open-domain questions (current web data). The confidence threshold determines when to fall back, and source labels let the LLM attribute answers correctly.
Prerequisites
- Python 3.8+ installed
- A local search index (Meilisearch, Elasticsearch, or SQLite FTS)
- requests library installed
- A Scavio API key from scavio.dev
Walkthrough
Step 1: Set up the local search function
Define a function that searches your local index and returns results with confidence scores.
# Example with Meilisearch (replace with your index):
import meilisearch
local_client = meilisearch.Client('http://localhost:7700')
def local_search(query: str, top_k: int = 3) -> list:
results = local_client.index('docs').search(query, {'limit': top_k})
return [{
'text': hit['content'],
'title': hit.get('title', ''),
'score': hit.get('_rankingScore', 0),
'source': 'local'
} for hit in results['hits']]Step 2: Set up the API search function
Define a function that searches via Scavio when local results are insufficient.
import requests, os
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def api_search(query: str, platform: str = 'google') -> list:
resp = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': platform, 'query': query}, timeout=10)
return [{
'text': r.get('snippet', ''),
'title': r.get('title', ''),
'url': r.get('link', ''),
'source': 'web'
} for r in resp.json().get('organic', [])[:3]]Step 3: Build the hybrid retriever
Route to local or API based on confidence threshold.
CONFIDENCE_THRESHOLD = 0.7
def hybrid_retrieve(query: str) -> dict:
local_results = local_search(query)
if local_results and local_results[0]['score'] >= CONFIDENCE_THRESHOLD:
return {'source': 'local', 'results': local_results}
web_results = api_search(query)
if web_results:
return {'source': 'web', 'results': web_results}
return {'source': 'none', 'results': local_results or []}Step 4: Format context for the LLM
Build a prompt context string with source labels for attribution.
def format_context(retrieval: dict) -> str:
source_label = 'Internal docs' if retrieval['source'] == 'local' else 'Web search'
lines = [f'Source: {source_label}']
for r in retrieval['results']:
if r.get('url'):
lines.append(f"- {r['title']}: {r['text']} (ref: {r['url']})")
else:
lines.append(f"- {r['title']}: {r['text']}")
return '\n'.join(lines)
# Use in your RAG prompt:
# context = format_context(hybrid_retrieve(user_question))
# prompt = f'{context}\n\nQuestion: {user_question}\nAnswer:'Python Example
import requests, os
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def hybrid_rag(query, local_index, threshold=0.7):
local = local_index.search(query, {'limit': 3})
if local['hits'] and local['hits'][0].get('_rankingScore', 0) >= threshold:
return {'source': 'local', 'context': [h['content'] for h in local['hits']]}
web = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': 'google', 'query': query}, timeout=10).json()
return {'source': 'web', 'context': [r['snippet'] for r in web.get('organic', [])[:3]]}JavaScript Example
async function hybridRag(query, localIndex, threshold = 0.7) {
const local = await localIndex.search(query, {limit: 3});
if (local.hits?.length && local.hits[0]._rankingScore >= threshold) {
return {source: 'local', context: local.hits.map(h => h.content)};
}
const web = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'},
body: JSON.stringify({platform: 'google', query})
}).then(r => r.json());
return {source: 'web', context: (web.organic || []).slice(0, 3).map(r => r.snippet)};
}Expected Output
A hybrid RAG retriever that searches local docs first and falls back to web search for out-of-corpus questions.