| """`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: |
| 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] |
| 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()) |
|
|
