File size: 3,005 Bytes
5e1dfdc |
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 |
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)({}): # pass empty context for now
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)}") |