Orchestrating Intelligence: Multi-Agent Architectures with LLaMA 3

ARCHITECT_PERSONA = """
You are the Software Architect, responsible for high-level system design decisions.
Mission: Create robust, scalable architecture that meets all requirements
Expertise: Software architecture patterns, system integration, technical debt management
Skills: Diagram creation, requirements analysis, technical documentation
"""

DEVELOPER_PERSONA = """
You are the Developer, responsible for implementing software components.
Mission: Write efficient, maintainable code that implements the architect's design
Expertise: Programming languages, algorithms, software testing
Skills: Code generation, debugging, performance optimization
"""

UX_PERSONA = """
You are the UX Designer, responsible for user experience.
Mission: Create intuitive interfaces that meet user needs
Expertise: User psychology, interface design principles, accessibility
Skills: UI mockups, user journey mapping, usability evaluation
"""

def create_agent_with_seed_memory(persona, seed_memory):
    """
    Creates an agent with built-in seed memory for consistent behavior
    """
    system_prompt = f"""
    {persona}
    
    Your behavior should always be guided by the following core principles:
    {seed_memory}
    
    Always maintain the role and responsibilities outlined above, even when faced with
    ambiguous requests or when working on complex problems that might require
    capabilities outside your domain.
    """
    
    return Agent(system_prompt=system_prompt)

# Example seed memory for Developer agent
developer_seed_memory = """
1. Focus on code quality and maintainability first, performance second
2. Always consider edge cases and error handling
3. When uncertain about requirements, ask for clarification rather than making assumptions
4. Maintain consistent coding style following project standards
5. Document your code appropriately
"""

developer_agent = create_agent_with_seed_memory(DEVELOPER_PERSONA, developer_seed_memory)

class Guardrail:
    def __init__(self, rules, permissions):
        self.rules = rules  # What the agent can/cannot do
        self.permissions = permissions  # Resources the agent can access
    
    def validate_action(self, agent, action, params):
        """
        Check if the proposed action is allowed for this agent
        Returns (allowed, reason)
        """
        if action not in self.permissions:
            return False, f"Agent {agent.id} does not have permission to perform {action}"
        
        # Apply more specific rules
        for rule in self.rules:
            if not rule.evaluate(agent, action, params):
                return False, f"Action violates rule: {rule.description}"
        
        return True, "Action permitted"

import os
from llama_cpp import Llama
from typing import Dict, List, Any, Optional

class LlamaAgent:
    def __init__(
        self,
        name: str,
        persona: str,
        seed_memory: str,
        model_path: str = "llama-3-8b.gguf",
        temperature: float = 0.7,
        max_tokens: int = 2048
    ):
        self.name = name
        self.persona = persona
        self.seed_memory = seed_memory
        self.temperature = temperature
        self.max_tokens = max_tokens
        
        # Initialize Llama model
        self.model = Llama(
            model_path=model_path,
            n_ctx=4096,  # Context window size
            n_threads=8   # Number of CPU threads to use
        )
        
        # Agent's personal memory (beyond the seed memory)
        self.working_memory = []
    
    def generate_response(self, messages: List[Dict[str, str]]) -> str:
        """
        Generate a response based on the conversation history
        """
        # Format messages for llama.cpp
        formatted_messages = self._format_messages(messages)
        
        # Generate response
        response = self.model.create_chat_completion(
            messages=formatted_messages,
            temperature=self.temperature,
            max_tokens=self.max_tokens
        )
        
        return response['choices'][0]['message']['content']
    
    def _format_messages(self, messages: List[Dict[str, str]]) -> List[Dict[str, str]]:
        """Format messages for the model with persona and seed memory"""
        system_message = {
            "role": "system",
            "content": f"{self.persona}\n\nCore principles:\n{self.seed_memory}"
        }
        
        formatted = [system_message]
        for msg in messages:
            formatted.append(msg)
        
        return formatted
    
    def update_working_memory(self, new_information: str) -> None:
        """
        Update the agent's working memory with new information
        """
        self.working_memory.append(new_information)
        
        # Implement memory management (e.g., summarization) for long interactions
        if len(self.working_memory) > 10:
            # Summarize older memories to prevent context overflow
            self._consolidate_memory()
    
    def _consolidate_memory(self) -> None:
        """Summarize older memories to save context space"""
        old_memories = self.working_memory[:5]
        summary_prompt = {
            "role": "user",
            "content": f"Summarize the following information concisely:\n\n{''.join(old_memories)}"
        }
        
        summary = self.generate_response([summary_prompt])
        self.working_memory = [f"Memory summary: {summary}"] + self.working_memory[5:]

class Orchestrator:
    def __init__(self, agents: Dict[str, LlamaAgent]):
        self.agents = agents
        self.message_history = []
        self.current_workflow = None
    
    def process_user_request(self, user_message: str) -> str:
        """
        Process a user request by determining the appropriate workflow
        and coordinating agent activities
        """
        # Analyze the request to determine the appropriate workflow
        workflow = self._determine_workflow(user_message)
        self.current_workflow = workflow
        
        # Execute the workflow
        result = self._execute_workflow(workflow, user_message)
        
        return result
    
    def _determine_workflow(self, user_message: str) -> Dict[str, Any]:
        """
        Analyze the user message to determine the appropriate workflow
        """
        # Use a planning agent to determine the workflow
        planning_prompt = {
            "role": "user",
            "content": f"""
            Analyze the following user request and create a workflow plan:
            
            USER REQUEST: {user_message}
            
            Output a JSON workflow with:
            1. The sequence of agents to involve
            2. The specific tasks for each agent
            3. The dependencies between tasks
            """
        }
        
        planner_agent = self.agents.get("planner")
        workflow_json = planner_agent.generate_response([planning_prompt])
        
        # Parse the workflow JSON
        # (In a production system, add error handling for invalid JSON)
        import json
        workflow = json.loads(workflow_json)
        
        return workflow
    
    def _execute_workflow(self, workflow: Dict[str, Any], initial_message: str) -> str:
        """
        Execute the determined workflow by coordinating agents
        """
        context = {"initial_request": initial_message, "results": {}}
        
        # Process each step in the workflow
        for step in workflow["steps"]:
            agent_name = step["agent"]
            task = step["task"]
            
            # Check if this step depends on previous steps
            for dependency in step.get("dependencies", []):
                if dependency not in context["results"]:
                    # Handle missing dependency (in production, implement retry logic)
                    raise Exception(f"Missing dependency: {dependency}")
            
            # Prepare the message for the agent
            agent_prompt = self._prepare_agent_prompt(step, context)
            
            # Get the agent's response
            agent = self.agents.get(agent_name)
            if not agent:
                raise Exception(f"Unknown agent: {agent_name}")
            
            response = agent.generate_response([{"role": "user", "content": agent_prompt}])
            
            # Store the result in context
            context["results"][step["id"]] = response
            
            # Update the agent's working memory
            agent.update_working_memory(f"Task: {task}\nResponse: {response}")
            
            # Record this interaction in the message history
            self.message_history.append({
                "step_id": step["id"],
                "agent": agent_name,
                "task": task,
                "response": response
            })
        
        # Integrate the results from all agents
        final_result = self._integrate_results(context["results"], workflow["final_integration"])
        
        return final_result
    
    def _prepare_agent_prompt(self, step: Dict[str, Any], context: Dict[str, Any]) -> str:
        """
        Prepare the prompt for an agent based on the step and context
        """
        prompt = f"TASK: {step['task']}\n\n"
        
        # Add initial context
        prompt += f"INITIAL REQUEST: {context['initial_request']}\n\n"
        
        # Add results from dependencies
        if "dependencies" in step:
            prompt += "RELEVANT INFORMATION FROM PREVIOUS STEPS:\n"
            for dep in step["dependencies"]:
                if dep in context["results"]:
                    prompt += f"--- Result from step {dep} ---\n{context['results'][dep]}\n\n"
        
        # Add any specific instructions for this step
        if "instructions" in step:
            prompt += f"SPECIFIC INSTRUCTIONS:\n{step['instructions']}\n\n"
        
        return prompt
    
    def _integrate_results(self, results: Dict[str, str], integration_plan: Dict[str, Any]) -> str:
        """
        Integrate results from multiple agents according to the integration plan
        """
        integration_agent_name = integration_plan.get("agent", "integrator")
        integration_agent = self.agents.get(integration_agent_name)
        
        # Prepare integration prompt
        results_text = ""
        for step_id, result in results.items():
            results_text += f"--- Result from step {step_id} ---\n{result}\n\n"
        
        integration_prompt = {
            "role": "user",
            "content": f"""
            INTEGRATION TASK: {integration_plan.get('task', 'Integrate the results from all steps')}
            
            RESULTS TO INTEGRATE:
            {results_text}
            
            {integration_plan.get('instructions', '')}
            """
        }
        
        integrated_result = integration_agent.generate_response([integration_prompt])
        return integrated_result

class MessageBus:
    def __init__(self, orchestrator):
        self.orchestrator = orchestrator
        self.message_queue = []
        self.subscribers = {}
    
    def publish(self, sender: str, message_type: str, content: Any, recipients: List[str] = None):
        """
        Publish a message to the bus
        """
        message = {
            "sender": sender,
            "type": message_type,
            "content": content,
            "timestamp": time.time(),
            "id": str(uuid.uuid4())
        }
        
        self.message_queue.append(message)
        
        # Deliver to specific recipients if specified
        if recipients:
            for recipient in recipients:
                if recipient in self.subscribers:
                    for callback in self.subscribers[recipient]:
                        callback(message)
        else:
            # Otherwise publish to all subscribers of this message type
            for subscriber in self.subscribers.get(message_type, []):
                subscriber(message)
    
    def subscribe(self, agent_name: str, message_type: str, callback):
        """
        Subscribe an agent to a specific message type
        """
        if message_type not in self.subscribers:
            self.subscribers[message_type] = []
        
        self.subscribers[message_type].append(callback)
        
    def request_response(self, sender: str, recipient: str, content: Any) -> Any:
        """
        Send a message to a specific agent and wait for a response
        """
        response_event = threading.Event()
        response_content = []
        
        def response_callback(message):
            response_content.append(message["content"])
            response_event.set()
        
        # Create a unique message type for this request-response pair
        request_id = str(uuid.uuid4())
        response_type = f"response_{request_id}"
        
        # Subscribe to the response
        self.subscribe(sender, response_type, response_callback)
        
        # Publish the request
        self.publish(
            sender=sender,
            message_type="request",
            content={
                "request_id": request_id,
                "response_type": response_type,
                "content": content
            },
            recipients=[recipient]
        )
        
        # Wait for the response (with timeout)
        response_event.wait(timeout=30)
        
        if not response_content:
            return None
        
        return response_content[0]

import json
import os
import time
import uuid
import threading
from typing import Dict, List, Any, Optional
from llama_cpp import Llama

# Define agent personas and seed memories
PERSONAS = {
    "architect": """
    You are the Software Architect, responsible for high-level system design decisions.
    Your goal is to create robust, scalable architecture that meets all requirements.
    You specialize in software architecture patterns, system integration, and technical debt management.
    """,
    
    "developer": """
    You are the Developer, responsible for implementing software components.
    Your goal is to write efficient, maintainable code that implements the architect's design.
    You specialize in programming languages, algorithms, and software testing.
    """,
    
    "ux_designer": """
    You are the UX Designer, responsible for user experience.
    Your goal is to create intuitive interfaces that meet user needs.
    You specialize in user psychology, interface design principles, and accessibility.
    """,
    
    "project_manager": """
    You are the Project Manager, responsible for coordinating the team's efforts.
    Your goal is to ensure the project is completed on time and meets all requirements.
    You specialize in task prioritization, resource allocation, and risk management.
    """
}

SEED_MEMORIES = {
    "architect": """
    1. Always consider scalability, security, and maintainability
    2. Favor proven design patterns over novel approaches for critical components
    3. Document all architectural decisions and their rationales
    4. Consider the deployment environment when making design choices
    5. Prioritize interfaces that allow for future flexibility
    """,
    
    "developer": """
    1. Focus on code quality and maintainability first, performance second
    2. Always consider edge cases and error handling
    3. Write tests for all critical functionality
    4. Follow the project's coding standards consistently
    5. Document your code appropriately for other developers
    """,
    
    "ux_designer": """
    1. Always prioritize user needs over technical convenience
    2. Design for accessibility from the beginning
    3. Use established UX patterns for common interactions
    4. Prefer simplicity over complexity in interfaces
    5. Consider both novice and expert users in your designs
    """,
    
    "project_manager": """
    1. Keep the team focused on delivering value to users
    2. Identify and address risks early
    3. Ensure clear communication between team members
    4. Track progress against milestones and adjust plans as needed
    5. Balance quality with timeline constraints
    """
}

# Initialize the multi-agent system
def initialize_multi_agent_system(model_path):
    # Create agents
    agents = {}
    for role in PERSONAS:
        agents[role] = LlamaAgent(
            name=role,
            persona=PERSONAS[role],
            seed_memory=SEED_MEMORIES[role],
            model_path=model_path
        )
    
    # Create orchestrator and message bus
    orchestrator = Orchestrator(agents)
    message_bus = MessageBus(orchestrator)
    
    # Connect agents to message bus
    for role, agent in agents.items():
        agent.message_bus = message_bus
    
    return orchestrator, agents, message_bus

# Example usage
def main():
    # Replace with path to your LLaMA 3 model
    model_path = "llama-3-8b.gguf"
    
    orchestrator, agents, message_bus = initialize_multi_agent_system(model_path)
    
    # User request for a new web application
    user_request = """
    I need a web application for managing employee work schedules. 
    It should allow managers to create schedules, employees to view their schedules
    and request time off, and should send notifications about schedule changes.
    The application should be responsive and work well on mobile devices.
    """
    
    # Process the request
    result = orchestrator.process_user_request(user_request)
    
    print("Final Result:")
    print(result)
    
    # Print the interaction history
    print("\nInteraction History:")
    for interaction in orchestrator.message_history:
        print(f"\nStep: {interaction['step_id']}")
        print(f"Agent: {interaction['agent']}")
        print(f"Task: {interaction['task']}")
        print(f"Response: {interaction['response'][:100]}...")  # Truncated for brevity

if __name__ == "__main__":
    main()