Los flujos de trabajo de agentes de varios pasos pasan resultados de búsqueda entre pasos, pero las respuestas API sin procesar desperdician tokens de ventana de contexto en campos irrelevantes. Un puente de contexto serializa los resultados de la búsqueda en el JSON estructurado mínimo que necesita cada paso del agente, lo que reduce el uso de tokens entre un 60 y un 80 % y al mismo tiempo preserva los datos importantes. Este tutorial crea un puente que comprime y formatea los resultados de búsqueda para el consumo de agentes.
Requisitos previos
- Python 3.8+
- solicita biblioteca
- Una clave API de Scavio de scavio.dev
- Un marco de agente (LangChain, CrewAI o personalizado)
Guia paso a paso
Paso 1: Definir estrategias de compresión de contexto
Cree niveles de compresión para diferentes tamaños de ventana de contexto de agente.
import os, requests, json
API_KEY = os.environ['SCAVIO_API_KEY']
SH = {'x-api-key': API_KEY, 'Content-Type': 'application/json'}
def compress_results(results, level='medium'):
"""Compress search results for agent context windows."""
if level == 'minimal':
# ~20 tokens per result (for 4K context models)
return [{'t': r.get('title', '')[:40], 'u': r.get('link', '').split('/')[2] if r.get('link') else ''}
for r in results[:3]]
elif level == 'medium':
# ~60 tokens per result (for 8K-16K context)
return [{'title': r.get('title', '')[:60], 'domain': r.get('link', '').split('/')[2] if r.get('link') else '',
'snippet': r.get('snippet', '')[:100]} for r in results[:5]]
else: # full
# ~120 tokens per result (for 32K+ context)
return [{'title': r.get('title', ''), 'link': r.get('link', ''),
'snippet': r.get('snippet', ''), 'position': r.get('position', 0)}
for r in results[:10]]
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=SH, json={'query': 'best serp api 2026', 'country_code': 'us'}).json()
results = data.get('organic_results', [])
for level in ['minimal', 'medium', 'full']:
compressed = compress_results(results, level)
tokens_est = len(json.dumps(compressed)) // 4
print(f'{level:8}: {len(compressed)} results, ~{tokens_est} tokens')Paso 2: Construya la clase puente de contexto
Cree un puente que administre el contexto de búsqueda entre los pasos del agente.
class ContextBridge:
def __init__(self, max_tokens=4000):
self.max_tokens = max_tokens
self.context = []
self.token_count = 0
self.searches = 0
self.cost = 0.0
def search(self, query, platform=None):
body = {'query': query, 'country_code': 'us'}
if platform: body['platform'] = platform
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=SH, json=body).json()
self.searches += 1
self.cost += 0.005
results = data.get('organic_results', [])
# Auto-select compression based on remaining budget
remaining = self.max_tokens - self.token_count
if remaining < 500: level = 'minimal'
elif remaining < 2000: level = 'medium'
else: level = 'full'
compressed = compress_results(results, level)
entry = {'query': query, 'platform': platform or 'google',
'results': compressed, 'level': level}
tokens = len(json.dumps(entry)) // 4
self.context.append(entry)
self.token_count += tokens
return entry
def get_context(self):
return json.dumps(self.context, indent=None)
def stats(self):
return f'{self.searches} searches, ~{self.token_count} tokens, ${self.cost:.3f}'
bridge = ContextBridge(max_tokens=4000)
bridge.search('best serp api 2026')
bridge.search('serp api user reviews', platform='reddit')
print(f'Stats: {bridge.stats()}')Paso 3: Agregar extracción de funciones SERP
Extraiga la descripción general de la IA y los datos de fragmentos destacados para el contexto del agente.
def extract_features(data):
"""Extract SERP features into agent-friendly format."""
features = {}
if data.get('ai_overview'):
ao = data['ai_overview']
features['ai_overview'] = {
'present': True,
'text': json.dumps(ao)[:200] if isinstance(ao, dict) else str(ao)[:200]
}
if data.get('answer_box'):
ab = data['answer_box']
features['answer_box'] = {
'title': ab.get('title', '')[:60],
'answer': ab.get('answer', ab.get('snippet', ''))[:150]
}
paa = data.get('related_questions', [])
if paa:
features['people_also_ask'] = [q.get('question', '')[:80] for q in paa[:4]]
return features
def search_with_features(query):
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=SH, json={'query': query, 'country_code': 'us', 'include_ai_overview': True}).json()
results = compress_results(data.get('organic_results', []), 'medium')
features = extract_features(data)
return {'query': query, 'results': results, 'features': features}
context = search_with_features('best python framework 2026')
print(json.dumps(context, indent=2)[:500])Paso 4: Serializar contexto para flujos de trabajo de varios pasos
Formatee el contexto acumulado para la transferencia entre los pasos del agente.
def format_for_agent(bridge, task_description):
"""Format accumulated search context for the next agent step."""
context_str = bridge.get_context()
prompt = f"""Based on the following search results gathered across {bridge.searches} searches:
{context_str}
Task: {task_description}
Provide your analysis based solely on the search data above. Note any gaps."""
token_est = len(prompt) // 4
print(f'Context prompt: ~{token_est} tokens ({bridge.stats()})')
return prompt
# Example multi-step workflow
bridge = ContextBridge(max_tokens=6000)
bridge.search('serp api pricing comparison 2026')
bridge.search('serp api developer reviews', platform='reddit')
bridge.search('serp api python tutorial', platform='youtube')
prompt = format_for_agent(bridge,
'Compare the top 3 SERP APIs by pricing, developer experience, and community sentiment.')
print(f'\nPrompt preview:\n{prompt[:300]}...')
print(f'\nTotal search cost: ${bridge.cost:.3f}')Ejemplo en Python
import os, requests, json
SH = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
def bridge_search(query, max_results=3):
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=SH, json={'query': query, 'country_code': 'us'}).json()
compressed = [{'title': r['title'][:50], 'snippet': r.get('snippet', '')[:80]}
for r in data.get('organic_results', [])[:max_results]]
tokens = len(json.dumps(compressed)) // 4
print(f'{query}: {len(compressed)} results, ~{tokens} tokens. Cost: $0.005')
return compressed
bridge_search('best serp api 2026')Ejemplo en JavaScript
const SH = { 'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json' };
async function bridgeSearch(query, maxResults = 3) {
const data = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: SH,
body: JSON.stringify({ query, country_code: 'us' })
}).then(r => r.json());
const compressed = (data.organic_results || []).slice(0, maxResults)
.map(r => ({ title: r.title.slice(0, 50), snippet: (r.snippet || '').slice(0, 80) }));
const tokens = Math.ceil(JSON.stringify(compressed).length / 4);
console.log(`${query}: ${compressed.length} results, ~${tokens} tokens`);
return compressed;
}
await bridgeSearch('best serp api 2026');Salida esperada
minimal : 3 results, ~45 tokens
medium : 5 results, ~190 tokens
full : 10 results, ~520 tokens
Stats: 2 searches, ~380 tokens, $0.010
Context prompt: ~450 tokens (3 searches, ~520 tokens, $0.015)
Prompt preview:
Based on the following search results gathered across 3 searches:
[{"query": "serp api pricing comparison 2026", "platform": "google", "results": [...
Total search cost: $0.015