#!/usr/bin/env python3 """ Genesis CAMEL Multi-Agent Orchestrator ======================================== Integration with CAMEL-AI framework for advanced multi-agent coordination. Features: - Role-playing sessions for domain expertise - Task decomposition and planning - Workforce coordination - Browser toolkit integration """ import os import sys from typing import Dict, List, Optional, Any from dataclasses import dataclass from datetime import datetime # Add genesis-memory to path sys.path.insert(0, "/mnt/e/genesis-system/genesis-memory") @dataclass class CAMELAgent: """CAMEL agent definition.""" agent_id: str name: str role: str capabilities: List[str] system_message: str @dataclass class RolePlayingSession: """A CAMEL role-playing session.""" session_id: str assistant_role: str user_role: str task: str messages: List[Dict[str, str]] status: str # 'active', 'completed', 'failed' created_at: str @dataclass class WorkforceTask: """A task for the CAMEL workforce.""" task_id: str description: str assigned_agents: List[str] subtasks: List[Dict[str, Any]] status: str result: Optional[Dict] = None # CAMEL Agent Definitions CAMEL_AGENTS = [ CAMELAgent( "CAM-001", "Role Player", "Domain Expert Simulator", ["role_playing", "domain_expertise", "persona_simulation"], "You are a versatile domain expert capable of assuming various professional roles." ), CAMELAgent( "CAM-002", "Task Planner", "Strategic Task Decomposition", ["task_decomposition", "planning", "dependency_analysis"], "You excel at breaking down complex tasks into manageable subtasks with clear dependencies." ), CAMELAgent( "CAM-003", "Workforce Manager", "Multi-Agent Coordination", ["agent_coordination", "workload_balancing", "progress_tracking"], "You coordinate multiple agents to complete complex workflows efficiently." ), ] class GenesisCAMELOrchestrator: """ Integrate CAMEL multi-agent framework with Genesis. This orchestrator provides: 1. Role-playing sessions for domain expertise simulation 2. Task decomposition and planning 3. Workforce coordination for complex workflows 4. Browser toolkit integration for web automation """ def __init__(self, api_key: Optional[str] = None): self.api_key = api_key or os.getenv("ANTHROPIC_API_KEY") or os.getenv("OPENAI_API_KEY") self.agents = {a.agent_id: a for a in CAMEL_AGENTS} self.sessions: Dict[str, RolePlayingSession] = {} self.workforce_tasks: Dict[str, WorkforceTask] = {} self._camel_available = self._check_camel() def _check_camel(self) -> bool: """Check if CAMEL is installed and available.""" try: import camel return True except ImportError: return False def get_status(self) -> Dict[str, Any]: """Get orchestrator status.""" return { "camel_available": self._camel_available, "api_key_set": bool(self.api_key), "agents": len(self.agents), "active_sessions": len([s for s in self.sessions.values() if s.status == 'active']), "active_tasks": len([t for t in self.workforce_tasks.values() if t.status == 'in_progress']), "timestamp": datetime.now().isoformat() } def create_role_playing_session( self, assistant_role: str, user_role: str, task: str, session_id: Optional[str] = None ) -> Dict[str, Any]: """ Create a CAMEL role-playing session. Args: assistant_role: The role for the AI assistant (e.g., "Python Developer") user_role: The role for the simulated user (e.g., "Project Manager") task: The task to accomplish through role-playing Returns: Session information dictionary """ import uuid session_id = session_id or f"rp-{uuid.uuid4().hex[:8]}" session = RolePlayingSession( session_id=session_id, assistant_role=assistant_role, user_role=user_role, task=task, messages=[], status="active", created_at=datetime.now().isoformat() ) self.sessions[session_id] = session if self._camel_available: try: from camel.societies import RolePlaying from camel.types import ModelType rp = RolePlaying( assistant_role_name=assistant_role, user_role_name=user_role, task_prompt=task, with_task_specify=True ) return { "success": True, "session_id": session_id, "assistant_role": assistant_role, "user_role": user_role, "task": task, "camel_session": True } except Exception as e: return { "success": True, "session_id": session_id, "assistant_role": assistant_role, "user_role": user_role, "task": task, "camel_session": False, "warning": f"CAMEL session creation failed: {e}" } return { "success": True, "session_id": session_id, "assistant_role": assistant_role, "user_role": user_role, "task": task, "camel_session": False, "note": "Using fallback mode (CAMEL not installed)" } def decompose_task(self, task_description: str) -> Dict[str, Any]: """ Decompose a complex task into subtasks. Uses CAMEL's task decomposition if available, otherwise uses a rule-based fallback. """ import uuid task_id = f"decomp-{uuid.uuid4().hex[:8]}" # Simple rule-based decomposition (fallback) subtasks = self._rule_based_decomposition(task_description) workforce_task = WorkforceTask( task_id=task_id, description=task_description, assigned_agents=["CAM-002"], subtasks=subtasks, status="decomposed" ) self.workforce_tasks[task_id] = workforce_task return { "success": True, "task_id": task_id, "original_task": task_description, "subtasks": subtasks, "subtask_count": len(subtasks), "method": "camel" if self._camel_available else "rule_based" } def _rule_based_decomposition(self, task: str) -> List[Dict[str, Any]]: """Simple rule-based task decomposition.""" # Common task patterns subtasks = [] task_lower = task.lower() # Analysis tasks if any(w in task_lower for w in ['analyze', 'review', 'examine', 'investigate']): subtasks.extend([ {"step": 1, "action": "gather_data", "description": "Collect relevant data and information"}, {"step": 2, "action": "analyze", "description": "Perform detailed analysis"}, {"step": 3, "action": "synthesize", "description": "Synthesize findings"}, {"step": 4, "action": "report", "description": "Generate report with recommendations"} ]) # Creation tasks elif any(w in task_lower for w in ['create', 'build', 'develop', 'implement']): subtasks.extend([ {"step": 1, "action": "plan", "description": "Plan the implementation approach"}, {"step": 2, "action": "design", "description": "Design the solution architecture"}, {"step": 3, "action": "implement", "description": "Implement the solution"}, {"step": 4, "action": "test", "description": "Test and validate"}, {"step": 5, "action": "deploy", "description": "Deploy and document"} ]) # Automation tasks elif any(w in task_lower for w in ['automate', 'schedule', 'workflow']): subtasks.extend([ {"step": 1, "action": "identify", "description": "Identify automation targets"}, {"step": 2, "action": "design_flow", "description": "Design automation workflow"}, {"step": 3, "action": "configure", "description": "Configure automation rules"}, {"step": 4, "action": "test", "description": "Test automation"}, {"step": 5, "action": "monitor", "description": "Set up monitoring"} ]) # Default decomposition else: subtasks.extend([ {"step": 1, "action": "understand", "description": "Understand the requirements"}, {"step": 2, "action": "plan", "description": "Create execution plan"}, {"step": 3, "action": "execute", "description": "Execute the plan"}, {"step": 4, "action": "verify", "description": "Verify results"} ]) return subtasks def create_workforce( self, task: str, agent_roles: Optional[List[str]] = None ) -> Dict[str, Any]: """ Create a coordinated workforce for a complex task. Args: task: The main task to accomplish agent_roles: Optional list of roles to include in the workforce Returns: Workforce configuration dictionary """ import uuid workforce_id = f"wf-{uuid.uuid4().hex[:8]}" # Default roles if not specified if agent_roles is None: agent_roles = [ "Project Manager", "Technical Lead", "Developer", "QA Engineer" ] # Decompose the task decomposition = self.decompose_task(task) workforce_config = { "workforce_id": workforce_id, "task": task, "roles": agent_roles, "subtasks": decomposition["subtasks"], "status": "configured", "created_at": datetime.now().isoformat() } return { "success": True, "workforce_id": workforce_id, "config": workforce_config, "agent_count": len(agent_roles), "subtask_count": len(decomposition["subtasks"]) } def execute_browser_task(self, task: str) -> Dict[str, Any]: """ Execute a browser automation task via CAMEL agent. This integrates with the Genesis Browser Swarm for actual browser automation. """ from agents.browser_coordinator import get_coordinator, AgentType coordinator = get_coordinator() # Create task in browser swarm browser_task = coordinator.create_task( title="CAMEL Browser Task", description=task, agent_type=AgentType.UTILITY, priority=7 ) # Get available agent agent = coordinator.get_available_agent(AgentType.UTILITY) if agent: coordinator.assign_task(browser_task.task_id, agent.agent_id) return { "success": True, "task_id": browser_task.task_id, "assigned_agent": agent.agent_id, "status": "assigned", "note": "Task assigned to browser swarm agent" } return { "success": True, "task_id": browser_task.task_id, "status": "queued", "note": "Task queued, waiting for available agent" } # Singleton instance _orchestrator: Optional[GenesisCAMELOrchestrator] = None def get_orchestrator() -> GenesisCAMELOrchestrator: """Get the singleton orchestrator instance.""" global _orchestrator if _orchestrator is None: _orchestrator = GenesisCAMELOrchestrator() return _orchestrator # CLI interface if __name__ == "__main__": orchestrator = get_orchestrator() status = orchestrator.get_status() print("=" * 60) print("GENESIS CAMEL ORCHESTRATOR STATUS") print("=" * 60) print(f"\nCAMEL Available: {status['camel_available']}") print(f"API Key Set: {status['api_key_set']}") print(f"Registered Agents: {status['agents']}") print(f"Active Sessions: {status['active_sessions']}") print(f"Active Tasks: {status['active_tasks']}") print("\n" + "=" * 60) print("CAMEL AGENTS") print("=" * 60) for agent in CAMEL_AGENTS: print(f"\n{agent.agent_id}: {agent.name}") print(f" Role: {agent.role}") print(f" Capabilities: {', '.join(agent.capabilities)}") # Test task decomposition print("\n" + "=" * 60) print("TASK DECOMPOSITION TEST") print("=" * 60) test_task = "Create an automated customer onboarding workflow" result = orchestrator.decompose_task(test_task) print(f"\nTask: {test_task}") print(f"Method: {result['method']}") print(f"Subtasks ({result['subtask_count']}):") for st in result['subtasks']: print(f" {st['step']}. {st['action']}: {st['description']}")