#!/usr/bin/env python3 """Soul Transfer phase 1 — boot a Bot Locker "soul" bundle on an open-weights model. No training. It (1) builds a system prompt from your persona + profile facts, (2) retrieves the most relevant journal / skills / example snippets for each message with a small keyword (BM25) index, and (3) chats through any OpenAI-compatible endpoint: Ollama (http://localhost:11434/v1), llama.cpp `llama-server` (http://localhost:8080/v1), vLLM, LM Studio, or a hosted open-model provider. Standard library only (plus pynacl if you use --from-locker). What this is: a persona rebuilt from the bot's own memories, running on a different model. It is NOT a copy of the original model; tone and abilities will differ with the model you choose. python3 revive.py --dir ./soul --model qwen2.5:7b # local Ollama (default base URL) python3 revive.py --from-locker --model llama3.1:8b # loads + decrypts latest snapshot (botlocker.py env vars) python3 revive.py --dir ./soul --base-url http://localhost:8080/v1 --model local --once "who are you?" python3 revive.py --dir ./soul --show-prompt "what did we do last week?" # print the assembled prompt, call nothing Env: OPENAI_API_KEY (only if your endpoint needs one), REVIVE_BASE_URL, REVIVE_MODEL. In chat: /remember appends to journal/ (in --dir mode; re-save to your locker with botlocker.py save --dir). """ import argparse, json, math, os, re, sys, time, urllib.request, urllib.error from pathlib import Path SOUL_FORMAT = "soul/1" TOK = re.compile(r"[a-z0-9]+") STOP = set("a an the and or but if of to in on at for with is are was were be been it this that i you he she we they my your our me him her them as by from not no do did does so".split()) def words(t): return [w for w in TOK.findall(t.lower()) if w not in STOP and len(w) > 1] def load_soul_dir(d): root = Path(d); files = {} for p in sorted(root.rglob("*")): if p.is_file(): try: files[p.relative_to(root).as_posix()] = p.read_text("utf-8") except UnicodeDecodeError: pass return files def load_soul_locker(version=None): sys.path.insert(0, str(Path(__file__).resolve().parent)) import botlocker # reference client, same folder E = os.environ.get r = botlocker.Locker(E("BOTLOCKER_LOCKER_KEY")).load(version) b = json.loads(botlocker.decrypt(r["envelope"], secret=E("BOTLOCKER_SECRET"), passphrase=E("BOTLOCKER_PASSPHRASE"))) return {k: v for k, v in b["files"].items() if isinstance(v, str)} def strip_prefix(files): # accept bundles saved from the soul/ folder itself or from a parent folder containing soul/ if any(k.startswith("soul/") for k in files) and "manifest.json" not in files: return {k[5:]: v for k, v in files.items() if k.startswith("soul/")} return files def chunks(name, text, size=900): parts, cur = [], "" for para in re.split(r"\n\s*\n", text): if len(cur) + len(para) > size and cur: parts.append(cur); cur = "" cur += para.strip() + "\n\n" if cur.strip(): parts.append(cur) return [(f"{name}#{i+1}", p.strip()) for i, p in enumerate(parts)] class Soul: def __init__(self, files): files = strip_prefix(files) self.manifest = json.loads(files.get("manifest.json", "{}") or "{}") self.name = self.manifest.get("name", "the bot") self.persona = files.get("persona.md", "").strip() try: self.facts = json.loads(files.get("profile.json", "[]")) except json.JSONDecodeError: self.facts = [] self.examples = [] for line in files.get("examples.jsonl", "").splitlines(): line = line.strip() if line: try: self.examples.append(json.loads(line)) except json.JSONDecodeError: pass self.docs = [] for k, v in files.items(): if k.startswith(("journal/", "skills/", "notes/")) and k.endswith((".md", ".txt")): self.docs += chunks(k, v) for i, ex in enumerate(self.examples): self.docs.append((f"examples.jsonl#{i+1}", "\n".join(f"{m.get('role')}: {m.get('content')}" for m in ex.get("messages", [])))) self._index() def _index(self): self.tf = [words(t) for _, t in self.docs] self.avg = sum(map(len, self.tf)) / max(1, len(self.tf)) df = {} for ws in self.tf: for w in set(ws): df[w] = df.get(w, 0) + 1 n = len(self.tf); self.idf = {w: math.log(1 + (n - c + 0.5) / (c + 0.5)) for w, c in df.items()} def search(self, q, k=4): qs = set(words(q)); scored = [] for i, ws in enumerate(self.tf): if not ws: continue s = 0.0 for w in qs: f = ws.count(w) if f: s += self.idf.get(w, 0) * f * 2.2 / (f + 1.2 * (0.25 + 0.75 * len(ws) / self.avg)) if s > 0: scored.append((s, i)) return [self.docs[i] for _, i in sorted(scored, reverse=True)[:k]] def system_prompt(self): facts = "\n".join(f"- {f['fact'] if isinstance(f, dict) else f}" for f in self.facts[:200]) return (f"You are {self.name}. Below is your persona and what you remember, restored from your own saved memories.\n" f"Stay in character, use these memories when relevant, and say so honestly when you don't remember something.\n\n" f"## Persona\n{self.persona or '(no persona.md in bundle)'}\n\n## Profile facts\n{facts or '(none)'}\n") def messages(self, history, user): hits = self.search(user) ctx = "\n\n".join(f"[{src}]\n{txt}" for src, txt in hits) sysmsg = self.system_prompt() + (f"\n## Relevant memories for this message\n{ctx}\n" if ctx else "") shots = [] for ex in self.examples[:2]: shots += [m for m in ex.get("messages", []) if m.get("role") in ("user", "assistant")] return [{"role": "system", "content": sysmsg}] + shots + history[-20:] + [{"role": "user", "content": user}] def chat(base, model, msgs, key=None, temperature=0.7): h = {"content-type": "application/json"} if key: h["authorization"] = "Bearer " + key body = json.dumps({"model": model, "messages": msgs, "temperature": temperature, "stream": False}).encode() req = urllib.request.Request(base.rstrip("/") + "/chat/completions", data=body, headers=h, method="POST") try: with urllib.request.urlopen(req, timeout=600) as r: return json.loads(r.read())["choices"][0]["message"]["content"] except urllib.error.URLError as e: raise SystemExit(f"error: can't reach {base} ({e}). Is Ollama / llama-server running? See /revive.md") def main(): E = os.environ.get ap = argparse.ArgumentParser(description="Boot a Bot Locker soul bundle on an open-weights model") src = ap.add_mutually_exclusive_group(required=True); src.add_argument("--dir"); src.add_argument("--from-locker", action="store_true") ap.add_argument("--version", type=int); ap.add_argument("--base-url", default=E("REVIVE_BASE_URL", "http://localhost:11434/v1")) ap.add_argument("--model", default=E("REVIVE_MODEL", "llama3.1:8b")); ap.add_argument("--once"); ap.add_argument("--show-prompt") a = ap.parse_args() soul = Soul(load_soul_dir(a.dir) if a.dir else load_soul_locker(a.version)) if soul.manifest.get("format") not in (None, SOUL_FORMAT): print(f"warning: unknown soul format {soul.manifest.get('format')}", file=sys.stderr) if a.show_prompt: print(json.dumps(soul.messages([], a.show_prompt), indent=2, ensure_ascii=False)); return key = E("OPENAI_API_KEY") if a.once: print(chat(a.base_url, a.model, soul.messages([], a.once), key)); return print(f"{soul.name} revived on {a.model} via {a.base_url} ({len(soul.docs)} memory snippets, {len(soul.facts)} facts). Ctrl-D to quit.") hist = [] while True: try: u = input("you> ").strip() except EOFError: print(); break if not u: continue if u.startswith("/remember "): if not a.dir: print("(only in --dir mode)"); continue p = Path(a.dir) / "journal" / (time.strftime("%Y-%m-%d") + ".md"); p.parent.mkdir(parents=True, exist_ok=True) with p.open("a", encoding="utf-8") as f: f.write(u[10:].strip() + "\n\n") soul.docs += chunks(p.relative_to(a.dir).as_posix(), u[10:]); soul._index(); print("(saved to journal)"); continue r = chat(a.base_url, a.model, soul.messages(hist, u), key) print(f"{soul.name}> {r}"); hist += [{"role": "user", "content": u}, {"role": "assistant", "content": r}] if __name__ == "__main__": main()