依赖于单个搜索后端的生产人工智能代理在中断期间面临着完全的输出降级。故障转移链按优先级顺序通过多个搜索平台路由查询，当主数据库返回错误或空结果时自动回退。本教程使用 Scavio 的多平台 API 构建故障转移链，其中单个密钥涵盖 Google、Reddit、YouTube、Amazon 和 Walmart。通过添加辅助提供商作为最终后备，相同的模式适用于供应商级冗余。
前置条件
- 安装了 Python 3.8+ 或 Node.js 18+
- requests 库 (Python) 或内置 fetch (JS)
- 来自 scavio.dev 的 Scavio API 密钥
- 对 try-catch 错误处理的基本了解
操作指南
步骤 1: 定义您的平台优先级
按照您的用例的偏好顺序列出搜索平台。谷歌涵盖了大多数查询； Reddit 和 YouTube 提供深度讨论和视频主题。
Python
PLATFORMS = ['google', 'reddit', 'youtube']
API_KEY = os.environ['SCAVIO_API_KEY']步骤 2: 构建故障转移功能
迭代平台，从第一个成功且非空结果的平台返回结果。
Python
import requests, os
def failover_search(query: str, min_results: int = 1) -> dict:
for platform in PLATFORMS:
try:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json={'platform': platform, 'query': query}, timeout=10)
resp.raise_for_status()
data = resp.json()
results = data.get('organic', [])
if len(results) >= min_results:
return {'platform': platform, 'results': results}
except (requests.RequestException, ValueError):
continue
return {'platform': 'none', 'results': []}步骤 3: 添加日志记录以提高可观察性
记录哪个平台为每个查询提供服务，以便您可以跟踪故障转移频率并识别降级的后端。
Python
import logging
logger = logging.getLogger('search_failover')
def logged_failover_search(query: str) -> dict:
for platform in PLATFORMS:
try:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json={'platform': platform, 'query': query}, timeout=10)
data = resp.json()
results = data.get('organic', [])
if results:
logger.info(f'Query "{query}" served by {platform} ({len(results)} results)')
return {'platform': platform, 'results': results}
logger.warning(f'{platform} returned 0 results for "{query}"')
except Exception as e:
logger.error(f'{platform} failed for "{query}": {e}')
logger.error(f'All platforms failed for "{query}"')
return {'platform': 'none', 'results': []}步骤 4: 标准化输出
确保下游 LLM 获得相同的格式，无论回答哪个平台。
Python
def normalize_result(result: dict, platform: str) -> dict:
return {
'title': result.get('title', ''),
'snippet': result.get('snippet', result.get('description', '')),
'url': result.get('link', result.get('url', '')),
'source_platform': platform
}
def search_for_agent(query: str) -> list:
data = logged_failover_search(query)
return [normalize_result(r, data['platform']) for r in data['results'][:5]]Python 示例
Python
import requests, os
PLATFORMS = ['google', 'reddit', 'youtube']
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def failover_search(query):
for p in PLATFORMS:
try:
r = requests.post('https://api.scavio.dev/api/v1/search',
headers=H, json={'platform': p, 'query': query}, timeout=10)
results = r.json().get('organic', [])
if results: return {'platform': p, 'results': results[:5]}
except: continue
return {'platform': 'none', 'results': []}
print(failover_search('best crm for startups'))JavaScript 示例
JavaScript
const PLATFORMS = ['google', 'reddit', 'youtube'];
async function failoverSearch(query) {
for (const p of PLATFORMS) {
try {
const r = 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: p, query})
});
const data = await r.json();
if (data.organic?.length) return {platform: p, results: data.organic.slice(0, 5)};
} catch { continue; }
}
return {platform: 'none', results: []};
}预期输出
JSON
A failover search function that automatically routes through Google, Reddit, and YouTube in order, returning normalized results from the first platform that succeeds.