r/PiCodingAgent 帖子询问了如何可靠地搜索编码代理。问题是：单一提供商搜索会默默失败。本教程构建了一个路由器，该路由器按优先级顺序尝试多个提供程序并自动回退。适用于任何代理运行时，而不仅仅是 Pi。
前置条件
- Scavio API 密钥
- Python 3.8+
- 可选：Tavily API 密钥、SearxNG 实例
操作指南
步骤 1: 定义提供者配置
列出具有端点和优先级的提供者。
Python
providers = [
{
'name': 'scavio',
'url': 'https://api.scavio.dev/api/v1/search',
'headers': lambda: {'x-api-key': os.environ['SCAVIO_API_KEY']},
'body': lambda q: {'platform': 'google', 'query': q},
'parse': lambda r: r.get('organic_results', []),
'cost': 0.005,
'priority': 1
},
{
'name': 'searxng',
'url': 'http://localhost:8080/search',
'headers': lambda: {},
'body': lambda q: {'q': q, 'format': 'json'},
'parse': lambda r: r.get('results', []),
'cost': 0,
'priority': 2
},
{
'name': 'tavily',
'url': 'https://api.tavily.com/search',
'headers': lambda: {},
'body': lambda q: {'api_key': os.environ.get('TAVILY_API_KEY', ''), 'query': q},
'parse': lambda r: r.get('results', []),
'cost': 0.008,
'priority': 3
},
]步骤 2: 构建具有后备功能的路由器
按优先顺序尝试每个提供程序，失败时再回退。
Python
import requests, os, time
def search_with_fallback(query, providers_list=providers):
sorted_providers = sorted(providers_list, key=lambda p: p['priority'])
errors = []
for provider in sorted_providers:
try:
resp = requests.post(provider['url'],
headers=provider['headers'](),
json=provider['body'](query),
timeout=10)
resp.raise_for_status()
results = provider['parse'](resp.json())
if results:
return {'provider': provider['name'], 'results': results, 'cost': provider['cost']}
errors.append(f"{provider['name']}: empty results")
except Exception as e:
errors.append(f"{provider['name']}: {str(e)}")
return {'provider': None, 'results': [], 'errors': errors}步骤 3: 添加延迟感知路由
跟踪响应时间并选择更快的提供商。
Python
provider_stats = {}
def update_stats(provider_name, latency, success):
if provider_name not in provider_stats:
provider_stats[provider_name] = {'total': 0, 'success': 0, 'avg_latency': 0}
stats = provider_stats[provider_name]
stats['total'] += 1
if success: stats['success'] += 1
stats['avg_latency'] = (stats['avg_latency'] * (stats['total'] - 1) + latency) / stats['total']
def smart_search(query):
start = time.time()
result = search_with_fallback(query)
latency = time.time() - start
if result['provider']:
update_stats(result['provider'], latency, True)
return result步骤 4: 日志提供商使用成本跟踪
跟踪哪个提供商为每个查询提供服务。
Python
import json, datetime
def log_search(query, result):
with open('search_router_log.jsonl', 'a') as f:
f.write(json.dumps({
'ts': datetime.datetime.now().isoformat(),
'query': query,
'provider': result['provider'],
'cost': result.get('cost', 0),
'result_count': len(result.get('results', []))
}) + '\n')Python 示例
Python
import os, requests
def search(query):
for provider in ['scavio', 'searxng', 'tavily']:
try:
if provider == 'scavio':
r = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': os.environ['SCAVIO_API_KEY']},
json={'platform': 'google', 'query': query}, timeout=10)
results = r.json().get('organic_results', [])
if results: return {'provider': 'scavio', 'results': results}
except: continue
return {'provider': None, 'results': []}JavaScript 示例
JavaScript
async function search(query) {
try {
const res = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'},
body: JSON.stringify({platform: 'google', query})
});
const data = await res.json();
if (data.organic_results?.length) return {provider: 'scavio', results: data.organic_results};
} catch(e) { /* fallback */ }
}预期输出
JSON
Multi-provider search router with automatic fallback. Priority: Scavio ($0.005) > SearxNG (free/self-hosted) > Tavily ($0.008). Latency tracking and JSONL cost log.