| import importlib |
| import json |
| import logging |
| import pandas as pd |
| from typing import Dict, Any |
| from feedback.loop_manager import LoopManager |
|
|
| class MultiplexingEngine: |
| def __init__(self, matrix_path: str): |
| self.matrix = pd.read_csv(matrix_path) |
| self.agents = {} |
| self.logger = logging.getLogger("MultiplexingEngine") |
| self.loop_enabled = True |
| self._validate_matrix() |
|
|
| def _validate_matrix(self): |
| required_columns = {'agent_id', 'role', 'activation_logic', 'config', 'behavior'} |
| missing = required_columns - set(self.matrix.columns) |
| if missing: |
| raise ValueError(f"Missing required columns: {missing}") |
|
|
| def _parse_json_field(self, field: str) -> Dict[str, Any]: |
| try: |
| return json.loads(field.replace("'", '"')) |
| except json.JSONDecodeError: |
| return {} |
|
|
| def _parse_behavior(self, behavior_str: str) -> Dict[str, Any]: |
| return self._parse_json_field(behavior_str) |
|
|
| def _parse_security(self, security_str: str) -> Dict[str, Any]: |
| return self._parse_json_field(security_str) |
|
|
| def _load_agent_class(self, role: str): |
| try: |
| module = importlib.import_module(f"agents.{role.lower()}") |
| return getattr(module, role) |
| except Exception as e: |
| self.logger.error(f"Agent class load failed: {str(e)}") |
| return None |
|
|
| def dispatch_tasks(self): |
| for _, row in self.matrix.iterrows(): |
| agent_id = row["agent_id"] |
| role = row["role"] |
| config = self._parse_json_field(row.get("config", "{}")) |
| behavior = self._parse_behavior(row.get("behavior", "{}")) |
| security = self._parse_security(row.get("security", "{}")) |
| activation_logic = row.get("activation_logic", "lambda metrics: True") |
|
|
| try: |
| if eval(activation_logic)({}): |
| AgentCls = self._load_agent_class(role) |
| if AgentCls: |
| agent_instance = AgentCls(**config) |
| if self.loop_enabled: |
| loop = LoopManager(agent_id) |
| result = loop.run_loop(f"Execute task for {agent_id}") |
| else: |
| result = agent_instance.execute(config=config, behavior=behavior) |
| print(f"{agent_id} executed: {result['output']}") |
| except Exception as e: |
| self.logger.error(f"Dispatch failed for {agent_id}: {str(e)}") |
|
|
| def process_feedback(self, agent_id: str, feedback_data: Dict[str, Any]): |
| try: |
| loop = LoopManager(agent_id) |
| result = loop.run_loop(feedback_data.get("task_input", "No input")) |
| print(f"[Feedback] Loop complete for {agent_id}: {result['output']}") |
| except Exception as e: |
| self.logger.error(f"Feedback processing failed for {agent_id}: {str(e)}") |
