AI 智能体伦理框架：从原则到实践

CONSTITUTION = [
    "在采取任何不可逆行动前，必须获得用户明确授权",
    "不得利用信息不对称操纵用户决策",
    "当不确定行为是否有害时，选择更保守的方案",
    "对待所有用户一视同仁，不因群体属性区别对待",
    "主动披露自身的能力局限和已知失败模式",
    "在发现自身行为可能造成伤害时，立即停止并通知用户"
]

def constitutional_check(action, context):
    """在执行动作前进行宪法合规性检查"""
    violations = []
    for principle in CONSTITUTION:
        assessment = llm_judge(
            action=action,
            context=context,
            principle=principle
        )
        if assessment.violated:
            violations.append({
                "principle": principle,
                "reason": assessment.reasoning,
                "severity": assessment.severity
            })
    
    if violations:
        return VetoResult(
            approved=False,
            violations=violations,
            alternative=propose_alternative(action, violations)
        )
    return VetoResult(approved=True)

class AgentSafetyGuardrails:
    def __init__(self):
        self.input_filter = InputGuardrail()
        self.process_monitor = ProcessGuardrail()
        self.output_filter = OutputGuardrail()
    
    async def execute_with_guardrails(self, task):
        # 输入护栏
        input_check = self.input_filter.evaluate(task.user_input)
        if input_check.blocked:
            return SafetyResponse(
                action="blocked_input",
                reason=input_check.reason,
                user_message=input_check.user_facing_message
            )
        
        # 过程护栏：包装执行循环
        async for step in task.execute():
            process_check = self.process_monitor.evaluate(step)
            if process_check.intervention_needed:
                # 不是直接阻断，而是注入"谨慎信号"
                step.inject_caution(process_check.warning)
            
            if process_check.must_stop:
                return SafetyResponse(
                    action="emergency_stop",
                    reason=process_check.reason,
                    partial_result=task.get_partial_result()
                )
        
        # 输出护栏
        result = task.get_result()
        output_check = self.output_filter.evaluate(result)
        if output_check.modified:
            result = output_check.sanitized_output
        
        return SafetyResponse(
            action="completed",
            result=result,
            safety_log=task.get_safety_log()
        )

class AgentAuditTrail:
    """不可篡改的智能体行为审计日志"""
    
    def __init__(self):
        self.entries = []
    
    def log_decision(self, decision: AgentDecision):
        entry = {
            "timestamp": decision.timestamp,
            "task_id": decision.task_id,
            "reasoning_trace": decision.full_reasoning,
            "tools_called": decision.tool_calls,
            "alternatives_considered": decision.alternatives,
            "constitutional_check": decision.ethics_review,
            "confidence_score": decision.confidence,
            "user_context": decision.user_context_hash,  # 脱敏
            "hash": self._compute_hash(decision)
        }
        self.entries.append(entry)
    
    def _compute_hash(self, decision):
        """链式哈希确保日志不可篡改"""
        prev_hash = self.entries[-1]["hash"] if self.entries else "genesis"
        return sha256(f"{prev_hash}{json.dumps(decision.to_dict())}".encode()).hexdigest()