AutoGen服务层组件深度解析：Gateway、Registry与Routing

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192

/// <summary>
/// AutoGen Gateway服务 - 统一入口和请求分发
/// </summary>
public class AutoGenGatewayService : BackgroundService
{
    private readonly IServiceRegistry _registry;
    private readonly ILoadBalancer _loadBalancer;
    private readonly IMessageRouter _messageRouter;
    private readonly ILogger<AutoGenGatewayService> _logger;
    private readonly GatewayConfiguration _config;

    public AutoGenGatewayService(
        IServiceRegistry registry,
        ILoadBalancer loadBalancer,
        IMessageRouter messageRouter,
        ILogger<AutoGenGatewayService> logger,
        IOptions<GatewayConfiguration> config)
    {
        _registry = registry;
        _loadBalancer = loadBalancer;
        _messageRouter = messageRouter;
        _logger = logger;
        _config = config.Value;
    }

    /// <summary>
    /// 处理客户端请求的核心方法
    /// </summary>
    public async Task<GatewayResponse> ProcessRequestAsync(
        GatewayRequest request, 
        CancellationToken cancellationToken = default)
    {
        var requestId = Guid.NewGuid().ToString();
        var startTime = DateTime.UtcNow;
        
        try
        {
            // 1. 请求验证和鉴权
            var authResult = await ValidateAndAuthorizeRequest(request);
            if (!authResult.IsValid)
            {
                return CreateErrorResponse(requestId, "认证失败", authResult.ErrorMessage);
            }
            
            // 2. 限流检查
            var rateLimitResult = await CheckRateLimit(request.ClientId, request.RequestType);
            if (rateLimitResult.IsBlocked)
            {
                return CreateErrorResponse(requestId, "请求过于频繁", "请稍后重试");
            }
            
            // 3. 查找目标代理
            var agentLocations = await _registry.FindAgentLocationsAsync(request.TargetAgent);
            if (!agentLocations.Any())
            {
                return CreateErrorResponse(requestId, "代理不存在", $"未找到代理: {request.TargetAgent}");
            }
            
            // 4. 负载均衡选择Worker
            var selectedWorker = await _loadBalancer.SelectWorkerAsync(
                agentLocations, 
                request.LoadBalancingStrategy
            );
            
            // 5. 转发请求到Worker
            var workerResponse = await ForwardRequestToWorker(selectedWorker, request, requestId, cancellationToken);
            
            // 6. 记录请求指标
            await RecordRequestMetrics(requestId, startTime, workerResponse.IsSuccess);
            
            return workerResponse;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "处理网关请求时发生错误: {RequestId}", requestId);
            return CreateErrorResponse(requestId, "内部错误", "系统暂时不可用");
        }
    }

    /// <summary>
    /// 转发请求到Worker节点
    /// </summary>
    private async Task<GatewayResponse> ForwardRequestToWorker(
        WorkerNode worker, 
        GatewayRequest request, 
        string requestId,
        CancellationToken cancellationToken)
    {
        var forwardingStartTime = DateTime.UtcNow;
        
        try
        {
            // 创建gRPC客户端
            using var channel = GrpcChannel.ForAddress(worker.Endpoint);
            var client = new AgentRpc.AgentRpcClient(channel);
            
            // 构造RPC请求
            var rpcRequest = new RpcRequest
            {
                RequestId = requestId,
                Target = new Protobuf.AgentId 
                { 
                    Type = request.TargetAgent.Type, 
                    Key = request.TargetAgent.Key 
                },
                Method = request.Method,
                Payload = SerializePayload(request.Message),
                Metadata = { request.Metadata }
            };
            
            // 设置超时和重试
            var callOptions = new CallOptions(
                deadline: DateTime.UtcNow.AddSeconds(_config.RequestTimeoutSeconds),
                cancellationToken: cancellationToken
            );
            
            // 发送请求
            var stream = client.OpenChannel(callOptions);
            await stream.RequestStream.WriteAsync(new Message { Request = rpcRequest });
            
            // 等待响应
            var responseTask = ReadResponseFromStream(stream.ResponseStream, requestId);
            var response = await responseTask.WaitAsync(cancellationToken);
            
            // 记录转发指标
            var forwardingDuration = DateTime.UtcNow - forwardingStartTime;
            await RecordForwardingMetrics(worker.Id, forwardingDuration, true);
            
            return CreateSuccessResponse(requestId, response);
        }
        catch (RpcException rpcEx)
        {
            _logger.LogWarning("gRPC调用失败: {WorkerId}, {Error}", worker.Id, rpcEx.Message);
            
            // 标记Worker为不健康
            await _registry.MarkWorkerUnhealthyAsync(worker.Id);
            
            return CreateErrorResponse(requestId, "Worker不可用", rpcEx.Message);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "转发请求到Worker失败: {WorkerId}", worker.Id);
            return CreateErrorResponse(requestId, "转发失败", ex.Message);
        }
    }

    /// <summary>
    /// 请求验证和鉴权
    /// </summary>
    private async Task<AuthorizationResult> ValidateAndAuthorizeRequest(GatewayRequest request)
    {
        // 1. 检查请求格式
        if (string.IsNullOrEmpty(request.ClientId))
        {
            return AuthorizationResult.Failed("客户端ID不能为空");
        }
        
        if (request.TargetAgent == null)
        {
            return AuthorizationResult.Failed("目标代理不能为空");
        }
        
        // 2. JWT Token验证
        if (!string.IsNullOrEmpty(request.AuthToken))
        {
            var tokenValidation = await ValidateJwtToken(request.AuthToken);
            if (!tokenValidation.IsValid)
            {
                return AuthorizationResult.Failed("Token验证失败");
            }
        }
        
        // 3. API Key验证
        if (!string.IsNullOrEmpty(request.ApiKey))
        {
            var apiKeyValidation = await ValidateApiKey(request.ApiKey);
            if (!apiKeyValidation.IsValid)
            {
                return AuthorizationResult.Failed("API Key无效");
            }
        }
        
        // 4. 权限检查
        var permissionCheck = await CheckPermissions(request.ClientId, request.TargetAgent);
        if (!permissionCheck.HasPermission)
        {
            return AuthorizationResult.Failed("权限不足");
        }
        
        return AuthorizationResult.Success();
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

class IntelligentLoadBalancer:
    """智能负载均衡器 - 基于多种策略的Worker选择"""
    
    def __init__(self):
        self.strategies = {
            'round_robin': self._round_robin_strategy,
            'least_connections': self._least_connections_strategy,
            'weighted_response_time': self._weighted_response_time_strategy,
            'cpu_based': self._cpu_based_strategy,
            'adaptive': self._adaptive_strategy
        }
        self.worker_stats = {}
        self.strategy_performance = {}
    
    async def select_worker(
        self, 
        available_workers: List[WorkerNode], 
        strategy: str = 'adaptive',
        request_context: Optional[RequestContext] = None
    ) -> WorkerNode:
        """
        选择最佳Worker节点
        
        Args:
            available_workers: 可用Worker节点列表
            strategy: 负载均衡策略
            request_context: 请求上下文信息
            
        Returns:
            WorkerNode: 选中的Worker节点
        """
        
        if not available_workers:
            raise NoAvailableWorkersException("没有可用的Worker节点")
        
        # 过滤健康的Worker
        healthy_workers = [w for w in available_workers if w.is_healthy]
        if not healthy_workers:
            # 如果没有健康的Worker，尝试使用状态未知的Worker
            healthy_workers = [w for w in available_workers if w.health_status != 'unhealthy']
        
        if not healthy_workers:
            raise NoHealthyWorkersException("没有健康的Worker节点")
        
        # 应用负载均衡策略
        if strategy in self.strategies:
            selected_worker = await self.strategies[strategy](healthy_workers, request_context)
        else:
            # 默认使用轮询策略
            selected_worker = await self._round_robin_strategy(healthy_workers, request_context)
        
        # 更新Worker统计信息
        await self._update_worker_stats(selected_worker.id, 'selected')
        
        return selected_worker
    
    async def _adaptive_strategy(
        self, 
        workers: List[WorkerNode], 
        context: Optional[RequestContext]
    ) -> WorkerNode:
        """
        自适应负载均衡策略
        
        综合考虑响应时间、CPU使用率、连接数等多个因素，
        动态选择最优的Worker节点
        """
        
        scores = {}
        
        for worker in workers:
            # 获取Worker实时统计
            stats = await self._get_worker_realtime_stats(worker.id)
            
            # 计算综合得分 (得分越低越好)
            score = 0.0
            
            # 响应时间因子 (权重: 40%)
            avg_response_time = stats.get('avg_response_time', 1.0)
            score += (avg_response_time / 10.0) * 0.4
            
            # CPU使用率因子 (权重: 30%)
            cpu_usage = stats.get('cpu_usage', 0.5)
            score += cpu_usage * 0.3
            
            # 连接数因子 (权重: 20%)
            connection_ratio = stats.get('active_connections', 0) / worker.max_connections
            score += connection_ratio * 0.2
            
            # 错误率因子 (权重: 10%)
            error_rate = stats.get('error_rate', 0.0)
            score += error_rate * 0.1
            
            scores[worker.id] = score
        
        # 选择得分最低的Worker
        best_worker_id = min(scores.keys(), key=lambda k: scores[k])
        return next(w for w in workers if w.id == best_worker_id)
    
    async def _weighted_response_time_strategy(
        self, 
        workers: List[WorkerNode], 
        context: Optional[RequestContext]
    ) -> WorkerNode:
        """基于加权响应时间的负载均衡"""
        
        # 收集所有Worker的响应时间统计
        response_times = {}
        for worker in workers:
            stats = await self._get_worker_realtime_stats(worker.id)
            response_times[worker.id] = stats.get('avg_response_time', 1.0)
        
        # 计算权重 (响应时间越短权重越高)
        max_response_time = max(response_times.values())
        weights = {}
        for worker_id, response_time in response_times.items():
            # 权重 = (最大响应时间 - 当前响应时间) / 最大响应时间
            weights[worker_id] = (max_response_time - response_time) / max_response_time + 0.1
        
        # 加权随机选择
        total_weight = sum(weights.values())
        random_value = random.random() * total_weight
        
        cumulative_weight = 0
        for worker in workers:
            cumulative_weight += weights[worker.id]
            if random_value <= cumulative_weight:
                return worker
        
        # 默认返回第一个Worker
        return workers[0]

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188

class DistributedAgentRegistry:
    """分布式代理注册表 - 管理代理和Worker的注册信息"""
    
    def __init__(self, storage_backend: StorageBackend, consistency_level: str = 'eventual'):
        self.storage = storage_backend
        self.consistency_level = consistency_level
        self.local_cache = TTLCache(maxsize=10000, ttl=300)
        self.update_listeners = []
        self.registry_lock = asyncio.RLock()
    
    async def register_agent_type(
        self, 
        worker_id: str, 
        agent_type: str, 
        capabilities: Dict[str, Any],
        metadata: Optional[Dict[str, Any]] = None
    ) -> RegistrationResult:
        """
        注册代理类型到注册表
        
        Args:
            worker_id: Worker节点标识
            agent_type: 代理类型名称
            capabilities: 代理能力描述
            metadata: 附加元数据
            
        Returns:
            RegistrationResult: 注册结果
        """
        
        async with self.registry_lock:
            try:
                # 1. 验证注册请求
                validation_result = await self._validate_registration(worker_id, agent_type, capabilities)
                if not validation_result.is_valid:
                    return RegistrationResult(
                        success=False,
                        error=f"注册验证失败: {validation_result.error}"
                    )
                
                # 2. 检查是否已存在
                existing_registration = await self._find_existing_registration(worker_id, agent_type)
                
                # 3. 创建注册记录
                registration_record = AgentRegistration(
                    worker_id=worker_id,
                    agent_type=agent_type,
                    capabilities=capabilities,
                    metadata=metadata or {},
                    registered_at=datetime.utcnow(),
                    last_heartbeat=datetime.utcnow(),
                    status='active'
                )
                
                # 4. 持久化注册信息
                if existing_registration:
                    await self.storage.update_registration(registration_record)
                else:
                    await self.storage.create_registration(registration_record)
                
                # 5. 更新本地缓存
                cache_key = f"agent:{agent_type}:workers"
                cached_workers = self.local_cache.get(cache_key, [])
                if worker_id not in [w['worker_id'] for w in cached_workers]:
                    cached_workers.append({
                        'worker_id': worker_id,
                        'capabilities': capabilities,
                        'registered_at': registration_record.registered_at.isoformat()
                    })
                    self.local_cache[cache_key] = cached_workers
                
                # 6. 通知监听器
                await self._notify_registration_listeners('agent_registered', registration_record)
                
                return RegistrationResult(
                    success=True,
                    registration_id=f"{worker_id}:{agent_type}"
                )
            
            except Exception as e:
                _logger.error(f"代理注册失败: {e}", exc_info=True)
                return RegistrationResult(
                    success=False,
                    error=f"注册过程中发生错误: {str(e)}"
                )
    
    async def find_agent_locations(self, agent_id: AgentId) -> List[WorkerLocation]:
        """
        查找代理的位置信息
        
        Args:
            agent_id: 目标代理ID
            
        Returns:
            List[WorkerLocation]: Worker位置列表
        """
        
        # 1. 检查本地缓存
        cache_key = f"agent:{agent_id.type}:locations"
        cached_locations = self.local_cache.get(cache_key)
        
        if cached_locations and self.consistency_level == 'eventual':
            return [WorkerLocation(**loc) for loc in cached_locations]
        
        # 2. 从持久化存储查询
        registrations = await self.storage.find_registrations_by_agent_type(agent_id.type)
        
        # 3. 过滤活跃的Worker
        active_registrations = [
            reg for reg in registrations 
            if reg.status == 'active' and self._is_worker_responsive(reg.worker_id)
        ]
        
        # 4. 构造位置信息
        locations = []
        for registration in active_registrations:
            worker_info = await self.storage.get_worker_info(registration.worker_id)
            if worker_info:
                location = WorkerLocation(
                    worker_id=registration.worker_id,
                    endpoint=worker_info.endpoint,
                    capabilities=registration.capabilities,
                    load_score=await self._calculate_load_score(registration.worker_id),
                    last_seen=registration.last_heartbeat
                )
                locations.append(location)
        
        # 5. 更新缓存
        if locations:
            self.local_cache[cache_key] = [loc.to_dict() for loc in locations]
        
        return locations
    
    async def heartbeat_update(self, worker_id: str, health_info: Dict[str, Any]) -> None:
        """
        更新Worker心跳信息
        
        Args:
            worker_id: Worker标识
            health_info: 健康状态信息
        """
        
        try:
            # 1. 更新心跳时间戳
            await self.storage.update_worker_heartbeat(worker_id, datetime.utcnow())
            
            # 2. 更新健康状态
            await self.storage.update_worker_health(worker_id, health_info)
            
            # 3. 检查Worker状态变化
            previous_status = await self._get_worker_previous_status(worker_id)
            current_status = health_info.get('status', 'unknown')
            
            if previous_status != current_status:
                # 状态发生变化，通知监听器
                await self._notify_registration_listeners('worker_status_changed', {
                    'worker_id': worker_id,
                    'previous_status': previous_status,
                    'current_status': current_status,
                    'health_info': health_info
                })
                
                # 清除相关缓存
                await self._invalidate_worker_cache(worker_id)
        
        except Exception as e:
            _logger.error(f"更新Worker心跳失败: {worker_id}, {e}")
    
    async def _calculate_load_score(self, worker_id: str) -> float:
        """计算Worker负载得分"""
        
        stats = await self.storage.get_worker_stats(worker_id)
        if not stats:
            return 1.0  # 默认负载得分
        
        # 综合多个指标计算负载得分
        cpu_score = stats.get('cpu_usage', 0.5)
        memory_score = stats.get('memory_usage', 0.5) 
        connection_score = stats.get('connection_ratio', 0.5)
        response_time_score = min(stats.get('avg_response_time', 1.0) / 5.0, 1.0)
        
        # 加权平均
        load_score = (cpu_score * 0.3 + 
                     memory_score * 0.2 + 
                     connection_score * 0.3 + 
                     response_time_score * 0.2)
        
        return max(0.0, min(1.0, load_score))  # 限制在[0,1]范围内

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178

class IntelligentMessageRouter:
    """智能消息路由器 - 基于内容和上下文的智能路由"""
    
    def __init__(self):
        self.routing_rules = RoutingRuleEngine()
        self.route_cache = TTLCache(maxsize=50000, ttl=600)  # 10分钟缓存
        self.routing_history = deque(maxlen=1000)  # 保留最近1000次路由记录
        self.ml_router = MachineLearningRouter()
    
    async def route_message(
        self, 
        message: Any, 
        sender: Optional[AgentId], 
        routing_hint: Optional[RoutingHint] = None
    ) -> RoutingDecision:
        """
        智能路由消息到最合适的代理
        
        Args:
            message: 要路由的消息
            sender: 发送方代理ID
            routing_hint: 路由提示信息
            
        Returns:
            RoutingDecision: 路由决策结果
        """
        
        routing_context = RoutingContext(
            message=message,
            sender=sender,
            timestamp=datetime.utcnow(),
            hint=routing_hint
        )
        
        # 1. 快速规则路由
        rule_decision = await self._apply_routing_rules(routing_context)
        if rule_decision.confidence > 0.9:
            await self._record_routing_decision(routing_context, rule_decision, 'rule_based')
            return rule_decision
        
        # 2. 基于历史的相似性路由  
        similarity_decision = await self._similarity_based_routing(routing_context)
        if similarity_decision.confidence > 0.8:
            await self._record_routing_decision(routing_context, similarity_decision, 'similarity_based')
            return similarity_decision
        
        # 3. 机器学习路由
        ml_decision = await self.ml_router.predict_best_agent(routing_context)
        if ml_decision.confidence > 0.7:
            await self._record_routing_decision(routing_context, ml_decision, 'ml_based')
            return ml_decision
        
        # 4. 默认路由 (负载均衡)
        default_decision = await self._default_load_balanced_routing(routing_context)
        await self._record_routing_decision(routing_context, default_decision, 'default')
        
        return default_decision
    
    async def _apply_routing_rules(self, context: RoutingContext) -> RoutingDecision:
        """应用路由规则引擎"""
        
        message_content = self._extract_message_content(context.message)
        
        # 内容分析路由规则
        content_rules = [
            {
                'pattern': r'天气|气温|温度',
                'target_agent_type': 'WeatherAgent',
                'confidence': 0.95
            },
            {
                'pattern': r'计算|数学|运算',
                'target_agent_type': 'CalculatorAgent', 
                'confidence': 0.95
            },
            {
                'pattern': r'代码|编程|程序',
                'target_agent_type': 'CoderAgent',
                'confidence': 0.9
            },
            {
                'pattern': r'翻译|translate',
                'target_agent_type': 'TranslatorAgent',
                'confidence': 0.9
            }
        ]
        
        # 应用规则匹配
        for rule in content_rules:
            if re.search(rule['pattern'], message_content, re.IGNORECASE):
                return RoutingDecision(
                    target_agent_type=rule['target_agent_type'],
                    confidence=rule['confidence'],
                    reasoning=f"匹配规则: {rule['pattern']}",
                    routing_strategy='rule_based'
                )
        
        # 发送方路由规则
        if context.sender:
            sender_rules = await self._get_sender_based_rules(context.sender.type)
            for rule in sender_rules:
                if rule.matches(context):
                    return RoutingDecision(
                        target_agent_type=rule.target_type,
                        confidence=rule.confidence,
                        reasoning=f"发送方规则: {rule.description}",
                        routing_strategy='sender_based'
                    )
        
        # 无匹配规则
        return RoutingDecision(
            target_agent_type=None,
            confidence=0.0,
            reasoning="无匹配的路由规则"
        )
    
    async def _similarity_based_routing(self, context: RoutingContext) -> RoutingDecision:
        """基于历史相似性的路由"""
        
        # 1. 提取消息特征
        message_features = await self._extract_message_features(context.message)
        
        # 2. 在历史记录中查找相似消息
        similar_records = []
        for historical_record in self.routing_history:
            similarity = await self._calculate_similarity(message_features, historical_record.features)
            if similarity > 0.7:  # 相似度阈值
                similar_records.append((historical_record, similarity))
        
        if not similar_records:
            return RoutingDecision(confidence=0.0, reasoning="无相似历史记录")
        
        # 3. 按相似度排序
        similar_records.sort(key=lambda x: x[1], reverse=True)
        
        # 4. 分析最相似记录的路由结果
        most_similar_record, similarity = similar_records[0]
        
        if most_similar_record.routing_success_rate > 0.8:
            return RoutingDecision(
                target_agent_type=most_similar_record.target_agent_type,
                confidence=similarity * most_similar_record.routing_success_rate,
                reasoning=f"基于相似度 {similarity:.2f} 的历史记录",
                routing_strategy='similarity_based'
            )
        
        return RoutingDecision(confidence=0.0, reasoning="相似记录成功率过低")
    
    async def update_routing_feedback(
        self, 
        routing_id: str, 
        success: bool, 
        response_time: float,
        user_satisfaction: Optional[float] = None
    ) -> None:
        """更新路由反馈信息，用于优化路由算法"""
        
        # 1. 查找路由记录
        routing_record = await self._find_routing_record(routing_id)
        if not routing_record:
            return
        
        # 2. 更新反馈信息
        routing_record.success = success
        routing_record.response_time = response_time
        routing_record.user_satisfaction = user_satisfaction
        routing_record.feedback_updated_at = datetime.utcnow()
        
        # 3. 更新路由成功率统计
        await self._update_routing_success_rate(
            routing_record.target_agent_type,
            routing_record.routing_strategy,
            success
        )
        
        # 4. 训练机器学习模型
        if len(self.routing_history) % 100 == 0:  # 每100次反馈重训练一次
            await self.ml_router.retrain_model(list(self.routing_history))

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190

class AdvancedSubscriptionManager:
    """高级订阅管理器 - 支持动态订阅和复杂匹配规则"""
    
    def __init__(self):
        self.subscriptions = {}
        self.subscription_index = SubscriptionIndex()  # 高效的订阅索引
        self.dynamic_subscriptions = {}  # 动态订阅
        self.subscription_stats = {}
    
    async def add_subscription(
        self, 
        subscription: Subscription,
        is_dynamic: bool = False,
        ttl: Optional[int] = None
    ) -> str:
        """
        添加订阅规则
        
        Args:
            subscription: 订阅规则
            is_dynamic: 是否为动态订阅
            ttl: 生存时间（秒）
            
        Returns:
            str: 订阅ID
        """
        
        subscription_id = subscription.id or str(uuid.uuid4())
        
        # 1. 验证订阅规则
        validation_result = await self._validate_subscription(subscription)
        if not validation_result.is_valid:
            raise ValueError(f"订阅验证失败: {validation_result.error}")
        
        # 2. 检查重复订阅
        existing_subscription = await self._find_duplicate_subscription(subscription)
        if existing_subscription:
            return existing_subscription.id
        
        # 3. 存储订阅
        subscription.id = subscription_id
        self.subscriptions[subscription_id] = subscription
        
        # 4. 更新索引
        await self.subscription_index.add_subscription(subscription)
        
        # 5. 处理动态订阅
        if is_dynamic:
            self.dynamic_subscriptions[subscription_id] = {
                'subscription': subscription,
                'created_at': datetime.utcnow(),
                'ttl': ttl,
                'access_count': 0
            }
            
            # 设置过期定时器
            if ttl:
                asyncio.create_task(self._schedule_subscription_expiry(subscription_id, ttl))
        
        # 6. 初始化统计信息
        self.subscription_stats[subscription_id] = {
            'match_count': 0,
            'success_count': 0,
            'average_processing_time': 0.0,
            'last_matched': None
        }
        
        return subscription_id
    
    async def match_subscriptions(self, topic_id: TopicId) -> List[SubscriptionMatch]:
        """
        匹配订阅规则
        
        Args:
            topic_id: 主题ID
            
        Returns:
            List[SubscriptionMatch]: 匹配的订阅列表
        """
        
        # 1. 使用索引快速查找候选订阅
        candidate_subscriptions = await self.subscription_index.find_candidates(topic_id)
        
        # 2. 精确匹配验证
        matches = []
        for subscription in candidate_subscriptions:
            if subscription.is_match(topic_id):
                try:
                    agent_id = subscription.map_to_agent(topic_id)
                    
                    match = SubscriptionMatch(
                        subscription_id=subscription.id,
                        subscription=subscription,
                        target_agent_id=agent_id,
                        match_confidence=self._calculate_match_confidence(subscription, topic_id)
                    )
                    matches.append(match)
                    
                    # 更新匹配统计
                    await self._update_subscription_stats(subscription.id, 'match')
                
                except Exception as e:
                    _logger.warning(f"订阅映射失败: {subscription.id}, {e}")
        
        # 3. 按匹配度排序
        matches.sort(key=lambda m: m.match_confidence, reverse=True)
        
        return matches
    
    async def optimize_subscriptions(self) -> OptimizationResult:
        """优化订阅规则 - 基于使用统计进行优化"""
        
        optimization_result = OptimizationResult()
        
        # 1. 分析低效订阅
        low_usage_subscriptions = []
        for sub_id, stats in self.subscription_stats.items():
            if stats['match_count'] == 0 and self._is_subscription_old(sub_id):
                low_usage_subscriptions.append(sub_id)
        
        # 2. 清理无用订阅
        for sub_id in low_usage_subscriptions:
            await self.remove_subscription(sub_id)
            optimization_result.removed_subscriptions.append(sub_id)
        
        # 3. 优化订阅索引
        await self.subscription_index.rebuild_index()
        
        # 4. 合并重复订阅
        duplicate_groups = await self._find_duplicate_subscription_groups()
        for group in duplicate_groups:
            merged_subscription = await self._merge_subscriptions(group)
            for old_sub_id in group[1:]:  # 保留第一个，删除其余
                await self.remove_subscription(old_sub_id)
                optimization_result.merged_subscriptions.append(old_sub_id)
        
        return optimization_result

class SubscriptionIndex:
    """订阅索引 - 提供高效的订阅查找"""
    
    def __init__(self):
        # 多维索引结构
        self.type_index = {}           # 按类型索引
        self.prefix_index = {}         # 按前缀索引  
        self.pattern_index = {}        # 按模式索引
        self.agent_type_index = {}     # 按代理类型索引
    
    async def add_subscription(self, subscription: Subscription) -> None:
        """添加订阅到索引"""
        
        if isinstance(subscription, TypeSubscription):
            # 类型订阅索引
            if subscription.topic_type not in self.type_index:
                self.type_index[subscription.topic_type] = []
            self.type_index[subscription.topic_type].append(subscription)
        
        elif isinstance(subscription, TypePrefixSubscription):
            # 前缀订阅索引
            prefix = subscription.topic_type_prefix
            if prefix not in self.prefix_index:
                self.prefix_index[prefix] = []
            self.prefix_index[prefix].append(subscription)
        
        # 代理类型反向索引
        agent_type = subscription.agent_type
        if agent_type not in self.agent_type_index:
            self.agent_type_index[agent_type] = []
        self.agent_type_index[agent_type].append(subscription)
    
    async def find_candidates(self, topic_id: TopicId) -> List[Subscription]:
        """查找候选订阅"""
        
        candidates = set()
        
        # 1. 精确类型匹配
        if topic_id.type in self.type_index:
            candidates.update(self.type_index[topic_id.type])
        
        # 2. 前缀匹配
        for prefix, subscriptions in self.prefix_index.items():
            if topic_id.type.startswith(prefix):
                candidates.update(subscriptions)
        
        # 3. 模式匹配 (如果有的话)
        for pattern, subscriptions in self.pattern_index.items():
            if re.match(pattern, topic_id.type):
                candidates.update(subscriptions)
        
        return list(candidates)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129

/// <summary>
/// AutoGen综合监控服务
/// </summary>
public class AutoGenMonitoringService : BackgroundService
{
    private readonly IMetricsCollector _metricsCollector;
    private readonly IHealthChecker _healthChecker;
    private readonly IAlertManager _alertManager;
    private readonly ILogger<AutoGenMonitoringService> _logger;
    
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        while (!stoppingToken.IsCancellationRequested)
        {
            try
            {
                // 1. 收集系统指标
                await CollectSystemMetrics();
                
                // 2. 执行健康检查
                await PerformHealthChecks();
                
                // 3. 分析异常模式
                await AnalyzeAnomalyPatterns();
                
                // 4. 生成智能告警
                await GenerateIntelligentAlerts();
                
                await Task.Delay(TimeSpan.FromSeconds(30), stoppingToken);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "监控服务执行异常");
                await Task.Delay(TimeSpan.FromMinutes(1), stoppingToken);
            }
        }
    }
    
    /// <summary>
    /// 收集系统指标
    /// </summary>
    private async Task CollectSystemMetrics()
    {
        // 代理级指标
        var agentMetrics = await _metricsCollector.CollectAgentMetrics();
        
        // 运行时指标
        var runtimeMetrics = await _metricsCollector.CollectRuntimeMetrics();
        
        // 网络指标
        var networkMetrics = await _metricsCollector.CollectNetworkMetrics();
        
        // 业务指标
        var businessMetrics = await _metricsCollector.CollectBusinessMetrics();
        
        // 聚合并存储指标
        var aggregatedMetrics = new SystemMetrics
        {
            AgentMetrics = agentMetrics,
            RuntimeMetrics = runtimeMetrics,
            NetworkMetrics = networkMetrics,
            BusinessMetrics = businessMetrics,
            Timestamp = DateTime.UtcNow
        };
        
        await _metricsCollector.StoreMetrics(aggregatedMetrics);
    }
    
    /// <summary>
    /// 执行健康检查
    /// </summary>
    private async Task PerformHealthChecks()
    {
        var healthCheckTasks = new[]
        {
            _healthChecker.CheckGatewayHealth(),
            _healthChecker.CheckRegistryHealth(),
            _healthChecker.CheckRoutingHealth(),
            _healthChecker.CheckWorkerNodesHealth(),
            _healthChecker.CheckDatabaseHealth(),
            _healthChecker.CheckCacheHealth()
        };
        
        var healthResults = await Task.WhenAll(healthCheckTasks);
        
        // 汇总健康状态
        var overallHealth = new OverallHealthStatus
        {
            ComponentHealth = healthResults.ToDictionary(r => r.ComponentName, r => r),
            OverallStatus = healthResults.All(r => r.IsHealthy) ? "健康" : "异常",
            LastChecked = DateTime.UtcNow
        };
        
        await _healthChecker.UpdateOverallHealthStatus(overallHealth);
        
        // 如果有组件不健康，触发告警
        var unhealthyComponents = healthResults.Where(r => !r.IsHealthy).ToList();
        if (unhealthyComponents.Any())
        {
            await _alertManager.TriggerHealthAlert(unhealthyComponents);
        }
    }
    
    /// <summary>
    /// 分析异常模式
    /// </summary>
    private async Task AnalyzeAnomalyPatterns()
    {
        // 1. 获取最近的错误日志
        var recentErrors = await _metricsCollector.GetRecentErrors(TimeSpan.FromHours(1));
        
        // 2. 分析错误模式
        var errorPatterns = AnalyzeErrorPatterns(recentErrors);
        
        // 3. 检测异常趋势
        var anomalyTrends = await DetectAnomalyTrends();
        
        // 4. 生成分析报告
        var anomalyReport = new AnomalyAnalysisReport
        {
            ErrorPatterns = errorPatterns,
            AnomalyTrends = anomalyTrends,
            RecommendedActions = GenerateRecommendedActions(errorPatterns, anomalyTrends),
            AnalysisTimestamp = DateTime.UtcNow
        };
        
        await _metricsCollector.StoreAnomalyReport(anomalyReport);
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118

class IntelligentAlertManager:
    """智能告警管理器 - 减少告警噪音，提高告警质量"""
    
    def __init__(self):
        self.alert_rules = AlertRuleEngine()
        self.alert_history = deque(maxlen=10000)
        self.silence_rules = {}
        self.escalation_policies = {}
        self.ml_anomaly_detector = AnomalyDetector()
    
    async def process_potential_alert(
        self, 
        metric_name: str, 
        current_value: float, 
        context: Dict[str, Any]
    ) -> Optional[Alert]:
        """
        处理潜在告警
        
        Args:
            metric_name: 指标名称
            current_value: 当前指标值
            context: 上下文信息
            
        Returns:
            Optional[Alert]: 如果需要告警则返回Alert对象
        """
        
        # 1. 检查静默规则
        if await self._is_silenced(metric_name, context):
            return None
        
        # 2. 应用告警规则
        alert_rule = self.alert_rules.get_rule(metric_name)
        if not alert_rule:
            return None
        
        # 3. 阈值检查
        if not alert_rule.is_threshold_breached(current_value):
            return None
        
        # 4. 异常检测
        is_anomaly = await self.ml_anomaly_detector.is_anomaly(
            metric_name, current_value, context
        )
        
        # 5. 告警降噪 - 检查是否为重复告警
        if await self._is_duplicate_alert(metric_name, current_value, context):
            return None
        
        # 6. 创建告警
        alert = Alert(
            id=str(uuid.uuid4()),
            metric_name=metric_name,
            current_value=current_value,
            threshold_value=alert_rule.threshold,
            severity=self._calculate_severity(current_value, alert_rule),
            context=context,
            is_anomaly=is_anomaly,
            created_at=datetime.utcnow()
        )
        
        # 7. 应用升级策略
        escalation_policy = self.escalation_policies.get(alert.severity)
        if escalation_policy:
            alert.escalation_policy = escalation_policy
        
        # 8. 记录告警历史
        self.alert_history.append(alert)
        
        return alert
    
    async def _is_duplicate_alert(
        self, 
        metric_name: str, 
        value: float, 
        context: Dict[str, Any],
        time_window: timedelta = timedelta(minutes=5)
    ) -> bool:
        """检查是否为重复告警"""
        
        current_time = datetime.utcnow()
        threshold_time = current_time - time_window
        
        # 查找最近的相似告警
        for alert in reversed(self.alert_history):
            if alert.created_at < threshold_time:
                break
            
            if (alert.metric_name == metric_name and
                abs(alert.current_value - value) / max(alert.current_value, value) < 0.1):  # 10%差异内
                return True
        
        return False
    
    async def auto_resolve_alerts(self) -> List[str]:
        """自动解决告警"""
        
        resolved_alerts = []
        active_alerts = [alert for alert in self.alert_history if not alert.is_resolved]
        
        for alert in active_alerts:
            # 检查告警条件是否仍然存在
            current_value = await self._get_current_metric_value(alert.metric_name)
            alert_rule = self.alert_rules.get_rule(alert.metric_name)
            
            if not alert_rule.is_threshold_breached(current_value):
                # 条件已恢复，自动解决告警
                alert.is_resolved = True
                alert.resolved_at = datetime.utcnow()
                alert.resolution_reason = "指标已恢复正常"
                
                resolved_alerts.append(alert.id)
                
                # 发送恢复通知
                await self._send_resolution_notification(alert)
        
        return resolved_alerts

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240