V6.5: 1000 samples (500+500), MTP val+entropy, EWC+W8A8 eval investigation, integrate gru-ring modules
c2992d0 verified | """ | |
| circular_orchestration — Orquestração circular distribuída granulável. | |
| Sistema de orquestração circular que resolve Tarefas decompostas em | |
| subtarefas com Ações Monitoráveis (feedback) e Predição de Estado (Evolução). | |
| Arquitetura circular | |
| -------------------- | |
| Ao contrário do PEV linear (Plan→Evolve→Verify), a orquestração circular | |
| é um loop contínuo onde cada agente pass a tarefa ao próximo, e o último | |
| retorna ao primeiro, formando um anel: | |
| Agent_0 → Agent_1 → ... → Agent_{N-1} → Agent_0 → ... | |
| Cada agente executa uma fase (Decompose, Execute, Monitor, Predict, Adjust) | |
| e passa o estado ao próximo. O loop continua até convergência. | |
| Análise matemática profunda | |
| --------------------------- | |
| 1. Álgebra de estado (espaço de estados): | |
| Estado do sistema: s_t = (task_state, agent_states, environment) | |
| Transição: s_{t+1} = Φ(s_t, a_t, ω_t) | |
| onde a_t é a ação tomada e ω_t é ruído/observação. | |
| Φ é composição circular: Φ = φ_{N-1} ∘ ... ∘ φ_1 ∘ φ_0 | |
| onde φ_i é a função de transição do agente i. | |
| 2. Predição de estado via Filtro de Kalman estendido: | |
| x̂_{t+1|t} = f(x̂_t, u_t) (predição) | |
| P_{t+1|t} = F_t P_t F_t^T + Q (covariância predita) | |
| x̂_{t+1|t+1} = x̂_{t+1|t} + K_t(z_t - h(x̂_{t+1|t})) (correção) | |
| P_{t+1|t+1} = (I - K_t H_t) P_{t+1|t} (covariância corrigida) | |
| onde K_t = P_{t+1|t} H_t^T (H_t P_{t+1|t} H_t^T + R)^{-1} (ganho de Kalman) | |
| 3. Estabilidade do anel circular: | |
| Seja G_i a matriz Jacobiana do agente i. O anel é estável se: | |
| ρ(G_{N-1} · ... · G_0) < 1 | |
| onde ρ(·) é o raio espectral. | |
| 4. Teoria de controle circular: | |
| Ação ótima via LQR circular: | |
| J = Σ_t (x_t^T Q x_t + u_t^T R u_t) | |
| u_t* = -K x_t onde K = R^{-1} B^T S | |
| S resolve a equação de Riccati: S = Q + A^T S A - A^T S B (R + B^T S B)^{-1} B^T S A | |
| 5. Convergência do anel: | |
| Erro após k voltas: E_k = ||s_k - s*|| | |
| E_{k+1} ≤ ρ_prod · E_k + ε | |
| onde ρ_prod = ∏_i ρ(G_i) é o produto dos raios espectrais. | |
| Convergência garantida quando ρ_prod < 1. | |
| """ | |
| from __future__ import annotations | |
| import math | |
| import time | |
| import threading | |
| from collections import deque, defaultdict | |
| from concurrent.futures import Future, ThreadPoolExecutor, as_completed | |
| from dataclasses import dataclass, field | |
| from enum import Enum | |
| from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union | |
| import numpy as np | |
| # ============================================================================ | |
| # Exceções | |
| # ============================================================================ | |
| class CircularOrchestrationError(RuntimeError): | |
| """Erro na orquestração circular.""" | |
| class AgentStateError(RuntimeError): | |
| """Erro de estado do agente.""" | |
| class PredictionError(RuntimeError): | |
| """Erro na predição de estado.""" | |
| class RingInstabilityError(RuntimeError): | |
| """Anel circular instável (raio espectral ≥ 1).""" | |
| # ============================================================================ | |
| # Estruturas de dados | |
| # ============================================================================ | |
| class AgentPhase(Enum): | |
| """Fases do ciclo de orquestração circular.""" | |
| DECOMPOSE = "decompose" # Decompor tarefa em sub-tarefas | |
| EXECUTE = "execute" # Executar sub-tarefas | |
| MONITOR = "monitor" # Monitorar ações (feedback) | |
| PREDICT = "predict" # Predizer próximo estado | |
| ADJUST = "adjust" # Ajustar baseado em predição vs realidade | |
| class TaskState(Enum): | |
| """Estado da tarefa no ciclo.""" | |
| INITIALIZED = "initialized" | |
| DECOMPOSED = "decomposed" | |
| EXECUTING = "executing" | |
| MONITORED = "monitored" | |
| PREDICTED = "predicted" | |
| ADJUSTED = "adjusted" | |
| CONVERGED = "converged" | |
| FAILED = "failed" | |
| class MonitoredAction: | |
| """Ação monitorável executada por um agente. | |
| Contém: | |
| - action_type: tipo da ação (decompose, execute, adjust, etc.) | |
| - input: entrada da ação | |
| - output: saída da ação | |
| - feedback: feedback monitorado (score, erro, métricas) | |
| - timestamp: quando foi executada | |
| - agent_id: qual agente executou | |
| """ | |
| action_type: str | |
| input: Any = None | |
| output: Any = None | |
| feedback: Dict[str, Any] = field(default_factory=dict) | |
| timestamp: float = field(default_factory=time.time) | |
| agent_id: str = "" | |
| # Feedback monitorado. | |
| success: bool = True | |
| error_magnitude: float = 0.0 # ||output - expected|| | |
| latency_ms: float = 0.0 | |
| resource_usage: Dict[str, float] = field(default_factory=dict) | |
| class StatePrediction: | |
| """Predição de estado do sistema. | |
| Usa filtro de Kalman estendido para predizer próximo estado. | |
| """ | |
| predicted_state: np.ndarray # x̂_{t+1|t} | |
| predicted_covariance: np.ndarray # P_{t+1|t} | |
| kalman_gain: np.ndarray # K_t | |
| innovation: np.ndarray # z_t - h(x̂) (resíduo) | |
| confidence: float # traço(P) / dim (incerteza média) | |
| timestamp: float = field(default_factory=time.time) | |
| class AgentState: | |
| """Estado de um agente no anel circular. | |
| Mantém: | |
| - phase: fase atual (DECOMPOSE, EXECUTE, MONITOR, PREDICT, ADJUST) | |
| - state_vector: vetor de estado (para filtro de Kalman) | |
| - covariance: matriz de covariância do estado | |
| - action_history: histórico de ações monitoradas | |
| - predictions: histórico de predições | |
| """ | |
| agent_id: str | |
| phase: AgentPhase = AgentPhase.DECOMPOSE | |
| state_vector: Optional[np.ndarray] = None # x_t | |
| covariance: Optional[np.ndarray] = None # P_t | |
| action_history: List[MonitoredAction] = field(default_factory=list) | |
| predictions: List[StatePrediction] = field(default_factory=list) | |
| # Estatísticas. | |
| n_actions: int = 0 | |
| n_successes: int = 0 | |
| n_failures: int = 0 | |
| total_error: float = 0.0 | |
| avg_error: float = 0.0 | |
| # Jacobiano para análise de estabilidade. | |
| jacobian: Optional[np.ndarray] = None # G_i | |
| class CircularTask: | |
| """Tarefa processada pelo anel circular. | |
| Uma tarefa é decomposta em sub-tarefas, cada uma processada | |
| por um agente no anel. O estado flui circularmente. | |
| """ | |
| id: str | |
| description: str | |
| subtasks: List[Dict[str, Any]] = field(default_factory=list) | |
| current_state: TaskState = TaskState.INITIALIZED | |
| # Estado global do anel. | |
| global_state: Optional[np.ndarray] = None | |
| global_covariance: Optional[np.ndarray] = None | |
| # Histórico de voltas (rounds) do anel. | |
| round_history: List[Dict[str, Any]] = field(default_factory=list) | |
| n_rounds: int = 0 | |
| converged: bool = False | |
| convergence_score: float = 0.0 | |
| # ============================================================================ | |
| # 1. KalmanPredictor — predição de estado via filtro de Kalman | |
| # ============================================================================ | |
| class KalmanPredictor: | |
| """Filtro de Kalman estendido para predição de estado do anel. | |
| Modelo: | |
| x_{t+1} = f(x_t, u_t) + w_t (estado) | |
| z_t = h(x_t) + v_t (observação) | |
| Predição: | |
| x̂_{t+1|t} = f(x̂_t, u_t) | |
| P_{t+1|t} = F_t P_t F_t^T + Q | |
| Correção: | |
| K_t = P_{t+1|t} H_t^T (H_t P_{t+1|t} H_t^T + R)^{-1} | |
| x̂_{t+1|t+1} = x̂_{t+1|t} + K_t (z_t - h(x̂_{t+1|t})) | |
| P_{t+1|t+1} = (I - K_t H_t) P_{t+1|t} | |
| onde: | |
| F_t = ∂f/∂x |_{x̂_t} (Jacobiano da dinâmica) | |
| H_t = ∂h/∂x |_{x̂_{t+1|t}} (Jacobiano da observação) | |
| Q = covariância do ruído do processo | |
| R = covariância do ruído de observação | |
| """ | |
| def __init__( | |
| self, | |
| state_dim: int, | |
| process_noise: float = 0.01, | |
| measurement_noise: float = 0.1, | |
| verbose: bool = False, | |
| ): | |
| self.state_dim = state_dim | |
| self.Q = np.eye(state_dim) * process_noise # ruído do processo | |
| self.R = np.eye(state_dim) * measurement_noise # ruído de medição | |
| self.verbose = verbose | |
| def predict( | |
| self, | |
| state: np.ndarray, | |
| covariance: np.ndarray, | |
| control_input: Optional[np.ndarray] = None, | |
| transition_fn: Optional[Callable] = None, | |
| ) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: | |
| """Prediz próximo estado. | |
| x̂_{t+1|t} = f(x̂_t, u_t) | |
| P_{t+1|t} = F_t P_t F_t^T + Q | |
| Args: | |
| state: estado atual x̂_t (state_dim,) | |
| covariance: covariância P_t (state_dim, state_dim) | |
| control_input: entrada de controle u_t | |
| transition_fn: função f(x, u) → x' (default: identidade) | |
| Retorna: (predicted_state, predicted_covariance, F_jacobian) | |
| """ | |
| if transition_fn is None: | |
| transition_fn = lambda x, u: x # identidade | |
| # Prediz estado. | |
| predicted_state = transition_fn(state, control_input) | |
| # Jacobiano F_t = ∂f/∂x (aproximação numérica). | |
| F = self._numerical_jacobian(transition_fn, state, control_input) | |
| # Prediz covariância: P_{t+1|t} = F P F^T + Q | |
| predicted_cov = F @ covariance @ F.T + self.Q | |
| if self.verbose: | |
| print(f" Kalman predict: x̂={predicted_state[:3]}..., " | |
| f"tr(P)={np.trace(predicted_cov):.4f}") | |
| return predicted_state, predicted_cov, F | |
| def correct( | |
| self, | |
| predicted_state: np.ndarray, | |
| predicted_cov: np.ndarray, | |
| observation: np.ndarray, | |
| measurement_fn: Optional[Callable] = None, | |
| ) -> Tuple[np.ndarray, np.ndarray, StatePrediction]: | |
| """Corrige predição com observação. | |
| K_t = P H^T (H P H^T + R)^{-1} | |
| x̂ = x̂_pred + K (z - h(x̂_pred)) | |
| P = (I - K H) P_pred | |
| Args: | |
| predicted_state: x̂_{t+1|t} | |
| predicted_cov: P_{t+1|t} | |
| observation: z_t (medição real) | |
| measurement_fn: função h(x) → z (default: identidade) | |
| Retorna: (corrected_state, corrected_cov, StatePrediction) | |
| """ | |
| if measurement_fn is None: | |
| measurement_fn = lambda x: x # identidade | |
| # Jacobiano H_t. | |
| H = self._numerical_jacobian(measurement_fn, predicted_state, None) | |
| # Ganho de Kalman: K = P H^T (H P H^T + R)^{-1} | |
| S = H @ predicted_cov @ H.T + self.R | |
| try: | |
| K = predicted_cov @ H.T @ np.linalg.inv(S) | |
| except np.linalg.LinAlgError: | |
| K = predicted_cov @ H.T @ np.linalg.pinv(S) | |
| # Inovação (resíduo): y = z - h(x̂_pred) | |
| predicted_obs = measurement_fn(predicted_state) | |
| innovation = observation - predicted_obs | |
| # Correção: x̂ = x̂_pred + K * y | |
| corrected_state = predicted_state + K @ innovation | |
| # Covariância corrigida: P = (I - K H) P_pred | |
| I = np.eye(self.state_dim) | |
| corrected_cov = (I - K @ H) @ predicted_cov | |
| # Confiança: traço(P) / dim (menor = mais confiante). | |
| confidence = 1.0 - min(np.trace(corrected_cov) / self.state_dim, 1.0) | |
| prediction = StatePrediction( | |
| predicted_state=predicted_state, | |
| predicted_covariance=predicted_cov, | |
| kalman_gain=K, | |
| innovation=innovation, | |
| confidence=float(confidence), | |
| ) | |
| if self.verbose: | |
| print(f" Kalman correct: innovation={innovation[:3]}..., " | |
| f"confidence={confidence:.4f}") | |
| return corrected_state, corrected_cov, prediction | |
| def _numerical_jacobian( | |
| self, | |
| fn: Callable, | |
| x: np.ndarray, | |
| u: Optional[np.ndarray], | |
| eps: float = 1e-5, | |
| ) -> np.ndarray: | |
| """Computa Jacobiano numericamente via diferenças finitas. | |
| F_ij = ∂f_i/∂x_j ≈ (f(x + ε·e_j, u) - f(x, u)) / ε | |
| """ | |
| n = len(x) | |
| J = np.zeros((n, n)) | |
| # Chama fn com tratamento uniforme de argumentos. | |
| def call(x_val): | |
| if u is not None: | |
| return fn(x_val, u) | |
| try: | |
| return fn(x_val) | |
| except TypeError: | |
| return fn(x_val, None) | |
| f0 = call(x) | |
| for j in range(n): | |
| x_plus = x.copy() | |
| x_plus[j] += eps | |
| f_plus = call(x_plus) | |
| J[:, j] = (f_plus - f0) / eps | |
| return J | |
| # ============================================================================ | |
| # 2. CircularAgent — agente no anel de orquestração | |
| # ============================================================================ | |
| class CircularAgent: | |
| """Agente no anel de orquestração circular. | |
| Cada agente: | |
| 1. Recebe estado do agente anterior | |
| 2. Executa sua fase (DECOMPOSE, EXECUTE, MONITOR, PREDICT, ADJUST) | |
| 3. Monitora a ação (feedback) | |
| 4. Prediz próximo estado (Kalman) | |
| 5. Passa estado ao próximo agente | |
| O agente mantém: | |
| - Estado interno (vetor + covariância) | |
| - Histórico de ações monitoradas | |
| - Predições de estado | |
| - Jacobiano (para análise de estabilidade do anel) | |
| """ | |
| def __init__( | |
| self, | |
| agent_id: str, | |
| phase: AgentPhase, | |
| execute_fn: Callable[[Any, AgentState], Any], | |
| state_dim: int = 4, | |
| kalman_predictor: Optional[KalmanPredictor] = None, | |
| verbose: bool = False, | |
| ): | |
| self.agent_id = agent_id | |
| self.phase = phase | |
| self.execute_fn = execute_fn | |
| self.state_dim = state_dim | |
| self.kalman = kalman_predictor or KalmanPredictor(state_dim, verbose=verbose) | |
| self.verbose = verbose | |
| # Estado interno. | |
| self.state = AgentState( | |
| agent_id=agent_id, | |
| phase=phase, | |
| state_vector=np.zeros(state_dim), | |
| covariance=np.eye(state_dim) * 0.1, | |
| ) | |
| def process( | |
| self, | |
| input_data: Any, | |
| task: CircularTask, | |
| ) -> Tuple[Any, MonitoredAction, Optional[StatePrediction]]: | |
| """Processa input na fase do agente. | |
| 1. Executa ação (execute_fn) | |
| 2. Monitora ação (feedback) | |
| 3. Prediz próximo estado (Kalman) | |
| 4. Atualiza estado interno | |
| Retorna: (output, monitored_action, prediction) | |
| """ | |
| t0 = time.time() | |
| self.state.n_actions += 1 | |
| # 1. Executa ação. | |
| try: | |
| output = self.execute_fn(input_data, self.state) | |
| success = True | |
| error_mag = 0.0 | |
| except Exception as e: | |
| output = None | |
| success = False | |
| error_mag = 1.0 | |
| if self.verbose: | |
| print(f" Agent {self.agent_id}: FALHOU: {e}") | |
| latency = (time.time() - t0) * 1000 | |
| # 2. Monitora ação (feedback). | |
| action = MonitoredAction( | |
| action_type=self.phase.value, | |
| input=input_data, | |
| output=output, | |
| agent_id=self.agent_id, | |
| success=success, | |
| error_magnitude=error_mag, | |
| latency_ms=latency, | |
| feedback={ | |
| "task_id": task.id, | |
| "round": task.n_rounds, | |
| "phase": self.phase.value, | |
| }, | |
| ) | |
| self.state.action_history.append(action) | |
| if success: | |
| self.state.n_successes += 1 | |
| else: | |
| self.state.n_failures += 1 | |
| self.state.total_error += error_mag | |
| self.state.avg_error = self.state.total_error / self.state.n_actions | |
| # 3. Prediz próximo estado (Kalman). | |
| prediction = None | |
| if success and output is not None: | |
| # Converte output para vetor de estado. | |
| obs = self._extract_state_vector(output, task) | |
| # Prediz. | |
| pred_state, pred_cov, F = self.kalman.predict( | |
| self.state.state_vector, | |
| self.state.covariance, | |
| control_input=obs, | |
| ) | |
| # Corrige com observação. | |
| corr_state, corr_cov, prediction = self.kalman.correct( | |
| pred_state, pred_cov, obs, | |
| ) | |
| # Atualiza estado. | |
| self.state.state_vector = corr_state | |
| self.state.covariance = corr_cov | |
| self.state.jacobian = F # para análise de estabilidade | |
| self.state.predictions.append(prediction) | |
| # 4. Atualiza estado da tarefa. | |
| task.current_state = self._next_task_state(task.current_state) | |
| if self.verbose: | |
| print(f" Agent {self.agent_id} ({self.phase.value}): " | |
| f"success={success}, latency={latency:.1f}ms, " | |
| f"error={error_mag:.4f}") | |
| return output, action, prediction | |
| def _extract_state_vector(self, output: Any, task: CircularTask) -> np.ndarray: | |
| """Extrai vetor de estado do output da ação.""" | |
| if isinstance(output, (int, float)): | |
| return np.array([float(output)] + [0.0] * (self.state_dim - 1)) | |
| if isinstance(output, (list, tuple)): | |
| vals = list(output)[:self.state_dim] | |
| while len(vals) < self.state_dim: | |
| vals.append(0.0) | |
| return np.array(vals, dtype=float) | |
| if isinstance(output, dict): | |
| # Usa valores numéricos do dict. | |
| vals = [float(v) for v in output.values() if isinstance(v, (int, float))][:self.state_dim] | |
| while len(vals) < self.state_dim: | |
| vals.append(0.0) | |
| return np.array(vals) | |
| if isinstance(output, str): | |
| # Hash → vetor. | |
| h = hash(output) % (10 ** self.state_dim) | |
| return np.array([float(h // (10**i) % 10) / 10 for i in range(self.state_dim)]) | |
| return np.zeros(self.state_dim) | |
| def _next_task_state(self, current: TaskState) -> TaskState: | |
| """Transição de estado da tarefa baseado na fase do agente.""" | |
| transitions = { | |
| (TaskState.INITIALIZED, AgentPhase.DECOMPOSE): TaskState.DECOMPOSED, | |
| (TaskState.DECOMPOSED, AgentPhase.EXECUTE): TaskState.EXECUTING, | |
| (TaskState.EXECUTING, AgentPhase.MONITOR): TaskState.MONITORED, | |
| (TaskState.MONITORED, AgentPhase.PREDICT): TaskState.PREDICTED, | |
| (TaskState.PREDICTED, AgentPhase.ADJUST): TaskState.ADJUSTED, | |
| (TaskState.ADJUSTED, AgentPhase.DECOMPOSE): TaskState.DECOMPOSED, | |
| } | |
| return transitions.get((current, self.phase), current) | |
| def get_spectral_radius(self) -> float: | |
| """Raio espectral do Jacobiano (para análise de estabilidade).""" | |
| if self.state.jacobian is None: | |
| return 0.0 | |
| eigenvalues = np.linalg.eigvals(self.state.jacobian) | |
| return float(max(abs(eigenvalues))) | |
| def get_stats(self) -> Dict[str, Any]: | |
| return { | |
| "agent_id": self.agent_id, | |
| "phase": self.phase.value, | |
| "n_actions": self.state.n_actions, | |
| "n_successes": self.state.n_successes, | |
| "n_failures": self.state.n_failures, | |
| "avg_error": self.state.avg_error, | |
| "spectral_radius": self.get_spectral_radius(), | |
| "state_confidence": ( | |
| 1.0 - min(np.trace(self.state.covariance) / self.state_dim, 1.0) | |
| if self.state.covariance is not None else 0.0 | |
| ), | |
| "n_predictions": len(self.state.predictions), | |
| } | |
| # ============================================================================ | |
| # 3. CircularOrchestrator — orquestração circular distribuída | |
| # ============================================================================ | |
| class CircularOrchestrator: | |
| """Orquestração circular distribuída granulável. | |
| Arquitetura: N agentes em anel, cada um com uma fase diferente. | |
| O estado flui circularmente: Agent_0 → Agent_1 → ... → Agent_{N-1} → Agent_0 | |
| Cada volta (round) do anel executa o ciclo completo: | |
| DECOMPOSE → EXECUTE → MONITOR → PREDICT → ADJUST → (volta a DECOMPOSE) | |
| Análise de estabilidade: | |
| ρ_prod = ∏_i ρ(G_i) < 1 (produto dos raios espectrais) | |
| Garante convergência do anel. | |
| Análise de convergência: | |
| E_{k+1} ≤ ρ_prod · E_k + ε | |
| k_max = ceil(log((E_0 - ε/(1-ρ_prod)) / (threshold - ε/(1-ρ_prod))) / log(ρ_prod)) | |
| Uso: | |
| >>> orch = CircularOrchestrator(n_agents=5, state_dim=4) | |
| >>> solution = orch.solve(task, executor_fns=[...]) | |
| """ | |
| def __init__( | |
| self, | |
| n_agents: int = 5, | |
| state_dim: int = 4, | |
| max_rounds: int = 10, | |
| convergence_threshold: float = 0.9, | |
| process_noise: float = 0.01, | |
| measurement_noise: float = 0.1, | |
| parallel: bool = False, # Se True, fases independentes rodam em paralelo | |
| verbose: bool = False, | |
| ): | |
| self.n_agents = n_agents | |
| self.state_dim = state_dim | |
| self.max_rounds = max_rounds | |
| self.convergence_threshold = convergence_threshold | |
| self.parallel = parallel | |
| self.verbose = verbose | |
| # Kalman compartilhado. | |
| self.kalman = KalmanPredictor( | |
| state_dim=state_dim, | |
| process_noise=process_noise, | |
| measurement_noise=measurement_noise, | |
| verbose=verbose, | |
| ) | |
| # Agentes (serão criados em solve). | |
| self.agents: List[CircularAgent] = [] | |
| # Estatísticas. | |
| self.round_history: List[Dict[str, Any]] = [] | |
| self.error_trajectory: List[float] = [] | |
| def solve( | |
| self, | |
| task: CircularTask, | |
| executor_fns: List[Callable[[Any, AgentState], Any]], | |
| ) -> CircularTask: | |
| """Resolve tarefa via orquestração circular. | |
| Args: | |
| task: tarefa a resolver | |
| executor_fns: lista de funções executoras (1 por agente) | |
| Retorna: task atualizada com resultado e estado final. | |
| """ | |
| assert len(executor_fns) == self.n_agents, \ | |
| f"Esperado {self.n_agents} executores, recebido {len(executor_fns)}" | |
| # Cria agentes (um por fase). | |
| phases = list(AgentPhase) | |
| # Se n_agents > len(phases), repete fases. | |
| while len(phases) < self.n_agents: | |
| phases.extend(AgentPhase) | |
| self.agents = [ | |
| CircularAgent( | |
| agent_id=f"agent_{i}", | |
| phase=phases[i % len(AgentPhase)], | |
| execute_fn=executor_fns[i], | |
| state_dim=self.state_dim, | |
| kalman_predictor=self.kalman, | |
| verbose=self.verbose, | |
| ) | |
| for i in range(self.n_agents) | |
| ] | |
| # Inicializa estado global. | |
| task.global_state = np.zeros(self.state_dim) | |
| task.global_covariance = np.eye(self.state_dim) * 0.1 | |
| if self.verbose: | |
| print(f"\n{'='*60}") | |
| print(f"CircularOrchestrator: {self.n_agents} agentes, " | |
| f"max_rounds={self.max_rounds}") | |
| print(f"Anel: {' → '.join(a.agent_id for a in self.agents)} → {self.agents[0].agent_id}") | |
| print(f"{'='*60}") | |
| # Loop circular. | |
| current_data: Any = task.description # input inicial | |
| for round_num in range(1, self.max_rounds + 1): | |
| task.n_rounds = round_num | |
| round_actions: List[MonitoredAction] = [] | |
| round_predictions: List[StatePrediction] = [] | |
| if self.verbose: | |
| print(f"\n--- Round {round_num}/{self.max_rounds} ---") | |
| # Cada agente processa em sequência (anel). | |
| for agent in self.agents: | |
| output, action, prediction = agent.process(current_data, task) | |
| round_actions.append(action) | |
| if prediction is not None: | |
| round_predictions.append(prediction) | |
| # Passa output ao próximo agente. | |
| current_data = output if output is not None else current_data | |
| # Analisa estabilidade do anel. | |
| spectral_radii = [a.get_spectral_radius() for a in self.agents] | |
| rho_prod = 1.0 | |
| for rho in spectral_radii: | |
| rho_prod *= max(rho, 0.01) # evita zero | |
| # Computa score de convergência. | |
| n_success = sum(1 for a in round_actions if a.success) | |
| success_rate = n_success / len(round_actions) | |
| avg_error = np.mean([a.error_magnitude for a in round_actions]) | |
| # Score combina success_rate e (1 - avg_error). | |
| conv_score = success_rate * (1.0 - avg_error) | |
| # Confiança média das predições. | |
| avg_confidence = ( | |
| np.mean([p.confidence for p in round_predictions]) | |
| if round_predictions else 0.5 | |
| ) | |
| # Score final pondera convergência e confiança. | |
| final_score = 0.7 * conv_score + 0.3 * avg_confidence | |
| task.convergence_score = final_score | |
| # Registra round. | |
| round_info = { | |
| "round": round_num, | |
| "n_actions": len(round_actions), | |
| "n_success": n_success, | |
| "success_rate": success_rate, | |
| "avg_error": avg_error, | |
| "avg_confidence": avg_confidence, | |
| "convergence_score": final_score, | |
| "spectral_radii": spectral_radii, | |
| "rho_product": rho_prod, | |
| "stable": rho_prod < 1.0, | |
| } | |
| task.round_history.append(round_info) | |
| self.round_history.append(round_info) | |
| self.error_trajectory.append(1.0 - final_score) | |
| if self.verbose: | |
| print(f" Score: {final_score:.4f}, " | |
| f"ρ_prod: {rho_prod:.4f}, " | |
| f"stable: {rho_prod < 1.0}") | |
| # Verifica convergência. | |
| if final_score >= self.convergence_threshold: | |
| task.converged = True | |
| task.current_state = TaskState.CONVERGED | |
| if self.verbose: | |
| print(f" ✓ Convergiu em {round_num} voltas!") | |
| break | |
| # Verifica instabilidade. | |
| if rho_prod >= 1.0 and round_num > 2: | |
| if self.verbose: | |
| print(f" ⚠ Anel instável (ρ_prod={rho_prod:.4f} ≥ 1)") | |
| # Não aborta, mas avisa. | |
| if not task.converged: | |
| if self.verbose: | |
| print(f"\n ✗ Não convergiu após {self.max_rounds} voltas") | |
| return task | |
| def analyze_stability(self) -> Dict[str, Any]: | |
| """Analisa estabilidade do anel circular. | |
| ρ_prod = ∏_i ρ(G_i) < 1 → estável | |
| """ | |
| if not self.agents: | |
| return {"stable": True, "reason": "sem agentes"} | |
| spectral_radii = [a.get_spectral_radius() for a in self.agents] | |
| rho_prod = 1.0 | |
| for rho in spectral_radii: | |
| rho_prod *= max(rho, 0.01) | |
| return { | |
| "n_agents": len(self.agents), | |
| "spectral_radii": spectral_radii, | |
| "rho_product": rho_prod, | |
| "stable": rho_prod < 1.0, | |
| "margin": 1.0 - rho_prod, # margem de estabilidade | |
| "agent_stats": [a.get_stats() for a in self.agents], | |
| } | |
| def analyze_convergence(self) -> Dict[str, Any]: | |
| """Analisa convergência do anel. | |
| E_{k+1} ≤ ρ_prod · E_k + ε | |
| """ | |
| if not self.error_trajectory: | |
| return {"converged": False, "reason": "sem dados"} | |
| from .distributed_reasoning_system import analyze_convergence as _analyze | |
| stability = self.analyze_stability() | |
| rho = stability.get("rho_product", 0.5) | |
| return _analyze(self.error_trajectory, rho=rho, epsilon=0.01) | |
| def get_all_stats(self) -> Dict[str, Any]: | |
| return { | |
| "n_rounds": len(self.round_history), | |
| "agents": [a.get_stats() for a in self.agents], | |
| "stability": self.analyze_stability(), | |
| "convergence": self.analyze_convergence(), | |
| "error_trajectory": self.error_trajectory, | |
| } | |
| def summary(self) -> str: | |
| lines = [ | |
| f"CircularOrchestrator ({self.n_agents} agentes):", | |
| f" Rounds: {len(self.round_history)}", | |
| f" Estabilidade: {self.analyze_stability()['stable']}", | |
| ] | |
| for a in self.agents: | |
| s = a.get_stats() | |
| lines.append( | |
| f" {s['agent_id']} ({s['phase']}): " | |
| f"actions={s['n_actions']}, " | |
| f"success={s['n_successes']}, " | |
| f"ρ={s['spectral_radius']:.4f}, " | |
| f"conf={s['state_confidence']:.4f}" | |
| ) | |
| return "\n".join(lines) | |
| # ============================================================================ | |
| # 4. Análise matemática | |
| # ============================================================================ | |
| def analyze_ring_stability( | |
| jacobians: List[np.ndarray], | |
| ) -> Dict[str, Any]: | |
| """Analisa estabilidade do anel circular. | |
| ρ_prod = ρ(G_{N-1} · ... · G_0) < 1 → estável | |
| Args: | |
| jacobians: lista de matrizes Jacobianas [G_0, G_1, ..., G_{N-1}] | |
| Retorna: análise de estabilidade | |
| """ | |
| if not jacobians: | |
| return {"stable": True, "rho_product": 0.0} | |
| # Produto dos Jacobianos. | |
| product = np.eye(jacobians[0].shape[0]) | |
| for J in jacobians: | |
| product = J @ product | |
| # Raio espectral do produto. | |
| eigenvalues = np.linalg.eigvals(product) | |
| rho_product = float(max(abs(eigenvalues))) | |
| # Raios espectrais individuais. | |
| individual_radii = [float(max(abs(np.linalg.eigvals(J)))) for J in jacobians] | |
| return { | |
| "n_agents": len(jacobians), | |
| "individual_spectral_radii": individual_radii, | |
| "rho_product": rho_product, | |
| "stable": rho_product < 1.0, | |
| "margin": 1.0 - rho_product, | |
| "eigenvalues": [complex(ev) for ev in eigenvalues], | |
| } | |
| def compute_optimal_ring_size( | |
| task_complexity: float, | |
| agent_overhead: float = 0.1, | |
| coordination_cost: float = 0.05, | |
| ) -> Dict[str, Any]: | |
| """Computa tamanho ótimo do anel (número de agentes). | |
| Análise: T(n) = C/n + α·n + β | |
| n* = sqrt(C / α) | |
| Speedup: S(n) = C / T(n) | |
| """ | |
| n_star = math.sqrt(task_complexity / max(agent_overhead, 1e-8)) | |
| n_star_int = max(2, min(20, int(round(n_star)))) | |
| t_seq = task_complexity | |
| t_par = task_complexity / n_star_int + agent_overhead * n_star_int + coordination_cost | |
| speedup = t_seq / max(t_par, 1e-8) | |
| return { | |
| "task_complexity": task_complexity, | |
| "agent_overhead": agent_overhead, | |
| "coordination_cost": coordination_cost, | |
| "n_star_theoretical": n_star, | |
| "n_star_empirical": n_star_int, | |
| "speedup": speedup, | |
| "t_sequential": t_seq, | |
| "t_parallel": t_par, | |
| } | |