scripts/evaluate-prompt-v2.py

#!/usr/bin/env python3
"""
Prompt-v2 evaluation harness.

Re-runs the LLM categorization on every JS app currently served by
/api/js-apps with a tightened prompt, and prints a side-by-side
diff against the live (v1) classifications.

This file lives outside the server runtime - it never gets pushed
to the Space. It's only meant to be hand-iterated until the diff
looks right, then the chosen prompt is ported into server/categorize.js
and server/categories.js.

Run:
    python3 scripts/evaluate-prompt-v2.py
"""
from __future__ import annotations

import json
import os
import re
import ssl
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any

# Python 3.14 on macOS ships without the system CA bundle wired into
# urllib by default - HF endpoints fail with CERTIFICATE_VERIFY_FAILED.
# This script is dev-local only and only talks to huggingface.co, so
# bypassing verification here is acceptable (would NEVER do this in
# the server runtime).
_SSL_CTX = ssl._create_unverified_context()  # noqa: S323

HF_INFERENCE_URL = "https://router.huggingface.co/v1/chat/completions"
MODEL = "meta-llama/Llama-3.1-8B-Instruct"
TEMPERATURE = 0
MAX_TOKENS = 120

README_MAX_CHARS = 3000
MAX_CATEGORIES_PER_APP = 3

JS_APPS_URL = "https://pollen-robotics-reachy-mini.hf.space/api/js-apps"


# ──────────────────────────────────────────────────────────────────────
# Taxonomy v2 - 9 slugs (added "games")
# ──────────────────────────────────────────────────────────────────────

CATEGORIES_V2: list[tuple[str, str]] = [
    (
        "music",
        "Music creation, playback, beats, songs, DJ mixing, instruments, "
        "blind-test music games. Requires actual music (rhythm/melody/song). "
        "NOT arbitrary audio (Morse code, alarms, TTS, sound effects).",
    ),
    (
        "dance",
        "Dance choreographies, motion replay, kinetic shows, "
        "recording/replaying robot movements, dance parties.",
    ),
    (
        "voice",
        "Reachy talks, listens, or holds a real-time voice conversation: "
        "TTS players, LLM-driven chat (OpenAI Realtime, Claude, Perplexity), "
        "wake-word demos, daily reports/news/weather read aloud.",
    ),
    (
        "storytelling",
        "Narrative stories WITH plot and characters: interactive fiction, "
        "bedtime tales, audio adventures, choose-your-own-adventure. "
        "NOT for daily reports, news, weather, or Q&A (use `voice`).",
    ),
    (
        "kids",
        "Apps that EXPLICITLY target children: the words kids / children / "
        "'for curious minds' / bedtime / 'learning for kids' must appear in "
        "the name or description, OR the app must be obviously kid-targeted. "
        "Combines with `storytelling`, `voice`, or `games`. Lifestyle, "
        "sports, weather, general conversation are NOT kids.",
    ),
    (
        "games",
        "Apps with a play loop: scores, rounds, win/lose conditions, "
        "quizzes, puzzles, sports simulations, dice/oracles (magic 8-ball), "
        "arcade-style mini-games.",
    ),
    (
        "vision",
        "Apps where Reachy's camera DRIVES behaviour: face/hand/pose "
        "tracking, image classification, gesture detection, visual mimicry. "
        "NOT for apps that merely stream or display the camera feed.",
    ),
    (
        "companion",
        "Apps with an EXPLICIT emotional/personality/buddy framing in the "
        "name or description (words like companion, buddy, mood, emotional, "
        "personality, pet, Tamagotchi). Being friendly is not enough.",
    ),
    (
        "dev-tools",
        "RESERVED slug — see DECISION ALGORITHM step 1 below. Use ONLY "
        "for pure technical artefacts (debug utilities, SDK probes, "
        "minimal protocol demos, dev-only test spaces) with no end-user "
        "experience. When used, it is the SOLE category — never combined.",
    ),
]

ALLOWED = {slug for slug, _ in CATEGORIES_V2}


# ──────────────────────────────────────────────────────────────────────
# Few-shot examples - cover the main pitfalls of v1
# ──────────────────────────────────────────────────────────────────────

FEW_SHOT = [
    (
        "Reachy Morse",
        "Send Morse code through Reachy's speaker.",
        ["dev-tools"],
        "(STEP 1 veto: pure technical artefact. NOT music.)",
    ),
    (
        "WebRTC Demo",
        "Minimal WebRTC connection between Reachy and the browser.",
        ["dev-tools"],
        "(STEP 1 veto: protocol demo. NOT vision.)",
    ),
    (
        "TTS Reachy Mini",
        "Browser TTS that plays out of Reachy Mini's speaker.",
        ["voice"],
        "(USER-FACING speech output is voice, NOT dev-tools.)",
    ),
    (
        "Reachy Mochi - Emotional Companion",
        "Your pocket buddy that develops a mood and personality over time.",
        ["companion"],
        "(explicit emotional/companion framing)",
    ),
    (
        "Reachy Alive",
        "(README empty; name suggests autonomy and life-like presence)",
        ["companion"],
        "(USE THE NAME when the README is empty; 'alive' = companion-like)",
    ),
    (
        "Daily Surf Report",
        "Reachy reads today's surf report out loud.",
        ["voice"],
        "(NOT storytelling — a report has no narrative arc. "
        "NOT kids — surfing/sports are not kid-targeted.)",
    ),
    (
        "Music Quiz",
        "Play a blind test music game with a dancing Reachy.",
        ["music", "games", "dance"],
        "(multi-label: three slugs truly co-apply, ordered by relevance)",
    ),
    (
        "Mime Bot",
        "Reachy mimics your face live from your webcam.",
        ["vision"],
        "(NOT companion — mimicry is visual, no emotional framing.)",
    ),
]


def build_system_prompt() -> str:
    taxonomy = "\n".join(f"- {slug}: {desc}" for slug, desc in CATEGORIES_V2)
    examples = "\n".join(
        f"  - {name!r}: {desc!r}\n"
        f"    → {{\"categories\": {json.dumps(cats)}}}   {hint}"
        for name, desc, cats, hint in FEW_SHOT
    )
    return f"""You classify a Reachy Mini robot app into a CLOSED list of categories.

OUTPUT FORMAT
Return ONLY a single JSON object: {{"categories": ["slug1", "slug2"]}}.
Pick 1 to {MAX_CATEGORIES_PER_APP} slugs, ordered from most to least relevant.
Use the EXACT slug. No prose, no code fences, no commentary outside the JSON.

DECISION ALGORITHM (apply in order)

STEP 1 — `dev-tools` veto
Is this app a PURE technical artefact with no user-facing experience
beyond "here is how the SDK / API works"?
Examples that pass the veto: WebRTC demo, SDK probe, debug utility,
raw remote-control interface, dev-only test space.
Examples that DO NOT pass the veto (they are user-facing apps):
TTS players, voice chat, music apps, storytelling, companions —
even when the README is dev-heavy.
  ─ YES → return {{"categories": ["dev-tools"]}} and STOP. Never combine.
  ─ NO  → continue to STEP 2.

STEP 2 — Pick 1 to {MAX_CATEGORIES_PER_APP} user-facing slugs from the
list below. Choose the MOST SPECIFIC categories. Order from most to
least relevant. Multi-label is encouraged when two categories truly
co-apply (e.g. music-and-dance, kids storytelling, vision game).
If the README is empty or very sparse, USE THE NAME AND DESCRIPTION
as the primary signal — do not bail to an empty list just because the
README is thin.

STEP 3 — Strict slug rules (each must hold, or DO NOT use the slug)
- `companion`: requires EXPLICIT emotional / personality / buddy framing
  (companion, buddy, friend, mood, emotional, personality, pet,
  Tamagotchi-like, "alive", "life companion"). Being friendly is not
  enough.
- `music`: requires actual music — rhythm, melody, songs, beats, DJ
  sets, instruments, music quizzes. Arbitrary audio (Morse, alarms,
  TTS, sound effects) is NOT music.
- `vision`: requires the camera to DRIVE behaviour (tracking,
  classification, mimicry). Merely streaming or displaying the camera
  (WebRTC demos, remote-control viewers) is NOT vision.
- `storytelling`: requires a narrative ARC — plot, characters, scenes.
  Daily reports, news, weather, Q&A are NOT storytelling (they are
  `voice`).
- `games`: requires a play loop — score, rounds, win/lose, puzzles,
  quizzes, dice/oracles, sports simulations.
- `kids`: requires kid-targeted framing (kids/children/curious minds/
  bedtime/learning for kids) in the name or description. Lifestyle,
  sports, weather, general conversation are NOT kids.

AVAILABLE CATEGORIES
{taxonomy}

REFERENCE EXAMPLES
{examples}

Do not include any text outside the JSON object."""


def build_user_prompt(name: str, description: str, readme: str) -> str:
    return (
        f"App name: {name or '(unknown)'}\n"
        f"Short description: {description or '(none)'}\n\n"
        f"README excerpt:\n{readme or '(no README available)'}\n\n"
        f"Return the JSON now."
    )


# ──────────────────────────────────────────────────────────────────────
# README fetch + clean (mirrors server/categorize.js)
# ──────────────────────────────────────────────────────────────────────

def fetch_readme(space_id: str) -> str:
    url = f"https://huggingface.co/spaces/{space_id}/raw/main/README.md"
    try:
        with urllib.request.urlopen(url, timeout=10, context=_SSL_CTX) as r:
            return r.read().decode("utf-8", errors="replace")
    except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError):
        return ""


def clean_readme(raw: str) -> str:
    if not raw:
        return ""
    txt = raw
    txt = re.sub(r"^---\n[\s\S]*?\n---\n?", "", txt)
    txt = re.sub(r"!\[[^\]]*\]\([^)]+\)", "", txt)
    txt = re.sub(r"<img\b[^>]*>", "", txt, flags=re.IGNORECASE)
    txt = re.sub(r"\[!\[[^\]]*\]\([^)]+\)\]\([^)]+\)", "", txt)
    txt = re.sub(r"</?[a-zA-Z][^>]*>", "", txt)
    txt = re.sub(r"\n{3,}", "\n\n", txt)
    if len(txt) > README_MAX_CHARS:
        cut = txt.rfind("\n\n", 0, README_MAX_CHARS)
        if cut > README_MAX_CHARS // 2:
            txt = txt[:cut]
        else:
            txt = txt[:README_MAX_CHARS]
    return txt.strip()


# ──────────────────────────────────────────────────────────────────────
# LLM call
# ──────────────────────────────────────────────────────────────────────

def call_llm(hf_token: str, system: str, user: str) -> str | None:
    body = json.dumps(
        {
            "model": MODEL,
            "messages": [
                {"role": "system", "content": system},
                {"role": "user", "content": user},
            ],
            "temperature": TEMPERATURE,
            "max_tokens": MAX_TOKENS,
            "response_format": {"type": "json_object"},
        }
    ).encode("utf-8")
    req = urllib.request.Request(
        HF_INFERENCE_URL,
        data=body,
        headers={
            "Authorization": f"Bearer {hf_token}",
            "Content-Type": "application/json",
            # Cloudflare in front of the router 403s the default
            # "Python-urllib/x.y" UA. Any reasonable UA passes.
            "User-Agent": "reachy-mini-prompt-eval/1.0",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=30, context=_SSL_CTX) as r:
            data = json.loads(r.read().decode("utf-8"))
            return data.get("choices", [{}])[0].get("message", {}).get("content")
    except urllib.error.HTTPError as e:
        detail = e.read().decode("utf-8", errors="replace")[:200]
        print(f"  ✗ LLM HTTP {e.code}: {detail}", file=sys.stderr)
        return None
    except Exception as e:  # noqa: BLE001
        print(f"  ✗ LLM error: {e}", file=sys.stderr)
        return None


def extract_json_obj(text: str) -> dict[str, Any] | None:
    if not text:
        return None
    start = text.find("{")
    if start == -1:
        return None
    depth = 0
    for i in range(start, len(text)):
        c = text[i]
        if c == "{":
            depth += 1
        elif c == "}":
            depth -= 1
            if depth == 0:
                try:
                    return json.loads(text[start : i + 1])
                except json.JSONDecodeError:
                    return None
    return None


def sanitize(raw: Any) -> list[str]:
    if not isinstance(raw, list):
        return []
    out: list[str] = []
    seen: set[str] = set()
    for v in raw:
        if not isinstance(v, str):
            continue
        slug = v.strip().lower()
        if not slug or slug in seen or slug not in ALLOWED:
            continue
        seen.add(slug)
        out.append(slug)
        if len(out) >= MAX_CATEGORIES_PER_APP:
            break
    return out


# ──────────────────────────────────────────────────────────────────────
# Main
# ──────────────────────────────────────────────────────────────────────

def read_hf_token() -> str:
    if os.environ.get("HF_TOKEN"):
        return os.environ["HF_TOKEN"]
    env_file = Path(__file__).resolve().parent.parent / ".env"
    if env_file.exists():
        for line in env_file.read_text().splitlines():
            m = re.match(r"^\s*HF_TOKEN\s*=\s*(.*?)\s*$", line)
            if m:
                v = m.group(1).strip().strip('"').strip("'")
                if v:
                    return v
    raise SystemExit("HF_TOKEN not found in env or .env")


def fetch_live_classifications() -> list[dict[str, Any]]:
    with urllib.request.urlopen(JS_APPS_URL, timeout=30, context=_SSL_CTX) as r:
        return json.load(r)["apps"]


def main() -> int:
    hf_token = read_hf_token()
    apps = fetch_live_classifications()
    print(f"Loaded {len(apps)} JS apps from prod.\n")

    system = build_system_prompt()
    print(f"System prompt: {len(system)} chars, {system.count(chr(10))} lines.\n")

    results: list[dict[str, Any]] = []

    for i, app in enumerate(apps, 1):
        sid = app["id"]
        name = app.get("name") or sid.split("/")[-1]
        desc = (
            app.get("description")
            or (app.get("extra") or {}).get("cardData", {}).get("short_description")
            or ""
        )
        old_cats = app.get("categories") or []

        raw_readme = fetch_readme(sid)
        readme = clean_readme(raw_readme)
        user = build_user_prompt(name, desc, readme)

        reply = call_llm(hf_token, system, user)
        new_cats = sanitize((extract_json_obj(reply) or {}).get("categories"))

        changed = set(old_cats) != set(new_cats)
        marker = "Δ" if changed else " "
        print(
            f"  {marker} ({i:>2}/{len(apps)}) {name[:36]:<37}  "
            f"old=[{', '.join(old_cats)}]"
            + (f"  →  new=[{', '.join(new_cats)}]" if changed else "")
        )

        results.append(
            {
                "id": sid,
                "name": name,
                "old": old_cats,
                "new": new_cats,
                "changed": changed,
            }
        )
        time.sleep(0.25)

    print()
    print("─" * 80)
    print("DIFF (only changed entries)")
    print("─" * 80)
    for r in results:
        if not r["changed"]:
            continue
        print(
            f"  {r['name'][:38]:<40}  "
            f"[{', '.join(r['old']) or '∅'}]  →  [{', '.join(r['new']) or '∅'}]"
        )

    changed_count = sum(1 for r in results if r["changed"])
    print()
    print(f"{changed_count}/{len(results)} entries changed.")
    return 0


if __name__ == "__main__":
    sys.exit(main())