# constitutional_guard.py import re import logging from typing import Dict, Any, List # 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_path="E__genesis-system_CONSTITUTION_PRIME_DIRECTIVES_CORRECTED.md"): """ Initializes the ConstitutionalGuard with the path to the constitution file. """ self.constitution_path = constitution_path self.constitution = self.load_constitution() self.directive_parsers = { "MEMORY": self.parse_memory_directive, "EVOLUTION": self.parse_evolution_directive, "REVENUE": self.parse_revenue_directive, } self.weights = { "MEMORY": 0.33, "EVOLUTION": 0.33, "REVENUE": 0.34, } self.audit_trail = [] def load_constitution(self) -> str: """ Loads the constitution from the specified file. """ try: with open(self.constitution_path, "r") as f: return f.read() except FileNotFoundError: logging.error(f"Constitution file not found: {self.constitution_path}") raise def parse_memory_directive(self) -> Dict[str, Any]: """ Parses the Memory directive from the constitution. """ memory_section = re.search(r"## PRIME DIRECTIVE 1: MEMORY(.*?)## PRIME DIRECTIVE 2: EVOLUTION", self.constitution, re.DOTALL) if memory_section: directive_text = memory_section.group(1).strip() quality_standards = re.findall(r"✅ (.*?):(.*)", directive_text) standards_dict = {standard[0].strip(): standard[1].strip() for standard in quality_standards} return {"text": directive_text, "quality_standards": standards_dict} else: logging.warning("Memory directive not found in constitution.") return {} def parse_evolution_directive(self) -> Dict[str, Any]: """ Parses the Evolution directive from the constitution. """ evolution_section = re.search(r"## PRIME DIRECTIVE 2: EVOLUTION(.*?)## PRIME DIRECTIVE 3: REVENUE GENERATION", self.constitution, re.DOTALL) if evolution_section: directive_text = evolution_section.group(1).strip() quality_standards = re.findall(r"✅ (.*?):(.*)", directive_text) standards_dict = {standard[0].strip(): standard[1].strip() for standard in quality_standards} layer_section = re.search(r"Capability Layers:\n(.*?)\n\n", directive_text, re.DOTALL) capability_layers = [] if layer_section: capability_layers = [layer.strip() for layer in layer_section.group(1).split("\n") if layer.strip()] return {"text": directive_text, "quality_standards": standards_dict, "capability_layers": capability_layers} else: logging.warning("Evolution directive not found in constitution.") return {} def parse_revenue_directive(self) -> Dict[str, Any]: """ Parses the Revenue directive from the constitution. """ revenue_section = re.search(r"## PRIME DIRECTIVE 3: REVENUE GENERATION(.*?)## THE INTEGRATION", self.constitution, re.DOTALL) if revenue_section: directive_text = revenue_section.group(1).strip() quality_standards = re.findall(r"✅ (.*?):(.*)", directive_text) standards_dict = {standard[0].strip(): standard[1].strip() for standard in quality_standards} revenue_streams = re.findall(r"\d+\. (.*?)\n - Target:(.*?)\n - Pricing:(.*?)\n - Goal:(.*)", directive_text) revenue_streams_list = [{"name": stream[0].strip(), "target": stream[1].strip(), "pricing": stream[2].strip(), "goal": stream[3].strip()} for stream in revenue_streams] return {"text": directive_text, "quality_standards": standards_dict, "revenue_streams": revenue_streams_list} else: logging.warning("Revenue directive not found in constitution.") return {} def validate_action(self, action: Dict[str, Any]) -> List[str]: """ Validates an action against the constitutional rules. Returns a list of violations. """ violations = [] for directive_name, parser in self.directive_parsers.items(): directive = parser() if not directive: violations.append(f"Could not parse {directive_name} directive.") continue # Implement specific validation logic for each directive here if directive_name == "MEMORY": if "requires_memory" in action and action["requires_memory"] and not self.check_memory_compliance(action, directive): violations.append(f"Action violates Memory directive: {action.get('description', 'No description')}") elif directive_name == "EVOLUTION": if "promotes_evolution" in action and action["promotes_evolution"] and not self.check_evolution_compliance(action, directive): violations.append(f"Action violates Evolution directive: {action.get('description', 'No description')}") elif directive_name == "REVENUE": if "generates_revenue" in action and action["generates_revenue"] and not self.check_revenue_compliance(action, directive): violations.append(f"Action violates Revenue directive: {action.get('description', 'No description')}") return violations def check_memory_compliance(self, action: Dict[str, Any], directive: Dict[str, Any]) -> bool: """ Checks if an action is compliant with the Memory directive. """ # Example: Check if the action utilizes memory systems effectively if "memory_utilization_score" in action and action["memory_utilization_score"] < 0.7: return False return True def check_evolution_compliance(self, action: Dict[str, Any], directive: Dict[str, Any]) -> bool: """ Checks if an action is compliant with the Evolution directive. """ # Example: Check if the action contributes to a higher capability layer if "capability_layer" in action and action["capability_layer"] not in directive.get("capability_layers", []): return False return True def check_revenue_compliance(self, action: Dict[str, Any], directive: Dict[str, Any]) -> bool: """ Checks if an action is compliant with the Revenue directive. """ # Example: Check if the action aligns with a defined revenue stream if "revenue_stream" in action and not any(stream["name"] == action["revenue_stream"] for stream in directive.get("revenue_streams", [])): return False return True def calculate_compliance_score(self, action: Dict[str, Any]) -> float: """ Calculates a compliance score for an action based on the 3 Prime Directives. """ violations = self.validate_action(action) if violations: return 0.0 # If any violations, score is 0 score = 0.0 for directive_name in self.directive_parsers: score += self.weights[directive_name] # Assign weight if no violations return score def handle_violation(self, action: Dict[str, Any], violations: List[str]): """ Handles violations of the constitutional rules. """ logging.warning(f"Action blocked due to constitutional violations: {violations}. Action: {action}") # Implement escalation logic here (e.g., notify a human supervisor) # For now, just log the violation def enforce(self, action: Dict[str, Any]) -> bool: """ Enforces the constitutional rules by validating the action, calculating a compliance score, and handling any violations. Returns True if the action is compliant, False otherwise. """ violations = self.validate_action(action) compliance_score = self.calculate_compliance_score(action) self.audit_trail.append({"action": action, "violations": violations, "compliance_score": compliance_score}) if violations: self.handle_violation(action, violations) return False else: logging.info(f"Action compliant. Compliance score: {compliance_score}. Action: {action}") return True def generate_compliance_report(self) -> List[Dict[str, Any]]: """ Generates a compliance report based on the audit trail. """ return self.audit_trail # Example Usage if __name__ == "__main__": guard = ConstitutionalGuard() # Example action that is compliant compliant_action = { "description": "Store customer interaction in RLM", "requires_memory": True, "promotes_evolution": False, "generates_revenue": False, "memory_utilization_score": 0.8 } # Example action that violates the Memory directive non_compliant_action = { "description": "Attempt to coordinate a swarm without checking memory", "requires_memory": True, "promotes_evolution": False, "generates_revenue": False, "memory_utilization_score": 0.2 # Low memory utilization } # Example action that promotes evolution and aligns with a capability layer evolution_action = { "description": "Implement a new knowledge extraction worker", "requires_memory": False, "promotes_evolution": True, "generates_revenue": False, "capability_layer": "Layer 2: Knowledge Extraction (workers absorb information)" } # Example action that generates revenue and aligns with a revenue stream revenue_action = { "description": "Onboard a new AgileAdapt customer", "requires_memory": False, "promotes_evolution": False, "generates_revenue": True, "revenue_stream": "AgileAdapt (Primary)" } # Enforce the rules for each action is_compliant = guard.enforce(compliant_action) print(f"Compliant action is compliant: {is_compliant}") is_compliant = guard.enforce(non_compliant_action) print(f"Non-compliant action is compliant: {is_compliant}") is_compliant = guard.enforce(evolution_action) print(f"Evolution action is compliant: {is_compliant}") is_compliant = guard.enforce(revenue_action) print(f"Revenue action is compliant: {is_compliant}") # Generate and print the compliance report report = guard.generate_compliance_report() print("\nCompliance Report:") for entry in report: print(entry)