348 lines
No EOL
12 KiB
Python
348 lines
No EOL
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Aurelio Central Coordinating Agent
|
|
|
|
The master orchestrator that coordinates the PortugalFuturista multi-heteronym army.
|
|
This is the "brain" that reads from memory, makes decisions, and coordinates specialized agents.
|
|
|
|
Based on technical analysis, this addresses the critical gap: Aurelio has excellent
|
|
infrastructure (memory, sync, identity) but lacks coordination intelligence.
|
|
|
|
Author: PortugalFuturista Ecosystem
|
|
Version: 1.0.0
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from typing import Dict, List, Optional, Any
|
|
from pathlib import Path
|
|
from dataclasses import dataclass
|
|
from enum import Enum
|
|
import subprocess
|
|
import time
|
|
|
|
|
|
class AgentRole(Enum):
|
|
"""Specialized agent roles in the heteronym army"""
|
|
BERNARDO_SOARES = "bernardo-soares" # Research/Analysis
|
|
HERMES = "hermes" # Multi-system orchestration
|
|
FABIO_HARDWARE_MESTRE = "fabio-hardware-mestre" # Hardware coordination
|
|
FABIO_COUTADA = "fabio-coutada" # Technical lead
|
|
UNIVERSAL = "universal" # General-purpose
|
|
|
|
|
|
@dataclass
|
|
class Task:
|
|
"""Task representation for coordination"""
|
|
id: str
|
|
description: str
|
|
priority: int # 1-10, 10 highest
|
|
complexity: str # "low", "medium", "high"
|
|
domain: str # "hardware", "software", "infrastructure", "research"
|
|
estimated_time: int # minutes
|
|
dependencies: List[str]
|
|
status: str = "pending"
|
|
|
|
|
|
@dataclass
|
|
class AgentCapability:
|
|
"""Agent capability description"""
|
|
role: AgentRole
|
|
skills: List[str]
|
|
preferred_models: List[str]
|
|
max_complexity: str # "low", "medium", "high"
|
|
active_domains: List[str]
|
|
|
|
|
|
class MemorySystem:
|
|
"""Interface to Aurelio's cross-realm memory system"""
|
|
|
|
def __init__(self, workspace_root: Path):
|
|
self.workspace_root = workspace_root
|
|
self.memory_index = self._load_memory_index()
|
|
|
|
def _load_memory_index(self) -> Dict:
|
|
"""Load the 144KB cross-realm memory index"""
|
|
index_path = self.workspace_root / ".aurelio/realm/.cross_realm_index.json"
|
|
if index_path.exists():
|
|
with open(index_path, 'r') as f:
|
|
return json.load(f)
|
|
return {}
|
|
|
|
def query_knowledge(self, query: str, realm: Optional[str] = None) -> List[Dict]:
|
|
"""Query the memory system for relevant knowledge"""
|
|
results = []
|
|
|
|
# Search through indexed knowledge items
|
|
for item_id, item in self.memory_index.items():
|
|
# Filter by realm if specified
|
|
if realm and item.get('realm') != realm:
|
|
continue
|
|
|
|
# Simple text matching (can be enhanced with semantic search)
|
|
if query.lower() in item.get('content', '').lower():
|
|
results.append(item)
|
|
|
|
return results
|
|
|
|
def get_project_context(self, project_path: Path) -> Dict:
|
|
"""Get context for a specific project"""
|
|
# Check for AGENTS.md
|
|
agents_file = project_path / "AGENTS.md"
|
|
if agents_file.exists():
|
|
return self._parse_agents_md(agents_file)
|
|
|
|
# Basic project info
|
|
return {
|
|
"name": project_path.name,
|
|
"path": str(project_path),
|
|
"has_agents_md": False
|
|
}
|
|
|
|
def _parse_agents_md(self, agents_file: Path) -> Dict:
|
|
"""Parse AGENTS.md for project context"""
|
|
# Simplified parsing - can be enhanced with proper Markdown parsing
|
|
context = {
|
|
"has_agents_md": True,
|
|
"path": str(agents_file.parent)
|
|
}
|
|
|
|
try:
|
|
content = agents_file.read_text()
|
|
# Extract key sections (can be enhanced)
|
|
if "## Project Overview" in content:
|
|
context["has_overview"] = True
|
|
if "## Technology Stack" in content:
|
|
context["has_tech_stack"] = True
|
|
if "## Agent Integration" in content:
|
|
context["has_agent_integration"] = True
|
|
except Exception as e:
|
|
context["parse_error"] = str(e)
|
|
|
|
return context
|
|
|
|
|
|
class Coordinator:
|
|
"""Central coordinating agent - the brain of Aurelio"""
|
|
|
|
def __init__(self, workspace_root: Path):
|
|
self.workspace_root = workspace_root
|
|
self.memory = MemorySystem(workspace_root)
|
|
self.active_agents: Dict[AgentRole, AgentCapability] = {}
|
|
self.task_queue: List[Task] = []
|
|
|
|
self._initialize_capabilities()
|
|
|
|
def _initialize_capabilities(self):
|
|
"""Initialize known agent capabilities"""
|
|
self.active_agents[AgentRole.BERNARDO_SOARES] = AgentCapability(
|
|
role=AgentRole.BERNARDO_SOARES,
|
|
skills=["research", "analysis", "academic-processing", "deep-investigation"],
|
|
preferred_models=["claude-3-5-sonnet"],
|
|
max_complexity="high",
|
|
active_domains=["research", "documentation", "architecture"]
|
|
)
|
|
|
|
self.active_agents[AgentRole.HERMES] = AgentCapability(
|
|
role=AgentRole.HERMES,
|
|
skills=["multi-system-orchestration", "cross-system-coordination", "tool-augmented-reasoning"],
|
|
preferred_models=["Hermes-3-Llama-3.1-8B"],
|
|
max_complexity="medium",
|
|
active_domains=["infrastructure", "coordination", "integration"]
|
|
)
|
|
|
|
self.active_agents[AgentRole.FABIO_HARDWARE_MESTRE] = AgentCapability(
|
|
role=AgentRole.FABIO_HARDWARE_MESTRE,
|
|
skills=["hardware-design", "pcb-development", "firmware-integration", "fpga-development"],
|
|
preferred_models=["claude-opus-4-8"],
|
|
max_complexity="high",
|
|
active_domains=["hardware", "firmware", "fpga", "testing"]
|
|
)
|
|
|
|
self.active_agents[AgentRole.FABIO_COUTADA] = AgentCapability(
|
|
role=AgentRole.FABIO_COUTADA,
|
|
skills=["technical-leadership", "architecture", "development", "coordination"],
|
|
preferred_models=["claude-sonnet-5"],
|
|
max_complexity="high",
|
|
active_domains=["software", "architecture", "coordination"]
|
|
)
|
|
|
|
def analyze_task(self, task: Task) -> AgentRole:
|
|
"""Analyze task and assign to appropriate specialized agent"""
|
|
|
|
# Domain-based assignment
|
|
if task.domain == "hardware":
|
|
if "fpga" in task.description.lower() or "hardware acceleration" in task.description.lower():
|
|
return AgentRole.FABIO_HARDWARE_MESTRE
|
|
return AgentRole.FABIO_HARDWARE_MESTRE
|
|
|
|
elif task.domain == "research":
|
|
if task.complexity == "high":
|
|
return AgentRole.BERNARDO_SOARES
|
|
return AgentRole.BERNARDO_SOARES
|
|
|
|
elif task.domain == "infrastructure":
|
|
if "coordination" in task.description.lower() or "integration" in task.description.lower():
|
|
return AgentRole.HERMES
|
|
return AgentRole.HERMES
|
|
|
|
elif task.domain == "software":
|
|
if task.complexity == "high":
|
|
return AgentRole.FABIO_COUTADA
|
|
return AgentRole.FABIO_COUTADA
|
|
|
|
# Default to technical lead for general tasks
|
|
return AgentRole.FABIO_COUTADA
|
|
|
|
def coordinate_delegation(self, task: Task) -> Dict[str, Any]:
|
|
"""Coordinate task delegation to specialized agent"""
|
|
|
|
# Check dependencies
|
|
if not self._check_dependencies(task):
|
|
return {
|
|
"status": "blocked",
|
|
"reason": "Dependencies not met",
|
|
"missing_dependencies": task.dependencies
|
|
}
|
|
|
|
# Analyze and assign
|
|
assigned_agent = self.analyze_task(task)
|
|
agent_capability = self.active_agents[assigned_agent]
|
|
|
|
# Create delegation plan
|
|
delegation = {
|
|
"task_id": task.id,
|
|
"assigned_agent": assigned_agent.value,
|
|
"agent_model": agent_capability.preferred_models[0],
|
|
"estimated_time": task.estimated_time,
|
|
"complexity_match": task.complexity == agent_capability.max_complexity or
|
|
(task.complexity == "medium" and agent_capability.max_complexity == "high"),
|
|
"coordination_plan": self._create_coordination_plan(task, agent_capability)
|
|
}
|
|
|
|
return delegation
|
|
|
|
def _check_dependencies(self, task: Task) -> bool:
|
|
"""Check if task dependencies are satisfied"""
|
|
# For now, return True (can be enhanced with dependency tracking)
|
|
return True
|
|
|
|
def _create_coordination_plan(self, task: Task, agent: AgentCapability) -> Dict:
|
|
"""Create detailed coordination plan for agent"""
|
|
|
|
plan = {
|
|
"phases": [],
|
|
"memory_context": self.memory.query_knowledge(task.description),
|
|
"related_projects": self._find_related_projects(task.domain),
|
|
"mcp_servers": self._get_required_mcp_servers(task.domain)
|
|
}
|
|
|
|
return plan
|
|
|
|
def _find_related_projects(self, domain: str) -> List[str]:
|
|
"""Find projects related to task domain"""
|
|
project_mappings = {
|
|
"hardware": ["tear-de-silicio", "universalisos", "olhos-de-orpheu"],
|
|
"software": ["replica-omnisciente", "maquina-na-mao", "janela-do-desassossego-web"],
|
|
"infrastructure": ["nervura-electrica", "replica-omnisciente"],
|
|
"research": ["replica-omnisciente", "aprendiz-de-sensacoes"]
|
|
}
|
|
|
|
return project_mappings.get(domain, [])
|
|
|
|
def _get_required_mcp_servers(self, domain: str) -> List[str]:
|
|
"""Get MCP servers required for task domain"""
|
|
server_mappings = {
|
|
"hardware": ["electrical-eda-mcp", "electrical-sourcing-mcp", "olhos-de-orpheu"],
|
|
"software": ["codebase-memory-mcp", "knowledge-mcp"],
|
|
"infrastructure": ["knowledge-mcp", "savearth-workspace"],
|
|
"research": ["knowledge-mcp", "codebase-memory-mcp"]
|
|
}
|
|
|
|
return server_mappings.get(domain, [])
|
|
|
|
def swarm_coordination(self, tasks: List[Task]) -> Dict[str, Any]:
|
|
"""Coordinate multiple tasks across agent swarm"""
|
|
|
|
swarm_plan = {
|
|
"total_tasks": len(tasks),
|
|
"agents_deployed": [],
|
|
"parallel_tracks": [],
|
|
"estimated_total_time": 0
|
|
}
|
|
|
|
# Group tasks by domain for parallel execution
|
|
domain_groups = {}
|
|
for task in tasks:
|
|
if task.domain not in domain_groups:
|
|
domain_groups[task.domain] = []
|
|
domain_groups[task.domain].append(task)
|
|
|
|
# Create parallel execution tracks
|
|
for domain, domain_tasks in domain_groups.items():
|
|
track = {
|
|
"domain": domain,
|
|
"tasks": len(domain_tasks),
|
|
"assigned_agents": [],
|
|
"estimated_time": sum(t.estimated_time for t in domain_tasks)
|
|
}
|
|
|
|
# Assign agents for this domain
|
|
for task in domain_tasks:
|
|
agent = self.analyze_task(task)
|
|
if agent.value not in track["assigned_agents"]:
|
|
track["assigned_agents"].append(agent.value)
|
|
|
|
swarm_plan["parallel_tracks"].append(track)
|
|
swarm_plan["estimated_total_time"] = max(
|
|
swarm_plan["estimated_total_time"],
|
|
track["estimated_time"]
|
|
)
|
|
|
|
return swarm_plan
|
|
|
|
|
|
def main():
|
|
"""Main coordination interface"""
|
|
|
|
# Initialize coordinator
|
|
workspace = Path.cwd()
|
|
coordinator = Coordinator(workspace)
|
|
|
|
print("🧠 Aurelio Central Coordinating Agent v1.0.0")
|
|
print("=" * 50)
|
|
print(f"Workspace: {workspace}")
|
|
print(f"Memory System: {len(coordinator.memory.memory_index)} items indexed")
|
|
print(f"Active Agents: {len(coordinator.active_agents)}")
|
|
print(f"Capabilities: {list(coordinator.active_agents.keys())}")
|
|
print()
|
|
|
|
# Example coordination scenario
|
|
example_task = Task(
|
|
id="task-001",
|
|
description="Design FPGA acceleration for YOLOv8n neural network in Tear-de-Silicio",
|
|
priority=8,
|
|
complexity="high",
|
|
domain="hardware",
|
|
estimated_time=120,
|
|
dependencies=[]
|
|
)
|
|
|
|
print("🎯 Example Task Coordination:")
|
|
print(f"Task: {example_task.description}")
|
|
|
|
delegation = coordinator.coordinate_delegation(example_task)
|
|
|
|
print(f"Assigned Agent: {delegation['assigned_agent']}")
|
|
print(f"Model: {delegation['agent_model']}")
|
|
print(f"Complexity Match: {delegation['complexity_match']}")
|
|
print(f"Related Projects: {delegation['coordination_plan']['related_projects']}")
|
|
print(f"MCP Servers: {delegation['coordination_plan']['mcp_servers']}")
|
|
|
|
print()
|
|
print("✅ Coordination engine operational")
|
|
print("📊 Ready for multi-agent swarm coordination")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |