通过将基础查询路由到搜索 API 作为后备,处理 Gemini 429 速率限制和 503 服务不可用错误,然后在 Gemini 再次可用时自动恢复。 Gemini 的接地功能很有用,但在负载下并不可靠:速率限制达到不可预测的阈值,503 可能会持续几分钟。搜索 API 回退可确保您的应用程序永远不会失去基础功能。本教程将 Gemini 调用包装在错误检测中,将故障路由到 Scavio,并实现一个运行状况检查循环,以便在 Gemini 恢复时恢复它。
前置条件
- 已安装 Python 3.8+
- 请求已安装库
- 来自 scavio.dev 的 Scavio API 密钥
- Google Gemini API 访问权限(或任何具有基础的法学硕士)
操作指南
步骤 1: 结束双子座接地电话
围绕 Gemini API 调用创建一个包装器,捕获 429 和 503 错误而不是崩溃。
import os, requests, time
GEMINI_KEY = os.environ.get('GEMINI_API_KEY', '')
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
gemini_healthy = True
gemini_last_check = 0
def gemini_ground(query: str) -> dict:
"""Call Gemini with grounding. Returns results or raises on failure."""
try:
resp = requests.post(
f'https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:generateContent?key={GEMINI_KEY}',
json={'contents': [{'parts': [{'text': query}]}]},
timeout=15)
if resp.status_code in (429, 503):
raise requests.exceptions.HTTPError(f'Gemini {resp.status_code}')
resp.raise_for_status()
return {'source': 'gemini', 'data': resp.json()}
except requests.exceptions.RequestException as e:
return {'source': 'gemini', 'error': str(e)}
print('Gemini wrapper ready')步骤 2: 检测错误并路由至 Scavio
当 Gemini 返回错误时,自动将基础查询路由到 Scavio 的搜索 API。
def scavio_ground(query: str) -> dict:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY},
json={'platform': 'google', 'query': query}, timeout=10)
resp.raise_for_status()
results = resp.json().get('organic_results', [])[:5]
context = '\n'.join(f"{r['title']}: {r.get('snippet', '')}" for r in results)
return {'source': 'scavio', 'context': context, 'results': results}
def grounded_query(query: str) -> dict:
global gemini_healthy
if gemini_healthy:
result = gemini_ground(query)
if 'error' not in result:
return result
print(f"Gemini failed: {result['error']}. Falling back to Scavio.")
gemini_healthy = False
return scavio_ground(query)
result = grounded_query('best project management tools 2026')
print(f"Grounded via: {result['source']}")步骤 3: 通往Scavio后备的路线
构建基于搜索的基础,用网络结果中的等效上下文取代 Gemini 的基础功能。
def format_grounding_context(search_results: list) -> str:
"""Format search results to match the context format Gemini grounding provides."""
if not search_results:
return ''
parts = []
for r in search_results:
title = r.get('title', '')
snippet = r.get('snippet', '')
url = r.get('link', '')
parts.append(f'[{title}]({url}): {snippet}')
return '\n'.join(parts)
def grounded_prompt(query: str) -> str:
result = grounded_query(query)
if result['source'] == 'scavio':
context = format_grounding_context(result.get('results', []))
return f'Use the following web context to answer:\n{context}\n\nQuestion: {query}'
return f'Question: {query}'
print(grounded_prompt('best CRM for startups 2026'))步骤 4: 双子座回来后自动恢复
定期测试双子座的健康状况,并在其恢复后将其恢复为主要接地源。
RECOVERY_INTERVAL = 60 # seconds between recovery checks
def check_gemini_health() -> bool:
global gemini_healthy, gemini_last_check
now = time.time()
if now - gemini_last_check < RECOVERY_INTERVAL:
return gemini_healthy
gemini_last_check = now
result = gemini_ground('test query')
if 'error' not in result:
gemini_healthy = True
print('Gemini recovered. Restoring as primary.')
else:
print(f'Gemini still down: {result["error"]}')
return gemini_healthy
def smart_ground(query: str) -> dict:
if not gemini_healthy:
check_gemini_health()
return grounded_query(query)
result = smart_ground('latest python release')
print(f"Source: {result['source']}")Python 示例
import requests, os
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def fallback_ground(query):
# Try Gemini first, fall back to Scavio
try:
# gemini_call(query) # Your Gemini grounding call
raise Exception('Simulated Gemini 429')
except Exception:
data = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': 'google', 'query': query}).json()
results = data.get('organic_results', [])[:3]
return '\n'.join(f"{r['title']}: {r.get('snippet', '')}" for r in results)
print(fallback_ground('best CRM 2026'))JavaScript 示例
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
async function fallbackGround(query) {
try {
// await geminiGround(query); // Your Gemini call
throw new Error('Simulated 429');
} catch {
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H, body: JSON.stringify({platform: 'google', query})
});
const results = (await r.json()).organic_results || [];
return results.slice(0, 3).map(r => `${r.title}: ${r.snippet}`).join('\n');
}
}
fallbackGround('best CRM 2026').then(console.log);预期输出
A grounding system that uses Gemini when available, automatically falls back to Scavio search on 429/503 errors, and auto-recovers when Gemini is healthy again.