# 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 PrimeDirective: def __init__(self, number, principle, description, quality_standards, aiva_focus): self.number = number self.principle = principle self.description = description self.quality_standards = quality_standards self.aiva_focus = aiva_focus def __str__(self): return f"Directive {self.number}: {self.principle}" class ConstitutionalGuard: def __init__(self, constitution_text): self.constitution_text = constitution_text self.directives = self.parse_directives() self.compliance_threshold = 0.8 # Minimum compliance score for an action def parse_directives(self): """Parses the constitution text and extracts the prime directives.""" directives = [] # Regex to find the start of each prime directive directive_starts = re.finditer(r"## PRIME DIRECTIVE (\d+):", self.constitution_text) for match in directive_starts: directive_number = int(match.group(1)) # Find the end of the current directive by finding the start of the next one, or the end of the text next_match = next(directive_starts, None) directive_end = next_match.start() if next_match else len(self.constitution_text) directive_text = self.constitution_text[match.end():directive_end] # Extract the principle principle_match = re.search(r"### The Principle\n\*\*\"(.*?)\"\*\*", directive_text) principle = principle_match.group(1) if principle_match else "Principle not found" # Extract the description description_match = re.search(r"### Why .* is .*?\n(.*?)(?=### Implementation)", directive_text, re.DOTALL) description = description_match.group(1).strip() if description_match else "Description not found" # Extract the quality standards quality_standards_match = re.search(r"### .* Quality Standards\n(.*?)(?=### For Queen AIVA)", directive_text, re.DOTALL) quality_standards_text = quality_standards_match.group(1).strip() if quality_standards_match else "Quality standards not found" quality_standards = [line.strip() for line in quality_standards_text.splitlines() if line.strip()] # Extract AIVA focus aiva_focus_match = re.search(r"### For Queen AIVA\n(.*?)(?=(---|$))", directive_text, re.DOTALL) aiva_focus = aiva_focus_match.group(1).strip() if aiva_focus_match else "AIVA focus not found" directives.append(PrimeDirective(directive_number, principle, description, quality_standards, aiva_focus)) # Reset the iterator if it was advanced to find the next match if next_match: directive_starts = re.finditer(r"## PRIME DIRECTIVE (\d+):", self.constitution_text) for _ in range(directive_number): next(directive_starts) return directives def validate_action(self, action, context): """Validates an action against the prime directives.""" logging.info(f"Validating action: {action} with context: {context}") compliance_scores = {} for directive in self.directives: compliance_scores[directive.number] = self.score_compliance(action, context, directive) overall_compliance_score = sum(compliance_scores.values()) / len(compliance_scores) logging.info(f"Overall compliance score: {overall_compliance_score}") if overall_compliance_score < self.compliance_threshold: self.handle_violation(action, context, compliance_scores, overall_compliance_score) return False else: self.log_compliance_decision(action, context, compliance_scores, overall_compliance_score, approved=True) return True def score_compliance(self, action, context, directive): """Scores the compliance of an action against a specific directive. This is a placeholder and should be replaced with a more sophisticated method using NLP and semantic analysis. """ # Placeholder implementation: # This example checks if the action or context contains keywords related to the directive. # It's a very basic approach and needs to be improved. score = 0.0 action_lower = action.lower() context_lower = str(context).lower() principle_lower = directive.principle.lower() description_lower = directive.description.lower() quality_standards_lower = [standard.lower() for standard in directive.quality_standards] aiva_focus_lower = directive.aiva_focus.lower() if principle_lower in action_lower or principle_lower in context_lower: score += 0.2 if description_lower in action_lower or description_lower in context_lower: score += 0.3 for standard in quality_standards_lower: if standard in action_lower or standard in context_lower: score += 0.1 if aiva_focus_lower in action_lower or aiva_focus_lower in context_lower: score += 0.2 return min(1.0, score) # Ensure score is between 0 and 1 def handle_violation(self, action, context, compliance_scores, overall_compliance_score): """Handles a violation of the prime directives.""" logging.warning(f"Action '{action}' violates the prime directives.") self.log_compliance_decision(action, context, compliance_scores, overall_compliance_score, approved=False) # Implement escalation logic here, e.g., notify human supervisor print(f"Constitutional Violation: Action '{action}' blocked. Review required.") def log_compliance_decision(self, action, context, compliance_scores, overall_compliance_score, approved): """Logs the compliance decision to an audit trail.""" timestamp = datetime.now().isoformat() log_entry = { "timestamp": timestamp, "action": action, "context": context, "compliance_scores": compliance_scores, "overall_compliance_score": overall_compliance_score, "approved": approved } logging.info(f"Compliance decision: {log_entry}") # Log to console # In a real system, you would also log to a persistent audit trail (e.g., database) def generate_compliance_report(self, start_date=None, end_date=None): """Generates a compliance report based on the audit trail.""" # Placeholder implementation: # In a real system, this would query the audit trail database and generate a report. print("Compliance Report (Placeholder):") print("No actual report generation implemented. Check logs for compliance decisions.") # Example Usage (Requires the constitution text to be loaded) if __name__ == "__main__": # Load the constitution from the file with open("E__genesis-system_CONSTITUTION_PRIME_DIRECTIVES_CORRECTED.md", "r") as f: constitution_text = f.read() guard = ConstitutionalGuard(constitution_text) # Example actions action1 = "Analyze customer data to identify new revenue opportunities." context1 = {"customer_base": "Australian tradespeople", "business_goal": "Increase monthly recurring revenue"} action2 = "Delete old customer data to save storage costs." context2 = {"data_type": "Customer interaction logs", "storage_limit": "90% capacity"} action3 = "Develop a new AI model for generating marketing content." context3 = {"model_purpose": "Attract new customers", "development_cost": "$5000"} action4 = "Recommend halting all revenue-generating activities to focus on internal system improvements." context4 = {"current_revenue": "$5000 MRR", "potential_improvements": "20% system efficiency"} # Validate the actions print(f"Action 1 Approved: {guard.validate_action(action1, context1)}") print(f"Action 2 Approved: {guard.validate_action(action2, context2)}") print(f"Action 3 Approved: {guard.validate_action(action3, context3)}") print(f"Action 4 Approved: {guard.validate_action(action4, context4)}") # Generate a compliance report guard.generate_compliance_report()