Les agents de production échouent lorsqu'un seul appel d'outil échoue. Une chaîne de repli garantit que votre agent renvoie toujours des résultats utiles en cascade via une recherche primaire, une source secondaire et des données mises en cache localement. Ce tutoriel construit une chaîne de repli à trois niveaux en utilisant l'API Scavio Search comme source primaire, avec un second passage de reclassement et un cache local comme filet de sécurité final.
Prérequis
- Python 3.11+ ou Node.js 20+
- Une clé API Scavio depuis https://scavio.dev
- Compréhension de base de la gestion d'erreurs try/catch
- Un fichier SQLite ou JSON local pour la mise en cache
Parcours
Étape 1: Configurer l'appel de recherche primaire
Effectuez la première demande de recherche à l'API Scavio. Si elle réussit, renvoyez les résultats directement. Enveloppez l'appel dans une gestion d'erreur pour que les échecs se propagent au niveau suivant.
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()Étape 2: Ajouter une recherche secondaire avec reclassement
Si l'appel primaire renvoie moins de trois résultats, lancez une requête plus large et reclassifiez localement. Cela couvre les cas limites où la requête d'origine était trop étroite.
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 dataÉtape 3: Implémenter le repli du cache local
Lisez à partir d'un fichier de cache JSON local comme dernier recours. Chaque recherche réussie écrit dans ce cache afin que les données obsolètes soient toujours disponibles lorsque les deux niveaux d'API échouent.
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)Étape 4: Câbler la chaîne de repli ensemble
Combinez les trois niveaux en une seule fonction. L'agent appelle cette fonction unique et obtient toujours un résultat, avec des métadonnées indiquant quel niveau l'a servi.
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"}Exemple 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', []))}")Exemple 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);
});Sortie attendue
JSON
Served from: primary
Results: 10