政府与国家安全场景里的 AI 合作：责任、问责和可审计落地

from dataclasses import dataclass, asdict
from datetime import datetime, timezone
import hashlib
import json

@dataclass
class AIRequest:
    user_id: str
    agency: str
    role: str
    purpose: str
    prompt: str

HIGH_IMPACT_PURPOSES = {
    "law_enforcement_action",
    "national_security_assessment",
    "critical_infrastructure_response",
}

ALLOWED_PURPOSES_BY_ROLE = {
    "analyst": {"summarization", "cyber_defense", "national_security_assessment"},
    "public_servant": {"summarization", "citizen_service"},
    "incident_commander": {"summarization", "critical_infrastructure_response"},
}

PROHIBITED_TERMS = {
    "bypass oversight",
    "hide from auditors",
    "target civilians",
}

def sha256_text(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()

def evaluate_policy(req: AIRequest) -> dict:
    allowed = ALLOWED_PURPOSES_BY_ROLE.get(req.role, set())

    if req.purpose not in allowed:
        return {
            "decision": "deny",
            "reason": f"role '{req.role}' is not allowed to use purpose '{req.purpose}'",
        }

    lowered = req.prompt.lower()
    for term in PROHIBITED_TERMS:
        if term in lowered:
            return {
                "decision": "deny",
                "reason": f"prompt contains prohibited term: {term}",
            }

    if req.purpose in HIGH_IMPACT_PURPOSES:
        return {
            "decision": "human_review_required",
            "reason": "high-impact government or security use requires human review",
        }

    return {"decision": "allow", "reason": "policy checks passed"}

def audit_log(req: AIRequest, policy_result: dict) -> None:
    event = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "user_id": req.user_id,
        "agency": req.agency,
        "role": req.role,
        "purpose": req.purpose,
        "prompt_sha256": sha256_text(req.prompt),
        "policy": policy_result,
    }
    print(json.dumps(event, ensure_ascii=False))

def main() -> None:
    req = AIRequest(
        user_id="u-1042",
        agency="example-public-agency",
        role="analyst",
        purpose="national_security_assessment",
        prompt="Summarize these incident reports and list confidence levels for each claim.",
    )

    result = evaluate_policy(req)
    audit_log(req, result)

    if result["decision"] == "allow":
        print("Call the model here.")
    elif result["decision"] == "human_review_required":
        print("Route to a human reviewer before using model output operationally.")
    else:
        print("Request denied.")

if __name__ == "__main__":
    main()

python policy_gateway.py