# constitutional_guard.py import re import logging from datetime import datetime # Configure logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') class Directive: def __init__(self, name, principle, description): self.name = name self.principle = principle self.description = description def __repr__(self): return f"Directive(name='{self.name}', principle='{self.principle[:20]}...', description='{self.description[:20]}...')" class ConstitutionalGuard: def __init__(self, constitution_file): self.constitution = self.load_constitution(constitution_file) self.directives = self.parse_directives(self.constitution) self.audit_trail = [] # List of dictionaries to store audit logs def load_constitution(self, constitution_file): try: with open(constitution_file, 'r') as f: return f.read() except FileNotFoundError: logging.error(f"Constitution file not found: {constitution_file}") raise except Exception as e: logging.error(f"Error loading constitution: {e}") raise def parse_directives(self, constitution): directives = [] # Regex to find the start of each prime directive directive_starts = re.finditer(r"PRIME DIRECTIVE \d+:", constitution) directive_matches = list(directive_starts) for i in range(len(directive_matches)): start = directive_matches[i].start() end = directive_matches[i+1].start() if i+1 < len(directive_matches) else len(constitution) directive_text = constitution[start:end] # Extract name (e.g., "MEMORY") name_match = re.search(r"PRIME DIRECTIVE \d+: (\w+)", directive_text) name = name_match.group(1).strip() if name_match else "UNKNOWN" # Extract principle principle_match = re.search(r"The Principle\n+\"*([^\"]+)\"*", directive_text) principle = principle_match.group(1).strip() if principle_match else "UNKNOWN" # Extract description. Taking the whole section for now. Can refine later. description = directive_text directives.append(Directive(name, principle, description)) return directives def validate_action(self, action, context): """ Validates an action against the constitutional directives. Args: action (str): A description of the action to be taken. context (dict): A dictionary containing relevant context for the action. Returns: bool: True if the action is compliant, False otherwise. """ compliance_scores = self.score_compliance(action, context) is_compliant = all(score > 0.5 for score in compliance_scores.values()) # Example threshold: 0.5 if not is_compliant: self.handle_violation(action, context, compliance_scores) self.log_compliance_decision(action, context, compliance_scores, is_compliant) return is_compliant def score_compliance(self, action, context): """ Scores the compliance of an action against each directive. Args: action (str): A description of the action. context (dict): A dictionary containing relevant context. Returns: dict: A dictionary of compliance scores, one for each directive. """ scores = {} for directive in self.directives: scores[directive.name] = self.evaluate_compliance(action, context, directive) return scores def evaluate_compliance(self, action, context, directive): """ Evaluates the compliance of an action against a specific directive. Args: action (str): A description of the action. context (dict): A dictionary containing relevant context. directive (Directive): The directive to evaluate against. Returns: float: A compliance score between 0 and 1. """ # Implement your compliance scoring logic here. This is a placeholder. # This could involve semantic similarity checks, keyword analysis, etc. # The example below is very basic and should be replaced with something more sophisticated. action_lower = action.lower() directive_name_lower = directive.name.lower() if directive_name_lower in action_lower: return 0.7 # Partially compliant else: return 0.3 # Potentially non-compliant def handle_violation(self, action, context, compliance_scores): """ Handles a violation of the constitutional directives. Args: action (str): A description of the action. context (dict): A dictionary containing relevant context. compliance_scores (dict): A dictionary of compliance scores. """ logging.warning(f"Constitutional violation detected: Action='{action}', Context='{context}', Scores='{compliance_scores}'") # Implement your violation handling logic here. # This could involve blocking the action, escalating to a human, etc. # For this example, we'll just log the violation. print(f"Action blocked: {action} due to constitutional violation.") raise Exception(f"Action blocked: {action} due to constitutional violation. Compliance scores: {compliance_scores}") def log_compliance_decision(self, action, context, compliance_scores, is_compliant): """ Logs the compliance decision and relevant information. Args: action (str): A description of the action. context (dict): A dictionary containing relevant context. compliance_scores (dict): A dictionary of compliance scores. is_compliant (bool): Whether the action is compliant. """ log_entry = { "timestamp": datetime.utcnow().isoformat(), "action": action, "context": context, "compliance_scores": compliance_scores, "is_compliant": is_compliant } self.audit_trail.append(log_entry) logging.info(f"Compliance decision logged: {log_entry}") def get_compliance_report(self): """ Generates a compliance report. Returns: list: A list of compliance log entries. """ return self.audit_trail # Example usage: if __name__ == '__main__': try: guard = ConstitutionalGuard("E__genesis-system_CONSTITUTION_PRIME_DIRECTIVES_CORRECTED.md") # Example 1: Compliant action action1 = "Store customer interaction in PostgreSQL RLM" context1 = {"customer_id": 123, "interaction_type": "email"} is_compliant1 = guard.validate_action(action1, context1) print(f"Action 1 is compliant: {is_compliant1}") # Example 2: Non-compliant action action2 = "Delete all customer data without backup" context2 = {"reason": "disk space"} try: is_compliant2 = guard.validate_action(action2, context2) print(f"Action 2 is compliant: {is_compliant2}") # This line shouldn't be reached except Exception as e: print(f"Action 2 validation failed: {e}") # Example 3: Generate revenue through AgileAdapt action3 = "Recommend AgileAdapt to a tradesperson" context3 = {"tradesperson_type": "plumber", "location": "Sydney"} is_compliant3 = guard.validate_action(action3, context3) print(f"Action 3 is compliant: {is_compliant3}") # Get compliance report report = guard.get_compliance_report() print("\nCompliance Report:") for entry in report: print(entry) except Exception as e: print(f"An error occurred: {e}")