qatosim.py

# Quantum Agent-Task Optimization (QATO) Simulation
# This script simulates how a quantum annealer would solve the optimal task
# assignment problem for the "Society of Minds" architecture.

# We use D-Wave's 'dimod' library, which provides tools for working with
# QUBO (Quadratic Unconstrained Binary Optimization) models, the native
# input format for quantum annealers.
# You can install it via pip: pip install dimod numpy

import numpy as np
import dimod
import time

def create_society_of_minds():
    """Defines the agents and their specialized roles for Phase I: The Town."""
    # As per the spec, we create 30 agents with specialized roles.
    # To make it interesting, we'll create a few of each type.
    roles = {
        "Data_Analyst": 5,
        "Creative_Writer": 3,
        "Debugger": 4,
        "Network_Security_Specialist": 5,
        "System_Administrator": 4,
        "UX_Designer": 3,
        "Project_Manager": 2,
        "Hardware_Technician": 4
    }
    agents = []
    for role, count in roles.items():
        for i in range(1, count + 1):
            agents.append(f"{role}_{i}")
    return agents

def create_job_board():
    """Defines the tasks and their required skills for the Task Marketplace."""
    # Tasks are designed to map to the agent specializations.
    tasks = {
        "Analyze_Q3_Login_Anomalies": "Data_Analyst",
        "Draft_Internal_Security_Bulletin": "Creative_Writer",
        "Patch_Kernel_Vulnerability_CVE-2024-XXXX": "Debugger",
        "Configure_New_Firewall_Rules": "Network_Security_Specialist",
        "Migrate_Database_Server": "System_Administrator",
        "Design_New_Dashboard_Mockup": "UX_Designer",
        "Coordinate_Software_Release_v2.1": "Project_Manager",
        "Replace_Failing_RAID_Controller": "Hardware_Technician",
        "Identify_Phishing_Campaign_Patterns": "Data_Analyst",
        "Review_Code_For_Buffer_Overflows": "Debugger",
        "Audit_VPN_Access_Logs": "Network_Security_Specialist",
        "Write_User_Documentation_For_New_App": "Creative_Writer",
    }
    return tasks

def calculate_suitability(agents, tasks):
    """
    Creates a suitability matrix (S).
    S[i, j] is a score representing how suitable agent i is for task j.
    A higher score is better.
    """
    num_agents = len(agents)
    num_task_keys = len(tasks.keys())
    task_list = list(tasks.keys())
    
    # Initialize a matrix of zeros
    suitability_matrix = np.zeros((num_agents, num_task_keys))

    for i, agent_name in enumerate(agents):
        agent_role = agent_name.split('_')[0]
        for j, task_name in enumerate(task_list):
            required_role = tasks[task_name]
            if agent_role == required_role:
                # High suitability if the role matches perfectly
                suitability_matrix[i, j] = 10
            else:
                # Low suitability otherwise
                suitability_matrix[i, j] = 1
                
    return suitability_matrix

def build_qubo(suitability_matrix):
    """
    Constructs the QUBO model from the suitability matrix.
    The goal is to MAXIMIZE the total suitability score. Quantum annealers
    MINIMIZE, so we formulate the problem as minimizing the NEGATIVE of the score.
    """
    num_agents, num_tasks = suitability_matrix.shape
    
    # Initialize the QUBO dictionary
    Q = {}

    # --- Objective Function ---
    # We want to maximize sum(S_ij * x_ij), where x_ij is a binary variable
    # that is 1 if agent i is assigned task j.
    # This is equivalent to minimizing sum(-S_ij * x_ij).
    # In QUBO, linear terms go on the diagonal.
    for i in range(num_agents):
        for j in range(num_tasks):
            # The variable is x_ij, which we can represent as a tuple (i, j)
            variable = (i, j)
            Q[(variable, variable)] = -suitability_matrix[i, j]

    # --- Constraints ---
    # We need to add penalties to ensure the solution is valid.
    # A large penalty term (gamma) ensures constraints are met.
    gamma = 20 # This value should be larger than any possible suitability score

    # Constraint 1: Each task must be assigned to exactly one agent.
    # For each task j, sum(x_ij for i in agents) == 1
    # The penalty term for this is gamma * (sum(x_ij) - 1)^2
    for j in range(num_tasks):
        for i in range(num_agents):
            # Linear part of the squared term
            var1 = (i, j)
            if (var1, var1) in Q:
                Q[(var1, var1)] += gamma * -2
            else:
                Q[(var1, var1)] = gamma * -2
            
            # Quadratic part of the squared term
            for k in range(i + 1, num_agents):
                var2 = (k, j)
                if (var1, var2) in Q:
                    Q[(var1, var2)] += gamma * 2
                else:
                    Q[(var1, var2)] = gamma * 2

    # Constraint 2: Each agent can be assigned at most one task.
    # For each agent i, sum(x_ij for j in tasks) <= 1
    # The penalty term for this is gamma * (sum(x_ij) * (sum(x_ij) - 1))
    # which simplifies to gamma * sum(x_ij * x_ik for j < k)
    for i in range(num_agents):
        for j in range(num_tasks):
            for k in range(j + 1, num_tasks):
                var1 = (i, j)
                var2 = (i, k)
                if (var1, var2) in Q:
                    Q[(var1, var2)] += gamma * 2
                else:
                    Q[(var1, var2)] = gamma * 2
                    
    return Q

def solve_and_interpret(qubo, agents, tasks, S):
    """Solves the QUBO using a simulated annealer and interprets the result."""
    task_list = list(tasks.keys())
    num_agents = len(agents)
    num_tasks = len(task_list)

    # Use dimod's classical solver that simulates the behavior of a quantum annealer
    sampler = dimod.SimulatedAnnealingSampler()
    
    # num_reads is like running the experiment multiple times to find the best result.
    # We've reduced this from 100 to 10 to significantly speed up the simulation.
    num_reads = 10
    
    print(f"** Solving with {num_reads} reads... (this may take a moment) **")
    start_time = time.time()
    response = sampler.sample_qubo(qubo, num_reads=num_reads)
    end_time = time.time()
    print(f"** Solving complete in {end_time - start_time:.2f} seconds. **\n")

    
    # The best solution is the one with the lowest energy
    best_solution = response.first.sample
    
    print("--- Optimal Task Assignment Found ---")
    total_suitability = 0
    assignments_found = 0
    
    for i in range(num_agents):
        for j in range(num_tasks):
            if best_solution.get((i, j), 0) == 1:
                agent_name = agents[i]
                task_name = task_list[j]
                suitability_score = S[i, j]
                total_suitability += suitability_score
                print(f"Assign -> Agent: {agent_name:<30} | Task: {task_name:<40} | Suitability: {suitability_score:.0f}")
                assignments_found += 1
    
    if assignments_found == 0:
        print("\n!! No valid assignments were found. The model may be too constrained or the penalty 'gamma' may need adjustment.")
    
    print("\n--- Total Suitability Score ---")
    print(f"The total score for this assignment is: {total_suitability:.0f}")


# --- Main Execution ---
if __name__ == "__main__":
    print("Initializing Project Chimera: Phase I (The Town)...")
    
    # 1. Create the agents and tasks
    agents = create_society_of_minds()
    tasks = create_job_board()
    print(f"Created {len(agents)} agents and {len(tasks)} tasks.\n")

    # 2. Define the problem by calculating suitability
    print("Calculating suitability matrix...")
    S = calculate_suitability(agents, tasks)
    
    # 3. Formulate the problem for the quantum computer (build the QUBO)
    print("Building QUBO model for quantum annealer...")
    qubo_model = build_qubo(S)
    
    # 4. Solve on the simulated quantum annealer and interpret the results
    print("Sending to simulated quantum annealer to find optimal assignment...\n")
    solve_and_interpret(qubo_model, agents, tasks, S)