Ajouter une capacité de recherche partagée à un système multi-agents permet à chaque agent d'ancrer son raisonnement sur des données web en direct sans dupliquer les requêtes ni épuiser les crédits API. Le modèle de base est une fine couche autour d'une API de recherche (comme Scavio) combinée à un cache en mémoire ou basé sur Redis, indexé par le hash de la requête. Chaque agent appelle l'outil partagé, et le cache garantit que les requêtes identiques entre agents retournent instantanément depuis le stockage local. Ce tutoriel construit l'outil de recherche partagé, l'intègre dans CrewAI comme exemple, et montre comment l'adapter pour AutoGen ou un orchestrateur personnalisé.
Prérequis
- Python 3.10+ installé
- bibliothèques crewai et requests installées (pip install crewai requests)
- Une clé API Scavio depuis scavio.dev
- Familiarité de base avec les frameworks multi-agents
Parcours
Étape 1: Construire l'outil de recherche en cache
Créez une fonction de recherche avec un cache en mémoire indexé par le hash SHA-256 de la requête et du pays. Tout agent appelant cette fonction avec la même requête obtient le résultat mis en cache, économisant des crédits et de la latence.
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 dataÉtape 2: Envelopper en tant qu'outil CrewAI
Les outils CrewAI sont des fonctions décorées avec un nom et une description. Enveloppez la fonction de recherche en cache pour que tout agent CrewAI puisse l'appeler.
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)Étape 3: Attribuer l'outil à plusieurs agents
Créez deux agents qui partagent le même outil de recherche. Lorsque les deux agents explorent des sujets qui se chevauchent, le cache empêche les appels API en double.
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)')Étape 4: Adapter pour AutoGen ou des agents personnalisés
La même fonction cached_search fonctionne avec n'importe quel orchestrateur. Pour AutoGen, enregistrez-la en tant que fonction. Pour des agents personnalisés, appelez-la directement. Le cache est partagé entre tous les appelants dans le même processus.
# 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 hitExemple 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)Exemple 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);Sortie attendue
[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