概述
Gemini 的接地搜索会在高峰流量期间定期返回 429 速率限制和 503 服务不可用错误，从而使您的代理失去网络数据。此工作流程实时检测 Gemini 接地故障，将相同的查询路由到 Scavio 的搜索 API 作为后备，记录每个后备事件以进行容量规划，针对 Gemini 运行定期运行状况检查，并在 Gemini 稳定后自动恢复到主要路径。您的代理人永远不会盲目，您可以全面了解 Gemini 的可靠性。
触发器
Gemini API 返回 429 或 503 错误
计划
事件驱动（关于双子座错误）
工作流步骤
1
检测 Gemini 错误
拦截来自 Gemini 接地 API 调用的 429（速率限制）或 503（服务不可用）响应。
2
通往Scavio后备的路线
通过 Scavio 的 API 重新执行相同的搜索查询。将 Gemini 接地格式映射到 Scavio 响应格式。
3
记录回退事件
记录时间戳、错误代码、查询和延迟增量，以进行容量规划和 SLA 跟踪。
4
双子座健康检查
每隔 60 秒，向 Gemini 发送一个轻型探测器。跟踪连续故障以检测长时间停机。
5
自动恢复为主
一旦 Gemini 返回 3 个连续的健康响应，请将流量从 Scavio 切换回 Gemini 作为主要流量。
Python 实现
Python
import requests, os, json, time
from datetime import datetime
H = {"x-api-key": os.environ["SCAVIO_API_KEY"]}
fallback_log = []
def search_with_fallback(query, primary="gemini"):
"""Try Gemini grounding first, fall back to Scavio on error."""
start = time.time()
if primary == "gemini":
# Simulate Gemini call - replace with actual Gemini grounding API
gemini_healthy = False # Set based on actual Gemini response
if not gemini_healthy:
# Fallback to Scavio
r = requests.post("https://api.scavio.dev/api/v1/search", headers=H,
json={"platform": "google", "query": query, "ai_overview": True},
timeout=10).json()
latency = round(time.time() - start, 3)
fallback_log.append({
"ts": datetime.utcnow().isoformat(),
"query": query,
"error": "gemini_429",
"fallback_latency_s": latency,
"scavio_results": len(r.get("organic", []))
})
return {"source": "scavio_fallback", "results": r.get("organic", [])[:5],
"ai_overview": r.get("ai_overview", {}).get("text", "")}
return {"source": "gemini_primary", "results": []}
def health_check_gemini():
"""Probe Gemini health. Returns True if healthy."""
# Replace with actual Gemini health check
return False
# Run fallback chain
queries = ["search api pricing 2026", "best serp api for agents", "web search api comparison"]
for q in queries:
result = search_with_fallback(q)
print(f"[{result['source'].upper()}] {q}: {len(result['results'])} results")
print(f"\nFallback events logged: {len(fallback_log)}")
for event in fallback_log:
print(f" {event['ts']} | {event['query']} | {event['fallback_latency_s']}s")JavaScript 实现
JavaScript
const H = {"x-api-key": process.env.SCAVIO_API_KEY, "Content-Type": "application/json"};
const fallbackLog = [];
async function searchWithFallback(query, primary = "gemini") {
const start = Date.now();
if (primary === "gemini") {
// Simulate Gemini call - replace with actual Gemini grounding API
const geminiHealthy = false; // Set based on actual Gemini response
if (!geminiHealthy) {
const r = await fetch("https://api.scavio.dev/api/v1/search", {
method: "POST", headers: H,
body: JSON.stringify({platform: "google", query, ai_overview: true})
}).then(r => r.json());
const latency = ((Date.now() - start) / 1000).toFixed(3);
fallbackLog.push({
ts: new Date().toISOString(), query,
error: "gemini_429", fallbackLatencyS: latency,
scavioResults: (r.organic || []).length
});
return {source: "scavio_fallback", results: (r.organic || []).slice(0, 5),
aiOverview: r.ai_overview?.text || ""};
}
}
return {source: "gemini_primary", results: []};
}
async function healthCheckGemini() {
// Replace with actual Gemini probe
return false;
}
(async () => {
const queries = ["search api pricing 2026", "best serp api for agents", "web search api comparison"];
for (const q of queries) {
const result = await searchWithFallback(q);
console.log(`[${result.source.toUpperCase()}] ${q}: ${result.results.length} results`);
}
console.log(`\nFallback events: ${fallbackLog.length}`);
fallbackLog.forEach(e => console.log(` ${e.ts} | ${e.query} | ${e.fallbackLatencyS}s`));
})();使用的平台
包含知识图谱、PAA和AI概览的网页搜索