Los agentes de IA que dependen de un único proveedor de búsqueda son frágiles. Se alcanzan los límites de tarifas, se activan los bloqueos de Cloudflare y se producen interrupciones. Una cadena alternativa prueba varios proveedores en orden de prioridad para que su agente siempre obtenga datos. Este tutorial crea una cadena alternativa de producción con Scavio como reintentos primarios, configurables, seguimiento de latencia y rotación automática de proveedores. Cada solicitud de Scavio cuesta $0,005 por crédito con 6 plataformas disponibles desde un punto final.
Requisitos previos
- Python 3.9+ instalado
- solicita biblioteca instalada
- Una clave API de Scavio de scavio.dev
- Opcional: claves del proveedor de respaldo para una cobertura alternativa completa
Guia paso a paso
Paso 1: Definir la interfaz del proveedor
Cree una clase base que cada proveedor de búsqueda debe implementar. Esto garantiza un formato de resultado uniforme independientemente del proveedor que responda.
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')Paso 2: Construir la cadena de respaldo
La cadena prueba a cada proveedor en orden. Si uno falla o arroja resultados vacíos, se pasa al siguiente. Realiza un seguimiento de qué proveedor tuvo éxito y la latencia.
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}')Paso 3: Agregar lógica de reintento con retroceso exponencial
Envuelva cada llamada al proveedor con lógica de reintento. Los fallos transitorios, como los tiempos de espera, se deben volver a intentar antes de pasar al siguiente proveedor.
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}')Paso 4: Agregar verificación de estado y rotación automática
Realice un seguimiento del estado de los proveedores y reste prioridad automáticamente a los proveedores que no están en buen estado. Vuelva a revisarlos periódicamente en caso de que se recuperen.
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())Ejemplo en 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']}")Ejemplo en 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)));Salida esperada
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