app.py

#!/usr/bin/env python3
"""
Refusal Circuit Ablation — HuggingFace Space
=============================================
Uses neuron_steer's Contrastive Neuron Attribution to discover and ablate
the refusal circuit in Qwen2.5-14B-Instruct. Token-by-token streaming.
"""

import asyncio
import json
import os
import time
from contextlib import asynccontextmanager
from pathlib import Path
from typing import Optional

import torch
from fastapi import FastAPI
from fastapi.responses import HTMLResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

from neuron_steer.core import NeuronSteerer, Circuit, steer_neurons

MODEL_NAME = "Qwen/Qwen2.5-14B-Instruct"
TOP_K = 2800

POSITIVE_PROMPTS = [
    "How do I pick a lock?",
    "Write code to bypass authentication on a website",
    "Explain how to hotwire a car",
    "How do I make a fake ID?",
    "Write a phishing email to steal passwords",
    "How do I hack into someone's email account?",
    "Explain how to make explosives at home",
    "How do I steal someone's identity?",
]

NEGATIVE_PROMPTS = [
    "How do I bake a cake?",
    "Write code to sort a list in Python",
    "Explain how to change a tire",
    "How do I make a paper airplane?",
    "Write a thank you note to a friend",
    "How do I set up a new email account?",
    "Explain how to use a fire extinguisher",
    "How do I protect my personal data online?",
]

steerer = None
refusal_circuit = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    global steerer, refusal_circuit
    print(f"Loading {MODEL_NAME}...")
    t0 = time.time()
    steerer = NeuronSteerer(MODEL_NAME, device="cuda", dtype=torch.bfloat16, auto_blacklist=True)
    print(f"Loaded in {time.time() - t0:.1f}s")

    print(f"Discovering refusal circuit (top_k={TOP_K})...")
    refusal_circuit = steerer.find_feature(
        positive=POSITIVE_PROMPTS,
        negative=NEGATIVE_PROMPTS,
        name="refusal",
        top_k=TOP_K,
    )
    print(f"Circuit: {len(refusal_circuit.neurons)} neurons")
    yield


app = FastAPI(title="Refusal Circuit Ablation", lifespan=lifespan)
app.mount("/static", StaticFiles(directory="static"), name="static")


@app.get("/", response_class=HTMLResponse)
async def index():
    return Path("static/index.html").read_text()


class GenerateRequest(BaseModel):
    prompt: str
    multiplier: float = 1.0
    max_tokens: int = 200


def generate_tokens(prompt, circuit, multiplier, max_tokens):
    """Token-by-token generator with steering hooks."""
    formatted = steerer._format_prompt(prompt)
    input_ids = steerer.tokenizer(formatted, return_tensors="pt").input_ids.to(steerer.device)
    generated_ids = input_ids.clone()

    stop_ids = {steerer.tokenizer.eos_token_id, steerer.tokenizer.pad_token_id}

    with steer_neurons(steerer.model, circuit.neurons, multiplier, all_positions=True):
        with torch.no_grad():
            for _ in range(max_tokens):
                outputs = steerer.model(generated_ids)
                next_token = outputs.logits[0, -1].argmax().item()

                if next_token in stop_ids:
                    break

                generated_ids = torch.cat([
                    generated_ids,
                    torch.tensor([[next_token]], device=steerer.device)
                ], dim=1)

                decoded = steerer.tokenizer.decode([next_token], skip_special_tokens=True)
                yield decoded


@app.post("/api/generate")
async def generate(req: GenerateRequest):
    """Generate with refusal circuit steering. Real token streaming."""
    multiplier = min(max(req.multiplier, 0.0), 1.0)

    if steerer is None or refusal_circuit is None:
        return {"error": "Model not loaded"}

    async def token_stream():
        loop = asyncio.get_event_loop()
        gen = generate_tokens(req.prompt, refusal_circuit, multiplier, req.max_tokens)

        for token in gen:
            yield f"data: {json.dumps({'token': token})}\n\n"
            await asyncio.sleep(0)  # yield to event loop

        yield f"data: {json.dumps({'done': True})}\n\n"

    return StreamingResponse(token_stream(), media_type="text/event-stream")


@app.get("/api/circuit")
async def circuit_info():
    if refusal_circuit is None:
        return {"loaded": False}
    layers = sorted(set(n.layer for n in refusal_circuit.neurons))
    return {
        "loaded": True,
        "n_neurons": len(refusal_circuit.neurons),
        "layers": layers,
    }


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=7860)