依赖于单一搜索提供商的人工智能代理是脆弱的。达到速率限制、Cloudflare 阻止触发并发生中断。后备链按优先级顺序尝试多个提供者，以便您的代理始终获取数据。本教程构建了一个以 Scavio 作为主要、可配置重试、延迟跟踪和自动提供程序轮换的生产后备链。每个 Scavio 请求的费用为每积分 0.005 美元，一个端点有 6 个可用平台。
前置条件
- 已安装 Python 3.9+
- 请求已安装库
- 来自 scavio.dev 的 Scavio API 密钥
- 可选：备份提供商密钥以实现完整的后备覆盖
操作指南
步骤 1: 定义提供者接口
创建每个搜索提供程序都必须实现的基类。这确保了统一的结果格式，无论哪个提供商响应。
Python
import os, time, requests
from dataclasses import dataclass
from typing import Optional
@dataclass
class SearchResult:
title: str
url: str
snippet: str
provider: str
class SearchProvider:
name: str
def search(self, query: str, num: int = 5) -> list[SearchResult]:
raise NotImplementedError
class ScavioProvider(SearchProvider):
name = 'scavio'
def __init__(self):
self.key = os.environ['SCAVIO_API_KEY']
def search(self, query: str, num: int = 5) -> list[SearchResult]:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': self.key, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us', 'num_results': num},
timeout=10)
resp.raise_for_status()
return [SearchResult(title=r['title'], url=r['link'],
snippet=r.get('snippet', ''), provider='scavio')
for r in resp.json().get('organic_results', [])]
print('Provider interface ready')步骤 2: 构建后备链
该链按顺序尝试每个提供商。如果其中一个失败或返回空结果，则会移至下一个。它跟踪哪个提供商成功了以及延迟。
Python
class FallbackChain:
def __init__(self, providers: list[SearchProvider]):
self.providers = providers
self.stats = {p.name: {'success': 0, 'fail': 0, 'total_ms': 0} for p in providers}
def search(self, query: str, num: int = 5) -> tuple[list[SearchResult], str]:
for provider in self.providers:
try:
start = time.time()
results = provider.search(query, num)
elapsed = (time.time() - start) * 1000
if results:
self.stats[provider.name]['success'] += 1
self.stats[provider.name]['total_ms'] += elapsed
return results, provider.name
except Exception as e:
self.stats[provider.name]['fail'] += 1
print(f'[fallback] {provider.name} failed: {e}')
continue
return [], 'none'
def report(self) -> str:
lines = ['Provider Stats:']
for name, s in self.stats.items():
total = s['success'] + s['fail']
avg_ms = s['total_ms'] / s['success'] if s['success'] else 0
lines.append(f" {name}: {s['success']}/{total} ok, {avg_ms:.0f}ms avg")
return '\n'.join(lines)
chain = FallbackChain([ScavioProvider()])
results, used = chain.search('python web framework 2026')
print(f'Got {len(results)} results from {used}')步骤 3: 添加具有指数退避的重试逻辑
使用重试逻辑包装每个提供程序调用。在转向下一个提供者之前，应重试超时等暂时性故障。
Python
class RetryProvider:
def __init__(self, provider: SearchProvider, max_retries: int = 2, base_delay: float = 0.5):
self.provider = provider
self.name = provider.name
self.max_retries = max_retries
self.base_delay = base_delay
def search(self, query: str, num: int = 5) -> list[SearchResult]:
last_error = None
for attempt in range(self.max_retries + 1):
try:
return self.provider.search(query, num)
except requests.exceptions.Timeout:
last_error = 'timeout'
delay = self.base_delay * (2 ** attempt)
print(f'[retry] {self.name} timeout, waiting {delay:.1f}s')
time.sleep(delay)
except requests.exceptions.HTTPError as e:
if e.response and e.response.status_code == 429:
delay = self.base_delay * (2 ** attempt)
print(f'[retry] {self.name} rate limited, waiting {delay:.1f}s')
time.sleep(delay)
last_error = 'rate_limit'
else:
raise
raise Exception(f'{self.name} failed after {self.max_retries} retries: {last_error}')
chain = FallbackChain([RetryProvider(ScavioProvider())])
results, used = chain.search('AI agent frameworks')
print(f'{len(results)} results via {used}')步骤 4: 添加健康检查和自动轮换
跟踪提供者的健康状况并自动降低不健康提供者的优先级。定期重新检查它们，以防它们恢复。
Python
class HealthAwareFallback(FallbackChain):
def __init__(self, providers, unhealthy_threshold=3, recheck_interval=60):
super().__init__(providers)
self.unhealthy_threshold = unhealthy_threshold
self.recheck_interval = recheck_interval
self.consecutive_fails = {p.name: 0 for p in providers}
self.last_recheck = {p.name: 0 for p in providers}
def search(self, query: str, num: int = 5) -> tuple[list[SearchResult], str]:
now = time.time()
for provider in self.providers:
name = provider.name if hasattr(provider, 'name') else str(provider)
# Skip unhealthy providers unless recheck interval passed
if self.consecutive_fails.get(name, 0) >= self.unhealthy_threshold:
if now - self.last_recheck.get(name, 0) < self.recheck_interval:
continue
self.last_recheck[name] = now
print(f'[health] rechecking {name}')
try:
start = time.time()
results = provider.search(query, num)
elapsed = (time.time() - start) * 1000
if results:
self.consecutive_fails[name] = 0
self.stats[name]['success'] += 1
self.stats[name]['total_ms'] += elapsed
return results, name
except Exception as e:
self.consecutive_fails[name] = self.consecutive_fails.get(name, 0) + 1
self.stats[name]['fail'] += 1
print(f'[health] {name} fail #{self.consecutive_fails[name]}: {e}')
return [], 'none'
chain = HealthAwareFallback([RetryProvider(ScavioProvider())])
results, used = chain.search('search api comparison')
print(f'{len(results)} results via {used}')
print(chain.report())Python 示例
Python
import os, time, requests
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
def scavio_search(query, num=5):
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us', 'num_results': num}, timeout=10)
resp.raise_for_status()
return [{'title': r['title'], 'url': r['link'], 'snippet': r.get('snippet', '')}
for r in resp.json().get('organic_results', [])]
def search_with_fallback(query, num=5, retries=2):
for attempt in range(retries + 1):
try:
return scavio_search(query, num)
except Exception as e:
if attempt < retries:
time.sleep(0.5 * (2 ** attempt))
else:
raise
results = search_with_fallback('AI agent search tools 2026')
for r in results:
print(f"{r['title']}\n {r['url']}")JavaScript 示例
JavaScript
const SCAVIO_KEY = process.env.SCAVIO_API_KEY;
async function search(query, num = 5, retries = 2) {
for (let i = 0; i <= retries; i++) {
try {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: 'us', num_results: num }),
signal: AbortSignal.timeout(10000)
});
if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
const data = await resp.json();
return (data.organic_results || []).map(r => ({ title: r.title, url: r.link, snippet: r.snippet || '' }));
} catch (e) {
if (i < retries) await new Promise(r => setTimeout(r, 500 * 2 ** i));
else throw e;
}
}
}
search('AI agent search tools 2026').then(r => r.forEach(x => console.log(x.title)));预期输出
JSON
Got 5 results from scavio
Provider Stats:
scavio: 1/1 ok, 320ms avg
AI Agent Frameworks Comparison 2026
https://example.com/ai-agent-frameworks
Building Reliable AI Agents with Search
https://blog.example.com/reliable-agents