Core Modules

Complete reference for Q-Memetic AI core modules and their interfaces.

Module Overview

Q-Memetic AI is organized into several core modules, each handling specific aspects of the memetic computing system:

Module	Purpose	Key Classes
core.engine	Main orchestration engine	MemeticEngine
core.meme	Meme representation and operations	Meme, MemeVector, MemeMetadata
core.entanglement	Quantum-inspired correlation networks	QuantumEntangler
core.evolution	Genetic algorithms and evolution	GeneticEvolver, EvolutionParameters
cognitive.models	User cognitive modeling	CognitiveFingerprint
federation.client	Federated learning client	FederatedClient
licensing.manager	License validation and management	LicenseManager
utils.persistence	Data storage and retrieval	DataPersistence
utils.security	Security and encryption	SecurityManager
visualization.noosphere	Interactive visualizations	NoosphereVisualizer

Quick Import Guide

# Core engine
from qmemetic_ai import MemeticEngine

# Individual components
from qmemetic_ai.core.meme import Meme, MemeVector, MemeMetadata
from qmemetic_ai.core.entanglement import QuantumEntangler
from qmemetic_ai.core.evolution import GeneticEvolver, EvolutionParameters
from qmemetic_ai.cognitive.models import CognitiveFingerprint
from qmemetic_ai.federation.client import FederatedClient
from qmemetic_ai.licensing.manager import LicenseManager
from qmemetic_ai.utils.persistence import DataPersistence
from qmemetic_ai.utils.security import SecurityManager
from qmemetic_ai.visualization.noosphere import NoosphereVisualizer

# Exception classes
from qmemetic_ai.licensing.manager import QMemeticLicenseError

Cognitive Models

CognitiveFingerprint

Models individual user preferences and cognitive patterns.

class CognitiveFingerprint:
    """
    Represents a user's cognitive patterns and preferences.

    Attributes:
        user_id: Unique user identifier
        preferences: Dictionary of user preferences
        interaction_history: Historical interaction data
        adaptation_rate: How quickly preferences change
    """

    def __init__(self, user_id: str, preferences: Optional[Dict] = None):
        """Initialize cognitive fingerprint for a user."""

    def update_preferences(self, interaction_data: Dict) -> None:
        """Update preferences based on user interactions."""

    def calculate_fit_score(self, meme: Meme) -> float:
        """Calculate how well a meme fits this user's cognitive profile."""

    def adapt_meme(self, meme: Meme) -> Meme:
        """Adapt a meme to better fit this user's preferences."""

    def to_dict(self) -> Dict[str, Any]:
        """Serialize cognitive fingerprint to dictionary."""

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> 'CognitiveFingerprint':
        """Deserialize cognitive fingerprint from dictionary."""

Usage Example

# Create cognitive fingerprint
profile = CognitiveFingerprint(
    user_id="researcher_123",
    preferences={
        "technical_level": 0.8,
        "domain_interests": ["AI", "quantum", "biology"],
        "communication_style": "formal",
        "complexity_preference": 0.7
    }
)

# Calculate meme fit
meme = engine.create_meme("Quantum computing applications in drug discovery")
fit_score = profile.calculate_fit_score(meme)
print(f"Meme fit score: {fit_score:.3f}")

# Adapt meme for user
adapted_meme = profile.adapt_meme(meme)
print(f"Adapted content: {adapted_meme.content}")

Federation Client

FederatedClient

Handles distributed memetic computing across multiple nodes.

class FederatedClient:
    """
    Client for federated memetic computing.

    Enables distributed evolution and sharing of memes
    across multiple computational nodes.
    """

    def __init__(self, node_id: str, config: Optional[Dict] = None):
        """Initialize federated client."""

    async def register_with_network(self) -> bool:
        """Register this node with the federated network."""

    async def discover_peers(self) -> List[str]:
        """Discover other nodes in the network."""

    async def sync_memes(self, mode: str = "bidirectional") -> Dict[str, Any]:
        """Synchronize memes with peer nodes."""

    async def federated_evolve(
        self, 
        memes: List[Meme], 
        generations: int = 5
    ) -> List[Meme]:
        """Perform distributed evolution across the network."""

    async def broadcast_meme(self, meme: Meme) -> List[str]:
        """Broadcast a meme to all connected peers."""

    def get_network_status(self) -> Dict[str, Any]:
        """Get status of federated network connections."""

Usage Example

# Initialize federated client
fed_client = FederatedClient(
    node_id="research_lab_1",
    config={
        "discovery_port": 8080,
        "max_peers": 10,
        "trust_threshold": 0.7
    }
)

# Register with network
await fed_client.register_with_network()

# Discover peers
peers = await fed_client.discover_peers()
print(f"Found {len(peers)} peer nodes")

# Synchronize memes
sync_result = await fed_client.sync_memes(mode="pull")
print(f"Received {sync_result['memes_received']} memes")

# Federated evolution
local_memes = [engine.create_meme("Local research insight")]
evolved = await fed_client.federated_evolve(local_memes, generations=5)

Licensing Manager

LicenseManager

Handles QuantumMeta license validation and feature access control.

class LicenseManager:
    """
    Manages QuantumMeta licensing and feature access.

    Provides strict enforcement of license terms with
    24-hour grace period for new users.
    """

    def __init__(self, license_key: Optional[str] = None):
        """Initialize license manager."""

    def validate_license(self) -> bool:
        """Validate the current license key."""

    def get_license_status(self) -> Dict[str, Any]:
        """Get detailed license status information."""

    def get_available_features(self) -> List[str]:
        """Get list of features available with current license."""

    def require_feature(self, feature_name: str) -> None:
        """Require a specific feature, raise exception if not available."""

    def show_license_info(self) -> None:
        """Display license information to console."""

    @staticmethod
    def licensed_class(tier_required: str = "core"):
        """Decorator to protect entire classes with license requirements."""

    @staticmethod
    def requires_license(features: List[str] = None):
        """Decorator to protect individual methods with license requirements."""

Usage Example

# Initialize license manager
license_manager = LicenseManager(license_key="your-key")

# Validate license
is_valid = license_manager.validate_license()
print(f"License valid: {is_valid}")

# Check available features
features = license_manager.get_available_features()
print(f"Available features: {features}")

# Require specific feature
try:
    license_manager.require_feature("federation_basic")
    print("Federation feature available")
except QMemeticLicenseError as e:
    print(f"License error: {e}")

# Use decorators
@LicenseManager.requires_license(features=["advanced_analytics"])
def advanced_analysis(data):
    """This function requires advanced analytics license."""
    return analyze_complex_data(data)

Utils: Persistence

DataPersistence

Handles data storage, backup, and retrieval operations.

class DataPersistence:
    """
    Manages persistent storage of memetic data.

    Supports multiple storage backends and automatic
    backup with compression.
    """

    def __init__(self, data_dir: str, config: Optional[Dict] = None):
        """Initialize data persistence manager."""

    def save_meme(self, meme: Meme) -> str:
        """Save a meme to persistent storage."""

    def load_meme(self, meme_id: str) -> Meme:
        """Load a meme from persistent storage."""

    def save_memes_batch(self, memes: List[Meme]) -> List[str]:
        """Save multiple memes efficiently."""

    def load_memes_batch(self, meme_ids: List[str]) -> List[Meme]:
        """Load multiple memes efficiently."""

    def export_data(self, data: Dict, file_path: str) -> None:
        """Export data to external file."""

    def import_data(self, file_path: str) -> Dict:
        """Import data from external file."""

    def create_backup(self) -> str:
        """Create a backup of all data."""

    def restore_backup(self, backup_path: str) -> None:
        """Restore data from backup."""

    def cleanup_old_data(self, retention_days: int = 30) -> int:
        """Clean up old data files."""

Usage Example

# Initialize persistence manager
persistence = DataPersistence(
    data_dir="./qmemetic_data",
    config={
        "compression": True,
        "backup_interval": 3600,
        "max_backup_files": 10
    }
)

# Save individual meme
meme = engine.create_meme("Important research finding")
meme_path = persistence.save_meme(meme)
print(f"Meme saved to: {meme_path}")

# Load meme
loaded_meme = persistence.load_meme(meme.meme_id)
print(f"Loaded: {loaded_meme.content}")

# Batch operations
memes = [engine.create_meme(f"Idea #{i}") for i in range(100)]
paths = persistence.save_memes_batch(memes)
print(f"Saved {len(paths)} memes")

# Create backup
backup_path = persistence.create_backup()
print(f"Backup created: {backup_path}")

Utils: Security

SecurityManager

Provides encryption, authentication, and security features.

class SecurityManager:
    """
    Manages security features including encryption,
    authentication, and access control.
    """

    def __init__(self, config: Optional[Dict] = None):
        """Initialize security manager."""

    def generate_session_id(self) -> str:
        """Generate a secure session identifier."""

    def encrypt_data(self, data: bytes, key: Optional[bytes] = None) -> bytes:
        """Encrypt data using AES-256-GCM."""

    def decrypt_data(self, encrypted_data: bytes, key: Optional[bytes] = None) -> bytes:
        """Decrypt data using AES-256-GCM."""

    def hash_password(self, password: str) -> str:
        """Hash password using bcrypt."""

    def verify_password(self, password: str, hashed: str) -> bool:
        """Verify password against bcrypt hash."""

    def generate_api_key(self) -> str:
        """Generate a secure API key."""

    def validate_api_key(self, api_key: str) -> bool:
        """Validate an API key."""

    def create_digital_signature(self, data: bytes) -> bytes:
        """Create digital signature for data integrity."""

    def verify_digital_signature(self, data: bytes, signature: bytes) -> bool:
        """Verify digital signature."""

Usage Example

# Initialize security manager
security = SecurityManager(config={
    "encryption_key": "your-encryption-key",
    "require_authentication": True
})

# Generate session
session_id = security.generate_session_id()
print(f"Session ID: {session_id}")

# Encrypt sensitive data
sensitive_data = b"confidential research data"
encrypted = security.encrypt_data(sensitive_data)
print(f"Encrypted {len(encrypted)} bytes")

# Decrypt data
decrypted = security.decrypt_data(encrypted)
print(f"Decrypted: {decrypted.decode()}")

# Password management
password_hash = security.hash_password("user_password")
is_valid = security.verify_password("user_password", password_hash)
print(f"Password valid: {is_valid}")

# API key generation
api_key = security.generate_api_key()
print(f"Generated API key: {api_key[:16]}...")

Error Handling

Exception Classes

# License-related exceptions
class QMemeticLicenseError(Exception):
    """Raised when license validation fails."""
    pass

# Evolution-related exceptions
class MemeticEvolutionError(Exception):
    """Raised when evolution process fails."""
    pass

# Network-related exceptions
class NetworkEntanglementError(Exception):
    """Raised when network operations fail."""
    pass

# Federation-related exceptions
class FederationError(Exception):
    """Raised when federated operations fail."""
    pass

# Data persistence exceptions
class DataPersistenceError(Exception):
    """Raised when data operations fail."""
    pass

Error Handling Patterns

from qmemetic_ai.exceptions import (
    QMemeticLicenseError,
    MemeticEvolutionError,
    NetworkEntanglementError
)

try:
    # Initialize engine
    engine = MemeticEngine(license_key="your-key")

    # Perform operations
    memes = engine.create_memes(content_list)
    evolved = engine.evolve(memes, generations=10)
    network = engine.entangle()

except QMemeticLicenseError as e:
    print(f"License error: {e}")
    print("Please check your license key or contact support")

except MemeticEvolutionError as e:
    print(f"Evolution error: {e}")
    print("Trying with reduced parameters...")
    evolved = engine.evolve(memes, generations=5, population_size=10)

except NetworkEntanglementError as e:
    print(f"Network error: {e}")
    print("Continuing without entanglement network")

except Exception as e:
    print(f"Unexpected error: {e}")
    engine.save_state("emergency_backup.json")

Type Hints and Validation

Q-Memetic AI uses comprehensive type hints for better development experience:

from typing import List, Dict, Optional, Union, Any, Tuple
from pydantic import BaseModel, Field, validator

# Type aliases
MemeID = str
UserID = str
Fitness = float
Vector = List[float]

# Pydantic models for validation
class MemeMetadata(BaseModel):
    author: Optional[str] = None
    domain: Optional[str] = None
    tags: List[str] = Field(default_factory=list)
    generation: int = 0
    created_at: float = Field(default_factory=time.time)

    @validator('tags')
    def validate_tags(cls, v):
        return [tag.strip().lower() for tag in v if tag.strip()]

class EvolutionParameters(BaseModel):
    population_size: int = Field(default=20, ge=2, le=1000)
    generations: int = Field(default=5, ge=1, le=100)
    mutation_rate: float = Field(default=0.25, ge=0.0, le=1.0)
    crossover_rate: float = Field(default=0.75, ge=0.0, le=1.0)

    @validator('mutation_rate', 'crossover_rate')
    def validate_probability(cls, v):
        if not 0.0 <= v <= 1.0:
            raise ValueError("Probability must be between 0 and 1")
        return v

Performance Monitoring

Built-in performance monitoring and profiling:

# Performance decorators
from qmemetic_ai.utils.performance import profile_function, monitor_memory

@profile_function
@monitor_memory
def expensive_operation(data):
    """Function with automatic performance monitoring."""
    return process_large_dataset(data)

# Context managers for profiling
with engine.profile_evolution() as profiler:
    evolved_memes = engine.evolve(memes, generations=10)

performance_report = profiler.get_report()
print(f"Evolution took {performance_report['total_time']:.2f} seconds")
print(f"Memory peak: {performance_report['memory_peak']:.1f} MB")

Configuration Integration

All modules support configuration through the unified config system:

# Module-specific configuration
config = {
    "cognitive": {
        "adaptation_rate": 0.1,
        "memory_decay": 0.95,
        "preference_weights": {
            "novelty": 0.3,
            "relevance": 0.4,
            "clarity": 0.3
        }
    },
    "federation": {
        "node_id": "research_node",
        "discovery_port": 8080,
        "trust_threshold": 0.7
    },
    "persistence": {
        "compression": True,
        "backup_interval": 3600,
        "retention_days": 30
    },
    "security": {
        "encryption_enabled": True,
        "audit_logging": True,
        "session_timeout": 7200
    }
}

# Apply configuration to engine
engine = MemeticEngine(license_key="your-key", config=config)

Testing and Development

Unit Testing

import unittest
from qmemetic_ai.core.meme import Meme, MemeMetadata

class TestMeme(unittest.TestCase):
    def setUp(self):
        self.metadata = MemeMetadata(
            author="test_user",
            domain="test",
            tags=["test", "example"]
        )

    def test_meme_creation(self):
        meme = Meme(
            content="Test meme content",
            metadata=self.metadata
        )
        self.assertEqual(meme.content, "Test meme content")
        self.assertEqual(meme.metadata.author, "test_user")

    def test_meme_serialization(self):
        meme = Meme("Test content", self.metadata)
        meme_dict = meme.to_dict()
        reconstructed = Meme.from_dict(meme_dict)
        self.assertEqual(meme.content, reconstructed.content)

Mock Testing

from unittest.mock import Mock, patch
from qmemetic_ai.core.engine import MemeticEngine

class TestMemeticEngine(unittest.TestCase):
    @patch('qmemetic_ai.licensing.manager.validate_qmemetic_license')
    def test_engine_initialization(self, mock_validate):
        mock_validate.return_value = True

        engine = MemeticEngine(license_key="test-key")
        self.assertIsNotNone(engine)
        mock_validate.assert_called_once()

For detailed API documentation of specific modules, see:

• Engine API
• Memes API
• Entanglement API
• Evolution API
• Visualization API