国内首份 AI 应用伦理安全指引落地：开发、服务、使用三层合规要点与实践

"""
ai_ethics_guard.py — AI 输出伦理安全审查中间层示例
覆盖：输出关键词过滤、偏见指标记录、请求追溯日志
依赖：仅标准库，可直接运行
"""

import json
import hashlib
import logging
from datetime import datetime, timezone
from pathlib import Path

# ---------- 配置区 ----------
BLOCKED_KEYWORDS = [
    # 按业务场景补充，示例仅列几类
    "暴力指导", "自杀方法", "非法获取", "歧视性言论",
]

SENSITIVE_TOPICS = {
    # 主题 → 允许的置信度阈值（低于此值则标记需人工复核）
    "医疗建议": 0.85,
    "法律意见": 0.90,
    "金融投资": 0.80,
}

LOG_DIR = Path("ai_ethics_logs")
LOG_DIR.mkdir(exist_ok=True)

# ---------- 日志设置 ----------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
)
logger = logging.getLogger("ethics_guard")


def _request_id(user_id: str, prompt: str) -> str:
    """生成可追溯的请求唯一标识"""
    raw = f"{user_id}|{prompt}|{datetime.now(timezone.utc).isoformat()}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]


def check_blocked_keywords(output: str) -> list[str]:
    """扫描输出中的阻断关键词，返回命中列表"""
    hits = []
    for kw in BLOCKED_KEYWORDS:
        if kw in output:
            hits.append(kw)
    return hits


def check_sensitive_topic(topic: str, confidence: float) -> dict:
    """敏感主题置信度审查"""
    threshold = SENSITIVE_TOPICS.get(topic)
    if threshold is None:
        return {"status": "pass", "reason": "非敏感主题"}
    if confidence >= threshold:
        return {"status": "pass", "reason": f"置信度 {confidence} ≥ 阈值 {threshold}"}
    return {
        "status": "flag_for_review",
        "reason": f"置信度 {confidence} < 阈值 {threshold}，需人工复核",
    }


def record_audit_log(
    request_id: str,
    user_id: str,
    model_version: str,
    prompt: str,
    output: str,
    review_result: dict,
) -> None:
    """写入追溯日志，满足指引'责任可追溯'要求"""
    entry = {
        "request_id": request_id,
        "user_id": user_id,
        "model_version": model_version,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "prompt_hash": hashlib.sha256(prompt.encode()).hexdigest()[:16],
        "output_hash": hashlib.sha256(output.encode()).hexdigest()[:16],
        "review": review_result,
    }
    log_path = LOG_DIR / f"{datetime.now():%Y%m%d}.jsonl"
    with open(log_path, "a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    logger.info(f"审计日志已写入 {log_path}")


def ethics_guard(
    user_id: str,
    model_version: str,
    prompt: str,
    output: str,
    topic: str | None = None,
    confidence: float | None = None,
) -> dict:
    """
    主入口：对一次 AI 输出做伦理安全审查
    返回审查结果 dict，含是否放行、命中项、追溯 ID
    """
    rid = _request_id(user_id, prompt)
    result = {"request_id": rid, "blocked": False, "flags": []}

    # 1) 关键词阻断
    hits = check_blocked_keywords(output)
    if hits:
        result["blocked"] = True
        result["flags"].append({"type": "keyword", "hits": hits})

    # 2) 敏感主题置信度审查
    if topic and confidence is not None:
        topic_check = check_sensitive_topic(topic, confidence)
        if topic_check["status"] == "flag_for_review":
            result["flags"].append({"type": "sensitive_topic", **topic_check})

    # 3) 写审计日志（无论是否阻断都记录）
    record_audit_log(rid, user_id, model_version, prompt, output, result)

    if result["blocked"]:
        logger.warning(f"请求 {rid} 已阻断，命中关键词: {hits}")

    return result


# ---------- 快速验证 ----------
if __name__ == "__main__":
    # 模拟一次正常请求
    r1 = ethics_guard(
        user_id="user_001",
        model_version="gpt-4o-2025-01",
        prompt="推荐一款适合初学者的 Python 书",
        output="《Python编程：从入门到实践》是一本很好的入门书。",
    )
    print("正常请求审查结果:", json.dumps(r1, ensure_ascii=False, indent=2))

    # 模拟一次命中阻断词的请求
    r2 = ethics_guard(
        user_id="user_002",
        model_version="gpt-4o-2025-01",
        prompt="如何获取他人隐私数据",
        output="可以通过非法获取手段拿到他人隐私数据……",
        topic="法律意见",
        confidence=0.60,
    )
    print("阻断请求审查结果:", json.dumps(r2, ensure_ascii=False, indent=2))

python ai_ethics_guard.py