File size: 7,814 Bytes
97c39f2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 | """TinyLiquid Analyst - headless terminal client (also scripts/SSH-friendly).
Usage:
.venv/bin/python tui/cli.py --ckpt ckpt/dpo # interactive REPL
.venv/bin/python tui/cli.py --ckpt ckpt/dpo --once "Verify: ..."
echo "Verify: ..." | .venv/bin/python tui/cli.py --ckpt ckpt/dpo
"""
import argparse
import sys
from tui.engine import AnalystEngine, parse_cmd, BANNER
HELP = """commands:
<text> chat with the current persona
/case <sop> <text> SOP-driven analysis (verdict + confidence + skeptic)
/skeptic <text> skeptic attack on a claim
/search <query> search the library/corpus index
/read <key> read a document from the index
/web <topic> search the live web (news/archive/wiki)
/fetch <url> pull one URL into the library
/pull <topic> search + pull top docs into the library
/tor on|off|status route .onion fetches through local Tor SOCKS
/sop [name] show one or all procedures
/persona <analyst|skeptic|none>
/rag on|off retrieval-grounded chat (auto search + context)
/memory on|off multi-turn conversation memory
/mem on|off|stats|<claim> DNA-helix long-term memory (prior-case recall)
/agent <task> run the procedural agent loop (search/read/verdict)
/agents <task> parallel research swarm: 4 agents, merged report
/synth [title] render case ledger into saved research documents
/chart <t>|<l>:<v,..>|... series chart + saved artifact (cross-synthesis)
/opinion <claim> dual-mind fusion: the model's calibrated opinion
/research <question> research-partner loop: pull, verify, opine, reply
/journal [name] full journalism suite -> CaseFile notebook
/gaps list the case's open questions (gap ledger)
/save <name> save this case
/load <name> load a case
/status show state
/help this help
/quit exit"""
def run_once(eng, text):
if not text.strip():
return
if text.startswith("/case"):
rest = text[5:].strip()
sop = ""
if rest and " " in rest[:60]:
maybe, body = rest.split(maxsplit=1)
if maybe in eng.__class__.__module__: # noqa - simple guard
pass
sop = maybe
text = body
elif rest:
text = rest
rep = eng.analyze(text, sop or None)
print(json_dump(rep))
elif text.startswith("/"):
print(handle_cmd(eng, text))
else:
print(eng.chat(text))
def json_dump(rep):
import json
return json.dumps(rep, indent=2, ensure_ascii=False)
def handle_cmd(eng, line):
cmd, arg = parse_cmd(line) or ("chat", line)
if cmd == "chat":
return eng.chat(arg)
if cmd == "case":
sop = ""
body = arg
if " " in body:
maybe, rest = body.split(maxsplit=1)
if maybe in {"claim_verification", "cross_source_discrepancy", "pattern_finding",
"timeline_reconstruction", "historical_truth", "politics_analysis",
"dark_web_research", "terminal_control", "source_triage"}:
sop, body = maybe, rest
return json_dump(eng.analyze(body, sop or None))
if cmd == "skeptic":
return eng.generate(arg, persona="skeptic")
if cmd == "search":
return "\n".join(f"{k} (score {s:.2f})" for k, s in eng.search(arg)) or "(no hits)"
if cmd == "read":
return eng.read(arg)
if cmd == "sop":
return eng.sop_text(arg) if arg else "\n".join(list_sops_str())
if cmd == "persona":
if arg in eng.PERSONA_ID if hasattr(eng, "PERSONA_ID") else arg in ("analyst", "skeptic", "none"):
eng.persona = arg
return f"persona -> {arg}"
return "persona: analyst|skeptic|none"
if cmd == "save":
return "saved " + eng.save_case(arg or "default")
if cmd == "load":
return eng.load_case(arg or "default")
if cmd == "rag":
eng.rag = arg not in ("off", "0", "false")
return f"retrieval-grounded chat: {'on' if eng.rag else 'off'}"
if cmd == "memory":
eng.memory = arg not in ("off", "0", "false")
return f"multi-turn memory: {'on' if eng.memory else 'off'}"
if cmd == "mem":
if arg in ("on", "1", "true", "yes"):
eng.helix_on = True
return "long-term helix memory: on"
if arg in ("off", "0", "false", "no"):
eng.helix_on = False
return "long-term helix memory: off"
if arg == "stats":
return json_dump(eng.mem_stats())
return eng.recall(arg)
if cmd == "agent":
return json_dump(eng.agent(arg))
if cmd == "agents":
return json_dump(eng.agents(arg, n=4))
if cmd == "synth":
return eng.synthesize(arg or None)
if cmd == "chart":
return eng.chart(arg)
if cmd == "opinion":
return json_dump(eng.opinion(arg))
if cmd == "research":
return json_dump(eng.research(arg))
if cmd == "journal":
return eng.journal(arg or None)
if cmd == "gaps":
return "\n".join(f" {g}" for g in eng.gaps())
if cmd == "web":
hits = eng.web_search(arg, n=8)
if not hits:
return "(no web hits)"
return "\n".join("[{0}] {1}\n {2}\n {3}".format(
r["source"], r["title"], r["url"], r["snippet"][:100]) for r in hits)
if cmd == "fetch":
return json_dump(eng.web_fetch(arg))
if cmd == "pull":
return json_dump(eng.web_pull(arg, n=3))
if cmd == "tor":
if arg in ("on", "1", "true", "yes"):
eng.tor = True
return "tor: on"
if arg in ("off", "0", "false", "no"):
eng.tor = False
return "tor: off"
return "tor: {0} | {1}".format(eng.tor, eng.tor_status_msg())
if cmd == "status":
return (f"persona={eng.persona} memory={'on' if eng.memory else 'off'} "
f"helix={'on' if getattr(eng, 'helix_on', False) else 'off'} "
f"rag={'on' if eng.rag else 'off'} case={eng.case.name} "
f"msgs={len(eng.case.chat)} notes={len(eng.case.ledger)} ckpt={eng.ckpt}")
if cmd in ("help", "?"):
return HELP
if cmd in ("quit", "exit"):
raise SystemExit(0)
return f"unknown command /{cmd}; try /help"
def list_sops_str():
from research.agent import SOP_ALIASES
return [f" {k:28s} {', '.join(SOP_ALIASES[k][:3])}" for k in SOP_ALIASES]
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--ckpt", default="ckpt/v8_lora/best.pt")
ap.add_argument("--tok", default="data/tokenizer.json")
ap.add_argument("--once", default=None)
ap.add_argument("--threads", type=int, default=4)
args = ap.parse_args()
eng = AnalystEngine(ckpt=args.ckpt, tok_path=args.tok, threads=args.threads)
print(BANNER, flush=True)
print(f"model: {eng.ckpt} | persona: analyst | type /help\n", flush=True)
if args.once:
run_once(eng, args.once)
return
if not sys.stdin.isatty():
for line in sys.stdin:
line = line.strip()
if not line:
continue
if line in ("/quit", "/exit"):
break
print(handle_cmd(eng, line), flush=True)
return
for line in sys.stdin:
line = line.strip()
if not line:
continue
try:
print(handle_cmd(eng, line), flush=True)
except SystemExit:
break
if __name__ == "__main__":
main()
|