Agregar capacidad de búsqueda compartida a un sistema de múltiples agentes permite que cada agente base su razonamiento en datos web en vivo sin duplicar consultas ni gastar créditos API. El patrón principal es una envoltura delgada alrededor de una API de búsqueda (como Scavio) combinada con un caché en memoria o respaldado por Redis codificado por hash de consulta. Cada agente llama a la herramienta compartida y el caché garantiza que consultas idénticas entre agentes regresen instantáneamente desde el almacén local. Este tutorial crea la herramienta de búsqueda compartida, la conecta a CrewAI como ejemplo y muestra cómo adaptarla para AutoGen o un orquestador personalizado.
Requisitos previos
- Python 3.10+ instalado
- Crewai y bibliotecas de solicitudes instaladas (solicitudes de instalación de pip Crewai)
- Una clave API de Scavio de scavio.dev
- Familiaridad básica con marcos multiagente
Guia paso a paso
Paso 1: Cree la herramienta de búsqueda en caché
Cree una función de búsqueda con un caché en memoria codificado por el hash SHA-256 de la consulta y el país. Cualquier agente que llame a esta función con la misma consulta obtiene el resultado almacenado en caché, lo que ahorra créditos y latencia.
Python
import requests, os, hashlib, json
API_KEY = os.environ.get('SCAVIO_API_KEY', 'your_scavio_api_key')
ENDPOINT = 'https://api.scavio.dev/api/v1/search'
_cache = {}
def cached_search(query: str, country: str = 'us') -> dict:
key = hashlib.sha256(f'{query}:{country}'.encode()).hexdigest()
if key in _cache:
print(f'[cache hit] {query}')
return _cache[key]
resp = requests.post(ENDPOINT,
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': country})
resp.raise_for_status()
data = resp.json()
_cache[key] = data
print(f'[cache miss] {query} -> {len(data.get("organic_results", []))} results')
return dataPaso 2: Envolver como herramienta CrewAI
Las herramientas de CrewAI son funciones decoradas con un nombre y una descripción. Ajuste la función de búsqueda en caché para que cualquier agente de CrewAI pueda llamarla.
Python
from crewai.tools import tool
@tool('web_search')
def web_search_tool(query: str) -> str:
'''Search the web for a query and return top 5 results with titles, links, and snippets.'''
data = cached_search(query)
results = data.get('organic_results', [])[:5]
lines = []
for r in results:
lines.append(f"Title: {r['title']}")
lines.append(f"URL: {r['link']}")
lines.append(f"Snippet: {r.get('snippet', '')}")
lines.append('')
return '\n'.join(lines)Paso 3: Asignar la herramienta a múltiples agentes
Crea dos agentes que compartan la misma herramienta de búsqueda. Cuando ambos agentes investigan temas superpuestos, el caché evita llamadas API duplicadas.
Python
from crewai import Agent, Task, Crew
researcher = Agent(
role='Market Researcher',
goal='Find the top 3 CRM tools by market share in 2026',
backstory='Expert in SaaS market analysis',
tools=[web_search_tool],
verbose=True
)
analyst = Agent(
role='Pricing Analyst',
goal='Compare pricing tiers of the top CRM tools',
backstory='Expert in SaaS pricing strategy',
tools=[web_search_tool],
verbose=True
)
task1 = Task(description='Research top CRM tools 2026', agent=researcher,
expected_output='List of top 3 CRM tools with market share')
task2 = Task(description='Compare CRM pricing tiers', agent=analyst,
expected_output='Pricing comparison table')
crew = Crew(agents=[researcher, analyst], tasks=[task1, task2], verbose=True)
result = crew.kickoff()
print(f'Cache size: {len(_cache)} queries (saved credits on duplicates)')Paso 4: Adaptarse para AutoGen o agentes personalizados
La misma función cached_search funciona con cualquier orquestador. Para AutoGen, regístrelo como una función. Para agentes personalizados, llámelo directamente. El caché se comparte entre todas las personas que llaman en el mismo proceso.
Python
# AutoGen example
from autogen import register_function
def autogen_search(query: str) -> str:
data = cached_search(query)
return json.dumps(data.get('organic_results', [])[:5], indent=2)
# Register for any AutoGen agent
# register_function(autogen_search, caller=assistant, executor=user_proxy,
# name='web_search', description='Search the web')
# Custom agent loop example
def agent_step(agent_name: str, query: str):
results = cached_search(query)
organic = results.get('organic_results', [])[:5]
print(f'[{agent_name}] {query} -> {len(organic)} results')
return organic
# Both agents share the cache
agent_step('researcher', 'best crm tools 2026')
agent_step('analyst', 'best crm tools 2026') # cache hitEjemplo en Python
Python
import requests, os, hashlib, json
from crewai import Agent, Task, Crew
from crewai.tools import tool
API_KEY = os.environ.get('SCAVIO_API_KEY', 'your_scavio_api_key')
ENDPOINT = 'https://api.scavio.dev/api/v1/search'
_cache = {}
def cached_search(query: str, country: str = 'us') -> dict:
key = hashlib.sha256(f'{query}:{country}'.encode()).hexdigest()
if key in _cache:
return _cache[key]
resp = requests.post(ENDPOINT,
headers={'x-api-key': API_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': country})
resp.raise_for_status()
_cache[key] = resp.json()
return _cache[key]
@tool('web_search')
def web_search_tool(query: str) -> str:
'''Search the web and return top 5 results.'''
data = cached_search(query)
results = data.get('organic_results', [])[:5]
return '\n'.join(f"{r['title']}: {r['link']}" for r in results)
researcher = Agent(role='Researcher', goal='Find top CRM tools',
backstory='Market analyst', tools=[web_search_tool])
analyst = Agent(role='Analyst', goal='Compare CRM pricing',
backstory='Pricing expert', tools=[web_search_tool])
task1 = Task(description='Research CRM tools 2026', agent=researcher,
expected_output='Top 3 CRM tools')
task2 = Task(description='Compare pricing', agent=analyst,
expected_output='Pricing table')
crew = Crew(agents=[researcher, analyst], tasks=[task1, task2])
result = crew.kickoff()
print(f'Unique queries: {len(_cache)}')
print(result)Ejemplo en JavaScript
JavaScript
const API_KEY = process.env.SCAVIO_API_KEY || 'your_scavio_api_key';
const ENDPOINT = 'https://api.scavio.dev/api/v1/search';
const cache = new Map();
async function cachedSearch(query, country = 'us') {
const key = `${query}:${country}`;
if (cache.has(key)) return cache.get(key);
const resp = await fetch(ENDPOINT, {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: country })
});
if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
const data = await resp.json();
cache.set(key, data);
return data;
}
async function agentSearch(agentName, query) {
const data = await cachedSearch(query);
const results = (data.organic_results || []).slice(0, 5);
console.log(`[${agentName}] ${query} -> ${results.length} results (cache size: ${cache.size})`);
return results;
}
async function main() {
await agentSearch('researcher', 'best crm tools 2026');
await agentSearch('analyst', 'best crm tools 2026');
await agentSearch('analyst', 'crm pricing comparison 2026');
console.log(`Total unique queries: ${cache.size}`);
}
main().catch(console.error);Salida esperada
JSON
[cache miss] best crm tools 2026 -> 10 results
[cache hit] best crm tools 2026
[cache miss] crm pricing comparison 2026 -> 10 results
Total unique queries: 2
Researcher found: Salesforce, HubSpot, Pipedrive
Analyst pricing table:
Salesforce: $25-$300/user/mo
HubSpot: $0-$150/user/mo
Pipedrive: $14-$99/user/mo