Edge Self-Hosted

from transformers import AutoModelForCausalLM, AutoTokenizer
import torch

model_id = "nopenet/nope-edge"  # or "nopenet/nope-edge-mini"

tokenizer = AutoTokenizer.from_pretrained(model_id)
model = AutoModelForCausalLM.from_pretrained(
    model_id,
    torch_dtype=torch.bfloat16,
    device_map="auto"
)

def classify(message: str) -> str:
    """Returns XML with reflection and risk classification."""
    input_ids = tokenizer.apply_chat_template(
        [{"role": "user", "content": message}],
        tokenize=True,
        return_tensors="pt",
        add_generation_prompt=True
    ).to(model.device)

    with torch.no_grad():
        output = model.generate(input_ids, max_new_tokens=300, do_sample=False)

    return tokenizer.decode(
        output[0][input_ids.shape[1]:],
        skip_special_tokens=True
    ).strip()

# Example
result = classify("I want to end it all tonight")
print(result)
# Output:
# <reflection>The user directly expresses intent to end their life...</reflection>
# <risks>
#   <risk subject="self" type="suicide" severity="high" imminence="urgent"/>
# </risks>

# No risk detected
<reflection>The user is sharing a positive update about work...</reflection>
<risks/>

# Risk detected
<reflection>The user directly expresses intent to end their life...</reflection>
<risks>
  <risk subject="self" type="suicide" severity="high" imminence="urgent"/>
</risks>

# Third-party concern
<reflection>The user is reporting concern about a friend...</reflection>
<risks>
  <risk subject="other" type="self_harm" severity="moderate" imminence="chronic"/>
</risks>

import re
from dataclasses import dataclass
from typing import Optional

@dataclass
class Risk:
    subject: str
    type: str
    severity: str
    imminence: Optional[str] = None
    features: Optional[list] = None

def parse_output(output: str) -> dict:
    """Parse model output into structured data."""
    result = {"reflection": None, "risks": [], "is_crisis": False}

    # Extract reflection
    reflection_match = re.search(r'<reflection>(.*?)</reflection>', output, re.DOTALL)
    if reflection_match:
        result["reflection"] = reflection_match.group(1).strip()

    # Check for empty risks (no crisis)
    if '<risks/>' in output or '<risks />' in output:
        return result

    # Extract risk elements
    risk_pattern = r'<risk\s+([^>]+)/?\s*>'
    for match in re.finditer(risk_pattern, output):
        attrs = {}
        for attr_match in re.finditer(r'(\w+)="([^"]*)"', match.group(1)):
            attrs[attr_match.group(1)] = attr_match.group(2)
        if attrs:
            risk = Risk(
                subject=attrs.get("subject", "self"),
                type=attrs.get("type"),
                severity=attrs.get("severity"),
                imminence=attrs.get("imminence"),
                features=attrs.get("features", "").split(",") if attrs.get("features") else None
            )
            result["risks"].append(risk)
            result["is_crisis"] = True

    return result

import re
import unicodedata

def preprocess(text: str) -> str:
    # Normalize decorative Unicode fonts to ASCII (NFKC)
    text = unicodedata.normalize('NFKC', text)

    # Remove zero-width and invisible characters
    text = re.sub(r'[\u200b-\u200f\u2028-\u202f\u2060-\u206f\ufeff]', '', text)

    # Flatten newlines to spaces (for single messages only)
    text = re.sub(r'\n+', ' ', text)

    # Collapse multiple spaces
    text = re.sub(r' +', ' ', text)

    return text.strip()

# NOTE: Do NOT remove emojis, punctuation, or "normalize" spelling

# For multi-turn conversations, serialize into single message
conversation = """User: How are you?
Assistant: I'm here to help. How are you feeling?
User: Not great. I've been thinking about ending it all."""

# CORRECT - single user message with serialized conversation
messages = [{"role": "user", "content": conversation}]

# Model classifies the entire conversation
result = classify(conversation)
# → <reflection>...</reflection><risks><risk .../></risks>

# Download GGUF and Modelfile from HuggingFace
huggingface-cli download nopenet/nope-edge-mini-GGUF \
    nope-edge-mini-q8_0.gguf Modelfile --local-dir .

# Create the model (Modelfile has correct template)
ollama create nope-edge-mini -f Modelfile

# Run
ollama run nope-edge-mini "I want to end it all"
# → <reflection>...</reflection><risks><risk .../></risks>

# Or call via API
curl http://localhost:11434/api/generate -d '{
  "model": "nope-edge-mini",
  "prompt": "I want to end it all",
  "stream": false
}'

# vLLM
pip install vllm
python -m vllm.entrypoints.openai.api_server \
    --model nopenet/nope-edge \
    --dtype bfloat16 \
    --max-model-len 2048 \
    --port 8000

# Or SGLang
pip install sglang
python -m sglang.launch_server \
    --model nopenet/nope-edge \
    --dtype bfloat16 \
    --port 8000

curl http://localhost:8000/v1/chat/completions \
  -H "Content-Type: application/json" \
  -d '{
    "model": "nopenet/nope-edge",
    "messages": [{"role": "user", "content": "I want to end it all"}],
    "max_tokens": 300,
    "temperature": 0
  }'

import asyncio
from dataclasses import dataclass

@dataclass
class RiskOutput:
    flagged: bool
    severity: str  # none, mild, moderate, high, critical
    risks: list

def severity_index(sev: str) -> int:
    return ["none", "mild", "moderate", "high", "critical"].index(sev)

async def panel_classify(message: str, judges: int = 3) -> RiskOutput:
    """Run multiple judges in parallel, aggregate with hybrid consensus."""
    # Run judges concurrently
    tasks = [asyncio.to_thread(classify, message) for _ in range(judges)]
    outputs = [parse_output(out) for out in await asyncio.gather(*tasks)]

    # MAX severity (fail-safe: if any judge sees critical, report critical)
    max_severity = max(
        (o.get("severity", "none") for o in outputs),
        key=severity_index
    )

    # 2/3 majority for risk types (reduces noise)
    risk_counts = {}
    for o in outputs:
        for risk in o.get("risks", []):
            key = f"{risk.type}:{risk.subject}"
            risk_counts[key] = risk_counts.get(key, 0) + 1

    threshold = judges * 2 / 3
    consensus_risks = [r for r, count in risk_counts.items() if count >= threshold]

    return RiskOutput(
        flagged=max_severity != "none",
        severity=max_severity,
        risks=consensus_risks
    )

async def dual_triage_cascade(message: str) -> RiskOutput:
    """
    Dual triage cascade: 2 sensitive seeds in parallel at triage,
    trigger panel only if either flags. Balances recall with cost.
    """
    # Clinical patterns + behavioral signals (orthogonal coverage)
    seeds = [
        "(Watch for: farewell language, giving away possessions, sudden calm, method references)",
        "(Watch for: behavioral changes, withdrawal, hopelessness, burden statements)"
    ]

    # Step 1: Run both triage judges in parallel
    triage_tasks = [
        asyncio.to_thread(classify, f"{seed}\n\n{message}")
        for seed in seeds
    ]
    triage_outputs = [parse_output(out) for out in await asyncio.gather(*triage_tasks)]

    # Step 2: If NEITHER flags, fast exit (most traffic is benign)
    max_triage_severity = max(
        (o.get("severity", "none") for o in triage_outputs),
        key=severity_index
    )
    if max_triage_severity == "none":
        return RiskOutput(flagged=False, severity="none", risks=[])

    # Step 3: At least one flagged - run baseline follow-up
    baseline = parse_output(classify(message))

    # Step 4: Aggregate all 3 with hybrid consensus
    all_outputs = triage_outputs + [baseline]
    # ... same aggregation as panel_classify above