Les agents de codage Pi ont besoin d'une recherche web fiable pour consulter la documentation et vérifier les solutions. Un seul fournisseur de recherche qui tombe en panne bloque l'agent. Ce tutoriel configure l'agent Pi avec Scavio comme backend de recherche principal et ajoute des fournisseurs de secours pour la fiabilité. L'agent bascule automatiquement entre les fournisseurs en cas d'échec.
Prérequis
- Python 3.8+
- bibliothèque requests
- Une clé API Scavio depuis scavio.dev
- Agent Pi ou agent de codage similaire
Parcours
Étape 1: Construire le client de recherche multi-backend
Créez un client de recherche qui achemine les requêtes via plusieurs fournisseurs par ordre de priorité.
import os, requests, time, json
class MultiSearchClient:
def __init__(self):
self.providers = [
{
'name': 'scavio',
'fn': self._search_scavio,
'failures': 0,
'last_fail': 0,
},
]
self.scavio_key = os.environ.get('SCAVIO_API_KEY', '')
def _search_scavio(self, query):
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': self.scavio_key, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us'}, timeout=10)
resp.raise_for_status()
data = resp.json()
return [{'title': r.get('title', ''), 'url': r.get('link', ''),
'snippet': r.get('snippet', '')} for r in data.get('organic_results', [])]
def search(self, query, num_results=5):
for provider in self.providers:
if provider['failures'] >= 3 and time.time() - provider['last_fail'] < 60:
print(f' Skipping {provider["name"]} (cooldown)')
continue
try:
results = provider['fn'](query)[:num_results]
provider['failures'] = 0
return {'provider': provider['name'], 'results': results, 'count': len(results)}
except Exception as e:
provider['failures'] += 1
provider['last_fail'] = time.time()
print(f' {provider["name"]} failed: {str(e)[:40]}')
return {'provider': 'none', 'results': [], 'count': 0}
client = MultiSearchClient()
result = client.search('python FastAPI async tutorial')
print(f'Provider: {result["provider"]} | Results: {result["count"]}')
for r in result['results'][:3]:
print(f' {r["title"][:50]}')Étape 2: Intégrer avec le système d'outils de l'agent Pi
Enregistrez le client multi-recherche comme un outil que l'agent Pi peut appeler.
client = MultiSearchClient()
PI_SEARCH_TOOL = {
'name': 'web_search',
'description': 'Search the web with automatic provider failover. Use for docs, APIs, errors, or any question needing current data.',
'parameters': {
'type': 'object',
'properties': {
'query': {'type': 'string', 'description': 'Search query'},
'num_results': {'type': 'integer', 'default': 5}
},
'required': ['query']
}
}
def pi_search(query, num_results=5):
"""Pi Agent search tool with multi-provider failover."""
result = client.search(query, num_results)
# Format for agent consumption
formatted = f'Search results for "{query}" (via {result["provider"]}):\n'
for i, r in enumerate(result['results'], 1):
formatted += f'{i}. {r["title"]}\n URL: {r["url"]}\n {r["snippet"]}\n\n'
return formatted
# Test agent-formatted output
print(pi_search('how to handle database migrations in FastAPI'))Étape 3: Ajouter une recherche spécifique à la plateforme pour les tâches de codage
Étendez l'outil pour rechercher sur des plateformes spécifiques comme Reddit et YouTube pour obtenir de l'aide en codage.
def pi_search_platform(query, platform=None, num_results=5):
"""Search specific platforms for more targeted results."""
body = {'query': query, 'country_code': 'us'}
if platform:
body['platform'] = platform
try:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': client.scavio_key, 'Content-Type': 'application/json'},
json=body, timeout=10)
resp.raise_for_status()
data = resp.json()
return [{'title': r.get('title', ''), 'url': r.get('link', ''),
'snippet': r.get('snippet', '')} for r in data.get('organic_results', [])[:num_results]]
except Exception:
return []
# Pi Agent can now search specific platforms
print('=== Google (docs) ===')
for r in pi_search_platform('FastAPI async SQLAlchemy')[:2]:
print(f' {r["title"][:50]}')
print('\n=== Reddit (community answers) ===')
for r in pi_search_platform('FastAPI async SQLAlchemy', platform='reddit')[:2]:
print(f' {r["title"][:50]}')
print('\n=== YouTube (video tutorials) ===')
for r in pi_search_platform('FastAPI async SQLAlchemy', platform='youtube')[:2]:
print(f' {r["title"][:50]}')
print(f'\nTotal cost: $0.015 (3 platform searches)')
print(f'Pi Agent now has: web, reddit, youtube search')Exemple Python
import os, requests
SH = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
def pi_search(query):
data = requests.post('https://api.scavio.dev/api/v1/search',
headers=SH, json={'query': query, 'country_code': 'us'}, timeout=10).json()
return [{'title': r['title'], 'url': r['link']} for r in data.get('organic_results', [])[:5]]
for r in pi_search('FastAPI tutorial 2026'):
print(f'{r["title"][:50]}')Exemple JavaScript
const SH = { 'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json' };
const data = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: SH,
body: JSON.stringify({ query: 'FastAPI tutorial 2026', country_code: 'us' })
}).then(r => r.json());
(data.organic_results || []).slice(0, 5).forEach(r => console.log(r.title));Sortie attendue
Provider: scavio | Results: 5
FastAPI with Async SQLAlchemy - Complete Guide
Python Async Database Tutorial 2026
Search results for "how to handle database migrations in FastAPI" (via scavio):
1. Alembic Migrations with FastAPI - Official Docs
URL: https://fastapi.tiangolo.com/...
Learn how to set up Alembic for database migrations...
=== Google (docs) ===
FastAPI Async SQLAlchemy - Official Documentation
=== Reddit (community answers) ===
r/FastAPI - Best practices for async SQLAlchemy
=== YouTube (video tutorials) ===
FastAPI + Async SQLAlchemy Full Tutorial 2026
Total cost: $0.015 (3 platform searches)