Maneje los errores de límite de velocidad 429 de Gemini y servicio 503 no disponible enrutando consultas de conexión a tierra a una API de búsqueda como alternativa y luego recuperándose automáticamente cuando Gemini vuelva a estar disponible. La función de conexión a tierra de Gemini es útil pero poco confiable bajo carga: los límites de velocidad alcanzan umbrales impredecibles y los 503 pueden durar minutos. Un respaldo de API de búsqueda garantiza que su aplicación nunca pierda la capacidad de conexión a tierra. Este tutorial envuelve la llamada de Gemini en la detección de errores, enruta las fallas a Scavio e implementa un bucle de verificación de estado que restaura Gemini cuando se recupera.
Requisitos previos
- Python 3.8+ instalado
- solicita biblioteca instalada
- Una clave API de Scavio de scavio.dev
- Acceso a la API de Google Gemini (o cualquier LLM con conexión a tierra)
Guia paso a paso
Paso 1: Termina la llamada a tierra de Géminis
Cree un contenedor alrededor de su llamada API de Gemini que detecte los errores 429 y 503 en lugar de fallar.
import os, requests, time
GEMINI_KEY = os.environ.get('GEMINI_API_KEY', '')
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
gemini_healthy = True
gemini_last_check = 0
def gemini_ground(query: str) -> dict:
"""Call Gemini with grounding. Returns results or raises on failure."""
try:
resp = requests.post(
f'https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:generateContent?key={GEMINI_KEY}',
json={'contents': [{'parts': [{'text': query}]}]},
timeout=15)
if resp.status_code in (429, 503):
raise requests.exceptions.HTTPError(f'Gemini {resp.status_code}')
resp.raise_for_status()
return {'source': 'gemini', 'data': resp.json()}
except requests.exceptions.RequestException as e:
return {'source': 'gemini', 'error': str(e)}
print('Gemini wrapper ready')Paso 2: Detectar errores y ruta a Scavio
Cuando Gemini devuelve un error, dirige automáticamente la consulta básica a la API de búsqueda de Scavio.
def scavio_ground(query: str) -> dict:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY},
json={'platform': 'google', 'query': query}, timeout=10)
resp.raise_for_status()
results = resp.json().get('organic_results', [])[:5]
context = '\n'.join(f"{r['title']}: {r.get('snippet', '')}" for r in results)
return {'source': 'scavio', 'context': context, 'results': results}
def grounded_query(query: str) -> dict:
global gemini_healthy
if gemini_healthy:
result = gemini_ground(query)
if 'error' not in result:
return result
print(f"Gemini failed: {result['error']}. Falling back to Scavio.")
gemini_healthy = False
return scavio_ground(query)
result = grounded_query('best project management tools 2026')
print(f"Grounded via: {result['source']}")Paso 3: Ruta hacia el repliegue de Scavio
Cree una conexión a tierra basada en búsquedas que reemplace la función de conexión a tierra de Gemini con un contexto equivalente a partir de los resultados web.
def format_grounding_context(search_results: list) -> str:
"""Format search results to match the context format Gemini grounding provides."""
if not search_results:
return ''
parts = []
for r in search_results:
title = r.get('title', '')
snippet = r.get('snippet', '')
url = r.get('link', '')
parts.append(f'[{title}]({url}): {snippet}')
return '\n'.join(parts)
def grounded_prompt(query: str) -> str:
result = grounded_query(query)
if result['source'] == 'scavio':
context = format_grounding_context(result.get('results', []))
return f'Use the following web context to answer:\n{context}\n\nQuestion: {query}'
return f'Question: {query}'
print(grounded_prompt('best CRM for startups 2026'))Paso 4: Recuperación automática cuando Gemini regrese
Pruebe periódicamente la salud de Gemini y restáurelo como fuente principal de conexión a tierra cuando se recupere.
RECOVERY_INTERVAL = 60 # seconds between recovery checks
def check_gemini_health() -> bool:
global gemini_healthy, gemini_last_check
now = time.time()
if now - gemini_last_check < RECOVERY_INTERVAL:
return gemini_healthy
gemini_last_check = now
result = gemini_ground('test query')
if 'error' not in result:
gemini_healthy = True
print('Gemini recovered. Restoring as primary.')
else:
print(f'Gemini still down: {result["error"]}')
return gemini_healthy
def smart_ground(query: str) -> dict:
if not gemini_healthy:
check_gemini_health()
return grounded_query(query)
result = smart_ground('latest python release')
print(f"Source: {result['source']}")Ejemplo en Python
import requests, os
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def fallback_ground(query):
# Try Gemini first, fall back to Scavio
try:
# gemini_call(query) # Your Gemini grounding call
raise Exception('Simulated Gemini 429')
except Exception:
data = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': 'google', 'query': query}).json()
results = data.get('organic_results', [])[:3]
return '\n'.join(f"{r['title']}: {r.get('snippet', '')}" for r in results)
print(fallback_ground('best CRM 2026'))Ejemplo en JavaScript
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
async function fallbackGround(query) {
try {
// await geminiGround(query); // Your Gemini call
throw new Error('Simulated 429');
} catch {
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H, body: JSON.stringify({platform: 'google', query})
});
const results = (await r.json()).organic_results || [];
return results.slice(0, 3).map(r => `${r.title}: ${r.snippet}`).join('\n');
}
}
fallbackGround('best CRM 2026').then(console.log);Salida esperada
A grounding system that uses Gemini when available, automatically falls back to Scavio search on 429/503 errors, and auto-recovers when Gemini is healthy again.