Agent 灰度发布策略：单个实例先上线，逐步切流量（续篇）-尧图网络科技

Agent 灰度发布策略：单个实例先上线，逐步切流量（续篇）

场景痛点

Agent新版本上线。v2版本新增了"多模型路由"功能。直接全量发布——5分钟后，20%的请求返回格式变了，下游系统解析失败。回滚到v1，影响已扩散。

全量发布风险太高。但Agent不是普通微服务——Agent是有状态的（会话状态、记忆数据、工具注册信息）。普通微服务的灰度发布用流量权重切换（10%流量走新版本）。Agent的灰度更复杂：一个用户的会话中途不能切换版本——会话状态不兼容。

核心矛盾：Agent灰度需要会话级隔离而非请求级切流。新版本Agent必须先跑少量会话，验证会话级行为正确，再逐步扩大覆盖范围。

底层机制与原理剖析

Agent灰度发布的三层隔离模型：

关键机制：

金丝雀测试。不是切流量——是手动指定5个测试用户使用新版本Agent。测试用户的会话全程在v2实例上运行。观察：会话级行为是否正常（对话是否连贯、工具调用是否成功、记忆数据是否兼容）。
会话级灰度路由。灰度路由器不是按请求权重切流，是按用户ID哈希分配版本。同一用户的所有会话始终走同一版本——保证会话一致性。新用户10%的概率被分配到v2。
会话自然过渡。不强制切换已有会话。v1用户继续使用v1直到会话自然结束。新会话逐步增加到v2的比例。48小时后，v1的活跃会话自然清零，v2接收100%流量。
版本状态标记。每个Agent实例有版本标记：canary（金丝雀）、gray（灰度）、stable（稳定）。路由器根据标记决定分配策略。

生产级代码实现

AgentGrayRouter：灰度路由器

// deployment/gray-router.ts import { Redis } from 'ioredis'; import { createHash } from 'crypto'; type AgentVersion = 'v1' | 'v2'; type InstanceStatus = 'canary' | 'gray' | 'stable' | 'offline'; type GrayPhase = 'canary' | 'gray_10' | 'gray_30' | 'gray_50' | 'full_rollout' | 'rollback'; interface AgentInstance { instanceId: string; version: AgentVersion; status: InstanceStatus; endpoint: string; activeSessions: number; maxSessions: number; // 单实例最大会话数 healthStatus: 'healthy' | 'degraded' | 'down'; registeredAt: string; } interface GrayConfig { currentPhase: GrayPhase; canaryUsers: string[]; // 金丝雀测试用户ID列表 grayPercentage: number; // 当前灰度比例(0~100) sessionAffinity: boolean; // 是否启用会话亲和（同一用户始终同一版本） observationWindowHours: number; // 灰度观察窗口(小时) rollbackThreshold: number; // 错误率超过此阈值自动回滚 autoProgressEnabled: boolean; // 是否自动推进灰度阶段 } class AgentGrayRouter { private redis: Redis; private config: GrayConfig; private instances: Map<string, AgentInstance> = new Map(); constructor(redisUrl: string, config: GrayConfig) { this.redis = new Redis(redisUrl); this.config = config; this.loadInstances(); } // 核心路由决策：为用户会话选择Agent版本 async routeSession(userId: string, sessionId: string): RouteResult { // 1. 检查是否已有活跃会话（会话亲和） // 为什么先检查已有会话：会话中途切换版本导致状态不兼容， // 同一用户的同一会话必须在同一版本上运行全程 const existingVersion = await this.getExistingSessionVersion(userId, sessionId); if (existingVersion) { return this.routeToVersion(existingVersion, userId); } // 2. 金丝雀测试用户：强制路由到v2 if (this.config.currentPhase === 'canary') { if (this.config.canaryUsers.includes(userId)) { return this.routeToVersion('v2', userId); } return this.routeToVersion('v1', userId); // 非金丝雀用户走v1 } // 3. 灰度阶段：按用户ID哈希分配版本 // 为什么用哈希而非随机数：哈希保证同一用户始终分配到同一版本， // 避免同一用户的不同会话走不同版本 const hash = this.hashUserId(userId); const grayThreshold = this.config.grayPercentage / 100; const targetVersion: AgentVersion = (hash < grayThreshold) ? 'v2' : 'v1'; // 4. 检查目标版本的实例是否有足够容量 const versionInstances = this.getHealthyInstances(targetVersion); if (versionInstances.length === 0) { // 目标版本没有可用实例——降级到另一版本 // 为什么降级而非拒绝：拒绝请求影响用户体验， // 降级到可用版本保证服务连续性 const fallbackVersion: AgentVersion = (targetVersion === 'v2') ? 'v1' : 'v2'; return this.routeToVersion(fallbackVersion, userId); } // 5. 选择负载最低的实例 const selected = this.selectLeastLoaded(versionInstances); // 记录会话版本映射 await this.recordSessionVersion(userId, sessionId, targetVersion); return { version: targetVersion, instanceId: selected.instanceId, endpoint: selected.endpoint, isNewSession: true }; } // 基于用户ID的确定性哈希 // 为什么确定性而非随机：同一用户多次访问始终分配到同一灰度组， // 避免同一用户在不同会话中体验不同版本 private hashUserId(userId: string): number { const hash = createHash('sha256').update(userId).digest('hex'); // 取前8位hex转为数值，再归一化到0~1 const numeric = parseInt(hash.substring(0, 8), 16); return numeric / 0xFFFFFFFF; } // 推进灰度阶段 async progressGrayPhase(): PhaseProgressResult { const metrics = await this.collectGrayMetrics(); // 检查是否需要回滚 // 为什么自动回滚而非人工判断：灰度期间错误率飙升时人工响应太慢， // 自动回滚在5分钟内完成，人工判断可能需要30分钟 if (metrics.errorRate > this.config.rollbackThreshold) { await this.rollback(); return { newPhase: 'rollback', reason: `错误率${metrics.errorRate.toFixed(2)}超过阈值${this.config.rollbackThreshold}`, metrics }; } // 检查观察窗口是否足够 const observationHours = (Date.now() - metrics.phaseStartTime) / 3600000; if (observationHours < this.config.observationWindowHours) { return { newPhase: this.config.currentPhase, reason: `观察窗口不足：${observationHours.toFixed(1)}h < ${this.config.observationWindowHours}h`, metrics }; } // 自动推进到下一阶段 const phaseProgression: Record<GrayPhase, GrayPhase> = { 'canary': 'gray_10', 'gray_10': 'gray_30', 'gray_30': 'gray_50', 'gray_50': 'full_rollout', 'full_rollout': 'full_rollout', 'rollback': 'canary' // 回滚后重新金丝雀测试 }; const nextPhase = phaseProgression[this.config.currentPhase]; const nextPercentage: Record<GrayPhase, number> = { 'canary': 0, 'gray_10': 10, 'gray_30': 30, 'gray_50': 50, 'full_rollout': 100, 'rollback': 0 }; this.config.currentPhase = nextPhase; this.config.grayPercentage = nextPercentage[nextPhase]; await this.saveConfig(); // 更新实例状态标记 this.updateInstanceStatuses(nextPhase); return { newPhase: nextPhase, reason: `观察窗口${observationHours.toFixed(1)}h充足，错误率${metrics.errorRate.toFixed(2)}低于阈值`, metrics }; } // 回滚：所有流量切回v1 private async rollback(): void { this.config.currentPhase = 'rollback'; this.config.grayPercentage = 0; // v2实例标记为offline for (const instance of this.instances.values()) { if (instance.version === 'v2') { instance.status = 'offline'; } } // 清除所有v2会话的版本映射 // 为什么清除而非保留：回滚后v2实例下线，保留映射导致路由指向不可用实例 const v2SessionKeys = await this.redis.keys('session_version:*'); for (const key of v2SessionKeys) { const version = await this.redis.get(key); if (version === 'v2') { await this.redis.del(key); } } await this.saveConfig(); } // 收集灰度指标 private async collectGrayMetrics(): GrayMetrics { // v2版本的错误率和延迟 const v2Metrics = await this.getVersionMetrics('v2'); const v1Metrics = await this.getVersionMetrics('v1'); return { v1ErrorRate: v1Metrics.errorRate, v2ErrorRate: v2Metrics.errorRate, v1P99Latency: v1Metrics.p99Latency, v2P99Latency: v2Metrics.p99Latency, errorRate: v2Metrics.errorRate, // 主要看v2的错误率 phaseStartTime: await this.getPhaseStartTime(), v2ActiveSessions: v2Metrics.activeSessions, v1ActiveSessions: v1Metrics.activeSessions }; } private selectLeastLoaded(instances: AgentInstance[]): AgentInstance { return instances.reduce((min, inst) => inst.activeSessions < min.activeSessions ? inst : min, instances[0]); } private getHealthyInstances(version: AgentVersion): AgentInstance[] { return [...this.instances.values()].filter( inst => inst.version === version && inst.healthStatus === 'healthy' && inst.status !== 'offline' && inst.activeSessions < inst.maxSessions ); } private async recordSessionVersion(userId: string, sessionId: string, version: AgentVersion): void { const key = `session_version:${userId}:${sessionId}`; // 会话版本映射保留24小时（会话最大生命周期） await this.redis.setex(key, 86400, version); } private async getExistingSessionVersion(userId: string, sessionId: string): AgentVersion | null { const key = `session_version:${userId}:${sessionId}`; const version = await this.redis.get(key); return version as AgentVersion | null; } private routeToVersion(version: AgentVersion, userId: string): RouteResult { const instances = this.getHealthyInstances(version); if (instances.length === 0) { throw new Error(`版本${version}没有可用实例`); } const selected = this.selectLeastLoaded(instances); return { version, instanceId: selected.instanceId, endpoint: selected.endpoint, isNewSession: false }; } private async loadInstances(): void { const keys = await this.redis.keys('agent_instance:*'); for (const key of keys) { const data = await this.redis.get(key); if (data) { const instance: AgentInstance = JSON.parse(data); this.instances.set(instance.instanceId, instance); } } } private async saveConfig(): void { await this.redis.set('gray_config', JSON.stringify(this.config)); } private async getPhaseStartTime(): number { const data = await this.redis.get('gray_phase_start_time'); return data ? parseInt(data) : Date.now(); } } interface RouteResult { version: AgentVersion; instanceId: string; endpoint: string; isNewSession: boolean; } interface GrayMetrics { v1ErrorRate: number; v2ErrorRate: number; v1P99Latency: number; v2P99Latency: number; errorRate: number; phaseStartTime: number; v2ActiveSessions: number; v1ActiveSessions: number; } interface PhaseProgressResult { newPhase: GrayPhase; reason: string; metrics: GrayMetrics; }

灰度观察指标收集

# deployment/gray_metrics_collector.py import time from prometheus_client import Counter, Histogram, Gauge # 灰度版本级指标 v1_error_counter = Counter('agent_v1_errors', 'v1 version errors') v2_error_counter = Counter('agent_v2_errors', 'v2 version errors') v1_latency = Histogram('agent_v1_latency_ms', 'v1 latency', buckets=[50, 100, 200, 500, 1000]) v2_latency = Histogram('agent_v2_latency_ms', 'v2 latency', buckets=[50, 100, 200, 500, 1000]) v1_active_sessions = Gauge('agent_v1_active_sessions', 'v1 active sessions') v2_active_sessions = Gauge('agent_v2_active_sessions', 'v2 active sessions') class GrayMetricsCollector: """收集灰度期间两个版本的对比指标""" def __init__(self): self.phase_start_time = time.time() self.v2_error_window = [] # 最近5分钟的v2错误记录 def record_v1_request(self, latency_ms: float, success: bool): v1_latency.observe(latency_ms) if not success: v1_error_counter.inc() def record_v2_request(self, latency_ms: float, success: bool): v2_latency.observe(latency_ms) if not success: v2_error_counter.inc() self.v2_error_window.append(time.time()) def get_v2_error_rate(self) -> float: """计算v2版本最近5分钟的错误率""" # 清理超过5分钟的记录 cutoff = time.time() - 300 self.v2_error_window = [t for t in self.v2_error_window if t > cutoff] # 计算总请求数和错误数 total_v2 = v2_latency._value.get() # 总请求数（近似） errors = len(self.v2_error_window) if total_v2 == 0: return 0.0 return errors / total_v2 def get_v2_p99_latency(self) -> float: """v2版本P99延迟""" # 为什么看P99而非平均：灰度期间关注极端情况， # P99反映最慢的1%请求延迟——这些请求可能是v2版本特有的问题 return v2_latency._value.get() def generate_comparison_report(self) -> str: """生成灰度对比报告""" v2_error_rate = self.get_v2_error_rate() v1_error_rate = v1_error_counter._value.get() / max(1, v1_latency._value.get()) report = [ "=== Agent灰度对比报告 ===", f"当前阶段: {self.get_current_phase()}", f"灰度时长: {(time.time() - self.phase_start_time) / 3600:.1f}小时", "", "版本对比:", f" v1错误率: {v1_error_rate:.4f}", f" v2错误率: {v2_error_rate:.4f}", f" v1 P99延迟: {self.get_v1_p99_latency():.0f}ms", f" v2 P99延迟: {self.get_v2_p99_latency():.0f}ms", f" v1活跃会话: {v1_active_sessions._value.get()}", f" v2活跃会话: {v2_active_sessions._value.get()}", "", "决策建议:", f" v2错误率 > 5%: 建议回滚" if v2_error_rate > 0.05 else f" v2错误率 < 1%: 建议推进下一灰度阶段", f" v2 P99延迟 > v1的2倍: 建议回滚" if self.get_v2_p99_latency() > 2 * self.get_v1_p99_latency() else f" v2延迟在合理范围内" ] return "\n".join(report) def get_current_phase(self) -> str: """从配置获取当前灰度阶段""" # 从Redis或环境变量读取 return "gray_10"

K8s灰度部署配置

# deployment/agent-gray-deployment.yaml # v1稳定版本（基础部署） apiVersion: apps/v1 kind: Deployment metadata: name: agent-v1 labels: app: agent version: v1 status: stable spec: replicas: 10 # 根据灰度比例动态调整 selector: matchLabels: app: agent version: v1 template: metadata: labels: app: agent version: v1 status: stable annotations: prometheus.io/scrape: "true" prometheus.io/port: "9090" spec: containers: - name: agent image: agent:v1.0.0 ports: - containerPort: 8080 - containerPort: 9090 # metrics env: - name: AGENT_VERSION value: "v1" - name: MAX_SESSIONS value: "50" resources: requests: cpu: 200m memory: 512Mi limits: cpu: 500m memory: 1Gi readinessProbe: httpGet: path: /health port: 8080 initialDelaySeconds: 10 periodSeconds: 10 --- # v2灰度版本（金丝雀→灰度→全量） apiVersion: apps/v1 kind: Deployment metadata: name: agent-v2 labels: app: agent version: v2 status: canary # 随灰度阶段更新：canary→gray→stable spec: replicas: 1 # 金丝雀阶段1个实例，灰度阶段逐步增加 selector: matchLabels: app: agent version: v2 template: metadata: labels: app: agent version: v2 status: canary annotations: prometheus.io/scrape: "true" prometheus.io/port: "9090" spec: containers: - name: agent image: agent:v2.0.0 ports: - containerPort: 8080 - containerPort: 9090 env: - name: AGENT_VERSION value: "v2" - name: MAX_SESSIONS value: "50" resources: requests: cpu: 200m memory: 512Mi limits: cpu: 500m memory: 1Gi readinessProbe: httpGet: path: /health port: 8080 initialDelaySeconds: 15 # v2启动可能更慢（新增模型加载） periodSeconds: 10 --- # 灰度路由服务（替代普通Service） # 为什么不用普通Service+权重：普通Service的权重切流是请求级， # 无法保证会话亲和。灰度路由需要会话级控制 apiVersion: v1 kind: Service metadata: name: agent-router spec: type: ClusterIP ports: - port: 8080 targetPort: 8080 selector: app: agent # 选择所有版本的agent pod # 灰度路由器基于pod label进行细粒度路由， # 不依赖Service的selector进行版本区分

灰度推进自动化

# .github/workflows/agent-gray-progress.yml name: Agent Gray Rollout Progress on: schedule: - cron: '0 */6 * * *' # 每6小时评估一次灰度进展 workflow_dispatch: jobs: evaluate: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Evaluate gray metrics run: | # 获取v2版本错误率 V2_ERROR_RATE=$(curl -s 'http://prometheus:9090/api/v1/query' \ --data-urlencode 'query=rate(agent_v2_errors[5m]) / rate(agent_v2_latency_ms_count[5m])' \ | jq -r '.data.result[0].value[1]') echo "v2错误率: ${V2_ERROR_RATE}" # 获取v2 P99延迟 V2_P99=$(curl -s 'http://prometheus:9090/api/v1/query' \ --data-urlencode 'query=histogram_quantile(0.99, rate(agent_v2_latency_ms_bucket[5m]))' \ | jq -r '.data.result[0].value[1]') echo "v2 P99延迟: ${V2_P99}ms" # 决策逻辑 # 为什么用CI做决策而非脚本：灰度推进影响生产流量， # 需要可追溯的决策记录。CI日志记录每次评估和决策 if [ "$(echo "${V2_ERROR_RATE} > 0.05" | bc)" -eq 1 ]; then echo "DECISION=ROLLBACK" >> $GITHUB_ENV echo "原因: v2错误率${V2_ERROR_RATE}超过5%阈值" elif [ "$(echo "${V2_P99} > 1000" | bc)" -eq 1 ]; then echo "DECISION=ROLLBACK" >> $GITHUB_ENV echo "原因: v2 P99延迟${V2_P99}ms超过1000ms阈值" else echo "DECISION=PROGRESS" >> $GITHUB_ENV echo "原因: v2指标正常，推进下一灰度阶段" fi - name: Progress gray phase if: env.DECISION == 'PROGRESS' run: | # 调用灰度路由器API推进阶段 curl -X POST http://agent-router:8080/admin/gray/progress - name: Rollback if: env.DECISION == 'ROLLBACK' run: | # 紧急回滚 curl -X POST http://agent-router:8080/admin/gray/rollback # 通知团队 curl -X POST http://slack-webhook/ \ -H 'Content-type: application/json' \ -d '{"text":"⚠️ Agent灰度回滚：v2错误率超标"}'