AI金丝雀发布：这次怎么落地的

金丝雀发布流量分流发布策略

作者: 醉月思发布日期: 2024-08-10 分类: 技术实践

最近在做一个 AI 应用的升级，从 v1.0 迁移到 v2.0。

我们的 AI 应用主要提供两类服务：

文本摘要：对长文本进行摘要，QPS 约 200
智能问答：基于文档的问答，QPS 约 50

原来的 v1.0 用的是 GPT-3.5-turbo，响应时间在 2-3 秒。

写在前面

最近在做一个 AI 应用的升级，从 v1.0 迁移到 v2.0。本以为只是换个模型的事儿，结果踩了一堆坑——全量发布后，新模型在某些场景下返回超时，用户投诉量飙升。痛定思痛，决定搞一套金丝雀发布机制。

这篇文章记录了我从"全量发布就完了"到"精细化流量控制"的完整实践过程，包括遇到的坑、解决的思路和最终的方案。

背景：全量发布的那天晚上

先说下当时的情况。我们的 AI 应用主要提供两类服务：

文本摘要：对长文本进行摘要，QPS 约 200
智能问答：基于文档的问答，QPS 约 50

原来的 v1.0 用的是 GPT-3.5-turbo，响应时间在 2-3 秒。新版本 v2.0 想升级到 GPT-4o-mini，预期响应时间在 1-2 秒，成本还能降低 30%。

那晚 10 点，我们直接全量切换了。监控看起来一切正常，QPS 稳定，响应时间也没问题。凌晨 2 点开始出问题了：

某些长文本（超过 10 万字）的摘要请求，响应时间飙升到 30 秒+
问答服务偶尔返回"模型超时"
客服系统收到了 37 条投诉

紧急回滚后分析发现：GPT-4o-mini 在处理长文本时，token 消耗超出预期，导致 API 限流。

这次教训让我意识到：即使是小版本升级，也需要金丝雀发布。

需求：我到底需要什么

分析完那次事故，我列出了金丝雀发布的需求：

核心需求

可分流的发布：不能一刀切，需要能控制流量比例
可观察：新老版本的指标要能对比看
可回滚：发现问题能快速切回来
场景区分：不同场景（摘要 vs 问答）可能需要不同的流量策略

真实限制

后端服务是单体应用：没有微服务的基础设施，所有逻辑在一个服务里
资源有限：没有专门的负载均衡器或 API Gateway
历史包袱：v1.0 的代码已经跑了 8 个月，不敢大改
预算约束：不能引入商业解决方案

基于这些限制，我设计了一套轻量级的金丝雀发布方案。

实现：分三个阶段搞定

阶段一：应用层流量切分

最简单的方式是在应用层做流量切分。代码长这样：

import random
from functools import wraps
from datetime import datetime

def canary_release(canary_ratio=0.1, whitelist=None):
    """
    金丝雀发布装饰器
    :param canary_ratio: 新版本流量比例，0-1
    :param whitelist: 白名单用户 ID 列表
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 白名单优先走新版本
            user_id = kwargs.get('user_id') or ''
            if whitelist and user_id in whitelist:
                return func(*args, **kwargs)

            # 按比例分流
            if random.random() < canary_ratio:
                return func(*args, **kwargs)
            else:
                # 调用老版本
                return legacy_func(*args, **kwargs)
        return wrapper
    return decorator

# 老版本函数（假设存在）
def legacy_func(*args, **kwargs):
    # 调用 GPT-3.5-turbo 的逻辑
    pass

# 新版本函数
@canary_release(canary_ratio=0.1, whitelist=['user_123', 'user_456'])
def new_version(*args, **kwargs):
    # 调用 GPT-4o-mini 的逻辑
    pass

这个方案的好处是：

简单直接：代码改动小，容易理解
白名单支持：可以先让内部用户尝鲜
可配置：通过环境变量或配置中心调整流量比例

但很快发现了问题：random.random() 是完全随机的，同一个用户可能这次走新版本，下次走老版本，体验不一致。

阶段二：基于用户的流量分流

为了解决一致性问题，改用基于用户 ID 的哈希分流：

import hashlib

def hash_based_canary(canary_ratio=0.1, whitelist=None):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            user_id = kwargs.get('user_id') or ''

            # 白名单优先
            if whitelist and user_id in whitelist:
                return func(*args, **kwargs)

            # 基于用户 ID 的哈希分流
            if not user_id:
                # 匿名用户使用随机分流
                return func(*args, **kwargs) if random.random() < canary_ratio else legacy_func(*args, **kwargs)

            hash_value = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
            if (hash_value % 100) < (canary_ratio * 100):
                return func(*args, **kwargs)
            else:
                return legacy_func(*args, **kwargs)
        return wrapper
    return decorator

这样同一个用户 ID 总是会路由到同一个版本，体验一致性有了保证。

阶段三：智能流量控制

阶段二的方案在测试时还行，但上线后发现还不够智能。比如：

时间段差异：白天流量大时，10% 的新版本流量已经不少了；晚上流量小时，10% 可能不够测试
场景差异：文本摘要场景复杂，需要更小心的放量；问答场景简单，可以快速放量
错误率感知：新版本错误率高时，应该自动减少流量

于是我搞了个动态流量控制器：

import time
from collections import deque
from typing import Optional

class CanaryTrafficController:
    def __init__(
        self,
        base_ratio: float = 0.1,
        max_ratio: float = 0.5,
        error_threshold: float = 0.05,
        window_size: int = 100
    ):
        self.base_ratio = base_ratio
        self.max_ratio = max_ratio
        self.error_threshold = error_threshold
        self.window_size = window_size
        self.errors = deque(maxlen=window_size)
        self.current_ratio = base_ratio

    def record_error(self):
        """记录一次错误"""
        self.errors.append(1)

    def record_success(self):
        """记录一次成功"""
        self.errors.append(0)

    def get_current_ratio(self) -> float:
        """获取当前流量比例"""
        if not self.errors:
            return self.current_ratio

        error_rate = sum(self.errors) / len(self.errors)

        # 错误率超过阈值，减少流量
        if error_rate > self.error_threshold:
            self.current_ratio = max(self.base_ratio * 0.5, self.current_ratio * 0.8)
        else:
            # 错误率低，可以缓慢增加流量
            self.current_ratio = min(self.max_ratio, self.current_ratio * 1.05)

        return self.current_ratio

# 使用示例
summary_controller = CanaryTrafficController(base_ratio=0.05, error_threshold=0.03)
qa_controller = CanaryTrafficController(base_ratio=0.2, error_threshold=0.05)

def smart_canary(controller: CanaryTrafficController):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            current_ratio = controller.get_current_ratio()

            user_id = kwargs.get('user_id') or ''
            if not user_id:
                return func(*args, **kwargs) if random.random() < current_ratio else legacy_func(*args, **kwargs)

            hash_value = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
            if (hash_value % 100) < (current_ratio * 100):
                try:
                    result = func(*args, **kwargs)
                    controller.record_success()
                    return result
                except Exception as e:
                    controller.record_error()
                    # 回退到老版本
                    return legacy_func(*args, **kwargs)
            else:
                return legacy_func(*args, **kwargs)
        return wrapper
    return decorator

这个控制器会：

监控新版本的错误率
错误率高时自动减少流量
错误率低时缓慢增加流量
异常时自动回退到老版本

流量流程图

graph TD A[用户请求] --> B{是否有用户ID?} B -->|有| C[计算用户哈希] B -->|无| D[随机分流] C --> E{哈希值 < 流量比例?} D --> E E -->|是| F[调用新版本] E -->|否| G[调用老版本] F --> H{调用成功?} H -->|是| I[记录成功] H -->|否| J[记录错误并回退] I --> K[动态调整流量比例] J --> K G --> K

踩坑：那些年我遇到的坑

坑一：哈希碰撞导致流量不均

最初用简单的 hash(user_id) % 100，后来发现不同用户 ID 的哈希值分布不均匀，导致某些用户总是走新版本，某些用户总是走老版本。

解决：改用 MD5 后再取模，分布更均匀。

坑二：流量控制器反应太慢

最初设置 window_size=1000，但实际 QPS 只有 200，需要 5 秒才能收集够数据，控制器反应太慢。

解决：根据实际 QPS 调整 window_size，或者改用时间窗口（比如最近 10 秒的数据）。

坑三：老版本代码耦合严重

在实现回退逻辑时，发现老版本的代码和新版本耦合严重，无法直接调用。

解决：重构老版本代码，提取公共逻辑，新老版本只替换核心 API 调用部分。

坑四：监控缺失

发布后才发现没有好的监控，无法直观看到新老版本的对比数据。

解决：添加埋点，记录每个请求的版本、响应时间、错误率等指标。

结果：后来怎么样了

这套金丝雀发布方案上线后，效果还不错：

发布风险降低：新版本先 5% 流量测试，观察 2 小时没问题后再逐步放量
用户体验稳定：基于用户 ID 的分流保证了体验一致性
自动容错：错误率高时自动减少流量，减少了人工干预
成本可控：通过精细化控制，避免了全量发布导致的大规模问题

数据对比

指标	全量发布	金丝雀发布
发布风险	高	低
回滚时间	10-15 分钟	1-2 分钟
用户投诉	37 次	2 次
流量控制精度	0% 或 100%	1%-100%
自动容错	无	有

写在最后

金丝雀发布不是什么高大上的技术，核心就是"谨慎、观察、控制"。对于 AI 应用这种不确定性强的服务，更需要谨慎发布。

这套方案虽然简单，但解决了我当时的实际问题。如果你的场景更复杂，比如有微服务、有 API Gateway，那可以考虑更专业的方案（比如 Istio、NGINX 的流量切分）。

但话说回来，最好的方案是能解决你当前问题的方案，而不是最复杂的方案。

参考资料

版权声明: 本文首发于指尖魔法屋-AI金丝雀发布：这次怎么落地的（https://blog.thinkmoon.cn/post/292-ai-canary-release-full-traffic-split-practice/）转载或引用必须申明原指尖魔法屋来源及源地址！

利用 Cursor ACP 模式搭建飞书助手机器人，无需任何大模型接口

用 lark-cli 收飞书消息，Python 桥接 Cursor ACP（agent acp），项目代码里不集成任何大模型 API；适合已有 Cursor 订阅、想在本机跑 Agent 的场景。

📁 技术实践

📅 2026-07-07

用 GolemBot 把 Codex 接到飞书机器人

这篇文章记录一种少写代码的方案：用 GolemBot 接飞书机器人，用 Codex 做 Agent，再通过飞书 CLI Skills 读取文档、需求清单和 bug 清单。

📁 技术实践

📅 2026-06-09

调试 OpenAI Responses API 的 Web Search 踩坑记录

本文记录一次在投资研究 Agent 中调试 OpenAI Responses API Web Search 工具的过程，重点梳理请求格式、流式 SSE 解析、工具名兼容性和结构化数据源设计。

📁 技术实践

📅 2026-04-30

AI金丝雀发布：这次怎么落地的

写在前面

背景：全量发布的那天晚上

需求：我到底需要什么

核心需求

真实限制

实现：分三个阶段搞定

阶段一：应用层流量切分

阶段二：基于用户的流量分流

阶段三：智能流量控制

流量流程图

踩坑：那些年我遇到的坑

坑一：哈希碰撞导致流量不均

坑二：流量控制器反应太慢

坑三：老版本代码耦合严重

坑四：监控缺失

结果：后来怎么样了

数据对比

写在最后

参考资料

评论

相关阅读