Los agentes de codificación Pi necesitan una búsqueda web confiable para buscar documentación y verificar soluciones. La caída de un único proveedor de búsqueda arruina al agente. Este tutorial configura Pi Agent con Scavio como motor de búsqueda principal y agrega proveedores alternativos para mayor confiabilidad. El agente cambia automáticamente de proveedor en caso de falla.
Requisitos previos
- Python 3.8+
- solicita biblioteca
- Una clave API de Scavio de scavio.dev
- Agente Pi o agente de codificación similar
Guia paso a paso
Paso 1: Cree el cliente de búsqueda multibackend
Cree un cliente de búsqueda que enrute consultas a través de múltiples proveedores en orden de prioridad.
Python
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]}')Paso 2: Integrar con el sistema de herramientas Pi Agent
Registre el cliente de búsqueda múltiple como una herramienta a la que el Agente Pi puede llamar.
Python
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'))Paso 3: Agregar búsqueda específica de plataforma para tareas de codificación
Amplíe la herramienta para buscar plataformas específicas como Reddit y YouTube para obtener ayuda con la codificación.
Python
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')Ejemplo en Python
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]}')Ejemplo en JavaScript
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));Salida esperada
JSON
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)