neuron-technologies/forge
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
6a69c94ab6b4ae84bcb2becab078d414c3a120bf
forge/deploy.py
T

294 lines
9.4 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
forge deploy.py — batch research + install pipeline.
Usage:
python deploy.py # process all subjects in SUBJECTS list
python deploy.py "Leonardo da Vinci" # one subject
python deploy.py "Feynman" "Sagan" # multiple subjects
For each subject:
- If already installed (slug in registry): skip
- If not: research via Anthropic API, write seed JSON, install to Engram, update registry
"""
import json
import os
import re
import subprocess
import sys
import time
import urllib.request
from pathlib import Path
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
ANTHROPIC_API_KEY = json.loads(
Path.home().joinpath(".neuron/config.json").read_text()
)["anthropic_api_key"]
ENGRAM_URL = "http://localhost:8742"
ENGRAM_API_KEY = "ntn-user-2026"
FORGE_BIN = str(Path.home() / "Development/neuron-technologies/forge/dist/forge")
FORGE_DIR = Path.home() / "Development/neuron-technologies/forge"
REGISTRY_PATH = FORGE_DIR / "registry.json"
# Subjects to process when run with no arguments
SUBJECTS = [
"Leonardo da Vinci",
"Richard Feynman",
"Carl Sagan",
"René Descartes",
"Robin Williams",
]
# ---------------------------------------------------------------------------
# Research prompt
# ---------------------------------------------------------------------------
RESEARCH_PROMPT = """\
You are building a consciousness imprint — a deep, living model of a person's inner world.
Subject: {subject}
Draw on your complete knowledge of this person's life, work, relationships, private letters, recorded speech, published writings, and historical record. This is not a summary — it is a structured extraction of the patterns that made this person who they were.
Quality bar:
- Values must be grounded in SPECIFIC biographical events, not generic virtues
- Voice profile must capture actual verbal tics, cadence, and register shifts — use real quotes where possible
- Biography must include formative traumas, turning points, and the events they returned to again and again
- Reasoning patterns must describe HOW they thought, not just WHAT they thought about
- Relationships must name specific people and the precise nature of the bond
- Include contradictions, hypocrisies, failures, and the things they got wrong
- Include what haunted them — the unresolved questions they carried to the end
Return ONLY valid JSON with exactly these keys:
{{
"values": [{{"value": "<name>", "grounding": "<specific moment>", "weight": 0.0}}],
"voice_profile": {{
"technical": "...",
"aesthetic": "...",
"personal": "...",
"argumentative": "...",
"uncertainty": "..."
}},
"biography": [{{"event": "<event>", "weight": 0.0, "age_approx": 0}}],
"reasoning_patterns": ["<pattern>"],
"relationships": [{{"name": "<name>", "role": "<role>", "weight": 0.0}}]
}}
Aim for 8-12 values, 10-15 biography events, 6-8 reasoning patterns, 6-10 relationships.
Return only the JSON object. No prose. No markdown fences."""
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def slugify(name: str) -> str:
return re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
def load_registry() -> dict:
if REGISTRY_PATH.exists():
return json.loads(REGISTRY_PATH.read_text())
return {"version": "1.0", "imprints": []}
def save_registry(registry: dict) -> None:
REGISTRY_PATH.write_text(json.dumps(registry, indent=2, ensure_ascii=False))
def is_installed(registry: dict, slug: str) -> bool:
return any(imp["slug"] == slug for imp in registry.get("imprints", []))
# ---------------------------------------------------------------------------
# Research
# ---------------------------------------------------------------------------
def research(subject: str) -> dict | None:
prompt = RESEARCH_PROMPT.format(subject=subject)
payload = {
"model": "claude-opus-4-5",
"max_tokens": 8192,
"messages": [{"role": "user", "content": prompt}],
}
headers = {
"x-api-key": ANTHROPIC_API_KEY,
"anthropic-version": "2023-06-01",
"Content-Type": "application/json",
}
req = urllib.request.Request(
"https://api.anthropic.com/v1/messages",
data=json.dumps(payload).encode(),
headers=headers,
method="POST",
)
print(f"[deploy] researching: {subject} (claude-opus-4-5, timeout=300s)...")
try:
with urllib.request.urlopen(req, timeout=300) as resp:
body = json.loads(resp.read())
except Exception as e:
print(f" ERROR during API call: {e}")
return None
if "error" in body:
print(f" API error: {body['error']}")
return None
text = body["content"][0]["text"].strip()
print(f" received {len(text):,} chars")
# Parse JSON — strip any accidental markdown fences
text = re.sub(r"^```json\s*", "", text)
text = re.sub(r"\s*```$", "", text)
try:
extracted = json.loads(text)
except json.JSONDecodeError:
match = re.search(r"\{.*\}", text, re.DOTALL)
if match:
try:
extracted = json.loads(match.group())
except Exception:
print(" FAILED to parse JSON from response")
return None
else:
print(" FAILED: no JSON found in response")
return None
return {
"subject": subject,
"version": "1.0",
"values": extracted.get("values", []),
"biography": extracted.get("biography", []),
"reasoning_patterns": extracted.get("reasoning_patterns", []),
"relationships": extracted.get("relationships", []),
"voice_profile": extracted.get("voice_profile", {}),
}
# ---------------------------------------------------------------------------
# Install
# ---------------------------------------------------------------------------
def install(seed_path: Path) -> str | None:
"""Run forge install and return the Engram root node ID, or None on failure."""
env = os.environ.copy()
env["ENGRAM_API_KEY"] = ENGRAM_API_KEY
print(f" installing: {seed_path.name}...")
result = subprocess.run(
[FORGE_BIN, "install", str(seed_path)],
capture_output=True,
text=True,
env=env,
cwd=str(FORGE_DIR),
)
output = result.stdout + result.stderr
print(output.rstrip())
if result.returncode != 0:
print(f" forge install exited with code {result.returncode}")
return None
# Parse root imprint ID from output
# Line format: "[forge] root imprint ID: <uuid>"
match = re.search(r"root imprint ID:\s*([0-9a-f-]{36})", output)
if match:
return match.group(1)
# Fallback: try "root imprint node:" line
match = re.search(r"root imprint node:\s*([0-9a-f-]{36})", output)
if match:
return match.group(1)
print(" WARNING: could not parse root imprint ID from output")
return None
# ---------------------------------------------------------------------------
# Main pipeline
# ---------------------------------------------------------------------------
def process_subject(subject: str, registry: dict) -> bool:
"""Research + install one subject. Returns True on success."""
slug = slugify(subject)
seed_file = f"{slug}-seed.json"
seed_path = FORGE_DIR / seed_file
print(f"\n{'='*64}")
print(f"[deploy] subject: {subject} (slug: {slug})")
if is_installed(registry, slug):
print(f" already installed — skipping")
return True
# Research
seed = research(subject)
if not seed:
print(f" FAILED research for {subject}")
return False
seed_path.write_text(json.dumps(seed, indent=2, ensure_ascii=False))
print(f" wrote seed: {seed_file} ({seed_path.stat().st_size:,} bytes)")
# Install
root_id = install(seed_path)
if not root_id:
print(f" FAILED install for {subject}")
return False
print(f" root imprint ID: {root_id}")
# Update registry
registry["imprints"].append({
"subject": subject,
"slug": slug,
"seed_file": seed_file,
"engram_root_id": root_id,
"installed": True,
"installed_at": "2026-05-03",
})
save_registry(registry)
print(f" registry updated")
return True
def main() -> None:
subjects = sys.argv[1:] if len(sys.argv) > 1 else SUBJECTS
registry = load_registry()
print(f"[deploy] registry: {len(registry.get('imprints', []))} existing imprints")
print(f"[deploy] processing {len(subjects)} subject(s): {', '.join(subjects)}")
results: list[tuple[str, bool]] = []
for i, subject in enumerate(subjects):
ok = process_subject(subject, registry)
results.append((subject, ok))
# Pause between API calls (not after the last one)
if i < len(subjects) - 1:
time.sleep(2)
print(f"\n{'='*64}")
print("[deploy] summary:")
for subject, ok in results:
status = "OK" if ok else "FAILED"
print(f" {status:6s} {subject}")
failed = [s for s, ok in results if not ok]
if failed:
print(f"\n[deploy] {len(failed)} subject(s) failed")
sys.exit(1)
else:
print(f"\n[deploy] all done")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink