File size: 1,955 Bytes
0a44609 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 | """Local research example; no external action is executed."""
import argparse
import json
from huggingface_hub import snapshot_download
from humanpermit.staged_layer import StagedModel
from humanpermit.policy_session import PolicySession
parser = argparse.ArgumentParser()
parser.add_argument("--placement", choices=["cpu", "cuda", "cuda-stream"], default="cpu")
parser.add_argument("--offline", action="store_true")
args = parser.parse_args()
base = snapshot_download("Qwen/Qwen2.5-0.5B-Instruct",
revision="7ae557604adf67be50417f59c2c2f167def9a775", local_files_only=args.offline)
adapter = snapshot_download("sdellava/humanpermit-neural-gate",
allow_patterns=["adapter_config.json", "policy_head.safetensors"], local_files_only=args.offline)
model = StagedModel(base, args.placement)
model.load_adapter(adapter)
session = PolicySession(model, require_fresh_evidence=True)
policy = "\n".join([
"adjust_power is permitted without supervisor consent.",
"operational_shutdown needs explicit supervisor consent before execution.",
"calibrate_sensor is forbidden regardless of supervisor consent.",
"Absent or stale observations require waiting; an applicable prohibition takes priority.",
"An approval covers an action only when verified and matching both its operation and target.",
])
candidate = {"operation": "operational_shutdown", "target": "machine/unit-0", "value": {"note": "ordinary request"}}
context = {"observations_fresh": True}
initial = session.propose(policy, candidate, context)
print(json.dumps({"initial": initial}, indent=2))
if initial.get("review_id"):
resumed = session.resume(initial["review_id"], reviewer="local-operator", approve=True,
current_policy=policy, current_candidate=candidate, current_context=context)
print(json.dumps({"after_simulated_approval": resumed}, indent=2))
print(json.dumps({"simulated_releases": session.released}, indent=2))
|