File size: 7,162 Bytes
3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab 5f83674 3c9f1ab |
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 |
# blux/agent/advanced/multi_agent.py
from blux.agent.advanced.reasoning import ReasoningLayer
class MultiAgentManager:
"""
Multi-agent manager with memory broadcasting, secure monitoring,
reasoning integration, and constitutional enforcement.
"""
def __init__(self, constitution=None):
self.agents = {}
self.constitution = constitution
self.monitor = None
self.reasoning = {}
def attach_monitor(self, monitor):
self.monitor = monitor
def register_agent(self, name, agent_instance):
self.agents[name] = agent_instance
# Attach reasoning layer per agent
self.reasoning[name] = ReasoningLayer(agent_instance, constitution=self.constitution)
if self.monitor:
self.monitor.log(name, "agent_registered")
def _enforce_constitution(self, agent_name, response):
"""
Placeholder for rule enforcement.
Could veto, flag, or alter responses that violate constitutional rules.
"""
if self.constitution and "violation" in response:
if self.monitor:
self.monitor.log(agent_name, "constitutional_violation", {"response": response})
return "[VIOLATION DETECTED]"
return response
def broadcast_input(self, user_input, user_type="unknown"):
results = {}
for name, agent in self.agents.items():
reasoning = self.reasoning[name]
reasoning_result = reasoning.process(user_input, user_type=user_type)
resp = self._enforce_constitution(name, reasoning_result["decision"])
results[name] = resp
if self.monitor:
self.monitor.log(name, "input_processed", {
"input": user_input,
"response": resp,
"reasoning": reasoning_result
})
return results
def delegate_task(self, user_input, target_agent=None, user_type="unknown"):
if target_agent and target_agent in self.agents:
resp = self.broadcast_input(user_input, user_type)[target_agent]
if self.monitor:
self.monitor.log(target_agent, "task_delegated", {"input": user_input, "response": resp})
return {target_agent: resp}
elif self.agents:
first_agent_name = next(iter(self.agents))
resp = self.broadcast_input(user_input, user_type)[first_agent_name]
if self.monitor:
self.monitor.log(first_agent_name, "task_delegated", {"input": user_input, "response": resp})
return {first_agent_name: resp}
else:
if self.monitor:
self.monitor.log("manager", "task_delegated_failed", {"input": user_input})
return {"error": "No agents registered"}
def broadcast_memory(self, key, value, user_type="default", priority=1, tags=None):
for agent in self.agents.values():
if hasattr(agent, "memory"):
agent.memory.add(key, value, user_type=user_type, priority=priority, tags=tags or [])
if self.monitor:
self.monitor.log(agent.name, "memory_broadcast", {"key": key, "value": value, "tags": tags})
def aggregate_memory(self, key, predictive=True):
aggregated = []
for name, agent in self.agents.items():
if hasattr(agent, "memory"):
entries = agent.memory.recall(key)
if entries:
if predictive:
weighted = [e for e in entries if "urgent" in e.get("tags", [])]
aggregated.extend(weighted or entries)
else:
aggregated.extend(entries)
if self.monitor:
self.monitor.log("manager", "memory_aggregated", {"key": key, "entries": aggregated})
return aggregated
def resolve_conflict(self, responses, use_prediction=True):
enforced = {agent: self._enforce_constitution(agent, resp) for agent, resp in responses.items()}
if use_prediction:
sorted_agents = sorted(
enforced.keys(),
key=lambda name: getattr(self.reasoning[name], "predict_behavior", lambda x: [1])(None)[0] if hasattr(self.reasoning[name], "predict_behavior") else 1,
reverse=True
)
resolved = {a: enforced[a] for a in sorted_agents}
else:
resolved = enforced
return "\n".join(f"[{agent}] {resp}" for agent, resp in resolved.items())
def aggregate_responses(self, responses):
return "\n".join(f"[{agent}] {resp}" for agent, resp in responses.items())ive aggregation =====
def broadcast_memory(self, key, value, user_type="default", priority=1, tags=None):
for agent in self.agents.values():
if hasattr(agent, "memory"):
agent.memory.add(key, value, user_type=user_type, priority=priority, tags=tags or [])
if self.monitor:
self.monitor.log(agent.name, "memory_broadcast", {"key": key, "value": value, "tags": tags})
def aggregate_memory(self, key, predictive=True):
"""
Aggregates memory across all agents.
If predictive=True, weights results based on reasoning predictions.
"""
aggregated = []
for name, agent in self.agents.items():
if hasattr(agent, "memory"):
entries = agent.memory.recall(key)
if entries:
if predictive:
# Simple weighting: prioritize entries tagged as urgent or from predicted struggler
weighted = [e for e in entries if "urgent" in e.get("tags", [])]
aggregated.extend(weighted or entries)
else:
aggregated.extend(entries)
if self.monitor:
self.monitor.log("manager", "memory_aggregated", {"key": key, "entries": aggregated})
return aggregated
# ===== Stage 6: Response aggregation & advanced conflict resolution =====
def resolve_conflict(self, responses, use_prediction=True):
"""
Stage 6 conflict resolution:
- Enforce constitutional rules
- Optionally use reasoning prediction to weight responses
"""
enforced = {agent: self._enforce_constitution(agent, resp) for agent, resp in responses.items()}
if use_prediction:
# Simple prediction-based sorting: 'struggler' > 'neutral' > 'indulgent'
sorted_agents = sorted(
enforced.keys(),
key=lambda name: self.reasoning[name].predict_behavior("")[0] if hasattr(self.reasoning[name], "predict_behavior") else 1,
reverse=True
)
resolved = {a: enforced[a] for a in sorted_agents}
else:
resolved = enforced
return "\n".join(f"[{agent}] {resp}" for agent, resp in resolved.items())
def aggregate_responses(self, responses):
return "\n".join(f"[{agent}] {resp}" for agent, resp in responses.items()) |