"""`thx02`: THX decisions from any terminal, including Termux on Android. thx02 "I was charged twice, refund please" --choice team=billing,support,sales --yesno "Refund request?" thx02 "Income 4200, debts 2800, limit 50%. Pass?" --choice verdict=pass,fail --think on --thoughts thx02 --image receipt.jpg --yesno "Is the receipt legible?" thx02 --batch tickets.txt --choice team=billing,shipping > answers.jsonl thx02 interactive: type states, get decisions thx02 models models your key can use thx02 serve ... run your own THX-02 server (pip install "thx02[server]") The API key comes from --key or $HALX_API_KEY. --local runs the model on this machine (pip install "thx02[local]"). """ from __future__ import annotations import argparse import json import os import sys import time from . import __version__ from .client import APIError, Client, image, top RUBY, DIM, BOLD, RESET = "\033[38;2;226;23;102m", "\033[2m", "\033[1m", "\033[0m" def _color() -> bool: return sys.stdout.isatty() and not os.environ.get("NO_COLOR") def _options(spec: str, flag: str): if "=" not in spec: raise SystemExit(f"{flag} needs NAME=option1,option2 (got {spec!r})") name, opts = spec.split("=", 1) items = [o.strip() for o in opts.split(",") if o.strip()] if len(items) < 2: raise SystemExit(f"{flag} {name}: give at least two options") return name.strip(), items def build_questions(a) -> dict: qs = {} for spec in a.choice or []: name, items = _options(spec, "--choice") crit = {} for it in items: # "key:description" or just "key" k, _, d = it.partition(":") crit[k.strip()] = (d or k).strip() qs[name] = {"type": "choice", "instructions": a.instructions or f"Which {name} applies?", "criteria": crit} for spec in a.score or []: name, items = _options(spec, "--score") qs[name] = {"type": "score", "instructions": a.instructions or f"How would you rate the {name}?", "criteria": items} for i, text in enumerate(a.yesno or []): name = text.strip().rstrip("?")[:60] # the question itself labels its answer qs[name if name not in qs else f"{name} ({i + 1})"] = {"type": "noul", "instructions": text} if a.questions: raw = open(a.questions, encoding="utf-8").read() if os.path.exists(a.questions) else a.questions qs.update(json.loads(raw)) return qs or {"intent": {"type": "choice", "instructions": "What does the user want?", "criteria": { "question": "Asks a question", "complaint": "Complains about something", "request": "Asks for an action", "other": "Something else"}}} def show(out: dict, color: bool) -> None: r, d, b, x = (RUBY, DIM, BOLD, RESET) if color else ("", "", "", "") for name, a in out.get("answers", {}).items(): if a.get("type") == "noul": dist = {"yes": a["noul"], "no": 1 - a["noul"]} else: legend = a.get("legend") or {} dist = {legend.get(k, k): v for k, v in a.get("probabilities", {}).items()} best, _ = top(a) print(f"\n{b}{name}{x}") for k, p in sorted(dist.items(), key=lambda kv: -kv[1]): bar = "█" * round(p * 24) mark = f"{r}›{x}" if k == best else " " print(f" {mark} {k[:24]:24s} {r if k == best else d}{bar:24s}{x} {p * 100:5.1f}%") if a.get("thought"): print(f" {d}thought:{x} {a['thought']}") usage = out.get("usage") or {} extra = f" · {usage.get('thoughts')} thought(s)" if usage.get("thoughts") else "" print(f"\n{d}{out.get('model', '')} · {out.get('latency_ms', '?')} ms{extra}{x}") def _local_runner(path: str): os.environ.setdefault("THX02_DEVICE", "cpu") try: import torch from . import THX02 except ImportError: raise SystemExit('--local needs the model runtime: pip install "thx02[local]"') from None device = "cuda" if torch.cuda.is_available() and os.environ.get("THX02_DEVICE") != "cpu" else "cpu" print(f"loading THX-02 on {device} ...", file=sys.stderr) m = THX02.load(path, device=device) return lambda body: m.decide(body["state"], body["questions"], images=body.get("images"), think=False, tta=body.get("tta")) def main(argv=None) -> int: argv = list(sys.argv[1:] if argv is None else argv) if argv[:1] == ["serve"]: try: from .serve import main as serve_main except ImportError: raise SystemExit('thx02 serve needs the server extra: pip install "thx02[server]"') from None sys.argv = ["thx02-serve", *argv[1:]] return serve_main() or 0 ap = argparse.ArgumentParser(prog="thx02", description="THX decisions from the terminal (api.hal-x.ai or local).", epilog="Docs: https://api.hal-x.ai/docs/thx-02/") ap.add_argument("state", nargs="?", help='text to decide about, or "models"; omit for interactive mode') ap.add_argument("--choice", action="append", metavar="NAME=a,b,c", help="pick one option (a:description works too)") ap.add_argument("--yesno", action="append", metavar="QUESTION", help="yes / no question") ap.add_argument("--score", action="append", metavar="NAME=low,mid,high", help="ordered scale, lowest first") ap.add_argument("--instructions", "-i", help="question text for --choice / --score") ap.add_argument("--questions", "-q", help="questions as JSON (a string or a file)") ap.add_argument("--image", action="append", help="image file to include (up to 4)") ap.add_argument("--think", choices=["off", "auto", "on"], default="off", help="thinking mode (default off)") ap.add_argument("--thoughts", action="store_true", help="show the thoughts") ap.add_argument("--tta", type=int, choices=[1, 4], help="read choice options in 4 orders and average") ap.add_argument("--batch", metavar="FILE", help="one state per line (or JSONL with a state field); prints JSONL") ap.add_argument("--model", default=os.environ.get("THX_MODEL", "thx-02"), help="thx-02 (default) or thx-01") ap.add_argument("--key", help="API key (default $HALX_API_KEY)") ap.add_argument("--server", default=os.environ.get("HALX_BASE_URL"), help="API base URL (default https://api.hal-x.ai)") ap.add_argument("--local", metavar="MODEL", help="run on this machine: a THX-02 folder or HAL-X-AI/THX-02") ap.add_argument("--json", action="store_true", help="print raw JSON") ap.add_argument("--version", action="version", version=f"thx02 {__version__}") a = ap.parse_args(argv) color = _color() and not a.json client = Client(api_key=a.key, base_url=a.server, model=a.model) if a.state == "models" and not a.local: try: print("\n".join(client.models())) except APIError as e: print(f"error: {e}", file=sys.stderr) return 1 return 0 qs = build_questions(a) images = [image(f) for f in (a.image or [])][:4] think = {"off": False, "auto": "auto", "on": True}[a.think] run_local = _local_runner(a.local) if a.local else None def body_for(state): b = {"state": state, "questions": qs} if images: b["images"] = images if think: b["think"] = think if a.thoughts: b["return_thoughts"] = True if a.tta: b["tta"] = a.tta return b def ask(state) -> int: t0 = time.time() try: if run_local: out = run_local(body_for(state)) else: b = body_for(state) out = client.decide(b["state"], b["questions"], images=b.get("images"), think=think, return_thoughts=a.thoughts, tta=a.tta) except APIError as e: print(f"error: {e}", file=sys.stderr) if e.status == 401 and not client.api_key: print("Set your key: export HALX_API_KEY=sk-... (or --key). Docs: https://api.hal-x.ai/docs/thx-02/", file=sys.stderr) return 1 out.setdefault("latency_ms", round((time.time() - t0) * 1000)) if a.json: print(json.dumps(out, ensure_ascii=False, indent=2)) else: show(out, color) return 0 if a.batch: lines = [ln.strip() for ln in open(a.batch, encoding="utf-8") if ln.strip()] states = [json.loads(ln).get("state", "") if ln.startswith("{") else ln for ln in lines] for i in range(0, len(states), 256): chunk = states[i:i + 256] try: results = client.batch([body_for(s) for s in chunk]) except APIError as e: print(f"error: {e}", file=sys.stderr) return 1 for s, res in zip(chunk, results): print(json.dumps({"state": s, **res}, ensure_ascii=False)) return 0 if a.state is not None: return ask(a.state) if images: return ask("") if not sys.stdin.isatty(): return ask(sys.stdin.read().strip()) r, d, x = (RUBY, DIM, RESET) if color else ("", "", "") print(f"{r}THX-02{x} interactive · questions: {', '.join(qs)} · empty line to quit") while True: try: s = input(f"\n{d}state>{x} ").strip() except (EOFError, KeyboardInterrupt): print() break if not s: break ask(s) return 0 if __name__ == "__main__": sys.exit(main())