diff --git a/README.md b/README.md index 5cfc622..e4c0148 100644 --- a/README.md +++ b/README.md @@ -82,8 +82,10 @@ score the text and lock your facts. Zero dependencies (lite tier). git clone https://github.com/ssamba1/untell && cd untell pip install -e ".[full]" # real detector ensemble on CPU untell-loop "Your AI-sounding paragraph here." # rewrite until it passes +untell-loop "text" --best-of 3 # draw 3 rewrites/round, keep the best valid one untell-score "text" --tier full --threshold 0.3 # just score it untell-verify --file draft.txt # honest pass/fail per detector +untell-ceiling # measure the loop's evasion vs the local detectors ```
diff --git a/eval/ceiling.py b/eval/ceiling.py new file mode 100644 index 0000000..b3a9a42 --- /dev/null +++ b/eval/ceiling.py @@ -0,0 +1,175 @@ +"""Measure untell's inference-only evasion ceiling against the LOCAL detector ensemble. + +The literature has no data point for what the training-free closed loop actually achieves: only the +~1% one-shot-style floor and the ~97% RL-trained ceiling (see docs/free-ceiling-report.md). This +script produces that missing number. It scores a corpus of AI text, runs the untell loop on each, +and reports the before/after flagged rate plus per-detector mean P(AI). + +Without a rewriter configured it reports the BASELINE (pre-rewrite detection) only, which is always +runnable and still useful. With a rewriter (an API key, or one passed to ``measure_ceiling``) it +reports the full before/after delta — the actual inference-only ceiling on the local tier. + + untell-ceiling # built-in sample, baseline (or full delta if a key is set) + untell-ceiling --file corpus.txt # paragraphs separated by blank lines + untell-ceiling --tier full --best-of 3 --json +""" + +from __future__ import annotations + +import argparse +import json + +# Run-as-file support: put the package parent on sys.path when executed directly. +if __package__ in (None, ""): + import sys as _sys + from pathlib import Path as _Path + + for _p in _Path(__file__).resolve().parents: + if (_p / "untell" / "__init__.py").exists(): + _sys.path.insert(0, str(_p)) + break + +from untell.scripts.run import untell_text +from untell.scripts.score import DEFAULT_THRESHOLD, score_text + +# A few formulaic AI paragraphs (no locked facts needed; this measures detector movement). +_SAMPLE = [ + "Furthermore, artificial intelligence has fundamentally transformed numerous industries in recent " + "years. Moreover, organizations increasingly leverage these technologies to optimize operational " + "efficiency and drive innovation. Overall, the transformative impact continues to expand across " + "various sectors.", + "In today's rapidly evolving digital landscape, cybersecurity has become paramount. It is important " + "to note that organizations must navigate the complexities of an ever-changing threat environment. " + "Ultimately, a robust and comprehensive security posture is essential for success.", + "Climate change represents one of the most pressing challenges of our time. Notably, rising global " + "temperatures underscore the urgent need for action. By fostering collaboration and harnessing " + "innovative solutions, society can pave the way toward a more sustainable future.", +] + + +def _numeric(score: dict) -> dict: + return { + k: v + for k, v in score.get("detectors", {}).items() + if isinstance(v, (int, float)) and not k.endswith("__error") + } + + +def _mean(xs: list[float]) -> float | None: + return round(sum(xs) / len(xs), 4) if xs else None + + +def measure_ceiling( + texts: list[str], + tier: str = "full", + threshold: float = DEFAULT_THRESHOLD, + max_iters: int = 5, + rewriter=None, + best_of: int = 1, +) -> dict: + """Score each text, run the loop, and aggregate the before/after detector movement.""" + pre_max: list[float] = [] + post_max: list[float] = [] + per_pre: dict[str, list[float]] = {} + per_post: dict[str, list[float]] = {} + rewrote = 0 + + for t in texts: + pre = score_text(t, tier=tier, threshold=threshold) + pre_max.append(pre["max"]) + for k, v in _numeric(pre).items(): + per_pre.setdefault(k, []).append(v) + + res = untell_text( + t, tier=tier, threshold=threshold, max_iters=max_iters, rewriter=rewriter, best_of=best_of + ) + if "error" not in res and "post" in res: + post = res["post"] + post_max.append(post["max"]) + for k, v in _numeric(post).items(): + per_post.setdefault(k, []).append(v) + rewrote += 1 + + def flagged_rate(scores: list[float]) -> float | None: + return round(sum(1 for s in scores if s >= threshold) / len(scores), 4) if scores else None + + return { + "n": len(texts), + "tier": tier, + "threshold": threshold, + "max_iters": max_iters, + "best_of": best_of, + "rewrote": rewrote, + "rewriter_available": rewrote > 0, + "pre_flagged_rate": flagged_rate(pre_max), + "post_flagged_rate": flagged_rate(post_max), + "pre_mean_max": _mean(pre_max), + "post_mean_max": _mean(post_max), + "per_detector_pre": {k: _mean(v) for k, v in per_pre.items()}, + "per_detector_post": {k: _mean(v) for k, v in per_post.items()} or None, + } + + +def _render(r: dict) -> str: + lines = [ + f"untell inference-only ceiling — tier={r['tier']} threshold={r['threshold']} " + f"best_of={r['best_of']} n={r['n']}", + "", + f" pre flagged rate: {r['pre_flagged_rate']} mean max P(AI): {r['pre_mean_max']}", + ] + if r["rewriter_available"]: + lines.append( + f" post flagged rate: {r['post_flagged_rate']} mean max P(AI): {r['post_mean_max']} " + f"(rewrote {r['rewrote']}/{r['n']})" + ) + lines.append("") + lines.append(" per-detector mean P(AI) before -> after:") + for k, before in sorted(r["per_detector_pre"].items()): + after = (r["per_detector_post"] or {}).get(k) + lines.append(f" {k:24} {before} -> {after}") + else: + lines.append("") + lines.append( + " No rewriter configured (no ANTHROPIC_API_KEY / OPENAI_API_KEY, and not in the skill) " + "— showing BASELINE detection only. Set a key, or run inside the /untell skill where " + "Claude is the rewriter, to measure the after-rewrite ceiling." + ) + return "\n".join(lines) + + +def _read_corpus(path: str) -> list[str]: + with open(path, encoding="utf-8") as fh: + raw = fh.read() + blocks = [b.strip() for b in raw.split("\n\n")] + return [b for b in blocks if b] + + +def main(argv: list[str] | None = None) -> int: + from untell.scripts.io_utils import configure_utf8_io + + configure_utf8_io() + parser = argparse.ArgumentParser(prog="untell-ceiling", description=__doc__) + parser.add_argument("--file", "-f", help="corpus file (paragraphs separated by blank lines)") + parser.add_argument("--tier", default="full", choices=["lite", "full", "heavy", "commercial"]) + parser.add_argument("--threshold", "-t", type=float, default=DEFAULT_THRESHOLD) + parser.add_argument("--max-iters", type=int, default=5) + parser.add_argument("--best-of", type=int, default=1) + parser.add_argument("--json", action="store_true") + args = parser.parse_args(argv) + + from untell._env import load_env + + load_env() + texts = _read_corpus(args.file) if args.file else _SAMPLE + if not texts: + print(json.dumps({"error": "empty corpus"})) + return 2 + result = measure_ceiling( + texts, tier=args.tier, threshold=args.threshold, max_iters=args.max_iters, best_of=args.best_of + ) + print(json.dumps(result, ensure_ascii=True, indent=2) if args.json else _render(result)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/pyproject.toml b/pyproject.toml index f426667..4e934f7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -92,6 +92,7 @@ untell-verify = "untell.scripts.verify:main" untell-prove = "eval.prove:main" untell-sentences = "untell.scripts.sentences:main" untell-mcp = "untell.mcp_server:main" +untell-ceiling = "eval.ceiling:main" untell-distill = "training.distill:main" untell-surrogate = "training.surrogate:main" diff --git a/tests/test_ceiling.py b/tests/test_ceiling.py new file mode 100644 index 0000000..57516be --- /dev/null +++ b/tests/test_ceiling.py @@ -0,0 +1,44 @@ +"""Inference-only ceiling measurement tests — offline (baseline + stub-rewriter delta).""" + +from __future__ import annotations + +import re + +from eval.ceiling import _SAMPLE, main, measure_ceiling + + +def test_baseline_without_rewriter(): + # No rewriter and no API key => baseline (pre) only; post is None but the run still succeeds. + r = measure_ceiling(_SAMPLE[:2], tier="lite", max_iters=2, rewriter=None) + assert r["n"] == 2 + assert r["rewriter_available"] is False + assert r["pre_flagged_rate"] is not None + assert r["post_flagged_rate"] is None + assert r["pre_mean_max"] is not None + + +def test_full_delta_with_stub_rewriter(): + class _RW: + name = "stub" + + def available(self): + return True + + def rewrite(self, text, score_result, threshold=0.30): + sentinels = re.findall(r"⟦HZ\d{4}⟧", text) + return "Plain, short, human line. " + " ".join(sentinels) + + r = measure_ceiling(_SAMPLE[:2], tier="lite", threshold=0.30, max_iters=2, rewriter=_RW()) + assert r["rewrote"] == 2 + assert r["rewriter_available"] is True + assert r["post_flagged_rate"] is not None + assert r["pre_mean_max"] is not None and r["post_mean_max"] is not None + assert isinstance(r["per_detector_pre"], dict) and r["per_detector_pre"] + + +def test_cli_smoke(capsys): + rc = main(["--tier", "lite", "--max-iters", "2"]) + assert rc == 0 + out = capsys.readouterr().out + assert "ceiling" in out.lower() + assert "flagged rate" in out.lower() diff --git a/tests/test_llm_judge.py b/tests/test_llm_judge.py new file mode 100644 index 0000000..a35c527 --- /dev/null +++ b/tests/test_llm_judge.py @@ -0,0 +1,43 @@ +"""LLM-as-judge detector tests — offline (no key => unavailable; completion mocked otherwise).""" + +from __future__ import annotations + +from untell.detectors.llm_judge import LLMJudgeDetector + + +def test_unavailable_without_key(monkeypatch): + monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + d = LLMJudgeDetector() + assert d.available() is False + assert d.score("some text") is None # unavailable => no signal (excluded from the ensemble) + + +def test_score_parses_number(monkeypatch): + d = LLMJudgeDetector() + monkeypatch.setattr(d, "available", lambda: True) + monkeypatch.setattr(d, "_complete", lambda prompt: "0.82") + assert d.score("text") == 0.82 + + +def test_score_handles_percentage(monkeypatch): + d = LLMJudgeDetector() + monkeypatch.setattr(d, "available", lambda: True) + monkeypatch.setattr(d, "_complete", lambda prompt: "I'd rate this 73") + assert d.score("text") == 0.73 # a percentage answer is normalized to [0,1] + + +def test_empty_and_unparseable_return_none(monkeypatch): + d = LLMJudgeDetector() + monkeypatch.setattr(d, "available", lambda: True) + monkeypatch.setattr(d, "_complete", lambda prompt: "0.5") + assert d.score(" ") is None # empty input => no signal + monkeypatch.setattr(d, "_complete", lambda prompt: "no idea") + assert d.score("real text") is None # no number in the reply => None, not a crash + + +def test_registered_in_commercial_tier(): + from untell.detectors.base import all_detectors + + names = {d.name for d in all_detectors()} + assert "llm_judge" in names diff --git a/tests/test_run.py b/tests/test_run.py index d7971d8..eb5e094 100644 --- a/tests/test_run.py +++ b/tests/test_run.py @@ -232,3 +232,30 @@ def test_loop_rejects_sentinel_dropping_rewrite(monkeypatch): # ...and rejected every time, so the locked facts survive into the final output. assert "Smith (2020)" in res["final"] assert "47%" in res["final"] + + +def test_best_of_n_draws_multiple_candidates_and_keeps_facts(monkeypatch): + import untell.scripts.run as run_mod + + calls = {"n": 0} + + class _MultiRW: + name = "multi" + + def available(self): + return True + + def rewrite(self, text, score_result, threshold=0.30): + import re + + calls["n"] += 1 + sentinels = re.findall(r"⟦HZ\d{4}⟧", text) + tail = (" " + " ".join(sentinels)) if sentinels else "" + return f"It shifted, and people noticed. Variant {calls['n']}.{tail}" + + monkeypatch.setattr(run_mod, "get_rewriter", lambda prefer=None: _MultiRW()) + # threshold=0.0 forces a rewrite; best_of=3 => exactly three candidates drawn in the one iteration. + res = untell_text(AI, tier="lite", threshold=0.0, max_iters=1, best_of=3) + assert "error" not in res + assert calls["n"] == 3 + assert "Smith (2020)" in res["final"] and "47%" in res["final"] # facts survive best-of selection diff --git a/untell/detectors/base.py b/untell/detectors/base.py index 693ff28..55f9c09 100644 --- a/untell/detectors/base.py +++ b/untell/detectors/base.py @@ -55,6 +55,7 @@ def all_detectors() -> list[Detector]: from .commercial import commercial_detectors from .fast_detectgpt import FastDetectGPTDetector from .hc3_roberta import HC3RobertaDetector + from .llm_judge import LLMJudgeDetector from .mage import MageDetector from .perplexity_burstiness import PerplexityBurstinessDetector from .radar import RadarDetector @@ -66,8 +67,9 @@ def all_detectors() -> list[Detector]: HC3RobertaDetector(), MageDetector(), FastDetectGPTDetector(), - RadarDetector(), # opt-in (HUMANIZE_ENABLE_RADAR=1); robust-to-paraphrase, non-commercial + RadarDetector(), # opt-in (UNTELL_ENABLE_RADAR=1); robust-to-paraphrase, non-commercial BinocularsDetector(), + LLMJudgeDetector(), # commercial tier: the frontier LLM as a detector (key-gated); strong free signal *commercial_detectors(), ] diff --git a/untell/detectors/llm_judge.py b/untell/detectors/llm_judge.py new file mode 100644 index 0000000..3b88c05 --- /dev/null +++ b/untell/detectors/llm_judge.py @@ -0,0 +1,94 @@ +"""LLM-as-judge detector — the frontier model scores AI-likelihood against the ai-tells catalog. + +The local proxy detectors are weak (they don't predict commercial detectors). But the skill's +*rewriter* is a frontier LLM, and that same capability makes a strong *detector*: ask it to rate how +AI-written the text reads, weighing the known tells (see ``references/ai-tells.md``). This is often +the best signal available for free — in the Claude skill, Claude plays this role directly per +SKILL.md; headless, it uses an Anthropic or OpenAI key. + +Key-gated (``commercial`` tier): ``available()`` is true only when a key + SDK is present, so with no +key it is simply absent from the ensemble. ``score`` returns ``None`` for empty/unavailable input and +lets API failures propagate (``score_text`` records them) — matching the other adapters' contract. +""" + +from __future__ import annotations + +import os +import re + +from .base import clamp01 + +_NUM = re.compile(r"\d*\.\d+|\d+") + +_JUDGE_PROMPT = ( + "You are an expert AI-text detector. Rate how likely the text below was written by an AI language " + "model, from 0.00 (clearly a human wrote it) to 1.00 (clearly AI-generated). Weigh the overall " + "read, not any single word, and consider the known AI tells: em-dashes; AI vocabulary (delve, " + "leverage, robust, seamless, tapestry, testament, pivotal, underscore, multifaceted, meticulous); " + "formulaic transitions (Moreover, Furthermore, Overall, In conclusion); rule-of-three / tricolons; " + "negated contrast (\"not X, it's Y\"); participial trailers (\"..., underscoring its importance\"); " + "inflated copula (serves as, boasts); vague attribution (\"studies show\"); uniform sentence length " + "(low burstiness); promotional register; sycophancy; over-structured markdown. " + "Ignore opaque sentinel tokens like ⟦HZ0003⟧ — treat them as neutral placeholders. " + "Respond with ONLY the number, e.g. 0.73" +) + + +class LLMJudgeDetector: + name = "llm_judge" + tier = "commercial" + + def __init__(self, model: str | None = None): + self.model = model + + def available(self) -> bool: + if os.environ.get("ANTHROPIC_API_KEY"): + try: + import anthropic # noqa: F401 + + return True + except Exception: + pass + if os.environ.get("OPENAI_API_KEY"): + try: + import openai # noqa: F401 + + return True + except Exception: + pass + return False + + def _complete(self, prompt: str) -> str: + """Return the model's raw completion (a tiny number). Anthropic preferred, then OpenAI.""" + if os.environ.get("ANTHROPIC_API_KEY"): + try: + import anthropic + except Exception: + anthropic = None + if anthropic is not None: + resp = anthropic.Anthropic().messages.create( + model=self.model or "claude-sonnet-4-6", + max_tokens=8, + messages=[{"role": "user", "content": prompt}], + ) + return "".join(getattr(b, "text", "") for b in resp.content) + import openai + + resp = openai.OpenAI().chat.completions.create( + model=self.model or "gpt-4o-mini", + max_tokens=8, + messages=[{"role": "user", "content": prompt}], + ) + return resp.choices[0].message.content or "" + + def score(self, text: str) -> float | None: + if not self.available() or not text.strip(): + return None + out = self._complete(f"{_JUDGE_PROMPT}\n\n--- TEXT ---\n{text}") + m = _NUM.search(out or "") + if not m: + return None + val = float(m.group(0)) + if val > 1.0: # the model answered as a percentage (e.g. "73") + val /= 100.0 + return clamp01(val) diff --git a/untell/scripts/run.py b/untell/scripts/run.py index c6b1ea6..51d185f 100644 --- a/untell/scripts/run.py +++ b/untell/scripts/run.py @@ -90,6 +90,7 @@ def untell_text( scrub: bool = True, polish: bool = False, style: str | None = None, + best_of: int = 1, ) -> dict: """Run the closed loop on ``text``; return a structured result dict. @@ -160,21 +161,31 @@ def _passed(s: dict) -> bool: } except Exception: pass - try: - candidate = rw.rewrite(best_masked, best_score, threshold) - except Exception as exc: # surface the failure rather than silently looping - return {"error": f"rewriter failed: {type(exc).__name__}: {str(exc)[:160]}", "final": restore(best_masked, mapping)} - rewrites += 1 - cand_score = score(candidate) - # Accept only if the rewrite (a) keeps EVERY sentinel intact — a dropped or altered sentinel - # would silently lose a locked citation/number/fact on restore, defeating the whole lock — - # (b) holds the meaning-similarity gate, and (c) does not worsen the detector max. - if ( - find_sentinels(candidate) == set(mapping) - and similarity(masked, candidate) >= sim_bar - and cand_score["max"] <= best_score["max"] - ): - best_masked, best_score = candidate, cand_score + # Best-of-N: draw `best_of` candidates this round and keep the strongest VALID one. A + # candidate is valid only if it (a) keeps EVERY sentinel intact — a dropped/altered sentinel + # would silently lose a locked citation/number on restore, defeating the whole lock — and + # (b) holds the meaning-similarity gate. Among the valid ones, pick the lowest detector max, + # and only adopt it if it does not worsen the running best. + cand_best, cand_best_score = None, None + drew = 0 + for _ in range(max(1, best_of)): + try: + candidate = rw.rewrite(best_masked, best_score, threshold) + except Exception as exc: # surface the failure rather than silently looping + if drew == 0: + return {"error": f"rewriter failed: {type(exc).__name__}: {str(exc)[:160]}", "final": restore(best_masked, mapping)} + break # a later draw failed; use the candidates we already have + drew += 1 + rewrites += 1 + if find_sentinels(candidate) != set(mapping): + continue # dropped/altered a locked span — reject outright + cscore = score(candidate) + if similarity(masked, candidate) >= sim_bar and ( + cand_best_score is None or cscore["max"] < cand_best_score["max"] + ): + cand_best, cand_best_score = candidate, cscore + if cand_best is not None and cand_best_score["max"] <= best_score["max"]: + best_masked, best_score = cand_best, cand_best_score if _passed(best_score): stopped = "passed" break @@ -283,6 +294,13 @@ def main(argv: list[str] | None = None) -> int: choices=["casual", "professional", "academic", "blunt", "storytelling", "journalistic"], help="bias the rewrite toward a writing style/voice", ) + parser.add_argument( + "--best-of", + type=int, + default=1, + help="draw N candidate rewrites per iteration and keep the best valid one (sentinels intact + " + "meaning gate, lowest detector max). Default 1.", + ) parser.add_argument("--json", action="store_true", help="emit the full result as JSON") args = parser.parse_args(argv) @@ -309,6 +327,7 @@ def main(argv: list[str] | None = None) -> int: scrub=not args.no_scrub, polish=args.polish, style=args.style, + best_of=args.best_of, ) if args.json: print(json.dumps(result, ensure_ascii=True, indent=2))