Reduzca la latencia de la API de búsqueda en agentes de IA aplicando cuatro técnicas: solicitudes paralelas para flujos de trabajo de consultas múltiples, almacenamiento en caché de resultados para consultas repetidas, poda de consultas para reducir el tamaño de la carga útil y agrupación de conexiones para eliminar la sobrecarga del protocolo de enlace. En los flujos de trabajo típicos de los agentes, las llamadas de búsqueda representan entre el 60% y el 80% del tiempo total de respuesta. Incluso las pequeñas reducciones de latencia se acumulan en cadenas de razonamiento de varios pasos. Este tutorial implementa cada optimización con la API de Scavio y mide el impacto antes/después.
Requisitos previos
- Python 3.8+ instalado
- solicita biblioteca instalada
- Una clave API de Scavio de scavio.dev
- Un flujo de trabajo de agente existente con llamadas de búsqueda
Guia paso a paso
Paso 1: Medir la latencia inicial
Establezca una línea de base cronometrando llamadas de búsqueda secuenciales para que pueda medir el impacto de cada optimización.
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')Paso 2: Paralelizar solicitudes de consultas múltiples
Utilice un grupo de subprocesos para enviar múltiples solicitudes de búsqueda simultáneamente, reduciendo el tiempo total del reloj de pared entre 2 y 3 veces.
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)Paso 3: Agregar un caché de resultados con TTL
Almacene en caché los resultados de la búsqueda por cadena de consulta con un tiempo de vida para evitar llamadas API redundantes para consultas repetidas.
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')Paso 4: Podar cargas de respuesta
Elimine los campos innecesarios de los resultados de búsqueda antes de pasarlos al contexto LLM para reducir el tiempo de procesamiento del token.
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))}%')Ejemplo en 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))Ejemplo en 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'));Salida esperada
Measurable latency reductions: parallel requests cut total time by 2-3x, caching eliminates repeated calls, and payload pruning reduces downstream token processing.