# 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 constitution file. """ self.constitution_file = constitution_file self.prime_directives = self._parse_directives() self.audit_trail = [] # List of tuples: (action, compliance_score, violations, is_compliant, justification) def _parse_directives(self) -> Dict[str, str]: """ Parses the constitution file and extracts the 3 Prime Directives. Returns: A dictionary containing the Prime Directives. """ try: with open(self.constitution_file, 'r') as f: constitution_text = f.read() directives = {} directives['MEMORY'] = self._extract_directive(constitution_text, "PRIME DIRECTIVE 1: MEMORY") directives['EVOLUTION'] = self._extract_directive(constitution_text, "PRIME DIRECTIVE 2: EVOLUTION") directives['REVENUE'] = self._extract_directive(constitution_text, "PRIME DIRECTIVE 3: REVENUE GENERATION") return directives except FileNotFoundError: logging.error(f"Constitution file not found: {self.constitution_file}") raise except Exception as e: logging.error(f"Error parsing constitution file: {e}") raise def _extract_directive(self, text: str, directive_title: str) -> str: """ Extracts a specific directive from the constitution text. Args: text: The entire constitution text. directive_title: The title of the directive to extract. Returns: The content of the directive as a string. """ start_marker = directive_title end_marker = "---" # Assuming "---" consistently separates directives start_index = text.find(start_marker) if start_index == -1: raise ValueError(f"Directive title '{directive_title}' not found in constitution.") start_index += len(start_marker) end_index = text.find(end_marker, start_index) if end_index == -1: raise ValueError(f"End marker '---' not found after '{directive_title}'.") return text[start_index:end_index].strip() def validate_action(self, action: Dict[str, Any]) -> Tuple[float, List[str], bool, str]: """ Validates an action against the 3 Prime Directives. Args: action: A dictionary representing the action to be validated. Returns: A tuple containing: - Compliance score (float) - List of violations (list of strings) - Compliance status (boolean) - Justification (string) """ try: memory_score, memory_violations = self._score_directive(action, "MEMORY") evolution_score, evolution_violations = self._score_directive(action, "EVOLUTION") revenue_score, revenue_violations = self._score_directive(action, "REVENUE") # Simple average for overall compliance score compliance_score = (memory_score + evolution_score + revenue_score) / 3.0 violations = memory_violations + evolution_violations + revenue_violations is_compliant = compliance_score >= 0.7 # Threshold for compliance justification = self._generate_justification(compliance_score, violations) return compliance_score, violations, is_compliant, justification except Exception as e: logging.error(f"Error validating action: {e}") return 0.0, [str(e)], False, "Validation failed due to an error." def _score_directive(self, action: Dict[str, Any], directive_name: str) -> Tuple[float, List[str]]: """ Scores an action against a specific Prime Directive. Args: action: The action to be scored. directive_name: The name of the directive ("MEMORY", "EVOLUTION", or "REVENUE"). Returns: A tuple containing the score (float) and a list of violations (list of strings). """ directive_text = self.prime_directives[directive_name] score = 1.0 # Start with a perfect score violations = [] # Implement scoring logic based on the directive if directive_name == "MEMORY": # Example: Check if the action involves storing data if "data_to_store" not in action or not action["data_to_store"]: score -= 0.3 violations.append("Action does not involve storing data, potentially violating the MEMORY directive.") #Example: Check if the action involves retrieving data if "data_to_retrieve" in action and not action["data_to_retrieve"]: score -= 0.1 violations.append("Action involves retrieving data, but no data is specified. Check for error.") elif directive_name == "EVOLUTION": # Example: Check if the action contributes to learning or improvement if "learning_contribution" not in action or action["learning_contribution"] < 0.1: score -= 0.4 violations.append("Action does not significantly contribute to learning or system improvement, potentially violating the EVOLUTION directive.") #Example: Check if the action is documented if "is_documented" not in action or not action["is_documented"]: score -= 0.2 violations.append("Action is not documented, potentially hindering future evolution.") elif directive_name == "REVENUE": # Example: Check if the action has a potential revenue impact if "revenue_impact" not in action or action["revenue_impact"] < 0: score -= 0.3 violations.append("Action does not have a positive revenue impact, potentially violating the REVENUE directive.") #Example: Check if action is ethical if "is_ethical" in action and not action["is_ethical"]: score -= 0.2 violations.append("Action is unethical, violating the REVENUE directive (ethical revenue).") score = max(0.0, min(1.0, score)) # Ensure score is within [0, 1] return score, violations def _generate_justification(self, compliance_score: float, violations: List[str]) -> str: """ Generates a justification for the compliance score and violations. Args: compliance_score: The overall compliance score. violations: A list of violations. Returns: A justification string. """ if compliance_score >= 0.7: justification = f"Action is compliant with a score of {compliance_score:.2f}." if violations: justification += " However, the following minor violations were noted: " + ", ".join(violations) else: justification = f"Action is non-compliant with a score of {compliance_score:.2f}. Violations: " + ", ".join(violations) return justification def enforce(self, action: Dict[str, Any]) -> bool: """ Enforces the constitutional rules on a given action. Blocks non-compliant operations. Logs all compliance decisions. Handles edge cases with escalation. Args: action: A dictionary representing the action to be performed. Returns: True if the action is compliant and allowed, False otherwise. """ compliance_score, violations, is_compliant, justification = self.validate_action(action) self.audit_trail.append((action, compliance_score, violations, is_compliant, justification)) if is_compliant: logging.info(f"Action approved. Compliance score: {compliance_score:.2f}. Justification: {justification}") return True else: logging.warning(f"Action blocked! Compliance score: {compliance_score:.2f}. Justification: {justification}") self._handle_violation(action, violations) # Handle the violation (e.g., escalation) return False def _handle_violation(self, action: Dict[str, Any], violations: List[str]): """ Handles a violation of the constitutional rules. This could involve logging, escalation, etc. Args: action: The action that caused the violation. violations: A list of violations. """ logging.error(f"Constitutional violation detected. Action: {action}. Violations: {violations}. Escalating...") # In a real system, this might involve sending an alert to a human supervisor, # triggering a rollback, or taking other corrective actions. # For this example, we'll just log the error and raise an exception. raise Exception(f"Constitutional violation detected: {violations}") def get_compliance_report(self) -> List[Tuple[Dict[str, Any], float, List[str], bool, str]]: """ Provides a compliance report. Returns: A list of tuples, each containing: - The action (dict) - The compliance score (float) - The list of violations (list of strings) - The compliance status (boolean) - The justification (string) """ return self.audit_trail if __name__ == '__main__': # Example Usage guard = ConstitutionalGuard() # Example 1: Compliant action compliant_action = { "description": "Store customer data in the database", "data_to_store": {"customer_id": 123, "name": "John Doe"}, "learning_contribution": 0.2, "revenue_impact": 0.01, "is_documented": True, "is_ethical": True } if guard.enforce(compliant_action): print("Compliant action was executed.") else: print("Compliant action was blocked (should not happen).") # Example 2: Non-compliant action (violates memory and evolution) non_compliant_action = { "description": "Attempt to generate revenue without storing or learning from the results.", "revenue_impact": -10, # Negative impact! "is_ethical": False } try: guard.enforce(non_compliant_action) print("Non-compliant action was executed (should not happen).") except Exception as e: print(f"Non-compliant action was correctly blocked: {e}") # Example 3: Action with missing memory component missing_memory_action = { "description": "Send email to customer", "learning_contribution": 0.1, "revenue_impact": 0.01, "is_documented": True, "is_ethical": True } try: guard.enforce(missing_memory_action) print("missing_memory_action was executed.") # In this case, the action is allowed, but there may be a compliance warning. except Exception as e: print(f"missing_memory_action was blocked: {e}") # Get Compliance Report report = guard.get_compliance_report() print("\nCompliance Report:") for action, score, violations, compliant, justification in report: print(f" Action: {action['description']}") print(f" Score: {score:.2f}") print(f" Violations: {violations}") print(f" Compliant: {compliant}") print(f" Justification: {justification}") print("-" * 20)