Un post sur r/PiCodingAgent demandait une recherche fiable pour les agents de codage. Le problème : une recherche mono-fournisseur échoue silencieusement. Ce tutoriel construit un routeur qui essaie plusieurs fournisseurs par ordre de priorité et bascule automatiquement. Utile pour tout runtime d'agent, pas seulement Pi.
Prérequis
- Clé API Scavio
- Python 3.8+
- Optionnel : clé API Tavily, instance SearxNG
Parcours
Étape 1: Définir la configuration du fournisseur
Lister les fournisseurs avec leurs points de terminaison et leur priorité.
Python
providers = [
{
'name': 'scavio',
'url': 'https://api.scavio.dev/api/v1/search',
'headers': lambda: {'x-api-key': os.environ['SCAVIO_API_KEY']},
'body': lambda q: {'platform': 'google', 'query': q},
'parse': lambda r: r.get('organic_results', []),
'cost': 0.005,
'priority': 1
},
{
'name': 'searxng',
'url': 'http://localhost:8080/search',
'headers': lambda: {},
'body': lambda q: {'q': q, 'format': 'json'},
'parse': lambda r: r.get('results', []),
'cost': 0,
'priority': 2
},
{
'name': 'tavily',
'url': 'https://api.tavily.com/search',
'headers': lambda: {},
'body': lambda q: {'api_key': os.environ.get('TAVILY_API_KEY', ''), 'query': q},
'parse': lambda r: r.get('results', []),
'cost': 0.008,
'priority': 3
},
]Étape 2: Construire le routeur avec repli
Essayer chaque fournisseur par ordre de priorité, basculer en cas d'échec.
Python
import requests, os, time
def search_with_fallback(query, providers_list=providers):
sorted_providers = sorted(providers_list, key=lambda p: p['priority'])
errors = []
for provider in sorted_providers:
try:
resp = requests.post(provider['url'],
headers=provider['headers'](),
json=provider['body'](query),
timeout=10)
resp.raise_for_status()
results = provider['parse'](resp.json())
if results:
return {'provider': provider['name'], 'results': results, 'cost': provider['cost']}
errors.append(f"{provider['name']}: empty results")
except Exception as e:
errors.append(f"{provider['name']}: {str(e)}")
return {'provider': None, 'results': [], 'errors': errors}Étape 3: Ajouter un routage sensible à la latence
Suivre les temps de réponse et préférer les fournisseurs plus rapides.
Python
provider_stats = {}
def update_stats(provider_name, latency, success):
if provider_name not in provider_stats:
provider_stats[provider_name] = {'total': 0, 'success': 0, 'avg_latency': 0}
stats = provider_stats[provider_name]
stats['total'] += 1
if success: stats['success'] += 1
stats['avg_latency'] = (stats['avg_latency'] * (stats['total'] - 1) + latency) / stats['total']
def smart_search(query):
start = time.time()
result = search_with_fallback(query)
latency = time.time() - start
if result['provider']:
update_stats(result['provider'], latency, True)
return resultÉtape 4: Journaliser l'utilisation des fournisseurs pour le suivi des coûts
Suivre quel fournisseur répond à chaque requête.
Python
import json, datetime
def log_search(query, result):
with open('search_router_log.jsonl', 'a') as f:
f.write(json.dumps({
'ts': datetime.datetime.now().isoformat(),
'query': query,
'provider': result['provider'],
'cost': result.get('cost', 0),
'result_count': len(result.get('results', []))
}) + '\n')Exemple Python
Python
import os, requests
def search(query):
for provider in ['scavio', 'searxng', 'tavily']:
try:
if provider == 'scavio':
r = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': os.environ['SCAVIO_API_KEY']},
json={'platform': 'google', 'query': query}, timeout=10)
results = r.json().get('organic_results', [])
if results: return {'provider': 'scavio', 'results': results}
except: continue
return {'provider': None, 'results': []}Exemple JavaScript
JavaScript
async function search(query) {
try {
const res = 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})
});
const data = await res.json();
if (data.organic_results?.length) return {provider: 'scavio', results: data.organic_results};
} catch(e) { /* fallback */ }
}Sortie attendue
JSON
Multi-provider search router with automatic fallback. Priority: Scavio ($0.005) > SearxNG (free/self-hosted) > Tavily ($0.008). Latency tracking and JSONL cost log.