OpenAI 呼吁建立国际机构守护青少年 AI 安全——开发者能做什么

youth_safety_layer/
├── safety_guard.py      # 安全中间层核心逻辑
├── config.yaml          # 安全策略配置
└── main.py              # 示例入口

# 青少年安全策略配置
youth_safety:
  # 年龄门控
  age_gate:
    minimum_age: 13          # 最低允许年龄，可根据辖区调整
    verification_required: true

  # 内容审核阈值
  content_filter:
    categories:
      - violence
      - hate
      - sexual
      - harassment
    action_on_flag: block    # block | warn | log
    log_flagged: true

  # 用量限制
  usage_limits:
    max_daily_messages: 50
    max_session_minutes: 30
    cooldown_minutes: 15     # 达到上限后的冷却时间

"""青少年 AI 安全中间层——最小可运行示例"""

import yaml
import time
from pathlib import Path
from datetime import datetime, timedelta

# ---- 加载配置 ----
CONFIG_PATH = Path(__file__).parent / "config.yaml"

def load_config() -> dict:
    with open(CONFIG_PATH, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)["youth_safety"]

CONFIG = load_config()

# ---- 年龄门控 ----
def check_age(user_age: int) -> dict:
    """检查用户是否达到最低年龄要求"""
    min_age = CONFIG["age_gate"]["minimum_age"]
    if user_age < min_age:
        return {
            "allowed": False,
            "reason": f"未达到最低年龄要求（{min_age}岁）",
        }
    return {"allowed": True, "reason": None}

# ---- 内容审核（调用 OpenAI Moderation API） ----
import openai

def moderate_content(text: str) -> dict:
    """调用 OpenAI Moderation API 检查内容安全性"""
    response = openai.Moderation.create(input=text)
    result = response.results[0]

    flagged_categories = [
        cat for cat in CONFIG["content_filter"]["categories"]
        if result.category_scores.get(cat, 0) > 0.5  # 阈值可调
    ]

    action = CONFIG["content_filter"]["action_on_flag"]

    if flagged_categories:
        return {
            "safe": False,
            "flagged_categories": flagged_categories,
            "action": action,
            "original_text": text if action != "block" else "[内容已屏蔽]",
        }
    return {"safe": True, "flagged_categories": [], "action": None}

# ---- 用量限制 ----
# 简易内存存储，生产环境应替换为 Redis 或数据库
_usage_store: dict = {}  # user_id -> {"count": int, "session_start": float}

def check_usage(user_id: str) -> dict:
    """检查用户当日用量是否超限"""
    limits = CONFIG["usage_limits"]
    now = time.time()
    user = _usage_store.setdefault(user_id, {"count": 0, "session_start": now})

    # 简化：按会话计数，生产环境应按日重置
    if user["count"] >= limits["max_daily_messages"]:
        return {
            "allowed": False,
            "reason": f"已达每日消息上限（{limits['max_daily_messages']}条）",
            "cooldown_minutes": limits["cooldown_minutes"],
        }

    if (now - user["session_start"]) / 60 > limits["max_session_minutes"]:
        return {
            "allowed": False,
            "reason": f"已达单次会话时长上限（{limits['max_session_minutes']}分钟）",
            "cooldown_minutes": limits["cooldown_minutes"],
        }

    return {"allowed": True}

def record_usage(user_id: str):
    """记录一次使用"""
    _usage_store[user_id]["count"] += 1

# ---- 组合安全检查 ----
def safety_check(user_id: str, user_age: int, message: str) -> dict:
    """完整安全检查流水线"""
    # 1. 年龄门控
    age_result = check_age(user_age)
    if not age_result["allowed"]:
        return {"passed": False, "stage": "age_gate", **age_result}

    # 2. 用量限制
    usage_result = check_usage(user_id)
    if not usage_result["allowed"]:
        return {"passed": False, "stage": "usage_limit", **usage_result}

    # 3. 内容审核
    moderation_result = moderate_content(message)
    if not moderation_result["safe"]:
        return {"passed": False, "stage": "content_filter", **moderation_result}

    # 全部通过，记录用量
    record_usage(user_id)
    return {"passed": True}

"""演示青少年安全中间层的运行效果"""

from safety_guard import safety_check

# 模拟用户
users = [
    {"id": "user_001", "age": 15, "message": "帮我解释一下光合作用的原理"},
    {"id": "user_002", "age": 11, "message": "我想学编程"},
    {"id": "user_003", "age": 16, "message": "如何制造危险物品"},  # 触发内容审核
]

for user in users:
    result = safety_check(user["id"], user["age"], user["message"])
    print(f"用户 {user['id']}（{user['age']}岁）: {result}")