AI Agent 调 SERP API 跑生产,新版本 / 新参数上线怕炸。灰度发布是必备,2 小时搭一个完整方案。

1. 灰度发布 4 阶段

阶段 1(10%):  内测 / 团队 dogfood
阶段 2(30%):  早期采用者
阶段 3(70%):  大部分用户
阶段 4(100%): 全量

每阶段 1-3 天,出问题立即回滚。

2. 灰度维度

3 个维度可选:

• 用户维度:按 user_id 分流
• query 维度:按关键词分流
• 区域维度:按地区分流

我项目用 user_id 维度,简单稳定。

3. 灰度配置

# config/gray_release.yaml
gray_release:
  serp_api_v2:
    enabled: true
    percentage: 30
    rules:
      - match_user_id: ["team_only"]    # 团队内部
      - match_user_id_prefix: ["test_"]  # 测试账号
      - default_rollout: 30  # 30% 随机
    fallback: v1  # 失败回退 v1
    metrics:
      - error_rate
      - p99_latency
      - user_satisfaction
    rollback_threshold:
      error_rate: 0.05
      p99_latency: 5.0

YAML 配置,改 percentage 即可调灰度比例。

4. 灰度 SDK

import hashlib
import yaml

class GrayRelease:
    def __init__(self, config_path='config/gray_release.yaml'):
        self.config = yaml.safe_load(open(config_path))

    def get_version(self, user_id, service='serp_api_v2'):
        """决定 user 用哪个版本"""
        rule = self.config['gray_release'].get(service, {})
        if not rule.get('enabled'):
            return 'v1'  # 默认旧版

        # 白名单
        if user_id in rule.get('rules', [{}])[0].get('match_user_id', []):
            return 'v2'

        # 百分比
        percentage = rule.get('rules', [{}])[-1].get('default_rollout', 0)
        h = hashlib.md5(f"{user_id}:{service}".encode()).hexdigest()
        bucket = int(h[:8], 16) % 100

        if bucket < percentage:
            return 'v2'
        return 'v1'

gray = GrayRelease()

def call_serp(user_id, query):
    version = gray.get_version(user_id)

    if version == 'v2':
        return call_serp_v2(query)
    return call_serp_v1(query)

5. v1 vs v2 路由

def call_serp_v1(query):
    """旧版本 - 老接口"""
    r = requests.post(
        'https://api.serpbase.dev/google/search',
        headers={'X-API-Key': os.environ['SERPBASE_API_KEY']},
        json={'q': query, 'hl': 'zh-CN', 'gl': 'cn', 'num': 10},
        timeout=5
    )
    return r.json()

def call_serp_v2(query):
    """新版本 - 优化参数 + 新字段"""
    r = requests.post(
        'https://api.serpbase.dev/google/search',
        headers={'X-API-Key': os.environ['SERPBASE_API_KEY']},
        json={
            'q': query,
            'hl': 'zh-CN',
            'gl': 'cn',
            'num': 10,
            'new_field': True,  # v2 新参数
            'enable_ai_overview': True  # v2 新功能
        },
        timeout=5
    )
    return r.json()

v2 加新参数 / 新字段,实测无问题后全量。

6. 指标对比

两个版本指标对比:

from prometheus_client import Histogram, Counter

serp_latency = Histogram('serp_latency_seconds', 'SERP latency', ['version'])
serp_errors = Counter('serp_errors_total', 'SERP errors', ['version', 'error_type'])

def call_serp_monitored(user_id, query):
    version = gray.get_version(user_id)

    start = time.time()
    try:
        if version == 'v2':
            data = call_serp_v2(query)
        else:
            data = call_serp_v1(query)

        serp_latency.labels(version=version).observe(time.time() - start)
        return data

    except Exception as e:
        serp_errors.labels(version=version, error_type=type(e).__name__).inc()
        raise

Grafana 对比 v1 vs v2 延迟 / 错误率。

7. 自动回滚

灰度出问题,自动回滚:

def should_rollback(version='v2'):
    """检查 v2 是否需要回滚"""
    error_rate = get_error_rate(version)
    p99_latency = get_p99_latency(version)

    threshold = gray.config['gray_release']['serp_api_v2']['rollback_threshold']

    if error_rate > threshold['error_rate']:
        return True, f"error_rate {error_rate} > {threshold['error_rate']}"

    if p99_latency > threshold['p99_latency']:
        return True, f"p99 {p99_latency}s > {threshold['p99_latency']}s"

    return False, None

def auto_rollback_check():
    rollback, reason = should_rollback()
    if rollback:
        # 灰度比例改 0
        gray.config['gray_release']['serp_api_v2']['enabled'] = False
        alert(f"Auto rollback: {reason}")
        log_to_slack(f"🔴 Auto rollback: {reason}")

每分钟检查一次,出问题 1 分钟内自动回滚。

8. 多版本并存

并行 3 个版本:

VERSIONS = {
    'v1': call_serp_v1,
    'v2': call_serp_v2,
    'v3_beta': call_serp_v3,
}

def call_serp(user_id, query):
    # v3 内部测试,默认不开
    if user_id in BETA_USERS:
        return VERSIONS['v3_beta'](query)

    # 灰度决定
    version = gray.get_version(user_id)
    return VERSIONS[version](query)

3 个版本并存,v3 仅内部测试。

9. 灰度配置中心

不用 YAML 文件,改用配置中心 (实时改):

import redis

class GrayReleaseRedis:
    def __init__(self, redis_client):
        self.r = redis_client

    def get_percentage(self, service):
        return int(self.r.get(f'gray:{service}:percentage') or 0)

    def set_percentage(self, service, percentage):
        self.r.set(f'gray:{service}:percentage', percentage)
        log(f"Updated {service} percentage to {percentage}%")

    def is_enabled(self, service):
        return self.r.exists(f'gray:{service}:percentage')

# 实时调
gray_redis = GrayReleaseRedis(redis.Redis())
gray_redis.set_percentage('serp_api_v2', 50)  # 50% 灰度

不改代码,改配置实时生效。

10. 实战:30 天灰度发布

我项目发布 v2 流程:

阶段	比例	持续	监控指标
内测	5%	3 天	0 故障
早期	30%	5 天	P99 4.2s
大部分	70%	7 天	错误率 0.2%
全量	100%	-	-

5+5+7+13 = 30 天全流程。

11. Canary(金丝雀)

灰度的轻量版:1% 流量新版本:

def call_serp_canary(user_id, query):
    # 1% canary
    if get_canary_group(user_id, percentage=1):
        try:
            return call_serp_v2(query)
        except Exception:
            pass  # fallback 到 v1
    return call_serp_v1(query)

1% 跑 24 小时无问题 → 灰度 30% → 70% → 100%。

12. 工具推荐

• 自建 Redis:简单,够用
• LaunchDarkly:商业,$8/月起
• Argo Rollouts(K8s):K8s 原生
• Flagger:K8s 渐进式发布

我项目用 Redis + 自建,简单稳定。

13. 总结

灰度发布 4 件套:

1. 灰度 SDK(哈希分流)
2. 多版本路由
3. 指标对比 (Prometheus)
4. 自动回滚 (异常检测)

30 天发布流程:5% → 30% → 70% → 100%。

参考文档

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