SERP API 长时间搜索 (AI Overview 生成 / 复杂查询),返回慢。两种获取方式:轮询 / Webhook,各有适用。

1. 两种方式

轮询 (Polling):客户端反复请求,直到结果就绪。

Webhook 回调:服务端处理完,主动 POST 结果到你的 URL。

2. 对比

3. 场景:长时间任务

SERP API 某些端点处理 > 5s(复杂 query / 批量):

• 轮询:每 2s 查一次状态
• Webhook:等服务 POST 结果

4. 轮询实现

import time
import requests

def search_with_polling(query, poll_interval=2, max_wait=60):
    """轮询等待结果"""
    # 1. 提交任务
    r = requests.post(
        'https://api.serpbase.dev/google/search/async',
        headers={'X-API-Key': os.environ['SERPBASE_API_KEY']},
        json={'q': query, 'num': 10}
    )
    task_id = r.json()['task_id']

    # 2. 轮询状态
    start = time.time()
    while time.time() - start < max_wait:
        r = requests.get(
            f'https://api.serpbase.dev/tasks/{task_id}',
            headers={'X-API-Key': os.environ['SERPBASE_API_KEY']}
        )
        status = r.json()

        if status['state'] == 'completed':
            return status['result']
        elif status['state'] == 'failed':
            raise Exception(f"Task failed: {status['error']}")

        time.sleep(poll_interval)

    raise TimeoutError(f"Task {task_id} timed out after {max_wait}s")

5. 轮询优化

指数退避轮询:

def search_polling_backoff(query, max_wait=60):
    r = requests.post(...)
    task_id = r.json()['task_id']

    start = time.time()
    interval = 1
    while time.time() - start < max_wait:
        status = check_status(task_id)
        if status['state'] == 'completed':
            return status['result']
        if status['state'] == 'failed':
            raise Exception(status['error'])

        time.sleep(interval)
        interval = min(interval * 1.5, 5)  # 1, 1.5, 2.25, 3.4, 5

    raise TimeoutError(f"Task {task_id} timed out")

6. Webhook 实现

6.1 提交 + 注册回调

def search_with_webhook(query, callback_url):
    """提交任务 + 注册回调"""
    r = requests.post(
        'https://api.serpbase.dev/google/search/async',
        headers={'X-API-Key': os.environ['SERPBASE_API_KEY']},
        json={
            'q': query,
            'num': 10,
            'webhook_url': callback_url,  # 结果 POST 到这里
            'webhook_secret': os.environ.get('WEBHOOK_SECRET', '')
        }
    )
    return r.json()['task_id']

6.2 接收回调 (FastAPI)

from fastapi import FastAPI, Request, Header
import hmac
import hashlib

app = FastAPI()
WEBHOOK_SECRET = os.environ.get('WEBHOOK_SECRET', '')

@app.post('/webhook/serp')
async def serp_callback(request: Request):
    """接收 SERP 结果回调"""
    # 1. 验证签名
    signature = request.headers.get('X-Webhook-Signature', '')
    body = await request.body()

    expected = hmac.new(
        WEBHOOK_SECRET.encode(), body, hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        return {'error': 'invalid signature'}, 401

    # 2. 处理结果
    data = await request.json()
    task_id = data['task_id']
    result = data['result']

    # 3. 存储 / 通知
    store_result(task_id, result)

    return {'status': 'ok'}

6.3 验证签名

import hmac
import hashlib

def verify_webhook_signature(body, signature, secret):
    """HMAC 签名验证"""
    expected = hmac.new(
        secret.encode(), body, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, expected)

7. Webhook 重试

Webhook 失败要重试:

# SERP API 端逻辑
def deliver_webhook(url, data, secret, max_retries=5):
    """投递 webhook,失败重试"""
    for attempt in range(max_retries):
        try:
            r = requests.post(
                url,
                json=data,
                headers={
                    'X-Webhook-Signature': hmac.new(
                        secret.encode(), json.dumps(data).encode(), hashlib.sha256
                    ).hexdigest()
                },
                timeout=5
            )
            if r.status_code < 300:
                return True
        except Exception:
            pass

        time.sleep(2 ** attempt)  # 1, 2, 4, 8, 16

    # 全部失败,进死信队列
    dead_letter_queue.put(data)
    return False

8. 死信队列

import redis

r = redis.Redis()

def dead_letter_queue_push(data):
    r.lpush('webhook_dead_letters', json.dumps(data))

def dead_letter_replay():
    """重放死信"""
    while True:
        data = r.rpop('webhook_dead_letters')
        if not data:
            break
        item = json.loads(data)
        deliver_webhook(item['url'], item['data'], item['secret'])

9. 实战对比

长时间任务 (平均 8s) 实测:

轮询多 4 倍请求,Webhook 更快 + 更省。

10. 选型建议

11. 混合方案

同步 + 超时 → 转轮询:

def search_hybrid(query, sync_timeout=5):
    """先同步,超时转轮询"""
    try:
        # 1. 同步等 5s
        result = requests.post(
            'https://api.serpbase.dev/google/search',
            headers={'X-API-Key': key},
            json={'q': query},
            timeout=5
        )
        if result.status_code == 200:
            return result.json()

        # 2. 拿到 task_id,转轮询
        task_id = result.json().get('task_id')
        if task_id:
            return search_polling(task_id)
    except requests.Timeout:
        # 3. 超时,重新提交异步
        task_id = submit_async(query)
        return search_polling(task_id)

12. 监控

from prometheus_client import Counter, Histogram

webhook_received = Counter('webhook_received_total', 'Webhooks received')
webhook_failed = Counter('webhook_failed_total', 'Webhook failures')
webhook_latency = Histogram('webhook_latency_seconds', 'Webhook latency')

@webhook_latency.time()
async def serp_callback(request):
    webhook_received.inc()
    # ... 处理

13. 总结

Webhook vs 轮询:

• 轮询:简单,多 4 倍请求,可靠
• Webhook:实时,省资源,需重试 + 死信

长时间任务生产推荐 Webhook,加 HMAC 签名 + 重试 + 死信队列。

参考文档

本文 API 示例参考 serpbase 文档,接口路径、参数和返回字段以官方文档为准。