2026 年 7 月，SERP API 调用的稳定性实战：超时、重试、降级

接 SERP API 跑 Agent第一个版本能 demo第二个版本能上线但从 demo 到生产之间最难的是稳定性。上游抖动、网络异常、限流、模型发起多次相同调用每一个点都可能让你的服务出问题。这篇文章把我踩过的坑整理一下怎么设计超时和重试怎么处理上游错误体怎么在搜索失败时让 Agent 不至于崩。下面的例子都以 serpbase 的接口为例。客户端要扛住的第一件事超时Google 这类上游偶发超时很正常一次调用设个 8s 大概够用。但要分开两层来设客户端到 serpbase 的 HTTP 超时requests的timeout参数整个 Agent tool call 的整体超时防止一个搜索拖垮整个对话我一般这样写importrequests HTTP_TIMEOUT8# 单次 HTTP 调用TOTAL_TIMEOUT15# 含重试的整体预算defserp_search(query,hlen,glus):returnrequests.post(https://api.serpbase.dev/google/search,json{q:query,hl:hl,gl:gl,page:1,device:default},headers{X-API-Key:API_KEY,Content-Type:application/json},timeoutHTTP_TIMEOUT,).json()TOTAL_TIMEOUT这层用signal.alarm或者asyncio.wait_for都可以看你跑在什么 runtime 里。重试不是越多越好重试要回答三个问题哪些状态码要重试重试几次退避怎么算我的经验是重试对象网络超时、429、5xx不重试4xx 业务错status401这种瞎重试只是浪费额度次数2 次够了退避指数0.5s, 1.5s再随机抖动一下避免雪崩importrandomimporttimeimportrequests RETRY_STATUS{429,500,502,503,504}defwith_retry(fn,max_retry2):foriinrange(max_retry1):try:rfn()except(requests.Timeout,requests.ConnectionError):ifimax_retry:raisetime.sleep(0.5*(2**i)random.uniform(0,0.2))continueifr.status_codeinRETRY_STATUSandimax_retry:time.sleep(0.5*(2**i)random.uniform(0,0.2))continuereturnrreturnr业务失败 vs 网络失败serpbase 的失败响应也是 JSON成功外壳和失败外壳字段一样只是status非 0并且多了error_code、message。客户端要分开处理这两类错defcall_serp(query):rwith_retry(lambda:requests.post(https://api.serpbase.dev/google/search,json{q:query,hl:en,gl:us,page:1,device:default},headers{X-API-Key:API_KEY,Content-Type:application/json},timeoutHTTP_TIMEOUT,))bodyr.json()ifbody.get(status)!0:raiseSerpBaseBizError(body.get(error_code),body.get(message),body.get(request_id))returnbody业务错要落日志request_id必须打进去——找客服或者自己排查都靠它。监控告警也要分开网络失败率看 HTTP 状态码业务失败率看status ! 0的占比这两类错的原因不同处理方式也不同别混在一起。并发控制SERP API 一般有 QPS 限制多用户同时触发容易打爆。生产里加一个信号量importasyncio semasyncio.Semaphore(5)# 最多 5 个并发asyncdefguarded_search(query):asyncwithsem:returnawaitcall_serp_async(query)Semaphore(5)不是越大越好要看你的套餐 QPS 限制。如果不确定先从 3 开始观察一周的 P95 延迟和成功率再调大。缓存模型爱确认性检索同一个 session 里可能搜两三次同样的东西。给搜索加一层缓存能省不少钱importhashlibimportjsonfromcachetoolsimportTTLCache cacheTTLCache(maxsize2000,ttl600)# 10 分钟defcached_serp_search(query,hl,gl,page):keyhashlib.sha256(json.dumps({q:query,hl:hl,gl:gl,p:page},sort_keysTrue).encode()).hexdigest()[:16]ifkeyincache:returncache[key]resultcall_serp(query,hl,gl,page)cache[key]resultreturnresult新闻类查询 TTL 给 5–10 分钟事实类查询可以给 30 分钟。top_stories变化快缓存要短。失败降级最关键的一点搜索失败的时候Agent 怎么办我现在的做法是三层降级重试 2 次后还是失败 → 返回结构化错误模型收到错误信息模型收到错误 → 改为基于已有知识回答并明确告诉用户以下信息未实时核实如果是降级回答 → 在 UI 上加个标识让用户知道这是猜测defagent_with_fallback(user_query):try:resultscached_serp_search(user_query,en,us,1)except(SerpBaseBizError,requests.HTTPError)ase:return{answer:None,degraded:True,reason:str(e),request_id:getattr(e,request_id,None),}return{answer:synthesize_with_results(user_query,results),degraded:False}监控和告警跑稳了之后要加监控。我一般盯这几个指标成功率HTTP 2xx 业务status0的占比。低于 95% 就要查。P95 延迟超过 SLA 阈值要告警。serpbase 这类服务一般 2s 内能回来。5xx 占比超过 1% 就要排查可能是 IP 池或上游问题。业务错码分布某个error_code突然飙升一般是参数问题。把这些指标用 Prometheus 或者你喜欢的监控系统打点告警阈值按历史数据调别一上来就设得很严。一些实际跑出来的经验request_id一定要落日志。我每次问客服第一个问题就是能不能给我 request_id。客户端和上游之间最好加一层代理哪怕只是一个简单的函数这样换上游服务的时候改动小。监控别只看平均值。平均值会把问题平均掉要看 P95、P99。失败降级的提示要诚实。让模型说这个我没查到比我编一个对用户友好得多。熔断和隔离SERP API 是外部依赖一旦上游出问题你的 Agent 也会跟着出问题。生产里我加了一层简单的熔断器importtimeclassCircuitBreaker:def__init__(self,failure_threshold5,recovery_time30):self.failures0self.thresholdfailure_threshold self.recovery_timerecovery_time self.open_sinceNonedefallow(self):ifself.open_sinceisNone:returnTrueiftime.time()-self.open_sinceself.recovery_time:self.open_sinceNoneself.failures0returnTruereturnFalsedefrecord_failure(self):self.failures1ifself.failuresself.threshold:self.open_sincetime.time()defrecord_success(self):self.failures0breakerCircuitBreaker(failure_threshold5,recovery_time30)defserp_with_breaker(query):ifnotbreaker.allow():raiseSerpBaseBizError(circuit_open,上游连续失败临时熔断,None)try:rwith_retry(lambda:requests.post(https://api.serpbase.dev/google/search,json{q:query,hl:en,gl:us,page:1,device:default},headers{X-API-Key:API_KEY,Content-Type:application/json},timeoutHTTP_TIMEOUT,))bodyr.json()ifbody.get(status)!0:breaker.record_failure()raiseSerpBaseBizError(body.get(error_code),body.get(message),body.get(request_id))breaker.record_success()returnbodyexcept(requests.Timeout,requests.ConnectionError):breaker.record_failure()raise熔断器打开时直接返回失败不打上游。30 秒后尝试半开恢复后继续。熔断的核心不是省几次调用而是防止故障扩散——上游可能因为被打爆而恢复更慢熔断可以让它喘口气。测试SERP API 的测试比较特殊因为它依赖外部服务。几个常用的做法录制回放用真实 API 录一份响应存到文件测试时回放。注意date这类时间字段要脱敏。Mock 客户端写一个假的serp_search函数返回固定 JSON。在单元测试里用。契约测试用 schemathesis 或者 pydantic 校验响应 JSON 符合预期 schema。这点对 SERP API 特别重要——上游偶尔会改字段schema 校验能第一时间发现。# 用 pydantic 校验响应frompydanticimportBaseModelclassOrganicResult(BaseModel):rank:inttitle:strlink:strsnippet:str|NoneNonedeftest_response_shape():bodycall_serp(python asyncio)organicbody[search][organic]foriteminorganic:OrganicResult.model_validate(item)# 不符合 schema 会抛错把这些都补上之后SERP API 的稳定性基本能撑住生产。下面以 serpbase 的接口为例文档它的错误体结构和成功外壳一致写客户端不用维护两套。上线前 checklist最后留个 checklist上线前对着过一遍HTTP 超时设置建议 5–10s重试逻辑2 次指数退避只重试 429/5xx业务错和 HTTP 错分开处理request_id进日志QPS 信号量缓存层5–30 分钟 TTL熔断器失败降级路径监控指标成功率、P95、5xx 占比告警阈值Pydantic schema 校验录制回放的测试数据按这个走完一遍稳定性这块基本就稳了。