企业级 AI Agent 金丝雀灰度发布与流量路由实战：面向 Prompt 迭代、模型升级与工具链变更的渐进式交付架构

/**
 * 企业级 AI Agent 金丝雀灰度路由与健康评估引擎
 * 核心功能：会话黏性持久化、确定性权重切流、多维 Agent 认知指标聚合与熔断决策
 */

import { createHash } from 'node:crypto';

// 部署版本类型枚举
export type AgentDeploymentVersion = 'baseline' | 'canary';

// 路由决策结果接口
export interface RoutingDecision {
  version: AgentDeploymentVersion; // 最终路由的版本
  isCanary: boolean;               // 是否属于金丝雀流量
  reason: string;                 // 路由判决原因说明
  assignedAt: number;             // 决策时间戳
}

// 智能体运行时遥测指标上报结构
export interface AgentExecutionMetrics {
  sessionId: string;
  version: AgentDeploymentVersion;
  stepCount: number;              // 规划执行轨迹步数
  totalTokens: number;            // 消耗 Token 总数
  toolCallSuccess: boolean;       // 工具调用是否成功
  paramValidationError: boolean;  // 参数解析是否失败
  hasSemanticLoop: boolean;       // 是否检测到语义循环
  latencyMs: number;              // 任务耗时（毫秒）
}

// 金丝雀版本健康度评估统计报表
export interface CanaryHealthReport {
  sampleCount: number;
  avgStepCount: number;
  avgTokens: number;
  paramErrorRate: number;         // 参数错误率
  toolFailureRate: number;        // 工具失败率
  loopDetectedRate: number;       // 死循环率
  isHealthy: boolean;             // 是否通过金丝雀健康检查
  reasonsForFailure: string[];    // 阻断或不达标的具体原因
}

// 模拟的高可用分布式状态存储（生产环境中替换为 Redis 客户端）
export class DistributedSessionStore {
  private store = new Map<string, { version: AgentDeploymentVersion; expiresAt: number }>();

  // 获取会话绑定的版本
  async getSessionVersion(sessionId: string): Promise<AgentDeploymentVersion | null> {
    const record = this.store.get(sessionId);
    if (!record) return null;
    if (Date.now() > record.expiresAt) {
      this.store.delete(sessionId);
      return null;
    }
    return record.version;
  }

  // 写入会话版本绑定关系并设置 TTL（默认 2 小时）
  async bindSessionVersion(sessionId: string, version: AgentDeploymentVersion, ttlSeconds = 7200): Promise<void> {
    this.store.set(sessionId, {
      version,
      expiresAt: Date.now() + ttlSeconds * 1000
    });
  }
}

// 金丝雀流量路由器
export class AgentCanaryRouter {
  private sessionStore: DistributedSessionStore;
  private canaryWeightPercentage: number; // 金丝雀流量百分比 (0 - 100)
  private canaryOverrideWhitelist: Set<string>; // 白名单用户或测试租户
  private circuitBreakerTripped: boolean = false; // 熔断开关

  constructor(sessionStore: DistributedSessionStore, initialWeight = 5) {
    this.sessionStore = sessionStore;
    this.canaryWeightPercentage = Math.min(100, Math.max(0, initialWeight));
    this.canaryOverrideWhitelist = new Set();
  }

  // 动态更新金丝雀切流权重
  public updateWeight(weightPercentage: number): void {
    if (this.circuitBreakerTripped) {
      throw new Error('金丝雀已触发安全熔断，禁止增加流量权重，请先排查故障并复位熔断状态');
    }
    this.canaryWeightPercentage = Math.min(100, Math.max(0, weightPercentage));
  }

  // 触发紧急熔断：瞬间将金丝雀权重置零
  public tripCircuitBreaker(reason: string): void {
    this.circuitBreakerTripped = true;
    this.canaryWeightPercentage = 0;
    console.error(`[CRITICAL] 智能体金丝雀发布已触发紧急熔断，所有新流量即刻回滚至 Baseline！原因: ${reason}`);
  }

  // 手动复位熔断状态
  public resetCircuitBreaker(): void {
    this.circuitBreakerTripped = false;
    this.canaryWeightPercentage = 0; // 复位后仍需从 0% 谨慎重新步进
  }

  // 添加内测白名单标识（如工号或特定租户 ID）
  public addWhitelistTenant(tenantOrUserId: string): void {
    this.canaryOverrideWhitelist.add(tenantOrUserId);
  }

  // 核心路由分发判决
  public async routeRequest(
    sessionId: string,
    callerIdentity?: string,
    overrideHeader?: boolean
  ): Promise<RoutingDecision> {
    const now = Date.now();

    // 1. 若触发了熔断，强行路由至 Baseline
    if (this.circuitBreakerTripped) {
      return {
        version: 'baseline',
        isCanary: false,
        reason: 'circuit_breaker_tripped_forced_baseline',
        assignedAt: now
      };
    }

    // 2. 显式 Header 调试覆盖或白名单特权租户
    if (overrideHeader || (callerIdentity && this.canaryOverrideWhitelist.has(callerIdentity))) {
      return {
        version: 'canary',
        isCanary: true,
        reason: 'whitelist_or_debug_header_match',
        assignedAt: now
      };
    }

    // 3. 检查会话黏性持久化记录（必须保证长会话中途不飘移）
    const existingVersion = await this.sessionStore.getSessionVersion(sessionId);
    if (existingVersion) {
      return {
        version: existingVersion,
        isCanary: existingVersion === 'canary',
        reason: 'sticky_session_affinity_hit',
        assignedAt: now
      };
    }

    // 4. 首轮全新会话：基于 SHA-256 哈希取模实现确定性离散切流
    const hashInput = `${sessionId}_bxagent_canary_salt_2026`;
    const hashValue = createHash('sha256').update(hashInput).digest('hex');
    // 取前 8 位十六进制转为整数并对 100 取模
    const bucket = parseInt(hashValue.substring(0, 8), 16) % 100;

    const assignedVersion: AgentDeploymentVersion = bucket < this.canaryWeightPercentage ? 'canary' : 'baseline';

    // 5. 写入分布式持久化缓存以保证后续轮次强一致黏性
    await this.sessionStore.bindSessionVersion(sessionId, assignedVersion);

    return {
      version: assignedVersion,
      isCanary: assignedVersion === 'canary',
      reason: `probabilistic_bucket_${bucket}_lt_weight_${this.canaryWeightPercentage}`,
      assignedAt: now
    };
  }
}

// 金丝雀健康评估与自动化门禁守卫
export class CanaryHealthEvaluator {
  private baselineMetrics: AgentExecutionMetrics[] = [];
  private canaryMetrics: AgentExecutionMetrics[] = [];

  // 上报执行遥测指标
  public recordExecution(metrics: AgentExecutionMetrics): void {
    if (metrics.version === 'canary') {
      this.canaryMetrics.push(metrics);
    } else {
      this.baselineMetrics.push(metrics);
    }
  }

  // 执行自动化金丝雀分析（ACA）并返回健康诊断
  public evaluateCanaryHealth(): CanaryHealthReport {
    const count = this.canaryMetrics.length;
    if (count < 20) {
      // 样本数过少时保持谨慎观察，不轻易放量也不盲目报错
      return {
        sampleCount: count,
        avgStepCount: 0,
        avgTokens: 0,
        paramErrorRate: 0,
        toolFailureRate: 0,
        loopDetectedRate: 0,
        isHealthy: true,
        reasonsForFailure: ['insufficient_sample_size_monitoring']
      };
    }

    // 计算金丝雀各项关键均值与异常率
    const totalSteps = this.canaryMetrics.reduce((sum, m) => sum + m.stepCount, 0);
    const totalTokens = this.canaryMetrics.reduce((sum, m) => sum + m.totalTokens, 0);
    const paramErrors = this.canaryMetrics.filter(m => m.paramValidationError).length;
    const toolFailures = this.canaryMetrics.filter(m => !m.toolCallSuccess).length;
    const loopErrors = this.canaryMetrics.filter(m => m.hasSemanticLoop).length;

    const avgStepCount = totalSteps / count;
    const avgTokens = totalTokens / count;
    const paramErrorRate = paramErrors / count;
    const toolFailureRate = toolFailures / count;
    const loopDetectedRate = loopErrors / count;

    const failures: string[] = [];

    // 门禁红线 1：参数解析失败率超过 1%
    if (paramErrorRate > 0.01) {
      failures.push(`参数解析校验失败率异常 (${(paramErrorRate * 100).toFixed(2)}% > 1.00%)`);
    }

    // 门禁红线 2：工具调用业务异常率超过 2%
    if (toolFailureRate > 0.02) {
      failures.push(`外部工具调用失败率超标 (${(toolFailureRate * 100).toFixed(2)}% > 2.00%)`);
    }

    // 门禁红线 3：死循环检测率超过 0.5%
    if (loopDetectedRate > 0.005) {
      failures.push(`检测到推理死循环或反复重复调用 (${(loopDetectedRate * 100).toFixed(2)}% > 0.50%)`);
    }

    // 门禁红线 4：若 Baseline 存在充足样本，对比步长与 Token 膨胀率
    if (this.baselineMetrics.length >= 20) {
      const baseSteps = this.baselineMetrics.reduce((sum, m) => sum + m.stepCount, 0) / this.baselineMetrics.length;
      const baseTokens = this.baselineMetrics.reduce((sum, m) => sum + m.totalTokens, 0) / this.baselineMetrics.length;

      // 轨迹步长膨胀超过 50%
      if (avgStepCount > baseSteps * 1.5) {
        failures.push(`推理轨迹平均步数严重劣化 (Canary: ${avgStepCount.toFixed(1)} vs Base: ${baseSteps.toFixed(1)})`);
      }

      // Token 消耗增长超过 30%
      if (avgTokens > baseTokens * 1.3) {
        failures.push(`平均 Token 消耗突增超出安全阈值 (Canary: ${avgTokens.toFixed(0)} vs Base: ${baseTokens.toFixed(0)})`);
      }
    }

    return {
      sampleCount: count,
      avgStepCount,
      avgTokens,
      paramErrorRate,
      toolFailureRate,
      loopDetectedRate,
      isHealthy: failures.length === 0,
      reasonsForFailure: failures
    };
  }
}