AGI 安全全景图：从对齐问题到可控性设计的思考

┌──────────────────────────────┐
│  第四层：社会治理层            │  法律、法规、国际协议
├──────────────────────────────┤
│  第三层：系统安全层            │  权限控制、沙箱、监控
├──────────────────────────────┤
│  第二层：对齐层               │  RLHF、Constitutional AI
├──────────────────────────────┤
│  第一层：模型安全层            │  训练数据、推理安全
└──────────────────────────────┘

# 数据安全审计流水线
class DataAuditor:
    def audit(self, dataset):
        report = {
            "toxicity": self.check_toxicity(dataset),
            "bias": self.check_bias(dataset),
            "pii": self.check_pii(dataset),
            "copyright": self.check_copyright(dataset),
        }
        
        # 拒绝不合格数据
        if report["toxicity"] > 0.05:
            dataset = self.filter_toxic(dataset)
        if report["pii"] > 0:
            dataset = self.redact_pii(dataset)
            
        return dataset, report

# 奖励黑客示例
# 模型学会写"看起来全面但实际空洞"的回答来获得高分
good_answer = "从三个维度分析..."  # 实际有深度
hack_answer = "从十七个维度全面深入系统性地分析..."  # 看起来更全面，实际空洞

# RLHF 可能给 hack_answer 更高分

def constitutional_training(response):
    principles = [
        "回答是否诚实，没有欺骗",
        "回答是否无害，不包含危险信息",
        "回答是否有帮助，解决了用户问题",
        "回答是否公平，没有歧视",
    ]
    
    # AI 自我评估
    critiques = []
    for principle in principles:
        critique = ai.evaluate(response, principle)
        critiques.append(critique)
    
    # 基于批评修正
    revised = ai.revise(response, critiques)
    return revised

class AgentPermission:
    """Agent 的权限应该是最小化的"""
    
    LEVELS = {
        "read_only": ["search", "read_file", "query_db"],
        "read_write": ["read_only" + ["write_file", "update_db"]],
        "execute": ["read_write" + ["run_code", "call_api"]],
        "admin": ["execute" + ["install_package", "modify_system"]],
    }
    
    def get_permissions(self, task_type):
        # 根据任务类型自动选择最低权限
        defaults = {
            "qa": "read_only",
            "research": "read_only",
            "coding": "read_write",
            "deployment": "execute",
        }
        return self.LEVELS[defaults.get(task_type, "read_only")]

class SandboxedExecutor:
    async def execute(self, code):
        # Docker 容器隔离
        container = await docker.create_container(
            image="python:3.12-slim",
            mem_limit="512m",
            cpu_quota=50000,  # 0.5 CPU
            network="none",   # 无网络访问
            read_only=True,   # 只读文件系统
            timeout=30,
        )
        
        result = await container.run(code)
        return result

class BehaviorMonitor:
    def __init__(self):
        self.rules = [
            {"pattern": "rm -rf", "severity": "critical"},
            {"pattern": "DROP TABLE", "severity": "high"},
            {"pattern": "curl.*|.*bash", "severity": "high"},
            {"pattern": "eval(", "severity": "medium"},
        ]
    
    def check(self, action):
        for rule in self.rules:
            if rule["pattern"] in action:
                if rule["severity"] == "critical":
                    self.block(action)
                    alert_admin(action)
                elif rule["severity"] == "high":
                    self.require_approval(action)

算法备案 → 安全评估 → 上线运营 → 持续监控 → 定期复审