Les outils de recherche dans les agents IA échouent plus souvent que les développeurs ne le pensent : résultats vides pour des requêtes trop spécifiques, timeouts sur des réseaux lents, limites de débit lors d'utilisation en rafale, et requêtes malformées provenant du LLM. Une chaîne de nouvelles tentatives gère ces échecs avec élégance en tentant des requêtes progressivement simplifiées, avec un backoff entre les tentatives, et en tombant en recours sur des résultats en cache quand toutes les tentatives échouent. Ce tutoriel construit une chaîne de nouvelles tentatives de qualité production qui encapsule n'importe quel appel API de recherche. Le modèle ajoute un surcoût quasi nul car les nouvelles tentatives ne se déclenchent qu'en cas d'échec.
Prérequis
- Python 3.9+ installé
- bibliothèque requests installée
- Une clé API Scavio depuis scavio.dev
- Un agent avec un outil de recherche à renforcer
Parcours
Étape 1: Construire la fonction de recherche de base avec gestion des erreurs
Commencez par une fonction de recherche qui capture et classe tous les modes d'échec : erreurs HTTP, timeouts, résultats vides et réponses malformées.
Python
import requests, os, time
from enum import Enum
API_KEY = os.environ['SCAVIO_API_KEY']
class SearchError(Enum):
SUCCESS = 'success'
EMPTY = 'empty_results'
TIMEOUT = 'timeout'
RATE_LIMIT = 'rate_limit'
AUTH_ERROR = 'auth_error'
SERVER_ERROR = 'server_error'
UNKNOWN = 'unknown'
def search_with_status(query: str, timeout: int = 15) -> tuple:
"""Returns (results, error_type)."""
try:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us'},
timeout=timeout)
if resp.status_code == 429:
return [], SearchError.RATE_LIMIT
if resp.status_code == 401:
return [], SearchError.AUTH_ERROR
if resp.status_code >= 500:
return [], SearchError.SERVER_ERROR
results = resp.json().get('organic_results', [])
if not results:
return [], SearchError.EMPTY
return results, SearchError.SUCCESS
except requests.exceptions.Timeout:
return [], SearchError.TIMEOUT
except Exception:
return [], SearchError.UNKNOWNÉtape 2: Construire la chaîne de simplification des requêtes
Lorsqu'une requête renvoie des résultats vides, essayez des versions progressivement plus simples. Supprimez les qualificatifs, raccourcissez la requête et élargissez le périmètre.
Python
def simplify_query(query: str) -> list:
"""Generate progressively simpler versions of a query."""
words = query.split()
variants = [query] # original first
# Remove quotes and special chars
cleaned = query.replace('"', '').replace("'", '')
if cleaned != query:
variants.append(cleaned)
# Remove year qualifiers
import re
no_year = re.sub(r'\b202[0-9]\b', '', query).strip()
if no_year != query:
variants.append(no_year)
# First N words (progressively shorter)
if len(words) > 5:
variants.append(' '.join(words[:5]))
if len(words) > 3:
variants.append(' '.join(words[:3]))
# Remove filler words
filler = {'the', 'a', 'an', 'is', 'are', 'was', 'were', 'for', 'and', 'or', 'but'}
no_filler = ' '.join(w for w in words if w.lower() not in filler)
if no_filler != query and len(no_filler) > 5:
variants.append(no_filler)
return list(dict.fromkeys(variants)) # dedupe preserving order
# Test:
for v in simplify_query('"best enterprise CRM software" for startups 2026'):
print(f' {v}')Étape 3: Construire la chaîne de nouvelles tentatives avec backoff
Enchaînez les tentatives avec un backoff exponentiel pour les erreurs transitoires et une simplification des requêtes pour les résultats vides.
Python
import logging
log = logging.getLogger('search_retry')
def search_retry_chain(query: str, max_retries: int = 3) -> dict:
"""Retry chain: try original, then simplified queries, with backoff."""
queries = simplify_query(query)
all_attempts = []
for attempt, q in enumerate(queries[:max_retries]):
results, error = search_with_status(q)
all_attempts.append({'query': q, 'error': error.value, 'result_count': len(results)})
if error == SearchError.SUCCESS:
return {
'results': results,
'query_used': q,
'attempts': len(all_attempts),
'all_attempts': all_attempts
}
if error == SearchError.AUTH_ERROR:
log.error('Authentication failed. Check your API key.')
break
if error in (SearchError.RATE_LIMIT, SearchError.SERVER_ERROR, SearchError.TIMEOUT):
wait = 2 ** attempt
log.warning(f'Transient error ({error.value}), retrying in {wait}s...')
time.sleep(wait)
# Retry same query for transient errors
results, error = search_with_status(q)
if error == SearchError.SUCCESS:
all_attempts.append({'query': q, 'error': 'retry_success', 'result_count': len(results)})
return {'results': results, 'query_used': q,
'attempts': len(all_attempts), 'all_attempts': all_attempts}
# For EMPTY, move to next simplified query
log.info(f'Attempt {attempt + 1}: "{q}" -> {error.value}')
return {'results': [], 'query_used': query,
'attempts': len(all_attempts), 'all_attempts': all_attempts}Étape 4: Ajouter un cache de résultats pour le repli
Mettez en cache les résultats de recherche réussis pour que l'agent ait un repli lorsque toutes les tentatives échouent. Un cache basé sur TTL empêche de servir des données trop obsolètes.
Python
import hashlib
_cache = {}
CACHE_TTL = 3600 # 1 hour
def cached_search(query: str) -> dict:
key = hashlib.md5(query.lower().strip().encode()).hexdigest()
now = time.time()
# Check cache first
if key in _cache:
cached, timestamp = _cache[key]
if now - timestamp < CACHE_TTL:
return {**cached, 'from_cache': True, 'cache_age_s': int(now - timestamp)}
# Run retry chain
result = search_retry_chain(query)
# Cache successful results
if result['results']:
_cache[key] = (result, now)
elif key in _cache:
# All retries failed, return stale cache with warning
cached, timestamp = _cache[key]
return {**cached, 'from_cache': True,
'cache_age_s': int(now - timestamp),
'warning': 'Serving stale cached results (all retries failed)'}
return {**result, 'from_cache': False}
# Test:
result = cached_search('best crm software 2026')
print(f'Results: {len(result["results"])}, cache: {result["from_cache"]}, '
f'attempts: {result["attempts"]}')Étape 5: Intégrer dans votre agent en tant qu'outil de recherche principal
Remplacez la fonction de recherche de base de votre agent par la chaîne de nouvelles tentatives. L'agent appelle la même interface mais bénéficie de tentatives automatiques et de mise en cache.
Python
def agent_search_tool(query: str) -> str:
"""Production-ready search tool for agents.
Handles retries, query simplification, and caching automatically."""
result = cached_search(query)
if not result['results']:
return f'Search returned no results after {result["attempts"]} attempts. Please rephrase your query.'
formatted = []
for r in result['results'][:5]:
formatted.append(f'Title: {r["title"]}\nURL: {r["link"]}\nSnippet: {r.get("snippet", "")}')
header = f'Found {len(result["results"])} results'
if result.get('from_cache'):
header += f' (cached, {result["cache_age_s"]}s old)'
if result['query_used'] != query:
header += f' (simplified query: "{result["query_used"]}")'
return f'{header}\n\n' + '\n\n'.join(formatted)
# Use in any agent framework:
# @tool
# def web_search(query: str) -> str:
# return agent_search_tool(query)
print(agent_search_tool('best enterprise CRM software pricing comparison 2026'))Exemple Python
Python
import os, requests, time, hashlib
API_KEY = os.environ['SCAVIO_API_KEY']
_cache = {}
def search(query, timeout=10):
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us'}, timeout=timeout)
return resp.json().get('organic_results', [])
def search_with_retry(query, max_retries=3):
queries = [query] + [' '.join(query.split()[:n]) for n in [5, 3]]
for q in queries[:max_retries]:
results = search(q)
if results:
key = hashlib.md5(query.encode()).hexdigest()
_cache[key] = results
return results
time.sleep(1)
return _cache.get(hashlib.md5(query.encode()).hexdigest(), [])
results = search_with_retry('best enterprise CRM pricing comparison 2026')
print(f'{len(results)} results')Exemple JavaScript
JavaScript
const API_KEY = process.env.SCAVIO_API_KEY;
const cache = new Map();
async function search(query) {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: 'us' })
});
return (await resp.json()).organic_results || [];
}
async function searchWithRetry(query, maxRetries = 3) {
const words = query.split(' ');
const queries = [query, words.slice(0, 5).join(' '), words.slice(0, 3).join(' ')];
for (const q of queries.slice(0, maxRetries)) {
const results = await search(q);
if (results.length) { cache.set(query, results); return results; }
await new Promise(r => setTimeout(r, 1000));
}
return cache.get(query) || [];
}
searchWithRetry('best enterprise CRM pricing 2026')
.then(r => console.log(`${r.length} results`));Sortie attendue
JSON
"best enterprise CRM software" for startups 2026
best enterprise CRM software for startups 2026
best enterprise CRM software for startups
best enterprise CRM
enterprise CRM software startups 2026
Attempt 1: "best enterprise CRM software..." -> empty_results
Attempt 2: "best enterprise CRM software for startups" -> success
Results: 10, cache: False, attempts: 2
Found 10 results (simplified query: "best enterprise CRM software for startups")