File size: 3,740 Bytes
8b8e59d | 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 | """Dual-mind forensic analysis pipeline.
Analyst pass: follows the analysis SOP with a scratchpad.
Skeptic pass: attacks the analyst's conclusions.
Outputs a JSON report.
Usage:
.venv/bin/python research/analyst.py --file doc.txt --ckpt ckpt/forensic
cat doc.txt | .venv/bin/python research/analyst.py --ckpt ckpt/forensic
"""
import argparse
import json
import sys
from pathlib import Path
import torch
from model.config import TinyLiquidConfig, CONFIGS
from model.utils import latest_ckpt
from model.tiny_liquid import TinyLiquid
from data.tokenizer import load_tokenizer
SOP = (
"Follow the analysis protocol exactly. 1) Extract every checkable claim. "
"2) Separate evidence from assertion; name what is missing. "
"3) Compare accounts and flag contradictions, ambiguities, and overclaims. "
"4) Look for patterns across events: clustering, escalation, common cause. "
"5) State a verdict and a confidence for every conclusion; prefer "
"'cannot confirm' over speculation. Use the scratchpad before the final answer."
)
def parse_args():
ap = argparse.ArgumentParser()
ap.add_argument("--file", default=None)
ap.add_argument("--ckpt", default="ckpt/forensic")
ap.add_argument("--tok", default="data/tokenizer.json")
ap.add_argument("--max-new", type=int, default=220)
ap.add_argument("--threads", type=int, default=8)
return ap.parse_args()
def load_model(args):
torch.set_num_threads(args.threads)
tok = load_tokenizer(args.tok)
ckpt = latest_ckpt(args.ckpt)
assert ckpt, f"no checkpoints in {args.ckpt}"
sd = torch.load(ckpt, map_location="cpu")
cfg_dict = dict(sd.get("config", CONFIGS["tiny10m"]))
cfg = TinyLiquidConfig(vocab_size=tok.get_vocab_size(), **{k: v for k, v in cfg_dict.items() if k != "vocab_size"})
model = TinyLiquid(cfg)
model.load_state_dict(sd["model"])
model.eval()
return tok, model
def run(model, tok, persona, persona_id, user_text, max_new):
p_token = {"analyst": "<|analyst|>", "skeptic": "<|skeptic|>"}[persona]
prompt = p_token + "<|user|>" + user_text + "<|assistant|>"
ids = tok.encode(prompt).ids
out = model.generate(tok, ids, persona_id=persona_id, max_new=max_new,
temperature=0.6, top_k=40, repetition_penalty=1.4, no_repeat_ngram_size=4)
return tok.decode(out[len(ids):]).strip()
def main():
args = parse_args()
if args.file:
text = Path(args.file).read_text(encoding="utf-8", errors="ignore")
else:
text = sys.stdin.read()
text = text.strip()
assert text, "no input text"
tok, model = load_model(args)
doc = text if len(text) <= 1200 else text[:1200] + " [truncated]"
analyst_user = f"{SOP}\n\nMaterial under analysis:\n{doc}"
analyst = run(model, tok, "analyst", 1, analyst_user, args.max_new)
skeptic_user = (
"Act as the skeptic. Tear down the analysis below: find unsupported "
"conclusions, overclaims, weak sourcing, and alternative explanations. "
"Keep only what survives.\n\nAnalysis:\n" + analyst
)
skeptic = run(model, tok, "skeptic", 2, skeptic_user, max(args.max_new // 2, 100))
report = {
"analyst": analyst,
"skeptic": skeptic,
"note": "TinyLiquid output is research support, not a verdict. "
"Every conclusion needs primary-source verification.",
}
print(json.dumps(report, indent=2, ensure_ascii=False))
out = Path("corpus/reports")
out.mkdir(parents=True, exist_ok=True)
(out / "latest_report.json").write_text(json.dumps(report, indent=2, ensure_ascii=False),
encoding="utf-8")
if __name__ == "__main__":
main()
|