AI agents that depend on a single search provider are fragile. Rate limits hit, Cloudflare blocks trigger, and outages happen. A fallback chain tries multiple providers in priority order so your agent always gets data. This tutorial builds a production fallback chain with Scavio as primary, configurable retries, latency tracking, and automatic provider rotation. Each Scavio request costs $0.005 per credit with 6 platforms available from one endpoint.
Prérequis
- Python 3.9+ installé
- bibliothèque requests installée
- Une clé API Scavio depuis scavio.dev
- Optionnel : clés de fournisseur de secours pour une couverture de repli complète
Parcours
Étape 1: Définir l'interface du fournisseur
Créez une classe de base que chaque fournisseur de recherche doit implémenter. Cela garantit un format de résultat uniforme quel que soit le fournisseur qui répond.
Python
import os, time, requests
from dataclasses import dataclass
from typing import Optional
@dataclass
class SearchResult:
title: str
url: str
snippet: str
provider: str
class SearchProvider:
name: str
def search(self, query: str, num: int = 5) -> list[SearchResult]:
raise NotImplementedError
class ScavioProvider(SearchProvider):
name = 'scavio'
def __init__(self):
self.key = os.environ['SCAVIO_API_KEY']
def search(self, query: str, num: int = 5) -> list[SearchResult]:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': self.key, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us', 'num_results': num},
timeout=10)
resp.raise_for_status()
return [SearchResult(title=r['title'], url=r['link'],
snippet=r.get('snippet', ''), provider='scavio')
for r in resp.json().get('organic_results', [])]
print('Provider interface ready')Étape 2: Construire la chaîne de repli
La chaîne essaie chaque fournisseur dans l'ordre. Si l'un échoue ou renvoie des résultats vides, elle passe au suivant. Elle enregistre quel fournisseur a réussi et la latence.
Python
class FallbackChain:
def __init__(self, providers: list[SearchProvider]):
self.providers = providers
self.stats = {p.name: {'success': 0, 'fail': 0, 'total_ms': 0} for p in providers}
def search(self, query: str, num: int = 5) -> tuple[list[SearchResult], str]:
for provider in self.providers:
try:
start = time.time()
results = provider.search(query, num)
elapsed = (time.time() - start) * 1000
if results:
self.stats[provider.name]['success'] += 1
self.stats[provider.name]['total_ms'] += elapsed
return results, provider.name
except Exception as e:
self.stats[provider.name]['fail'] += 1
print(f'[fallback] {provider.name} failed: {e}')
continue
return [], 'none'
def report(self) -> str:
lines = ['Provider Stats:']
for name, s in self.stats.items():
total = s['success'] + s['fail']
avg_ms = s['total_ms'] / s['success'] if s['success'] else 0
lines.append(f" {name}: {s['success']}/{total} ok, {avg_ms:.0f}ms avg")
return '\n'.join(lines)
chain = FallbackChain([ScavioProvider()])
results, used = chain.search('python web framework 2026')
print(f'Got {len(results)} results from {used}')Étape 3: Ajouter une logique de tentative avec backoff exponentiel
Enveloppez chaque appel de fournisseur avec une logique de tentative. Les échecs transitoires comme les timeouts doivent être réessayés avant de passer au fournisseur suivant.
Python
class RetryProvider:
def __init__(self, provider: SearchProvider, max_retries: int = 2, base_delay: float = 0.5):
self.provider = provider
self.name = provider.name
self.max_retries = max_retries
self.base_delay = base_delay
def search(self, query: str, num: int = 5) -> list[SearchResult]:
last_error = None
for attempt in range(self.max_retries + 1):
try:
return self.provider.search(query, num)
except requests.exceptions.Timeout:
last_error = 'timeout'
delay = self.base_delay * (2 ** attempt)
print(f'[retry] {self.name} timeout, waiting {delay:.1f}s')
time.sleep(delay)
except requests.exceptions.HTTPError as e:
if e.response and e.response.status_code == 429:
delay = self.base_delay * (2 ** attempt)
print(f'[retry] {self.name} rate limited, waiting {delay:.1f}s')
time.sleep(delay)
last_error = 'rate_limit'
else:
raise
raise Exception(f'{self.name} failed after {self.max_retries} retries: {last_error}')
chain = FallbackChain([RetryProvider(ScavioProvider())])
results, used = chain.search('AI agent frameworks')
print(f'{len(results)} results via {used}')Étape 4: Ajouter une vérification de santé et une rotation automatique
Suivez la santé du fournisseur et dépriorisez automatiquement les fournisseurs défaillants. Vérifiez-les périodiquement au cas où ils se rétablissent.
Python
class HealthAwareFallback(FallbackChain):
def __init__(self, providers, unhealthy_threshold=3, recheck_interval=60):
super().__init__(providers)
self.unhealthy_threshold = unhealthy_threshold
self.recheck_interval = recheck_interval
self.consecutive_fails = {p.name: 0 for p in providers}
self.last_recheck = {p.name: 0 for p in providers}
def search(self, query: str, num: int = 5) -> tuple[list[SearchResult], str]:
now = time.time()
for provider in self.providers:
name = provider.name if hasattr(provider, 'name') else str(provider)
# Skip unhealthy providers unless recheck interval passed
if self.consecutive_fails.get(name, 0) >= self.unhealthy_threshold:
if now - self.last_recheck.get(name, 0) < self.recheck_interval:
continue
self.last_recheck[name] = now
print(f'[health] rechecking {name}')
try:
start = time.time()
results = provider.search(query, num)
elapsed = (time.time() - start) * 1000
if results:
self.consecutive_fails[name] = 0
self.stats[name]['success'] += 1
self.stats[name]['total_ms'] += elapsed
return results, name
except Exception as e:
self.consecutive_fails[name] = self.consecutive_fails.get(name, 0) + 1
self.stats[name]['fail'] += 1
print(f'[health] {name} fail #{self.consecutive_fails[name]}: {e}')
return [], 'none'
chain = HealthAwareFallback([RetryProvider(ScavioProvider())])
results, used = chain.search('search api comparison')
print(f'{len(results)} results via {used}')
print(chain.report())Exemple Python
Python
import os, time, requests
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
def scavio_search(query, num=5):
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us', 'num_results': num}, timeout=10)
resp.raise_for_status()
return [{'title': r['title'], 'url': r['link'], 'snippet': r.get('snippet', '')}
for r in resp.json().get('organic_results', [])]
def search_with_fallback(query, num=5, retries=2):
for attempt in range(retries + 1):
try:
return scavio_search(query, num)
except Exception as e:
if attempt < retries:
time.sleep(0.5 * (2 ** attempt))
else:
raise
results = search_with_fallback('AI agent search tools 2026')
for r in results:
print(f"{r['title']}\n {r['url']}")Exemple JavaScript
JavaScript
const SCAVIO_KEY = process.env.SCAVIO_API_KEY;
async function search(query, num = 5, retries = 2) {
for (let i = 0; i <= retries; i++) {
try {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: 'us', num_results: num }),
signal: AbortSignal.timeout(10000)
});
if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
const data = await resp.json();
return (data.organic_results || []).map(r => ({ title: r.title, url: r.link, snippet: r.snippet || '' }));
} catch (e) {
if (i < retries) await new Promise(r => setTimeout(r, 500 * 2 ** i));
else throw e;
}
}
}
search('AI agent search tools 2026').then(r => r.forEach(x => console.log(x.title)));Sortie attendue
JSON
Got 5 results from scavio
Provider Stats:
scavio: 1/1 ok, 320ms avg
AI Agent Frameworks Comparison 2026
https://example.com/ai-agent-frameworks
Building Reliable AI Agents with Search
https://blog.example.com/reliable-agents