Los agentes de producción fallan cuando falla una sola llamada de herramienta. Una cadena de respaldo garantiza que su agente siempre devuelva resultados útiles al conectarse en cascada a través de una búsqueda primaria, una fuente secundaria y datos almacenados en caché local. Este tutorial construye una cadena alternativa de tres niveles utilizando la API de búsqueda de Scavio como fuente principal, con un pase de reclasificación secundario y un caché local como red de seguridad final.
Requisitos previos
- Python 3.11+ o Node.js 20+
- Una clave API de Scavio de https://scavio.dev
- Comprensión básica del manejo de errores try/catch
- Un archivo SQLite o JSON local para almacenamiento en caché
Guia paso a paso
Paso 1: Configurar la llamada de búsqueda principal
Realice la solicitud de búsqueda inicial a la API de Scavio. Si esto tiene éxito, devolverá los resultados directamente. Envuelva la llamada en el manejo de errores para que las fallas se transmitan en cascada al siguiente nivel.
Python
import httpx
import json
from pathlib import Path
SCAVIO_API_KEY = "your-api-key"
CACHE_FILE = Path("search_cache.json")
async def primary_search(query: str) -> dict:
"""Tier 1: Live Scavio search."""
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": query, "num_results": 10}
)
resp.raise_for_status()
return resp.json()Paso 2: Agregar una búsqueda secundaria con reclasificación
Si la llamada principal arroja menos de tres resultados, inicie una consulta más amplia y vuelva a clasificar localmente. Esto detecta casos extremos en los que la consulta original era demasiado limitada.
Python
async def secondary_search(query: str) -> dict:
"""Tier 2: Broader query with local re-rank."""
broad_query = " ".join(query.split()[:4]) # simplify query
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": broad_query, "num_results": 20}
)
resp.raise_for_status()
data = resp.json()
# Re-rank: prefer results containing original query terms
terms = set(query.lower().split())
results = data.get("results", [])
scored = []
for r in results:
title_lower = r.get("title", "").lower()
score = sum(1 for t in terms if t in title_lower)
scored.append((score, r))
scored.sort(key=lambda x: x[0], reverse=True)
data["results"] = [r for _, r in scored[:10]]
return dataPaso 3: Implementar el respaldo de caché local
Lea desde un archivo de caché JSON local como último recurso. Cada búsqueda exitosa se escribe en esta caché, de modo que los datos obsoletos siempre estén disponibles cuando ambos niveles de API fallan.
Python
def write_cache(query: str, data: dict):
cache = json.loads(CACHE_FILE.read_text()) if CACHE_FILE.exists() else {}
cache[query] = data
CACHE_FILE.write_text(json.dumps(cache, indent=2))
def read_cache(query: str) -> dict | None:
if not CACHE_FILE.exists():
return None
cache = json.loads(CACHE_FILE.read_text())
return cache.get(query)Paso 4: Conecte la cadena de reserva
Combine los tres niveles en una sola función. El agente llama a esta función y siempre obtiene un resultado, con metadatos que indican qué nivel le sirvió.
Python
async def fallback_search(query: str) -> dict:
# Tier 1: primary search
try:
data = await primary_search(query)
if len(data.get("results", [])) >= 3:
write_cache(query, data)
return {"tier": "primary", **data}
except Exception as e:
print(f"Primary search failed: {e}")
# Tier 2: broader secondary search
try:
data = await secondary_search(query)
if data.get("results"):
write_cache(query, data)
return {"tier": "secondary", **data}
except Exception as e:
print(f"Secondary search failed: {e}")
# Tier 3: local cache
cached = read_cache(query)
if cached:
return {"tier": "cache", **cached}
return {"tier": "none", "results": [], "error": "All tiers exhausted"}Ejemplo en Python
Python
import asyncio
import httpx
import json
from pathlib import Path
SCAVIO_API_KEY = "your-api-key"
CACHE_FILE = Path("search_cache.json")
async def primary_search(query: str) -> dict:
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": query, "num_results": 10}
)
resp.raise_for_status()
return resp.json()
async def secondary_search(query: str) -> dict:
broad_query = " ".join(query.split()[:4])
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": broad_query, "num_results": 20}
)
resp.raise_for_status()
data = resp.json()
terms = set(query.lower().split())
results = data.get("results", [])
scored = [(sum(1 for t in terms if t in r.get("title", "").lower()), r) for r in results]
scored.sort(key=lambda x: x[0], reverse=True)
data["results"] = [r for _, r in scored[:10]]
return data
def write_cache(query: str, data: dict):
cache = json.loads(CACHE_FILE.read_text()) if CACHE_FILE.exists() else {}
cache[query] = data
CACHE_FILE.write_text(json.dumps(cache, indent=2))
def read_cache(query: str) -> dict | None:
if not CACHE_FILE.exists():
return None
return json.loads(CACHE_FILE.read_text()).get(query)
async def fallback_search(query: str) -> dict:
try:
data = await primary_search(query)
if len(data.get("results", [])) >= 3:
write_cache(query, data)
return {"tier": "primary", **data}
except Exception:
pass
try:
data = await secondary_search(query)
if data.get("results"):
write_cache(query, data)
return {"tier": "secondary", **data}
except Exception:
pass
cached = read_cache(query)
if cached:
return {"tier": "cache", **cached}
return {"tier": "none", "results": []}
result = asyncio.run(fallback_search("best headless CMS 2026"))
print(f"Served from: {result['tier']}")
print(f"Results: {len(result.get('results', []))}")Ejemplo en JavaScript
JavaScript
const SCAVIO_API_KEY = "your-api-key";
const fs = require("fs");
const CACHE_FILE = "search_cache.json";
async function primarySearch(query) {
const resp = await fetch("https://api.scavio.dev/api/v1/search", {
method: "POST",
headers: { "x-api-key": SCAVIO_API_KEY, "Content-Type": "application/json" },
body: JSON.stringify({ query, num_results: 10 })
});
if (!resp.ok) throw new Error("Primary search failed: " + resp.status);
return resp.json();
}
async function secondarySearch(query) {
const broad = query.split(" ").slice(0, 4).join(" ");
const resp = await fetch("https://api.scavio.dev/api/v1/search", {
method: "POST",
headers: { "x-api-key": SCAVIO_API_KEY, "Content-Type": "application/json" },
body: JSON.stringify({ query: broad, num_results: 20 })
});
if (!resp.ok) throw new Error("Secondary search failed");
const data = await resp.json();
const terms = new Set(query.toLowerCase().split(" "));
data.results = data.results
.map(r => ({ score: [...terms].filter(t => (r.title || "").toLowerCase().includes(t)).length, ...r }))
.sort((a, b) => b.score - a.score)
.slice(0, 10);
return data;
}
function writeCache(query, data) {
const cache = fs.existsSync(CACHE_FILE) ? JSON.parse(fs.readFileSync(CACHE_FILE)) : {};
cache[query] = data;
fs.writeFileSync(CACHE_FILE, JSON.stringify(cache, null, 2));
}
function readCache(query) {
if (!fs.existsSync(CACHE_FILE)) return null;
return JSON.parse(fs.readFileSync(CACHE_FILE))[query] || null;
}
async function fallbackSearch(query) {
try {
const data = await primarySearch(query);
if ((data.results || []).length >= 3) { writeCache(query, data); return { tier: "primary", ...data }; }
} catch {}
try {
const data = await secondarySearch(query);
if (data.results?.length) { writeCache(query, data); return { tier: "secondary", ...data }; }
} catch {}
const cached = readCache(query);
if (cached) return { tier: "cache", ...cached };
return { tier: "none", results: [] };
}
fallbackSearch("best headless CMS 2026").then(r => {
console.log("Served from:", r.tier);
console.log("Results:", (r.results || []).length);
});Salida esperada
JSON
Served from: primary
Results: 10