Las herramientas de búsqueda en los agentes de IA fallan con más frecuencia de lo que esperan los desarrolladores: resultados vacíos de consultas demasiado específicas, tiempos de espera en redes lentas, límites de velocidad durante el uso en ráfagas y consultas con formato incorrecto del LLM. Una cadena de reintentos maneja estos errores con elegancia al intentar consultas progresivamente más simples, con pausas entre intentos y recurrir a los resultados almacenados en caché cuando todos los reintentos fallan. Este tutorial crea una cadena de reintentos de nivel de producción que engloba cualquier llamada a la API de búsqueda. El patrón agrega costos generales casi nulos, ya que los reintentos solo se activan en caso de falla.
Requisitos previos
- Python 3.9+ instalado
- solicita biblioteca instalada
- Una clave API de Scavio de scavio.dev
- Un agente con una herramienta de búsqueda para endurecer
Guia paso a paso
Paso 1: Cree la función de búsqueda base con manejo de errores
Comience con una función de búsqueda que detecte y clasifique todos los modos de falla: errores HTTP, tiempos de espera, resultados vacíos y respuestas con formato incorrecto.
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.UNKNOWNPaso 2: Construya la cadena de simplificación de consultas
Cuando una consulta arroja resultados vacíos, pruebe con versiones progresivamente más simples. Elimine calificadores, acorte la consulta y amplíe el alcance.
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}')Paso 3: Construya la cadena de reintento con retroceso
Encadene reintentos con retroceso exponencial para errores transitorios y simplificación de consultas para resultados vacíos.
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}Paso 4: Agregar un caché de resultados como respaldo
Almacene en caché los resultados de búsqueda exitosos para que el agente tenga un respaldo cuando todos los reintentos fallen. La caché basada en TTL evita que se proporcionen datos muy obsoletos.
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"]}')Paso 5: Intégrelo en su agente como principal herramienta de búsqueda
Reemplace la función de búsqueda básica de su agente con la cadena de reintento. El agente llama a la misma interfaz pero obtiene reintentos y almacenamiento en caché automáticos.
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'))Ejemplo en 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')Ejemplo en 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`));Salida esperada
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")