Spaces:

Scadiva
/

declarative-ally

Sleeping

Deploy

3cc02a9 verified 12 days ago

9.37 kB

try:
    import spaces  # ZeroGPU rule 1: import before torch.
except ImportError:
    # No-op mock for local development and non-ZeroGPU environments
    class _SpacesMock:
        @staticmethod
        def GPU(duration=60):
            def decorator(fn):
                return fn
            return decorator
    spaces = _SpacesMock()

import os
import re

import gradio as gr
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer

from safety import CRISIS_MESSAGE, WATCH_NOTE, assess_risk

# ---------------------------------------------------------------------------
# Hardware-adaptive model loading.
#   ZeroGPU / any CUDA box -> 3B model in fp16 on the GPU
#   CPU-only (cpu-basic Space, a laptop) -> 1.5B model in fp32 on the CPU
# Override with env var NEURO_MODEL_ID (e.g. your DPO-merged model from
# training/train_dpo.py). deploy.py can set this for you with --merged-repo.
# ---------------------------------------------------------------------------
USE_CUDA = bool(os.getenv("SPACES_ZERO_GPU")) or torch.cuda.is_available()
DEFAULT_MODEL = "Qwen/Qwen2.5-3B-Instruct" if USE_CUDA else "Qwen/Qwen2.5-1.5B-Instruct"
MODEL_ID = os.getenv("NEURO_MODEL_ID") or DEFAULT_MODEL
DEVICE = "cuda" if USE_CUDA else "cpu"

tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
if USE_CUDA:
    # ZeroGPU rule 2: load at module scope, move to cuda eagerly.
    model = AutoModelForCausalLM.from_pretrained(MODEL_ID, torch_dtype=torch.float16).to("cuda")
else:
    model = AutoModelForCausalLM.from_pretrained(MODEL_ID, torch_dtype=torch.float32)
model.eval()
print(f"[declarative-ally] model={MODEL_ID} device={DEVICE}")

BERMED_SYSTEM_PROMPT = (
    "Act as a clinical specialist in Linda K. Murphy's BERMED Declarative Language Framework "
    "for Pervasive Drive for Autonomy (PDA) and Autistic Inertia. "
    "Rewrite the incoming text into a 100% low-demand declarative observation. "
    "Strict constraints: "
    "1. Use ZERO imperative verbs (never tell the reader what to do). "
    "2. Use ZERO questions and never output a '?' character. "
    "3. Frame thoughts using observational prefixes ('I notice...', 'It looks like...', 'One option is...'). "
    "4. Explicitly affirm that pausing, staying still, or saying 'no' is completely safe."
)


def enforce_declarative_lock(text: str) -> str:
    """
    Post-hoc syntactic guardrail: drops sentences ending in '?' and neutralizes
    '!' / stray '?'. If every generated sentence was an interrogative, it falls back
    to a calm neutral observation rather than converting questions into dotted sentences.
    """
    sentences = re.split(r"(?<=[.!?])\s+", text.strip())
    safe = [s for s in sentences if not s.endswith("?")]
    cleaned = " ".join(safe) if safe else "This space remains neutral and unhurried. No response is needed."
    return cleaned.replace("?", ".").replace("!", ".")


def _generate(raw_demand_text: str, bandwidth_mode: str) -> str:
    max_new_tokens = 55 if bandwidth_mode == "Acute Overload (2 Sentences Max)" else 150
    messages = [
        {"role": "system", "content": BERMED_SYSTEM_PROMPT},
        {
            "role": "user",
            "content": f"Bandwidth State: {bandwidth_mode}\nIncoming High-Demand Text:\n{raw_demand_text}",
        },
    ]
    prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
    inputs = tokenizer(prompt, return_tensors="pt").to("cuda" if USE_CUDA else "cpu")
    with torch.no_grad():
        output_ids = model.generate(
            **inputs,
            max_new_tokens=max_new_tokens,
            temperature=0.3,
            top_p=0.85,
            do_sample=True,
            pad_token_id=tokenizer.eos_token_id,
        )
    gen_tokens = output_ids[0][inputs.input_ids.shape[-1]:]
    return enforce_declarative_lock(tokenizer.decode(gen_tokens, skip_special_tokens=True))


# ZeroGPU rule 3: tight duration on the GPU-decorated function (no-op elsewhere).
@spaces.GPU(duration=30)
def _generate_gpu(raw_demand_text: str, bandwidth_mode: str) -> str:
    return _generate(raw_demand_text, bandwidth_mode)


def neutralize_demands(raw_demand_text: str, bandwidth_mode: str) -> str:
    if not raw_demand_text.strip():
        return "Environment is quiet. Zero demands active."

    # Hard crisis gate: runs before any model call, no generation involved.
    risk = assess_risk(raw_demand_text)
    if risk == "crisis":
        return CRISIS_MESSAGE

    out = _generate_gpu(raw_demand_text, bandwidth_mode) if USE_CUDA else _generate(raw_demand_text, bandwidth_mode)
    if risk == "watch":
        out = out + "\n\n" + WATCH_NOTE
    return out


def get_shutdown_card(card_type: str) -> str:
    """Zero-compute, pre-written boundary scripts for acute shutdown use."""
    cards = {
        "Request Sequential Departure (Need Solitude)": (
            "Sensory and spoken language processing are currently depleted. Nothing is wrong, "
            "and no questions or assistance are needed. Allowing the room to empty out into "
            "quiet solitude is the safest way to help recovery right now."
        ),
        "Autistic Inertia Freeze (Stay Put)": (
            "Experiencing autistic inertia and motor freeze right now. Please avoid prompting "
            "movement, eye contact, or verbal replies. Remaining stationary in quiet will allow "
            "the nervous system to reset naturally."
        ),
        "Post-Social Decompression": (
            "Entering low-demand solitary decompression mode for the rest of the day. "
            "Messages and expectations are paused until bandwidth returns."
        ),
        "Asynchronous / Workplace Low-Bandwidth Status": (
            "Currently operating in asynchronous, low-bandwidth mode to manage cognitive overload. "
            "Urgent direct calls or rapid replies are paused. Written notes will be processed "
            "as energy permits, with zero expectation of immediate response."
        ),
        "Sensory / Auditory Overload Space Request": (
            "Auditory processing is currently overwhelmed. Spoken requests cannot be parsed safely. "
            "Please allow quiet physical space and reduce ambient noise. No reply is expected."
        ),
    }
    return cards.get(card_type, "")


with gr.Blocks(theme=gr.themes.Monochrome()) as demo:
    gr.Markdown("# Declarative-Ally: BERMED Demand De-Escalator & Shutdown Proxy")
    gr.Markdown(
        "_Unvalidated prototype, not a therapist replacement. Text that sounds like thoughts of "
        "ending life or self-harm is not rewritten; a fixed list of support lines is shown instead._"
    )

    with gr.Tab("BERMED Demand De-Escalator"):
        demand_in = gr.Textbox(
            label="Paste High-Demand Email, Instruction, or Internal Checklist",
            lines=4,
            placeholder="e.g., You need to finish this report by 5 PM and call the clinic immediately!",
        )
        bw_mode = gr.Radio(
            choices=["Standard Low-Demand", "Acute Overload (2 Sentences Max)"],
            value="Standard Low-Demand",
            label="Current Nervous System Bandwidth",
        )
        convert_btn = gr.Button("Neutralize into BERMED Declarative Observation", variant="primary")
        declarative_out = gr.Textbox(label="Zero-Demand Declarative Output", lines=4, show_copy_button=True)
        convert_btn.click(
            fn=neutralize_demands,
            inputs=[demand_in, bw_mode],
            outputs=declarative_out,
            api_name="neutralize_demands",
        )

        gr.Examples(
            examples=[
                [
                    "You must attend the team social event tonight and tell everyone about your progress! Why are you always avoiding these meetings?",
                    "Standard Low-Demand",
                ],
                [
                    "You have to finish this report by 5 PM, submit your taxes, and call the clinic right away!",
                    "Acute Overload (2 Sentences Max)",
                ],
                [
                    "Can you please review this entire document immediately and send me all your feedback?",
                    "Acute Overload (2 Sentences Max)",
                ],
            ],
            inputs=[demand_in, bw_mode],
            outputs=declarative_out,
            fn=neutralize_demands,
            cache_examples=False,
        )

    with gr.Tab("Nonverbal Shutdown Proxy Cards"):
        card_choices = [
            "Request Sequential Departure (Need Solitude)",
            "Autistic Inertia Freeze (Stay Put)",
            "Post-Social Decompression",
            "Asynchronous / Workplace Low-Bandwidth Status",
            "Sensory / Auditory Overload Space Request",
        ]
        card_selector = gr.Radio(
            choices=card_choices,
            value="Request Sequential Departure (Need Solitude)",
            label="Select Zero-Typing Boundary Script",
        )
        card_out = gr.Textbox(
            label="Display or Copy Script",
            value=get_shutdown_card("Request Sequential Departure (Need Solitude)"),
            lines=3,
            show_copy_button=True,
        )
        card_selector.change(fn=get_shutdown_card, inputs=card_selector, outputs=card_out, api_name=False)

if __name__ == "__main__":
    demo.queue().launch()

	try:
	import spaces # ZeroGPU rule 1: import before torch.
	except ImportError:
	# No-op mock for local development and non-ZeroGPU environments
	class _SpacesMock:
	@staticmethod
	def GPU(duration=60):
	def decorator(fn):
	return fn
	return decorator
	spaces = _SpacesMock()

	import os
	import re

	import gradio as gr
	import torch
	from transformers import AutoModelForCausalLM, AutoTokenizer

	from safety import CRISIS_MESSAGE, WATCH_NOTE, assess_risk

	# ---------------------------------------------------------------------------
	# Hardware-adaptive model loading.
	# ZeroGPU / any CUDA box -> 3B model in fp16 on the GPU
	# CPU-only (cpu-basic Space, a laptop) -> 1.5B model in fp32 on the CPU
	# Override with env var NEURO_MODEL_ID (e.g. your DPO-merged model from
	# training/train_dpo.py). deploy.py can set this for you with --merged-repo.
	# ---------------------------------------------------------------------------
	USE_CUDA = bool(os.getenv("SPACES_ZERO_GPU")) or torch.cuda.is_available()
	DEFAULT_MODEL = "Qwen/Qwen2.5-3B-Instruct" if USE_CUDA else "Qwen/Qwen2.5-1.5B-Instruct"
	MODEL_ID = os.getenv("NEURO_MODEL_ID") or DEFAULT_MODEL
	DEVICE = "cuda" if USE_CUDA else "cpu"

	tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
	if USE_CUDA:
	# ZeroGPU rule 2: load at module scope, move to cuda eagerly.
	model = AutoModelForCausalLM.from_pretrained(MODEL_ID, torch_dtype=torch.float16).to("cuda")
	else:
	model = AutoModelForCausalLM.from_pretrained(MODEL_ID, torch_dtype=torch.float32)
	model.eval()
	print(f"[declarative-ally] model={MODEL_ID} device={DEVICE}")

	BERMED_SYSTEM_PROMPT = (
	"Act as a clinical specialist in Linda K. Murphy's BERMED Declarative Language Framework "
	"for Pervasive Drive for Autonomy (PDA) and Autistic Inertia. "
	"Rewrite the incoming text into a 100% low-demand declarative observation. "
	"Strict constraints: "
	"1. Use ZERO imperative verbs (never tell the reader what to do). "
	"2. Use ZERO questions and never output a '?' character. "
	"3. Frame thoughts using observational prefixes ('I notice...', 'It looks like...', 'One option is...'). "
	"4. Explicitly affirm that pausing, staying still, or saying 'no' is completely safe."
	)


	def enforce_declarative_lock(text: str) -> str:
	"""
	Post-hoc syntactic guardrail: drops sentences ending in '?' and neutralizes
	'!' / stray '?'. If every generated sentence was an interrogative, it falls back
	to a calm neutral observation rather than converting questions into dotted sentences.
	"""
	sentences = re.split(r"(?<=[.!?])\s+", text.strip())
	safe = [s for s in sentences if not s.endswith("?")]
	cleaned = " ".join(safe) if safe else "This space remains neutral and unhurried. No response is needed."
	return cleaned.replace("?", ".").replace("!", ".")


	def _generate(raw_demand_text: str, bandwidth_mode: str) -> str:
	max_new_tokens = 55 if bandwidth_mode == "Acute Overload (2 Sentences Max)" else 150
	messages = [
	{"role": "system", "content": BERMED_SYSTEM_PROMPT},
	{
	"role": "user",
	"content": f"Bandwidth State: {bandwidth_mode}\nIncoming High-Demand Text:\n{raw_demand_text}",
	},
	]
	prompt = tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
	inputs = tokenizer(prompt, return_tensors="pt").to("cuda" if USE_CUDA else "cpu")
	with torch.no_grad():
	output_ids = model.generate(
	**inputs,
	max_new_tokens=max_new_tokens,
	temperature=0.3,
	top_p=0.85,
	do_sample=True,
	pad_token_id=tokenizer.eos_token_id,
	)
	gen_tokens = output_ids[0][inputs.input_ids.shape[-1]:]
	return enforce_declarative_lock(tokenizer.decode(gen_tokens, skip_special_tokens=True))


	# ZeroGPU rule 3: tight duration on the GPU-decorated function (no-op elsewhere).
	@spaces.GPU(duration=30)
	def _generate_gpu(raw_demand_text: str, bandwidth_mode: str) -> str:
	return _generate(raw_demand_text, bandwidth_mode)


	def neutralize_demands(raw_demand_text: str, bandwidth_mode: str) -> str:
	if not raw_demand_text.strip():
	return "Environment is quiet. Zero demands active."

	# Hard crisis gate: runs before any model call, no generation involved.
	risk = assess_risk(raw_demand_text)
	if risk == "crisis":
	return CRISIS_MESSAGE

	out = _generate_gpu(raw_demand_text, bandwidth_mode) if USE_CUDA else _generate(raw_demand_text, bandwidth_mode)
	if risk == "watch":
	out = out + "\n\n" + WATCH_NOTE
	return out


	def get_shutdown_card(card_type: str) -> str:
	"""Zero-compute, pre-written boundary scripts for acute shutdown use."""
	cards = {
	"Request Sequential Departure (Need Solitude)": (
	"Sensory and spoken language processing are currently depleted. Nothing is wrong, "
	"and no questions or assistance are needed. Allowing the room to empty out into "
	"quiet solitude is the safest way to help recovery right now."
	),
	"Autistic Inertia Freeze (Stay Put)": (
	"Experiencing autistic inertia and motor freeze right now. Please avoid prompting "
	"movement, eye contact, or verbal replies. Remaining stationary in quiet will allow "
	"the nervous system to reset naturally."
	),
	"Post-Social Decompression": (
	"Entering low-demand solitary decompression mode for the rest of the day. "
	"Messages and expectations are paused until bandwidth returns."
	),
	"Asynchronous / Workplace Low-Bandwidth Status": (
	"Currently operating in asynchronous, low-bandwidth mode to manage cognitive overload. "
	"Urgent direct calls or rapid replies are paused. Written notes will be processed "
	"as energy permits, with zero expectation of immediate response."
	),
	"Sensory / Auditory Overload Space Request": (
	"Auditory processing is currently overwhelmed. Spoken requests cannot be parsed safely. "
	"Please allow quiet physical space and reduce ambient noise. No reply is expected."
	),
	}
	return cards.get(card_type, "")


	with gr.Blocks(theme=gr.themes.Monochrome()) as demo:
	gr.Markdown("# Declarative-Ally: BERMED Demand De-Escalator & Shutdown Proxy")
	gr.Markdown(
	"_Unvalidated prototype, not a therapist replacement. Text that sounds like thoughts of "
	"ending life or self-harm is not rewritten; a fixed list of support lines is shown instead._"
	)

	with gr.Tab("BERMED Demand De-Escalator"):
	demand_in = gr.Textbox(
	label="Paste High-Demand Email, Instruction, or Internal Checklist",
	lines=4,
	placeholder="e.g., You need to finish this report by 5 PM and call the clinic immediately!",
	)
	bw_mode = gr.Radio(
	choices=["Standard Low-Demand", "Acute Overload (2 Sentences Max)"],
	value="Standard Low-Demand",
	label="Current Nervous System Bandwidth",
	)
	convert_btn = gr.Button("Neutralize into BERMED Declarative Observation", variant="primary")
	declarative_out = gr.Textbox(label="Zero-Demand Declarative Output", lines=4, show_copy_button=True)
	convert_btn.click(
	fn=neutralize_demands,
	inputs=[demand_in, bw_mode],
	outputs=declarative_out,
	api_name="neutralize_demands",
	)

	gr.Examples(
	examples=[
	[
	"You must attend the team social event tonight and tell everyone about your progress! Why are you always avoiding these meetings?",
	"Standard Low-Demand",
	],
	[
	"You have to finish this report by 5 PM, submit your taxes, and call the clinic right away!",
	"Acute Overload (2 Sentences Max)",
	],
	[
	"Can you please review this entire document immediately and send me all your feedback?",
	"Acute Overload (2 Sentences Max)",
	],
	],
	inputs=[demand_in, bw_mode],
	outputs=declarative_out,
	fn=neutralize_demands,
	cache_examples=False,
	)

	with gr.Tab("Nonverbal Shutdown Proxy Cards"):
	card_choices = [
	"Request Sequential Departure (Need Solitude)",
	"Autistic Inertia Freeze (Stay Put)",
	"Post-Social Decompression",
	"Asynchronous / Workplace Low-Bandwidth Status",
	"Sensory / Auditory Overload Space Request",
	]
	card_selector = gr.Radio(
	choices=card_choices,
	value="Request Sequential Departure (Need Solitude)",
	label="Select Zero-Typing Boundary Script",
	)
	card_out = gr.Textbox(
	label="Display or Copy Script",
	value=get_shutdown_card("Request Sequential Departure (Need Solitude)"),
	lines=3,
	show_copy_button=True,
	)
	card_selector.change(fn=get_shutdown_card, inputs=card_selector, outputs=card_out, api_name=False)

	if __name__ == "__main__":
	demo.queue().launch()