背景

SERP API 在生产环境跑 1 个月后,遇到了这些错:

• 网络抖动 (1-2 次/周)
• 429 限流 (2-3 次/周,峰值时段)
• vendor 5xx(2-3 次/月,短暂故障)
• 4xx 客户端错 (罕见,但有)
• 解析失败 (vendor 改 schema,1-2 次/月)

下面是 5 个生产级处理模式,踩坑后总结的。

模式 1:retry with exponential backoff

import time
import random

def fetch_with_retry(url, headers, body, max_retries=3):
    """指数 backoff + jitter,只对 5xx/timeout 重试"""
    for attempt in range(max_retries):
        try:
            r = requests.post(url, headers=headers, json=body, timeout=10)
            r.raise_for_status()
            return r.json()
        except requests.exceptions.Timeout:
            if attempt == max_retries - 1:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(wait)
        except requests.exceptions.HTTPError as e:
            if 400 <= e.response.status_code < 500:
                raise  # 4xx 不重试(参数错,重试也错)
            if e.response.status_code == 429:
                # 429 看 Retry-After 头
                retry_after = int(e.response.headers.get("Retry-After", 1))
                if retry_after > 60:
                    raise  # 限流过久,放弃
                time.sleep(retry_after)
                continue
            # 5xx 重试
            if attempt == max_retries - 1:
                raise
            time.sleep(2 ** attempt)

关键:4xx 不重试 (参数错,重试也错),5xx 才重试。429 看 Retry-After 头。

模式 2:circuit breaker(熔断)

class CircuitBreaker:
    def __init__(self, failure_threshold=10, cooldown=30):
        self.failures = 0
        self.last_failure = 0
        self.failure_threshold = failure_threshold
        self.cooldown = cooldown
        self.state = "CLOSED"  # CLOSED / OPEN / HALF_OPEN

    def call(self, func, *args, **kwargs):
        if self.state == "OPEN":
            if time.time() - self.last_failure > self.cooldown:
                self.state = "HALF_OPEN"
            else:
                raise Exception("Circuit open, fast-fail")
        try:
            result = func(*args, **kwargs)
            if self.state == "HALF_OPEN":
                self.state = "CLOSED"
                self.failures = 0
            return result
        except Exception:
            self.failures += 1
            self.last_failure = time.time()
            if self.failures >= self.failure_threshold:
                self.state = "OPEN"
            raise

场景:vendor 故障时 5xx 飙到 100%,不熔断的话 30 秒内几万请求全失败,雪崩。熔断后 30 秒内所有调用 fast-fail,等 vendor 恢复。

模式 3:fallback to cache

def fetch_with_fallback(query):
    cache_key = f"serp:{hash(query)}"

    # 1. 试 API
    try:
        data = fetch_serp_api(query)
        cache.set(cache_key, data, ttl=300)
        return data
    except Exception as e:
        # 2. 失败返旧缓存(允许 stale)
        cached = cache.get(cache_key, allow_stale=True)
        if cached:
            return cached
        # 3. 没缓存返 None
        return None

场景:API 故障但需要响应,给个 5-10 分钟前的旧数据 (标记 stale)。比返错好。

模式 4:graceful degradation

def fetch_or_fail(query):
    """失败返降级结构(部分字段),不是 None"""
    try:
        return fetch_serp_api(query)
    except Exception:
        # 返最小可用结构
        return {
            "organic": [],  # 空数组
            "people_also_ask": [],
            "_stale": True,
            "_error": "SERP API unavailable",
        }

场景:LLM 收到降级结构 (organic: []),不会因缺字段崩。_stale: True 让 LLM 知道数据可能不准。

模式 5:dead letter queue

def fetch_with_dlq(query, dlq):
    try:
        return fetch_serp_api(query)
    except Exception as e:
        # 失败 3 次后进死信队列,人工 review
        retry_count = redis.incr(f"serp:retry:{hash(query)}")
        if retry_count > 3:
            dlq.push({
                "query": query,
                "error": str(e),
                "ts": time.time(),
            })
            redis.expire(f"serp:retry:{hash(query)}", 86400)
        raise

场景:失败不丢失,3 次后进死信队列,人工 review 或 fallback 缓存。

5 模式组合

实际生产里 5 个模式我都用,按这个顺序:

def robust_fetch(query, dlq):
    """综合 5 个模式"""

    # 1. Cache check
    cached = cache.get(f"serp:{query}")
    if cached:
        return cached

    # 2. Circuit breaker
    try:
        return circuit_breaker.call(fetch_with_retry, query)
    except Exception as e:
        # 3. Fallback to stale cache
        stale = cache.get(f"serp:{query}", allow_stale=True)
        if stale:
            return stale

        # 4. Graceful degradation
        if retry_count(query) < 3:
            redis.incr(f"serp:retry:{query}")
            raise

        # 5. Dead letter queue
        dlq.push({"query": query, "error": str(e), "ts": time.time()})
        return {
            "organic": [],
            "people_also_ask": [],
            "_stale": True,
            "_error": str(e),
        }

监控告警

指标	阈值	告警
failure_rate (5min)	> 5%	Slack
circuit_breaker_open	任意	Slack + oncall
dlq 长度	> 100	Slack
avg_latency	> 3s	Slack(vendor 慢)

实测数据

错误类型	频率	处理结果
网络抖动	35 / 周	retry 100% 成功
vendor 5xx	12 / 周	60% retry 成功,40% fallback 缓存
429 限流	8 / 周	retry_after 等待
4xx 客户端错	3 / 月	不重试,直接 fail
解析失败	2 / 月	通知 vendor + fallback 缓存

SERP API 特有技巧

auto-refund 利用:很多 SERP API(包括我用的这个) 失败时自动退 credit。如果 vendor 故障 1 小时,我的 retry 都失败,fallback 缓存兜底,vendor 恢复后失败那批调用都退了 credit,没浪费。

vendor 切换:SERP API 1 家故障 30 分钟以上时,可临时切到备选 vendor(类似服务)。但 vendor 切换是大事,先有降级和熔断,后考虑切换。

故障时仍记 log:失败时 log 到 ELK(请求 ID + 错误),事后分析原因。SERP API 返回 request_id 字段,方便 vendor 排查。

小结

SERP API 错误处理 5 模式按这个优先级:

1. Cache(成本最低)
2. Retry with backoff(网络抖动)
3. Circuit breaker(vendor 故障)
4. Fallback to cache / degradation(持续故障)
5. Dead letter queue(人工 review)

5 模式不是堆叠,而是按场景组合:

• 正常:cache → API → log
• 抖动:cache miss → API retry → success
• 故障:circuit open → fallback cache → success(用户感知不到)
• 长期故障:cache stale → degradation 字段 → 部分 success

SerpBase 的 auto-refund 帮我省了 30% SERP 成本 (失败调用不扣 credit)。综合 5 模式,SERP 调用成本 + 0.001 美元 / 次,延迟 p99 < 3s。