Réduisez la latence des API de recherche dans les agents d'IA en appliquant quatre techniques : requêtes parallèles pour les workflows multi-requêtes, mise en cache des résultats pour les requêtes répétées, élagage des requêtes pour réduire la taille des charges utiles, et mise en commun des connexions pour éliminer les frais de poignée de main. Dans les workflows d'agents typiques, les appels de recherche représentent 60 à 80 % du temps de réponse total. Même de petites réductions de latence se cumulent dans les chaînes de raisonnement en plusieurs étapes. Ce tutoriel implémente chaque optimisation avec l'API Scavio et mesure l'impact avant/après.
Prérequis
- Python 3.8+ installé
- Bibliothèque requests installée
- Une clé API Scavio depuis scavio.dev
- Un workflow d'agent existant avec des appels de recherche
Parcours
Étape 1: Mesurer la latence de base
Établissez une référence en chronométrant les appels de recherche séquentiels afin de mesurer l'impact de chaque optimisation.
import requests, os, time
from concurrent.futures import ThreadPoolExecutor
API_KEY = os.environ['SCAVIO_API_KEY']
SESSION = requests.Session()
SESSION.headers.update({'x-api-key': API_KEY})
def timed_search(query: str) -> tuple:
start = time.monotonic()
resp = SESSION.post('https://api.scavio.dev/api/v1/search',
json={'platform': 'google', 'query': query}, timeout=10)
latency = (time.monotonic() - start) * 1000
return query, round(latency, 1), len(resp.json().get('organic_results', []))
# Baseline: sequential
queries = ['best crm 2026', 'python async tutorial', 'react vs vue']
start = time.monotonic()
for q in queries:
_, ms, _ = timed_search(q)
print(f'{q}: {ms}ms')
print(f'Sequential total: {(time.monotonic() - start)*1000:.0f}ms')Étape 2: Paralléliser les requêtes multi-requêtes
Utilisez un pool de threads pour envoyer plusieurs requêtes de recherche simultanément, réduisant le temps total d'exécution de 2 à 3 fois.
def parallel_search(queries: list, max_workers: int = 3) -> list:
start = time.monotonic()
with ThreadPoolExecutor(max_workers=max_workers) as pool:
results = list(pool.map(timed_search, queries))
total = (time.monotonic() - start) * 1000
for q, ms, count in results:
print(f'{q}: {ms}ms ({count} results)')
print(f'Parallel total: {total:.0f}ms')
return results
parallel_search(queries)Étape 3: Ajouter un cache de résultats avec TTL
Mettez en cache les résultats de recherche par chaîne de requête avec une durée de vie pour éviter les appels API redondants pour les requêtes répétées.
import hashlib
cache = {}
CACHE_TTL = 300 # seconds
def cached_search(query: str, platform: str = 'google') -> dict:
key = hashlib.md5(f'{platform}:{query}'.encode()).hexdigest()
now = time.time()
if key in cache and now - cache[key]['ts'] < CACHE_TTL:
return cache[key]['data']
resp = SESSION.post('https://api.scavio.dev/api/v1/search',
json={'platform': platform, 'query': query}, timeout=10)
data = resp.json()
cache[key] = {'data': data, 'ts': now}
return data
# First call: network
start = time.monotonic()
cached_search('best crm 2026')
print(f'First call: {(time.monotonic() - start)*1000:.0f}ms')
# Second call: cache
start = time.monotonic()
cached_search('best crm 2026')
print(f'Cache hit: {(time.monotonic() - start)*1000:.0f}ms')Étape 4: Élaguer les charges utiles de réponse
Supprimez les champs inutiles des résultats de recherche avant de les transmettre au contexte du LLM pour réduire le temps de traitement des jetons.
def pruned_search(query: str) -> list:
data = cached_search(query)
results = data.get('organic_results', [])
return [{
'title': r.get('title', ''),
'snippet': r.get('snippet', '')[:200],
'url': r.get('link', ''),
} for r in results[:5]]
# Compare payload sizes:
import json
full = cached_search('best crm 2026')
pruned = pruned_search('best crm 2026')
print(f'Full response: {len(json.dumps(full))} chars')
print(f'Pruned response: {len(json.dumps(pruned))} chars')
print(f'Reduction: {100 - len(json.dumps(pruned)) * 100 // len(json.dumps(full))}%')Exemple Python
import requests, os, time, hashlib
from concurrent.futures import ThreadPoolExecutor
S = requests.Session()
S.headers.update({'x-api-key': os.environ['SCAVIO_API_KEY']})
cache = {}
def fast_search(query):
key = hashlib.md5(query.encode()).hexdigest()
if key in cache and time.time() - cache[key]['ts'] < 300:
return cache[key]['data']
data = S.post('https://api.scavio.dev/api/v1/search',
json={'platform': 'google', 'query': query}).json()
cache[key] = {'data': data, 'ts': time.time()}
return data
def parallel(queries):
with ThreadPoolExecutor(3) as pool:
return list(pool.map(fast_search, queries))Exemple JavaScript
const cache = new Map();
async function fastSearch(query) {
const key = query;
const cached = cache.get(key);
if (cached && Date.now() - cached.ts < 300000) return cached.data;
const r = 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 r.json();
cache.set(key, {data, ts: Date.now()});
return data;
}
async function parallel(queries) {
return Promise.all(queries.map(fastSearch));
}
parallel(['best crm 2026', 'react tutorial']).then(r => console.log(r.length + ' results'));Sortie attendue
Measurable latency reductions: parallel requests cut total time by 2-3x, caching eliminates repeated calls, and payload pruning reduces downstream token processing.