| """cli - Oracle7 from the terminal. Everything the Workbench page does that makes sense in a terminal, with the same code, |
| so an answer here is the answer the page gives. `python oracle7.py help` lists the commands. |
| |
| organs which capabilities this install carries, 4.8 to 10.0 |
| lesson "question" 7.5 a physics answer opened into a checked lesson (steps, worked example, practice) |
| compute [length] 5.5 the composition space's units, counted exactly and decoded by sampling |
| grow [rounds] [questions] 6.0 it builds parts from the questions it could not answer (your own workspace copy) |
| packet publish <id> | verify <file> 6.5 parts between places; every check runs at the receiver |
| place 7.0 the Slow Fire rules on a simulated oak (greeting, consent, sensors, tampering) |
| discover 8.5 hypotheses written from the formula library before any data is fetched |
| ledger ask "question" | ledger replay 9.5 answers kept against a content-addressed state, replayed as of then |
| air "question" 10.0 the air's carbon from NOAA's records, cited, computed two ways |
| audit <ledger> [--reread] 10.0 re-verify a conversation ledger without trusting it |
| ask "question" [--json] one question: the answer, its status and the path it took (your specialists included) |
| chat a conversation (follow-ups work); /help, /path, /new, /quit inside |
| specialists [word] the 267 admitted specialists (and yours), with held-out scores |
| card <id> one specialist's card |
| systems the five systems built the same way |
| verify [--chain] check this install against RELEASE_HASHES.json (and the root against its Solana anchor) |
| mine your specialists and their state |
| new <id> [words...] scaffold a specialist in workbench/local/specialists/<id> |
| example install the worked example (habitable zone) |
| exam <id> dev|heldout run its exam (held-out: once) and print the verdict |
| update [--apply] check Hugging Face for a newer anchored release of this install; --apply downloads the |
| changed parts, checks each against the release's hashes, and writes them (your |
| specialists, memory and conversations are kept) |
| swarm point <target.json> point the swarm at something: it builds, examines, represents and combines parts, then does |
| the work (ask / plan / code); also: swarm parts, swarm paths (swarm/point.py) |
| build "q1" "q2" ... [--id x] build a specialist from 3-6 example questions: the swarm works out what they need, names |
| the gap, and (for a table of named things) builds it from Wikidata; then run its exam |
| build --file questions.txt the same, one example question per line |
| compose --file examples.txt [--id x] compose a specialist from 8+ worked examples (a formula): inputs and output with |
| units, one row per line "1.050, 1.010 -> 5.25"; half write the program, half examine it |
| memory [export <file> | import <file>] your group's shared specialist memory (4.2): admitted specialists, filed on the |
| Tree of Life spine; builds recall from it first; imports are checked against their chain |
| sensor-ideal <id> <quantity> <low> <high> the ideal band for one of a sensor's quantities |
| sensor-pair <id> <quantity a> <quantity b> declare two of its quantities physically coupled (checked on their history) |
| sensor-history <id> <file.csv> import past readings (a time column, one column per quantity) for the baseline reader |
| scan read this computer (system, CPU, memory, disks, uptime, top processes) |
| sensors your sensors: read each once now and print the newest values |
| sensor-test <kind> <address> read a sensor once and list the numbers found (kinds: http mqtt serial file simulated) |
| history [id] saved conversations, or one conversation's turns |
| essay "topic" [options] ProseFold: an essay composed from this model's own records, no language model; options |
| --length short|medium|long|full --level young|general|advanced --depth basic|standard|technical |
| --style plain|encyclopedic|textbook|prometheus7 --voice <your writing.txt> |
| registry <id> a specialist's on-chain registrations and attestations, with each signer's $ORACLE7 |
| registry-memo <id> register|attest the memo to sign (in the Workbench's Share tab, or any wallet that signs memos); |
| then: registry-record <id> register|attest <signature> |
| serve [port] | workbench [port] | setup | help |
| Add --json to ask, specialists, verify, mine, scan and sensors for machine-readable output.""" |
| from __future__ import annotations |
|
|
| import json |
| import re |
| import sys |
| import time |
| from pathlib import Path |
|
|
| HERE = Path(__file__).resolve().parent |
| sys.path.insert(0, str(HERE)) |
| STAGE = {"L0": "Intake", "L1": "Parse", "L2": "Route", "L3": "Conversation", "L4": "Speak & verify", "L5": "Self-model", |
| "S": "Specialists", "E": "Engines", "V": "Knowledge vault"} |
| COMMANDS = {"organs", "lesson", "compute", "grow", "packet", "place", "discover", "ledger", "air", "audit", "update", "swarm", "ask", "chat", "specialists", "card", "systems", "verify", "mine", "new", "example", "exam", "scan", "sensors", |
| "sensor-test", "history", "help", "registry", "registry-memo", "essay", "build", "compose", "memory", "sensor-ideal", |
| "sensor-pair", "sensor-history"} |
|
|
|
|
| def out(obj, as_json, text): |
| print(json.dumps(obj, indent=1, ensure_ascii=False, default=str) if as_json else text) |
|
|
|
|
| def path_line(steps): |
| return " -> ".join("%s: %s" % (STAGE.get(s, s), m) if m else STAGE.get(s, s) for s, m in steps) |
|
|
|
|
| class Session: |
| """the Workbench's own answering (core + your specialists, readers first when addressed, history, help)""" |
|
|
| def __init__(self, engine): |
| import server |
| self.server = server |
| server.WB = server.Workbench(engine) |
| self.wb = server.WB |
|
|
| def ask(self, text, sid): |
| r = self.wb.chat(text, sid) |
| st = r.get("status") |
| why = None if st == "answered" else self.server.why_declined(r.get("path"))[1] |
| steps = self.server.trace(r.get("path"), r.get("response"), st) |
| if self.server.settings().get("keep_conversations") and self.server.CID.match(sid): |
| self.server.conv_add(sid, {"t": int(time.time()), "q": text, "status": st, "response": str(r.get("response", "")), |
| "path": str(r.get("path", "")), "trace": steps, "why": why, "ms": 0}) |
| return {"status": st, "response": str(r.get("response", "")), "path": str(r.get("path", "")), "why": why, "trace": steps} |
|
|
|
|
| def show_answer(r, as_json): |
| if as_json: |
| return out(r, True, "") |
| print(r["response"].strip() or "(no answer)") |
| if r["why"]: |
| print(" why: %s" % r["why"]) |
| print(" [%s] %s" % (r["status"], path_line(r["trace"]))) |
|
|
|
|
| def specialists_data(): |
| import server |
| return server.specialists_payload() |
|
|
|
|
| def main(argv, engine_factory): |
| a = list(argv) |
| as_json = "--json" in a |
| a = [x for x in a if x != "--json"] |
| cmd, rest = a[0], a[1:] |
| if cmd == "help": |
| print(__doc__.split("\n\n", 1)[1]) |
| return 0 |
| if cmd == "specialists": |
| d = specialists_data() |
| w = " ".join(rest).lower() |
| items = [x for x in d["items"] if not w or w in (x["name"] + " " + x["id"] + " " + " ".join(x["takes"])).lower()] |
| lines = ["%-34s %-17s %s" % (x["id"][:34], x["group"][:17], ("%s/%s vs core %s" % (x["heldout"][0], x["heldout"][2], x["heldout"][1])) if x["heldout"] else "") |
| for x in items] |
| lines += ["%-34s %-17s %s" % (y["id"][:34], "yours", y["state"]) for y in d["yours"] if not w or w in y["id"]] |
| out({"items": items, "yours": d["yours"]}, as_json, "\n".join(lines) + "\n%d specialists%s" % (len(items), " matching '%s'" % w if w else "")) |
| return 0 |
| if cmd == "card": |
| if not rest: |
| print("usage: card <id>") |
| return 2 |
| f = HERE / "data" / "cards" / (rest[0] + ".md") |
| if f.exists(): |
| print(f.read_text(encoding="utf-8")) |
| return 0 |
| x = next((x for x in specialists_data()["items"] if x["id"] == rest[0]), None) |
| if not x: |
| print("no specialist %s (try: specialists <word>)" % rest[0]) |
| return 1 |
| print("%s\n%s 路 takes: %s\nheld-out: %s" % (x["name"], x["group"], ", ".join(x["takes"]), x["heldout"])) |
| return 0 |
| if cmd == "systems": |
| s = json.loads((HERE / "data" / "systems.json").read_text(encoding="utf-8")) |
| out(s, as_json, "\n\n".join("%s (%s)\n%s\n%s" % (x["name"], x["id"], x["does"], "\n".join(" - " + m for m in x["measured"])) for x in s["items"])) |
| return 0 |
| if cmd == "verify": |
| import verify |
| v = verify.verify_install() |
| ok = v["root_matches"] and not v["changed"] |
| txt = "%s: %d of %d parts, %d specialist folders\npublished root %s\ncomputed root %s" % ( |
| "VERIFIED" if ok else "DOES NOT MATCH", v["verified"], v["parts"], v["specialists_verified"], v["published_root"], v["computed_root"]) |
| if v["changed"]: |
| txt += "\nchanged: " + ", ".join(v["changed"]) |
| if v["missing"]: |
| txt += "\nmissing: " + ", ".join(v["missing"]) |
| if "--chain" in rest: |
| try: |
| c = verify.check_anchor() |
| v["anchor_check"] = c |
| txt += "\nSolana anchor: %s (slot %s) %s" % ("root is on-chain" if c["root_on_chain"] else "root NOT in that transaction", c["slot"], c["explorer"]) |
| except Exception as e: |
| txt += "\nSolana anchor: %s" % e |
| out(v, as_json, txt) |
| return 0 if ok else 1 |
| if cmd == "mine": |
| d = specialists_data()["yours"] |
| out(d, as_json, "\n".join("%-36s %-9s %s" % (y["id"], y["state"], y.get("note", "")) for y in d) or "none yet (try: new <id>, example, or the Workbench)") |
| return 0 |
| if cmd == "new": |
| import server |
| if not rest or not server.ID.match(rest[0]): |
| print("usage: new <id> [words...] (id: lower-case letters, digits, hyphens)") |
| return 2 |
| d = server.scaffold(rest[0], rest[0], [w.lower() for w in rest[1:]]) |
| print("made %s\nedit %s, write held-out items in panel.json, then: exam %s dev" % (d, d / ("organ_%s.py" % rest[0].replace("-", "_")), rest[0])) |
| return 0 |
| if cmd == "build": |
| sys.path.insert(0, str(HERE / "swarm")) |
| import run as swarm_run |
| sid = None |
| anyway = "--anyway" in rest |
| rest = [x for x in rest if x != "--anyway"] |
| if "--id" in rest: |
| i = rest.index("--id") |
| sid = rest[i + 1] if i + 1 < len(rest) else None |
| rest = rest[:i] + rest[i + 2:] |
| if rest[:1] == ["--file"] and len(rest) > 1: |
| qs = [ln.strip() for ln in Path(rest[1]).read_text(encoding="utf-8").splitlines() if ln.strip()] |
| else: |
| qs = rest |
| if len(qs) < 3: |
| print('usage: build "question 1" "question 2" "question 3" ... [--id name] (or: build --file questions.txt)') |
| return 2 |
| print("Asking the swarm about %d example questions..." % len(qs)) |
| r = swarm_run.build(qs, sid, anyway=anyway) |
| b = r.get("brief") or {} |
| if b and b.get("outcome") not in ("refuse",): |
| print(" outcome %s 路 type %s 路 neighbours %s" % (b.get("outcome"), b.get("type"), ", ".join(b.get("neighbors") or []) or "none")) |
| if b.get("gap"): |
| print(" gap %s" % b["gap"]) |
| print(r["message"]) |
| if as_json: |
| print(json.dumps({k: v for k, v in r.items() if k != "build"}, indent=1, ensure_ascii=False, default=str)) |
| return 0 if r["outcome"] != "error" else 2 |
| if cmd == "memory": |
| sys.path.insert(0, str(HERE / "swarm")) |
| import memory as mem |
| if rest[:1] == ["export"] and len(rest) > 1: |
| print("exported %d shareable entries to %s" % (mem.export(rest[1]), rest[1])) |
| return 0 |
| if rest[:1] == ["import"] and len(rest) > 1: |
| try: |
| n, name = mem.import_shared(rest[1]) |
| except ValueError as e: |
| print(str(e)) |
| return 2 |
| print("imported %d entries as the shared memory '%s' (every chain hash checked)" % (n, name)) |
| return 0 |
| d = mem.listing() |
| if as_json: |
| print(json.dumps(d, indent=1, ensure_ascii=False)) |
| return 0 |
| print("your group: %s, %d admitted specialists in its memory" % (d["group"], len(d["entries"]))) |
| for e in d["entries"]: |
| print(" %-34s %-9s %s" % (e["specialist"], e["node"], "shared" if e["shared"] else "private")) |
| print("shared memories imported: %s" % (", ".join(d["shared_memories"]) or "none")) |
| print("memory export <file> | memory import <file>") |
| return 0 |
| if cmd == "compose": |
| sys.path.insert(0, str(HERE / "swarm")) |
| import run as swarm_run |
| sid = None |
| anyway = "--anyway" in rest |
| rest = [x for x in rest if x != "--anyway"] |
| if "--id" in rest: |
| i = rest.index("--id") |
| sid = rest[i + 1] if i + 1 < len(rest) else None |
| rest = rest[:i] + rest[i + 2:] |
| if rest[:1] != ["--file"] or len(rest) < 2: |
| print("usage: compose --file examples.txt [--id name]\n\nexamples.txt:\n title: Alcohol by volume\n" |
| " inputs: original gravity (SG), final gravity (SG)\n output: ABV (%)\n 1.050, 1.010 -> 5.25\n ... (8 or more rows)") |
| return 2 |
| print("Composing a program from your worked examples (up to a minute)...") |
| r = swarm_run.compose(Path(rest[1]).read_text(encoding="utf-8"), sid, anyway=anyway) |
| print(r["message"]) |
| if r.get("program") and not as_json: |
| print("\n" + r["program"]) |
| if as_json: |
| print(json.dumps(r, indent=1, ensure_ascii=False, default=str)) |
| return 0 if r["outcome"] == "composed" else (1 if r["outcome"] == "declined" else 2) |
| if cmd == "example": |
| import shutil |
| import server |
| dst = server.SPECS / "habitable-zone" |
| if dst.exists(): |
| print("already installed: %s" % dst) |
| return 0 |
| shutil.copytree(HERE / "examples" / "habitable-zone", dst) |
| print("installed %s\nnext: exam habitable-zone dev, then exam habitable-zone heldout" % dst) |
| return 0 |
| if cmd == "scan": |
| import sources |
| rd = sources.scan_local() |
| out(rd, as_json, "\n".join("%-22s %s [%s]" % ("%s %s" % (r["entity"], r["signal"]), r["text"], r["source"]) for r in rd)) |
| return 0 |
| if cmd in ("sensor-ideal", "sensor-pair", "sensor-history"): |
| import sensors |
| cfgs = sensors.configs() |
| if not rest or rest[0] not in cfgs: |
| print("usage: %s <sensor id> ...; sensors: %s" % (cmd, ", ".join(sorted(cfgs)) or "none connected")) |
| return 2 |
| c = cfgs[rest[0]] |
| qs = {f["quantity"] for f in c["fields"]} |
| if cmd == "sensor-ideal": |
| if len(rest) < 4 or rest[1] not in qs: |
| print("usage: sensor-ideal <id> <quantity> <low> <high>; quantities: %s" % ", ".join(sorted(qs))) |
| return 2 |
| lo, hi = float(rest[2]), float(rest[3]) |
| c.setdefault("ideal", {})[rest[1]] = [min(lo, hi), max(lo, hi)] |
| sensors.save_config(c) |
| print("ideal band for %s on %s: %s to %s" % (rest[1], rest[0], min(lo, hi), max(lo, hi))) |
| return 0 |
| if cmd == "sensor-pair": |
| if len(rest) < 3 or rest[1] not in qs or rest[2] not in qs or rest[1] == rest[2]: |
| print("usage: sensor-pair <id> <quantity a> <quantity b>; quantities: %s" % ", ".join(sorted(qs))) |
| return 2 |
| c.setdefault("pairs", []) |
| if [rest[1], rest[2]] not in c["pairs"]: |
| c["pairs"].append([rest[1], rest[2]]) |
| sensors.save_config(c) |
| print("declared %s and %s on %s a coupled pair; the reader checks the coupling on their history before it uses it" |
| % (rest[1], rest[2], rest[0])) |
| return 0 |
| if len(rest) < 2: |
| print("usage: sensor-history <id> <file.csv> (a time column, one column per quantity)") |
| return 2 |
| try: |
| n = sensors.import_history(rest[0], rest[1]) |
| except ValueError as e: |
| print(str(e)) |
| return 2 |
| print("imported %d 15-minute readings into %s's history" % (n, rest[0])) |
| return 0 |
| if cmd == "sensor-test": |
| import sensors |
| if len(rest) < 1: |
| print("usage: sensor-test <http|mqtt|serial|file|simulated> <address>") |
| return 2 |
| vals, where = sensors.read_source({"kind": rest[0], "address": rest[1] if len(rest) > 1 else ""}) |
| out({"source": where, "fields": vals}, as_json, "read from %s\n%s" % (where, "\n".join(" %-30s %-12s guess: %s" % (k, v, sensors.guess_quantity(k)) for k, v in vals.items()))) |
| return 0 |
| if cmd == "sensors": |
| import sensors |
| res = [] |
| for sid, c in sorted(sensors.configs().items()): |
| try: |
| rows = sensors.poll_once(c) |
| res.append({"id": sid, "name": c["name"], "rows": rows}) |
| except Exception as e: |
| res.append({"id": sid, "name": c["name"], "error": str(e)}) |
| txt = "\n".join("%s: %s" % (x["name"], x.get("error") or "; ".join("%s %s %s%s" % (r["q"], r["v"], r["unit"], "" if r["ok"] else " (REJECTED: %s)" % r["why"]) |
| for r in x["rows"])) for x in res) |
| out(res, as_json, txt or "no sensors (connect one in the Workbench, Sensors tab)") |
| return 0 |
| if cmd in ("registry", "registry-memo", "registry-record"): |
| import registry |
| import server |
| if not rest or not (server.SPECS / rest[0]).is_dir(): |
| print("usage: registry <id> | registry-memo <id> register|attest | registry-record <id> register|attest <signature>") |
| return 2 |
| d = server.SPECS / rest[0] |
| if cmd == "registry-memo": |
| print(registry.memo_for(d, rest[1] if len(rest) > 1 else "register")) |
| return 0 |
| if cmd == "registry-record": |
| recs = registry.record(d, rest[1], rest[2]) |
| print("recorded: %d record(s) for %s" % (len(recs), rest[0])) |
| return 0 |
| c = registry.check(d) |
| txt = "%s: %s, %d attestation(s), vouching wallets hold %s $ORACLE7\n" % ( |
| c["id"], "registered" if c["registered"] else "not registered", c["attestations"], c["vouching_oracle7"]) |
| txt += "\n".join(" %-8s %s %s %s %s" % (r.get("kind"), (r.get("signer") or "")[:12] + "...", time.strftime("%Y-%m-%d %H:%M", time.gmtime(r["time"])) if r.get("time") else "", |
| "matches" if r.get("matches_files") else ("ERROR " + r["error"] if r.get("error") else "DOES NOT MATCH"), |
| "%s $ORACLE7" % r.get("oracle7", "")) for r in c["records"]) |
| out(c, as_json, txt) |
| return 0 |
| if cmd == "history": |
| import server |
| if rest: |
| c = server.conv_load(rest[0]) |
| print("\n".join("> %s\n %s [%s]" % (t["q"], str(t.get("response", ""))[:300].replace("\n", " "), t.get("status")) for t in c["turns"]) or "empty") |
| else: |
| print("\n".join("%s %s %d turns %s" % (c["id"], time.strftime("%Y-%m-%d %H:%M", time.localtime(c["updated"])), c["turns"], c["title"]) |
| for c in server.conv_list()) or "no saved conversations") |
| return 0 |
| if cmd == "update": |
| import update |
| try: |
| update.local() |
| except ValueError as e: |
| print("update: %s" % e) |
| return 1 |
| if "--apply" in rest: |
| r = update.apply() |
| print(json.dumps(r, indent=1) if as_json else "%s\n%s" % (r.get("note", ""), ("UPDATED to %s; verified %s. %s" % ( |
| r["published_root"][:16], r.get("verified"), r["why"])) if r.get("applied") else "not applied: %s" % r.get("why"))) |
| return 0 if r.get("applied") or r.get("up_to_date") else 1 |
| c = update.check() |
| print(json.dumps(c, indent=1) if as_json else "this install %s %s\npublished %s %s (%s)\n%s%s" % ( |
| c["this_root"][:16], c["this_head"], c["published_root"][:16], c["published_head"], |
| "anchored %s" % c["published_anchored"] if c["published_signed"] else "not anchored", c["note"], |
| "\nchanged parts: %s\nrun: python oracle7.py update --apply" % ", ".join(c["changed"][:12]) if c["newer"] else "")) |
| return 0 |
| if cmd in ("organs", "lesson", "compute", "grow", "packet", "place", "discover", "ledger", "air", "audit"): |
| sys.path.insert(0, str(HERE / "swarm")) |
| import cli10 |
| return cli10.main(cmd, rest, as_json) |
| |
| eng = engine_factory() |
| if cmd == "swarm": |
| sys.path.insert(0, str(HERE / "swarm")) |
| import point |
| return point.main(rest, eng, as_json) |
| if cmd == "essay": |
| import prosefold.options as PO |
| opt = {"length": "medium", "level": "general", "depth": "standard", "style": "plain"} |
| voice = None |
| for flag in ("--length", "--level", "--depth", "--style", "--voice"): |
| if flag in rest: |
| i = rest.index(flag) |
| val = rest[i + 1] if i + 1 < len(rest) else "" |
| rest = rest[:i] + rest[i + 2:] |
| if flag == "--voice": |
| voice = val |
| else: |
| opt[flag[2:]] = val |
| allowed = {"length": PO.LENGTHS, "level": PO.LEVELS, "depth": PO.DEPTHS, "style": PO.STYLES} |
| bad = [k for k, v in opt.items() if v not in allowed[k]] |
| if not rest or bad: |
| print('usage: essay "topic" [--length short|medium|long|full] [--level young|general|advanced]\n' |
| ' [--depth basic|standard|technical] [--style plain|encyclopedic|textbook|prometheus7] [--voice file]') |
| return 2 |
| r = PO.write(eng.chat, " ".join(rest), voice_path=voice, **opt) |
| if as_json: |
| return out(r, True, "") or 0 |
| print(r["essay_md"]) |
| print("[prosefold] %d sentences from %d records, %d of %d checked word for word against their sources; reading grade %s " |
| "(asked %s%s); no language model" % (r["sentences"], r["records_used"], r["faithful"], r["sentences"], r["reading_grade"], |
| opt["level"], "" if r["level_reached"] else ", the record could not get closer")) |
| return 0 if r["sentences"] else 1 |
| sess = Session(eng) |
| if cmd == "ask": |
| if not rest: |
| print('usage: ask "question" [--json]') |
| return 2 |
| show_answer(sess.ask(" ".join(rest), "cli%d" % int(time.time())), as_json) |
| return 0 |
| if cmd == "exam": |
| import exam |
| import server |
| if len(rest) < 2 or rest[1] not in ("dev", "heldout"): |
| print("usage: exam <id> dev|heldout") |
| return 2 |
| d = server.SPECS / rest[0] |
| if not d.is_dir(): |
| print("no specialist %s in %s" % (rest[0], server.SPECS)) |
| return 1 |
| if rest[1] == "heldout" and input("The held-out set is used once. Run it? [y/N] ").strip().lower() != "y": |
| return 1 |
| v = exam.run(d, rest[1], eng) |
| print("%s EXAM %s: %s" % (rest[1].upper(), rest[0], v["decision"])) |
| print(" domain %d/%d with it, %d/%d core alone (gain needed %d)" % (v["with_specialist"]["ok"], v["n"], v["head"]["ok"], v["n"], v["gain_needed"])) |
| print(" false yes %d (core %d) 路 wrong %d 路 lane %d/%d (cap %d) 路 self %s 路 median %.1f ms" % ( |
| v["with_specialist"]["fy"], v["head"]["fy"], v["with_specialist"]["wrong"], v["core"]["taken"], v["core"]["n"], v["core"]["cap"], |
| "ok" if v["self_ok"] else "FAIL", v["p50_ms"])) |
| for r in v["reasons"]: |
| print(" - " + r) |
| for f in v["fails"][:8]: |
| print(" not right: %s | expected %s | got %s" % (f["q"][:60], f["expected"][:30], f["got"][:60])) |
| return 0 |
| if cmd == "chat": |
| sid = "cli%d" % int(time.time()) |
| print("Oracle7 (%s). Ask anything. /path shows the last path, /new starts over, /help lists Workbench features, /quit leaves." % eng.tip) |
| last = None |
| while True: |
| try: |
| q = input("> ").strip() |
| except (EOFError, KeyboardInterrupt): |
| print() |
| return 0 |
| if not q: |
| continue |
| if q in ("/quit", "/exit"): |
| return 0 |
| if q == "/new": |
| sid = "cli%d" % int(time.time()) |
| print("(new conversation)") |
| continue |
| if q == "/path": |
| print(path_line(last["trace"]) if last else "(nothing asked yet)") |
| continue |
| last = sess.ask(q, sid) |
| print(last["response"].strip() or "(no answer)") |
| if last["why"]: |
| print(" why: %s" % last["why"]) |
| print("unknown command %s; try: help" % cmd) |
| return 2 |
|
|
