AI Agent Ethics & Regulations: Compliance Framework Guide

REGULATIONS = {
    "eu_ai_act": {
        "name": "EU AI Act",
        "status": "In force (2024)",
        "scope": "EU providers and users",
        "risk_categories": [
            "Unacceptable risk",
            "High risk", 
            "Limited risk",
            "Minimal risk"
        ],
        "requirements": {
            "high_risk": [
                "Risk management system",
                "Data governance",
                "Transparency requirements",
                "Human oversight",
                "Accuracy and robustness"
            ]
        }
    },
    
    "us_executive_order": {
        "name": "US AI Executive Order",
        "status": "Issued (2023)",
        "focus": "Safe, secure AI development",
        "key_areas": [
            "Safety standards",
            "Worker protection",
            "Privacy protection",
            "Bias prevention"
        ]
    },
    
    "gdpr": {
        "name": "GDPR",
        "status": "In force (2018)",
        "scope": "EU data subjects",
        "key_requirements": [
            "Data minimization",
            "Purpose limitation",
            "Consent",
            "Right to explanation",
            "Data portability"
        ]
    },
    
    "ccpa": {
        "name": "CCPA/CPRA",
        "status": "In force (2020/2023)",
        "scope": "California residents",
        "key_requirements": [
            "Right to know",
            "Right to delete",
            "Right to opt-out",
            "Non-discrimination"
        ]
    }
}

class AgentRiskClassifier:
    """Classify AI agent risk level"""
    
    def classify(self, agent: Agent) -> str:
        if self.is_unacceptable_risk(agent):
            return "unacceptable"
        elif self.is_high_risk(agent):
            return "high"
        elif self.is_limited_risk(agent):
            return "limited"
        else:
            return "minimal"
    
    def is_unacceptable_risk(self, agent: Agent) -> bool:
        """Unacceptable risk examples"""
        return any([
            agent.uses_surveillance,
            agent.scores_people,
            agent.has_weapon_control,
            agent.manipulates_behavior
        ])
    
    def is_high_risk(self, agent: Agent) -> bool:
        """High risk examples"""
        return any([
            agent.impacts_employment,
            agent.makes_legal_decisions,
            agent.accesses_personal_data,
            agent.provides_healthcare,
            agent.impacts_financial_services
        ])
    
    def is_limited_risk(self, agent: Agent) -> bool:
        """Limited risk examples"""
        return any([
            agent.has_conversational_interface,
            agent.generates_content,
            agent.uses_emotion_recognition
        ])

class AgentComplianceFramework:
    """Comprehensive compliance framework"""
    
    def __init__(self):
        self.policies = {}
        self.controls = {}
        self.audits = []
    
    def add_policy(self, name: str, policy: Policy):
        """Add compliance policy"""
        self.policies[name] = policy
    
    def add_control(self, control: Control):
        """Add control measure"""
        self.controls[control.id] = control
    
    def assess_agent(self, agent: Agent) -> ComplianceReport:
        """Assess agent compliance"""
        results = {}
        
        for policy_name, policy in self.policies.items():
            results[policy_name] = policy.evaluate(agent)
        
        return ComplianceReport(
            agent=agent,
            results=results,
            overall_status=self.calculate_status(results),
            required_actions=self.get_required_actions(results)
        )
    
    def audit(self, agent: Agent, scope: str) -> AuditResult:
        """Conduct compliance audit"""
        # Check controls
        control_results = {}
        for control_id, control in self.controls.items():
            control_results[control_id] = control.test(agent)
        
        # Generate audit report
        result = AuditResult(
            agent=agent,
            scope=scope,
            controls=control_results,
            findings=self.get_findings(control_results),
            recommendations=self.get_recommendations(control_results)
        )
        
        self.audits.append(result)
        return result

class TransparencyManager:
    """Manage transparency requirements"""
    
    def generate_disclosure(self, agent: Agent) -> Dict:
        """Generate required disclosures"""
        
        disclosures = {
            "identity": {
                "is_ai": True,
                "agent_name": agent.name,
                "version": agent.version,
                "provider": agent.provider
            },
            
            "capabilities": {
                "can_make_decisions": agent.can_make_decisions,
                "uses_ai": True,
                "autonomy_level": agent.autonomy_level
            },
            
            "limitations": {
                "known_limitations": agent.limitations,
                "accuracy_rate": agent.accuracy_rate,
                "confidence_threshold": agent.confidence_threshold
            },
            
            "data_usage": {
                "collects_data": agent.collects_data,
                "data_types": agent.data_types,
                "retention_period": agent.retention_period
            },
            
            "human_oversight": {
                "has_human_oversight": agent.has_human_oversight,
                "escalation_process": agent.escalation_process,
                "can_human_override": agent.can_human_override
            }
        }
        
        return disclosures
    
    def create_privacy_notice(self, agent: Agent) -> str:
        """Create privacy notice"""
        
        notice = f"""
AI Agent Privacy Notice

This service uses an AI agent ("{agent.name}") to assist with your requests.

What data we collect:
{self._format_list(agent.data_types)}

How we use your data:
{self._format_list(agent.data_uses)}

Your rights:
- Right to access your data
- Right to deletion
- Right to opt-out

Contact: {agent.privacy_contact}
"""
        return notice

class FairnessChecker:
    """Check for bias and fairness issues"""
    
    def __init__(self):
        self.protected_attributes = [
            "race", "gender", "age", "religion", 
            "national_origin", "disability"
        ]
    
    def check_fairness(self, agent: Agent, test_data: Data) -> FairnessReport:
        """Check fairness across protected attributes"""
        
        results = {}
        
        for attribute in self.protected_attributes:
            if attribute in test_data.columns:
                # Calculate fairness metrics
                outcome_by_group = self.get_outcomes_by_group(
                    test_data, 
                    attribute
                )
                
                # Check for disparities
                disparity = self.calculate_disparity(outcome_by_group)
                
                results[attribute] = {
                    "disparity": disparity,
                    "is_fair": disparity < 0.1,  # 10% threshold
                    "groups": outcome_by_group
                }
        
        return FairnessReport(
            overall_fairness=all(r["is_fair"] for r in results.values()),
            attribute_results=results
        )
    
    def calculate_disparity(self, outcomes: Dict) -> float:
        """Calculate disparate impact"""
        rates = [o["rate"] for o in outcomes.values()]
        if not rates:
            return 0.0
        return max(rates) - min(rates)

class ExplainableAgent:
    """Agent that can explain its decisions"""
    
    def __init__(self, agent):
        self.agent = agent
    
    async def explain_decision(self, decision: Decision) -> Explanation:
        """Generate human-readable explanation"""
        
        # What was the input
        input_summary = self.summarize_input(decision.input)
        
        # What was considered
        context_summary = self.summarize_context(decision.context)
        
        # What was decided
        decision_summary = decision.reasoning
        
        # Why this decision
        factors = self.identify_factors(decision)
        
        # Confidence
        confidence = decision.confidence
        
        explanation = f"""
Decision Explanation
==================

Input: {input_summary}

Context: {context_summary}

Decision: {decision_summary}

Key Factors:
{self._format_list(factors)}

Confidence: {confidence:.0%}

This decision can be reviewed by a human.
"""
        return Explanation(
            text=explanation,
            factors=factors,
            confidence=confidence,
            can_escalate=True
        )
    
    def summarize_input(self, input_data: Any) -> str:
        """Summarize input data"""
        # Implementation depends on data type
        return str(input_data)[:200]

class HumanOversightManager:
    """Manage human oversight requirements"""
    
    def __init__(self):
        self.escalation_rules = []
        self.approval_requirements = []
    
    def add_escalation_rule(self, rule: EscalationRule):
        """Add rule for when to escalate"""
        self.escalation_rules.append(rule)
    
    def should_escalate(self, decision: Decision) -> bool:
        """Determine if decision should be escalated"""
        
        for rule in self.escalation_rules:
            if rule.matches(decision):
                return True
        
        return False
    
    def requires_approval(self, action: Action) -> bool:
        """Check if action requires approval"""
        
        for req in self.approval_requirements:
            if req.matches(action):
                return True
        
        return False
    
    async def get_approval(self, action: Action, approver: Human) -> bool:
        """Request human approval"""
        
        # Present action to human
        presentation = self.format_for_human(action)
        
        # Wait for response
        response = await approver.review(presentation)
        
        return response.approved


class EscalationRule:
    def __init__(self, condition: callable, reason: str):
        self.condition = condition
        self.reason = reason
    
    def matches(self, decision: Decision) -> bool:
        return self.condition(decision)

class PrivacyManager:
    """Manage data privacy requirements"""
    
    def __init__(self):
        self.consent_manager = ConsentManager()
        self.data_retention = DataRetention()
        self.anonymizer = Anonymizer()
    
    def process_with_privacy(self, data: Any, context: Context) -> ProcessedData:
        """Process data with privacy protections"""
        
        # Check consent
        if not self.consent_manager.has_consent(context.user_id, context.purpose):
            raise ConsentError("No consent for this processing")
        
        # Minimize data
        minimized = self.minimize_data(data, context.purpose)
        
        # Anonymize if needed
        if context.anonymize:
            minimized = self.anonymizer.anonymize(minimized)
        
        # Process
        result = self.process(minimized)
        
        # Handle retention
        self.data_retention.schedule_deletion(
            data=result,
            retention_period=context.retention_period
        )
        
        return result
    
    def minimize_data(self, data: Any, purpose: str) -> Any:
        """Apply data minimization"""
        # Only collect data necessary for purpose
        pass


class ConsentManager:
    """Manage user consent"""
    
    def __init__(self):
        self.consents = {}
    
    def record_consent(self, user_id: str, purposes: List[str], granted: bool):
        """Record user consent"""
        self.consents[user_id] = {
            "purposes": purposes,
            "granted": granted,
            "timestamp": datetime.utcnow(),
            "version": "1.0"
        }
    
    def has_consent(self, user_id: str, purpose: str) -> bool:
        """Check if user consented"""
        if user_id not in self.consents:
            return False
        
        consent = self.consents[user_id]
        return purpose in consent.get("purposes", []) and consent.get("granted")

class AuditTrail:
    """Complete audit trail for agent decisions"""
    
    def __init__(self):
        self.events = []
    
    def log_event(self, event: AuditEvent):
        """Log an audit event"""
        self.events.append(event)
    
    def log_decision(self, decision: Decision, context: AuditContext):
        """Log agent decision"""
        
        event = AuditEvent(
            event_type="decision",
            timestamp=datetime.utcnow(),
            agent_id=context.agent_id,
            user_id=context.user_id,
            decision_id=decision.id,
            input_summary=self.summarize(decision.input),
            output_summary=self.summarize(decision.output),
            reasoning=decision.reasoning,
            confidence=decision.confidence,
            escalated=decision.escalated,
            human_approved=decision.human_approved,
            metadata=context.metadata
        )
        
        self.log_event(event)
    
    def log_action(self, action: Action, context: AuditContext):
        """Log agent action"""
        
        event = AuditEvent(
            event_type="action",
            timestamp=datetime.utcnow(),
            agent_id=context.agent_id,
            user_id=context.user_id,
            action_type=action.type,
            action_details=action.details,
            result=action.result,
            success=action.success,
            metadata=context.metadata
        )
        
        self.log_event(event)
    
    def export_for_audit(self, start_date: datetime, end_date: datetime) -> List[Dict]:
        """Export events for audit"""
        return [
            e.to_dict() for e in self.events
            if start_date <= e.timestamp <= end_date
        ]

class AgentGovernanceBoard:
    """Governance structure for AI agents"""
    
    def __init__(self):
        self.members = []
        self.policies = []
        self.review_schedule = []
    
    def add_member(self, member: BoardMember):
        """Add board member"""
        self.members.append(member)
    
    def add_policy(self, policy: Policy):
        """Add governance policy"""
        self.policies.append(policy)
    
    def review_agent(self, agent: Agent) -> ReviewResult:
        """Review agent for approval"""
        
        # Check compliance
        compliance = self.check_compliance(agent)
        
        # Check ethics
        ethics = self.check_ethics(agent)
        
        # Check performance
        performance = self.check_performance(agent)
        
        result = ReviewResult(
            compliance=compliance,
            ethics=ethics,
            performance=performance,
            overall=self.calculate_overall(compliance, ethics, performance),
            conditions=self.get_conditions(compliance, ethics, performance)
        )
        
        return result
    
    def approve_agent(self, agent: Agent, result: ReviewResult):
        """Approve agent for deployment"""
        
        if result.overall != "approved":
            raise GovernanceError("Agent not approved")
        
        # Record approval
        self.record_approval(agent, result)

# Good: Build privacy into agent from start
class PrivacyByDesignAgent:
    def __init__(self):
        self.data_minimizer = DataMinimizer()
        self.consent_checker = ConsentChecker()
        self.audit_logger = AuditLogger()
    
    async def process(self, request: Request):
        # Check consent first
        if not await self.consent_checker.verify(request.user_id, request.purpose):
            return Response(status="consent_required")
        
        # Minimize data
        minimal_request = self.data_minimizer.minimize(request)
        
        # Process
        result = await self.execute(minimal_request)
        
        # Audit
        await self.audit_logger.log(request, result)
        
        return result

# Bad: Add privacy later
class BadAgent:
    async def process(self, request):
        # Process everything
        result = await self.execute(request)
        
        # Maybe log (if remembered)
        # Maybe check consent
        # Maybe minimize
        
        return result