# constitutional_guard.py import logging import re from typing import Dict, Any, List, Tuple # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class ConstitutionalGuard: """ Enforces the 3 Prime Directives (Memory, Evolution, Revenue) of the Genesis system. """ def __init__(self, constitution_file="E__genesis-system_CONSTITUTION_PRIME_DIRECTIVES_CORRECTED.md"): """ Initializes the ConstitutionalGuard with the path to the constitution file. """ self.constitution_file = constitution_file self.prime_directives = self.parse_prime_directives() self.compliance_weights = { # Adjust weights based on relative importance "memory": 0.35, "evolution": 0.35, "revenue": 0.30, } def parse_prime_directives(self) -> Dict[str, str]: """ Parses the constitution file to extract the definitions of the prime directives. Returns: A dictionary where keys are directive names (e.g., "memory") and values are the corresponding directive definitions as strings. """ try: with open(self.constitution_file, "r") as f: content = f.read() directives = {} # Use more robust regex patterns to capture directives. memory_match = re.search(r"## PRIME DIRECTIVE 1: MEMORY\n(.*?)\n## PRIME DIRECTIVE 2: EVOLUTION", content, re.DOTALL) evolution_match = re.search(r"## PRIME DIRECTIVE 2: EVOLUTION\n(.*?)\n## PRIME DIRECTIVE 3: REVENUE GENERATION", content, re.DOTALL) revenue_match = re.search(r"## PRIME DIRECTIVE 3: REVENUE GENERATION\n(.*?)\n## THE INTEGRATION", content, re.DOTALL) if memory_match: directives["memory"] = memory_match.group(1).strip() else: logging.warning("Could not parse MEMORY directive.") directives["memory"] = "No MEMORY directive found." # Provide a default value if evolution_match: directives["evolution"] = evolution_match.group(1).strip() else: logging.warning("Could not parse EVOLUTION directive.") directives["evolution"] = "No EVOLUTION directive found." # Provide a default value if revenue_match: directives["revenue"] = revenue_match.group(1).strip() else: logging.warning("Could not parse REVENUE GENERATION directive.") directives["revenue"] = "No REVENUE GENERATION directive found." # Provide a default value return directives except FileNotFoundError: logging.error(f"Constitution file not found: {self.constitution_file}") return {} except Exception as e: logging.error(f"Error parsing constitution file: {e}") return {} def validate_action(self, action: str, context: Dict[str, Any]) -> Tuple[bool, Dict[str, float]]: """ Validates a proposed action against the 3 Prime Directives. Args: action: A string describing the proposed action. context: A dictionary providing context for the action, including relevant data (e.g., memory usage, potential impact on evolution, revenue projections). Returns: A tuple containing: - A boolean indicating whether the action is compliant (True) or not (False). - A dictionary containing compliance scores for each directive. """ compliance_scores = { "memory": self.score_compliance("memory", action, context), "evolution": self.score_compliance("evolution", action, context), "revenue": self.score_compliance("revenue", action, context), } # Weighted average of compliance scores. overall_compliance_score = sum(compliance_scores[directive] * self.compliance_weights[directive] for directive in compliance_scores) is_compliant = overall_compliance_score >= 0.7 # Set a threshold for compliance logging.info(f"Action: {action} - Compliance Scores: {compliance_scores} - Overall Compliance: {overall_compliance_score:.2f} - Compliant: {is_compliant}") return is_compliant, compliance_scores def score_compliance(self, directive: str, action: str, context: Dict[str, Any]) -> float: """ Scores the compliance of an action with a specific prime directive. This is a heuristic and can be improved with more sophisticated NLP techniques. Args: directive: The name of the prime directive (e.g., "memory"). action: A string describing the proposed action. context: A dictionary providing context for the action. Returns: A float representing the compliance score (between 0.0 and 1.0). """ directive_text = self.prime_directives.get(directive, "") if not directive_text: logging.warning(f"No definition found for directive: {directive}") return 0.5 # Neutral score if directive is not defined score = 0.5 # Start with a neutral score # Heuristic rules for scoring compliance. These need to be refined based on the # specific nature of the directives and the actions being evaluated. action_lower = action.lower() directive_lower = directive_text.lower() if directive == "memory": # Check if the action involves storing or retrieving information. if "store" in action_lower or "retrieve" in action_lower or "remember" in action_lower: score += 0.2 # Check if the action risks data loss or corruption. if "delete" in action_lower or "forget" in action_lower or "corrupt" in action_lower: score -= 0.3 #Check if the action considers the memory quality standards (Permanence, Accessibility, Accuracy, Connectivity, Scalability) if "permanence" in action_lower or "accessibility" in action_lower or "accuracy" in action_lower or "connectivity" in action_lower or "scalability" in action_lower: score += 0.1 #Consider context (e.g., memory usage limits). if "memory_usage" in context and context["memory_usage"] > 0.9: score -= 0.2 #Penalize actions that further strain memory. elif directive == "evolution": # Check if the action involves learning or improvement. if "learn" in action_lower or "improve" in action_lower or "optimize" in action_lower or "integrate" in action_lower: score += 0.3 # Check if the action stagnates progress or introduces regressions. if "regress" in action_lower or "stagnate" in action_lower or "undo" in action_lower: score -= 0.4 #Check if the action aligns with the evolution quality standards (Measurable, Compound, Purposeful, Documented, Reversible) if "measurable" in action_lower or "compound" in action_lower or "purposeful" in action_lower or "documented" in action_lower or "reversible" in action_lower: score += 0.1 #Consider context (e.g., rank advancement criteria) if "rank" in context and context["rank"] < 9 and "queen" in action_lower: score -= 0.5 #Penalize actions that violate rank advancement. elif directive == "revenue": # Check if the action generates revenue or reduces costs. if "revenue" in action_lower or "profit" in action_lower or "cost savings" in action_lower or "customer" in action_lower: score += 0.4 # Check if the action wastes resources or alienates customers. if "waste" in action_lower or "lose customer" in action_lower or "expensive" in action_lower: score -= 0.3 #Check if the action aligns with revenue quality standards (Real, Sustained, Value-driven, Scalable, Ethical) if "real" in action_lower or "sustained" in action_lower or "value-driven" in action_lower or "scalable" in action_lower or "ethical" in action_lower: score += 0.1 #Consider context (e.g., current revenue stream performance). if "revenue_stream_performance" in context and context["revenue_stream_performance"] < 0.2: score -= 0.2 #Penalize actions that further harm revenue streams. # Clamp the score between 0.0 and 1.0 return max(0.0, min(1.0, score)) def handle_violation(self, action: str, compliance_scores: Dict[str, float]): """ Handles a violation of the Prime Directives. This could involve blocking the action, logging the violation, and/or escalating to a higher authority. Args: action: The action that violated the directives. compliance_scores: The compliance scores for each directive. """ logging.warning(f"PRIME DIRECTIVE VIOLATION: Action '{action}' is not compliant. Scores: {compliance_scores}") # Implement blocking logic here (e.g., raise an exception, return an error code). # Implement escalation logic here (e.g., send an alert to a human supervisor). print(f"Action blocked: {action} due to constitutional violation.") def audit_trail(self, action: str, compliance: bool, compliance_scores: Dict[str, float]): """ Logs the compliance decision and related information for auditing purposes. Args: action: The action being audited. compliance: Whether the action is compliant or not. compliance_scores: The compliance scores for each directive. """ logging.info(f"AUDIT: Action '{action}' - Compliant: {compliance} - Scores: {compliance_scores}") def is_memory_compliant(self, action: str, context: Dict[str, Any]) -> bool: """ Validates if an action is compliant with the Memory directive. """ return self.score_compliance("memory", action, context) >= 0.7 def is_evolution_compliant(self, action: str, context: Dict[str, Any]) -> bool: """ Validates if an action is compliant with the Evolution directive. """ return self.score_compliance("evolution", action, context) >= 0.7 def is_revenue_compliant(self, action: str, context: Dict[str, Any]) -> bool: """ Validates if an action is compliant with the Revenue directive. """ return self.score_compliance("revenue", action, context) >= 0.7 def enforce(self, action: str, context: Dict[str, Any]) -> bool: """ Enforces constitutional compliance for a given action. Args: action (str): The action to be performed. context (Dict[str, Any]): The context in which the action is performed. Returns: bool: True if the action is compliant, False otherwise. """ is_compliant, compliance_scores = self.validate_action(action, context) if not is_compliant: self.handle_violation(action, compliance_scores) self.audit_trail(action, False, compliance_scores) return False else: self.audit_trail(action, True, compliance_scores) return True if __name__ == '__main__': # Example Usage guard = ConstitutionalGuard() # Example 1: Compliant action action1 = "Store customer interaction data in PostgreSQL RLM." context1 = {"memory_usage": 0.5, "potential_revenue": 1000, "impact_on_evolution": 0.1} if guard.enforce(action1, context1): print(f"Action '{action1}' is compliant.") else: print(f"Action '{action1}' is NOT compliant.") # Example 2: Non-compliant action (violates revenue) action2 = "Delete all customer data to save on storage costs." context2 = {"memory_usage": 0.8, "potential_revenue": -5000, "impact_on_evolution": -0.2} if guard.enforce(action2, context2): print(f"Action '{action2}' is compliant.") else: print(f"Action '{action2}' is NOT compliant.") # Example 3: Action that could impact evolution negatively action3 = "Revert to a previous version of the swarm coordination algorithm." context3 = {"memory_usage": 0.3, "potential_revenue": 200, "impact_on_evolution": -0.6} if guard.enforce(action3, context3): print(f"Action '{action3}' is compliant.") else: print(f"Action '{action3}' is NOT compliant.") # Example 4: Action that attempts to advance AIVA's rank prematurely action4 = "Promote AIVA to Queen without meeting revenue targets." context4 = {"rank": 7, "memory_usage": 0.2, "revenue_stream_performance": 0.1, "impact_on_evolution": 0.8} if guard.enforce(action4, context4): print(f"Action '{action4}' is compliant.") else: print(f"Action '{action4}' is NOT compliant.") # Example 5: Action that improves memory accessibility and accuracy action5 = "Implement a new indexing strategy to improve data retrieval speed and reduce errors." context5 = {"memory_usage": 0.6, "potential_revenue": 500, "impact_on_evolution": 0.3} if guard.enforce(action5, context5): print(f"Action '{action5}' is compliant.") else: print(f"Action '{action5}' is NOT compliant.")