Gérez les erreurs de limite de débit 429 de Gemini et les erreurs de service indisponible 503 en redirigeant les requêtes d'ancrage vers une API de recherche comme solution de repli, puis en rétablissant automatiquement Gemini lorsqu'il redevient disponible. La fonction d'ancrage de Gemini est utile mais peu fiable en cas de charge : les limites de débit sont atteintes à des seuils imprévisibles et les erreurs 503 peuvent durer plusieurs minutes. Un repli vers une API de recherche garantit que votre application ne perd jamais la capacité d'ancrage. Ce tutoriel encapsule l'appel Gemini dans une détection d'erreur, redirige les échecs vers Scavio et implémente une boucle de vérification de l'état qui restaure Gemini lorsqu'il se rétablit.
Prérequis
- Python 3.8+ installé
- bibliothèque requests installée
- Une clé API Scavio depuis scavio.dev
- Accès à l'API Google Gemini (ou tout LLM avec ancrage)
Parcours
Étape 1: Encapsulez l'appel d'ancrage Gemini
Créez un wrapper autour de votre appel API Gemini qui intercepte les erreurs 429 et 503 au lieu de planter.
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')Étape 2: Détectez les erreurs et redirigez vers Scavio
Lorsque Gemini renvoie une erreur, redirigez automatiquement la requête d'ancrage vers l'API de recherche 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']}")Étape 3: Redirigez vers le repli Scavio
Construisez l'ancrage basé sur la recherche qui remplace la fonction d'ancrage de Gemini par un contexte équivalent issu des résultats 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'))Étape 4: Rétablissement automatique lorsque Gemini revient
Testez périodiquement l'état de Gemini et restaurez-le comme source d'ancrage principale lorsqu'il se rétablit.
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']}")Exemple 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'))Exemple 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);Sortie attendue
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.