""" 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" @dataclass 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) @dataclass 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) @dataclass 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 @dataclass 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, }