"""Autonomous Innovation Engine. A 3-layer multi-agent system: Layer 1 (Generation) - divergent generator agents propose ideas. Layer 2 (Evaluation) - an adversarial critic council filters and refines. Layer 3 (Memory Plane)- a persistent record of outcomes that calibrates the critics over rounds and across runs. Usage: python engine_demo.py --dry-run # full loop with a mock LLM (no GPU, no torch) python engine_demo.py --smoke # 1 tiny round on real hardware python engine_demo.py # full demo run """ import argparse import json import os import re import sys import time import uuid from difflib import SequenceMatcher # -------------------------------------------------------------------------- # Config # -------------------------------------------------------------------------- MODEL_ID = "Qwen/Qwen3-1.7B" ROUNDS = 3 IDEAS_PER_GEN = 2 ACCEPT_THRESHOLD = 7.0 # weighted aggregate needed to be accepted REDTEAM_MIN = 6.0 # idea must survive the red-team attack REFINE_TOP_K = 2 # best ideas per round get one refinement pass MAX_NEW_TOKENS = 512 CALL_LIMIT = 120 # hard cap on LLM calls (guards runaway loops) MEMORY_DATASET = "LoveLogicAI/innovation-engine-memory" RESULTS_REPO = "LoveLogicAI/autonomous-innovation-engine" TRACKIO_SPACE_ID = os.environ.get("TRACKIO_SPACE_ID", "LoveLogicAI/autonomous-innovation-engine-trackio") THEME = ( "2026 innovation concepts spanning: autonomous software agents, " "self-healing infrastructure, knowledge & research automation, " "human-AI interaction, and global-scale scientific systems" ) DOMAINS = ["software", "infrastructure", "research", "human-ai", "science", "scale"] # -------------------------------------------------------------------------- # Layer 1: generator personas # -------------------------------------------------------------------------- GENERATORS = [ { "id": "pragmatist", "name": "The Systems Pragmatist", "temperature": 0.7, "brief": ( "You propose innovations that are buildable within 12 months with " "existing technology. You care about concrete systems, integration " "points, and shippable scope." ), }, { "id": "frontier", "name": "The Frontier Researcher", "temperature": 0.95, "brief": ( "You propose innovations that draw on bleeding-edge research and " "speculative-but-plausible capabilities. You reach one step beyond " "the current state of the art." ), }, { "id": "contrarian", "name": "The Contrarian Designer", "temperature": 1.0, "brief": ( "You deliberately avoid obvious or fashionable ideas. You look for " "inversions, overlooked niches, and counterintuitive approaches " "that others would dismiss." ), }, ] GEN_SYSTEM = ( "You are {name}, a generator agent in a multi-agent innovation engine. " "{brief}" ) GEN_USER = ( "Round {round} of the innovation engine. Theme: {theme}\n\n" "Ideas already proposed (do NOT duplicate them, propose something distinct): " "{existing}\n\n" "{memory_hint}" "Propose exactly {n} distinct innovation concepts as a JSON array. " "Each object must have exactly these keys:\n" ' "name" (short name, <=6 words),\n' ' "domain" (one of: {domains}),\n' ' "pitch" (<=30 words),\n' ' "mechanism" (how it actually works, <=40 words),\n' ' "differentiator" (why nobody has built it this way, <=20 words).\n' "Output ONLY the JSON array, no other text." ) # -------------------------------------------------------------------------- # Layer 2: critic council # -------------------------------------------------------------------------- CRITICS = [ { "id": "feasibility", "name": "The Feasibility Critic", "weight": 0.25, "rubric": ( "Judge technical feasibility: does the mechanism actually work, " "are the dependencies real, and is the engineering plausible " "within 12-24 months? 10 = buildable now, 1 = fantasy." ), "key": "score", }, { "id": "novelty", "name": "The Novelty Critic", "weight": 0.25, "rubric": ( "Judge novelty: is this genuinely distinct from existing products " "and from the memory plane's prior ideas? 10 = genuinely new, " "1 = a rename of something that exists." ), "key": "score", }, { "id": "market", "name": "The Market Critic", "weight": 0.25, "rubric": ( "Judge market viability: is there a real user, a real willingness " "to pay, and a wedge into the market? 10 = clear demand, " "1 = no one would use it." ), "key": "score", }, { "id": "redteam", "name": "The Red Team Executioner", "weight": 0.25, "rubric": ( "You attack every idea. Find the strongest reason it fails: " "hidden dependency, fatal edge case, abuse vector, or scaling " "wall. Score = how well the idea SURVIVES your attack " "(10 = survives easily, 1 = collapses immediately). Also state " "the single sharpest weakness." ), "key": "attack_score", }, ] CRITIC_SYSTEM = ( "You are {name}, a critic agent in an adversarial evaluation council. " "{rubric}" ) CRITIC_USER = ( "Score each of these ideas from the innovation engine (round {round}).\n" "Ideas:\n{ideas}\n\n" "{memory_hint}" "Return a JSON array with one object per idea: " '{{"id": , "{key}": , "weakness": "<=15 words>"}}. ' "Output ONLY the JSON array." ) # -------------------------------------------------------------------------- # Enhance mode: user's think-tank concepts as generation seeds # -------------------------------------------------------------------------- SEED_CONCEPTS = [ {"name": "Self-Evolving Codebase Ecosystem", "domain": "software", "pitch": "Software that autonomously refactors itself and deploys new features based on real-time market feedback and live user behavior.", "mechanism": "Telemetry-driven loop: agents propose refactors, validate in sandboxed CI, canary-deploy, roll back on regression."}, {"name": "Self-Healing Infrastructure Agent", "domain": "infrastructure", "pitch": "An autonomous remediation engine that executes secure, sandboxed code to patch vulnerabilities, rotate keys, and deploy firewall rules.", "mechanism": "Continuous vuln scanning triggers sandboxed remediation playbooks with approval gates and an audit trail."}, {"name": "Vibe-to-Build Prototype Generator", "domain": "software", "pitch": "Translates natural-language prompts directly into scoped repositories with local environments and automated CI/CD pipelines.", "mechanism": "Prompt to spec extraction to repo scaffolding with tests, containerized env, CI/CD, one-command local run."}, {"name": "Multi-Agent Type Protocol", "domain": "software", "pitch": "A framework enforcing strict schema validation and type safety between autonomous agents to prevent context drift and hallucinations.", "mechanism": "Typed message contracts with runtime validators; violations are quarantined and fed back for correction."}, {"name": "Intelligent Codebase Archaeology", "domain": "software", "pitch": "AI agents sift through legacy systems to deconstruct architectural patterns, reconstruct business logic, and generate modern refactoring plans.", "mechanism": "Static analysis plus LLM reasoning reconstructs intent from code and history; emits dependency maps and staged migration plans."}, {"name": "Autonomous Knowledge Synthesis Network", "domain": "research", "pitch": "An agent suite that scours the internet for new scientific data and automatically generates code prototypes for novel solutions.", "mechanism": "Crawler agents triage new papers and data; synthesis agents distill claims; prototypers generate and benchmark implementations."}, {"name": "Dynamic Context Synthesis Agent", "domain": "research", "pitch": "Monitors global technical documentation, synthesizes new findings, and drafts code to solve emerging compatibility issues.", "mechanism": "Diffs documentation releases across ecosystems, flags breaking changes, drafts migration patches validated against tests."}, {"name": "Dynamic Semantic Reality Layer", "domain": "human-ai", "pitch": "An interface adapting digital experiences and visual data presentation to a user's neuro-cognitive state in real-time.", "mechanism": "Multimodal attention and load estimation adapts layout, density, and modality with bounded, explainable adjustments."}, {"name": "Neuro-Cognitive Knowledge Bridge", "domain": "human-ai", "pitch": "Direct, real-time data streaming between large language models and human neural interfaces.", "mechanism": "Decoded neural intent mapped to a constrained query protocol over LLMs; latency-tolerant, consent-first design."}, {"name": "Autonomous Emotional Harmony Engine", "domain": "human-ai", "pitch": "An agent that detects cognitive dissonance in team communication and subtly adjusts interaction paths to mediate conflict.", "mechanism": "Discourse-level dissonance detection proposes reframings, pacing changes, or async handoffs; never alters content covertly."}, {"name": "Reality Verification Protocol", "domain": "human-ai", "pitch": "An automated system continuously verifying the physical authenticity of digital content to combat synthetic media.", "mechanism": "Sensor-level provenance capture plus cryptographic signing; verifiers cross-check against physical and source evidence."}, {"name": "Autonomous Molecular Discovery Engine", "domain": "science", "pitch": "A system that predicts new materials and autonomously controls robotic laboratories to synthesize them.", "mechanism": "Generative materials model proposes candidates; robotic lab runs closed-loop synthesis with active learning."}, {"name": "Decentralized Cognitive Cloud", "domain": "scale", "pitch": "A network pooling idle community devices to run private, localized AI models with total data sovereignty.", "mechanism": "Scheduler shards workloads across idle devices; raw data stays on-device, exchanging only outputs."}, {"name": "Quantum-Classical Hybrid Engine", "domain": "scale", "pitch": "A computational system optimizing global energy grids and logistics networks in real-time.", "mechanism": "Classical solvers handle the bulk; quantum-inspired annealing targets the hard combinatorial core, validated classically."}, ] ENHANCE_USER = ( "Enhance these innovation concepts from a think-tank. For each concept:\n" "- sharpen the pitch (<=30 words)\n" "- make the mechanism concrete and buildable (<=45 words)\n" "- state the true differentiator (<=20 words)\n" '- add "extension": the boldest credible next step beyond the concept (<=25 words)\n\n' "Concepts:\n{seeds}\n\n" "{memory_hint}" "Return a JSON array with one object per concept: " '{{"seed_name": , "name": , "domain": , ' '"pitch": ..., "mechanism": ..., "differentiator": ..., "extension": ...}}. ' "Output ONLY the JSON array." ) def validate_enhanced(obj): if not isinstance(obj, dict): return False if not all(k in obj for k in ("seed_name", "name", "domain", "pitch", "mechanism", "differentiator")): return False return obj["domain"] in DOMAINS # -------------------------------------------------------------------------- # JSON extraction (stdlib only, so --dry-run needs nothing else) # -------------------------------------------------------------------------- def _balanced_slice(text, start_ch, end_ch): """Return the first balanced [..] or {..} block starting at/after start.""" start = text.find(start_ch) if start == -1: return None depth = 0 in_str = False esc = False for i in range(start, len(text)): c = text[i] if in_str: if esc: esc = False elif c == "\\": esc = True elif c == '"': in_str = False continue if c == '"': in_str = True elif c in (start_ch, end_ch): depth += 1 if c == start_ch else -1 if depth == 0: return text[start : i + 1] return None def extract_json_array(text): """Extract the first JSON array from model output; also handles a bare object.""" block = _balanced_slice(text, "[", "]") if block is None: block = _balanced_slice(text, "{", "}") if block is not None: block = "[" + block + "]" if block is None: return None try: data = json.loads(block) except json.JSONDecodeError: return None if isinstance(data, dict): data = [data] if not isinstance(data, list): return None return data # -------------------------------------------------------------------------- # Layer 3: memory plane # -------------------------------------------------------------------------- class MemoryPlane: """Persistent record of every idea, its scores and its decision. Two jobs: 1. Calibration - critics get historical baselines (per-domain mean score, acceptance rate, best/worst examples) in their prompts, so filtering sharpens as the record grows. 2. Novelty - near-duplicate detection against past pitches. """ def __init__(self, path="memory_plane.jsonl"): self.path = path self.records = [] if os.path.exists(path): with open(path) as f: for line in f: line = line.strip() if line: try: self.records.append(json.loads(line)) except json.JSONDecodeError: pass def reload(self): self.records = [] if os.path.exists(self.path): with open(self.path) as f: for line in f: line = line.strip() if line: try: self.records.append(json.loads(line)) except json.JSONDecodeError: pass def add(self, record): record = dict(record) record.setdefault("idea_id", uuid.uuid4().hex[:8]) record["ts"] = round(time.time(), 3) self.records.append(record) with open(self.path, "a") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") return record["idea_id"] def _pitch_text(self, r): return f"{r.get('name', '')} {r.get('pitch', '')}".lower() def existing_names(self): return [r["name"] for r in self.records if "name" in r] def similar(self, pitch_text, k=3): scored = [] for r in self.records: ratio = SequenceMatcher(None, self._pitch_text(r), pitch_text.lower()).ratio() scored.append((ratio, r)) scored.sort(key=lambda x: -x[0]) return [ {"name": r["name"], "score": r.get("aggregate"), "pitch": r.get("pitch")} for _, r in scored[:k] ] def baselines(self): """Per-domain statistics that calibrate the critics.""" by_domain = {} for r in self.records: d = r.get("domain", "unknown") if isinstance(r.get("aggregate"), (int, float)): by_domain.setdefault(d, []).append(r["aggregate"]) stats = { d: {"n": len(v), "mean": round(sum(v) / len(v), 2), "max": max(v)} for d, v in by_domain.items() } accepted = [r for r in self.records if r.get("decision") == "accepted"] rejected = [r for r in self.records if r.get("decision") == "rejected"] overall = { "total_ideas": len(self.records), "accept_rate": round(len(accepted) / max(1, len(accepted) + len(rejected)), 2), } return {"overall": overall, "by_domain": stats} def critic_hint(self, k=2): """Context injected into critic prompts: baselines + best and worst.""" b = self.baselines() if b["overall"]["total_ideas"] == 0: return "" accepted = sorted( (r for r in self.records if r.get("decision") == "accepted"), key=lambda r: -(r.get("aggregate") or 0), ) rejected = sorted( (r for r in self.records if r.get("decision") == "rejected"), key=lambda r: (r.get("aggregate") or 10), ) best = [ f'{r["name"]} (score {r.get("aggregate")})' for r in accepted[:k] ] worst = [f'{r["name"]} (score {r.get("aggregate")})' for r in rejected[:k]] return ( f"Memory plane so far: {b['overall']['total_ideas']} ideas, " f"accept rate {b['overall']['accept_rate']}, " f"per-domain means: {json.dumps(b['by_domain'])}. " f"Best past ideas: {best}. Weakest past ideas: {worst}. " "Calibrate your scores against this history.\n\n" ) def generator_hint(self): if not self.records: return "" accepted = [r["name"] for r in self.records if r.get("decision") == "accepted"] return ( f"Previously ACCEPTED ideas (build on them or attack them, do not repeat): " f"{accepted}\n\n" if accepted else "" ) def sync_to_hub(self, repo_id): from huggingface_hub import HfApi api = HfApi() api.upload_file( path_or_fileobj=self.path, path_in_repo="memory_plane.jsonl", repo_id=repo_id, repo_type="dataset", commit_message="Sync memory plane", ) return len(self.records) # -------------------------------------------------------------------------- # LLM backend (torch/transformers imported lazily so --dry-run stays light) # -------------------------------------------------------------------------- class LLM: def __init__(self, model_id=MODEL_ID, max_new_tokens=MAX_NEW_TOKENS): import torch from transformers import AutoModelForCausalLM, AutoTokenizer self.torch = torch self.tok = AutoTokenizer.from_pretrained(model_id) dtype = torch.float16 if torch.cuda.is_available() else torch.float32 self.model = _from_pretrained(model_id, dtype, device_map="auto") self.model.eval() self.max_new_tokens = max_new_tokens self.calls = 0 self.output_tokens = 0 def _build(self, system, user): messages = [ {"role": "system", "content": system}, {"role": "user", "content": user}, ] try: return self.tok.apply_chat_template( messages, tokenize=False, add_generation_prompt=True, enable_thinking=False, ) except TypeError: # template without thinking toggle return self.tok.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) def chat(self, system, user, temperature=0.8): torch = self.torch prompt = self._build(system, user) inputs = self.tok(prompt, return_tensors="pt").to(self.model.device) with torch.inference_mode(): out = self.model.generate( **inputs, max_new_tokens=self.max_new_tokens, do_sample=temperature > 0, temperature=max(temperature, 1e-4), top_p=0.95, pad_token_id=self.tok.eos_token_id, ) text = self.tok.decode( out[0][inputs["input_ids"].shape[1] :], skip_special_tokens=True ) self.calls += 1 self.output_tokens += int(out.shape[1] - inputs["input_ids"].shape[1]) return text class MockLLM: """Deterministic stand-in that produces valid JSON for --dry-run.""" def __init__(self): self.calls = 0 self.output_tokens = 0 self._n = 0 def chat(self, system, user, temperature=0.8): self.calls += 1 self._n += 1 if "Propose exactly" in user: n = int(re.search(r"exactly (\d+)", user).group(1)) ideas = [] for i in range(n): k = self._n * n + i ideas.append( { "name": f"Mock Concept {k}", "domain": DOMAINS[k % len(DOMAINS)], "pitch": f"Mock pitch number {k} for dry-run testing.", "mechanism": f"Mock mechanism {k}: parse JSON, score, refine.", "differentiator": f"Mock differentiator {k}.", } ) return json.dumps(ideas) if "Enhance these innovation concepts" in user: seeds_json = user.split("Concepts:\n")[1].split("\n\n")[0] seeds = extract_json_array(seeds_json) or [] return json.dumps( [ {"seed_name": s["name"], "name": s["name"] + " (enhanced)", "domain": s["domain"], "pitch": s["pitch"], "mechanism": s["mechanism"], "differentiator": "mock differentiator", "extension": "mock extension"} for s in seeds ] ) if "Score each of these ideas" in user: ideas = extract_json_array(user.split("Ideas:\n")[1].split("\n\n")[0]) out = [] for j, idea in enumerate(ideas or []): score = 8 - (j % 4) out.append({"id": idea["id"], "score": score, "attack_score": score, "weakness": "mock weakness"}) return json.dumps(out) # refinement call return json.dumps( { "name": "Mock Concept (refined)", "domain": "software", "pitch": "Refined mock pitch.", "mechanism": "Refined mock mechanism.", "differentiator": "Refined mock differentiator.", } ) def _from_pretrained(model_id, dtype, **kwargs): """transformers>=4.56 renamed torch_dtype -> dtype; support both.""" from transformers import AutoModelForCausalLM try: return AutoModelForCausalLM.from_pretrained(model_id, dtype=dtype, **kwargs) except TypeError: return AutoModelForCausalLM.from_pretrained(model_id, torch_dtype=dtype, **kwargs) def ask(llm, system, user, temperature, retries=1): """LLM call with one retry on unparseable JSON.""" last = None for attempt in range(retries + 1): text = llm.chat(system, user + ( "\n\nYour previous output was not valid JSON. " "Output ONLY the JSON." if attempt else "" ), temperature=temperature) last = extract_json_array(text) if last is not None: return last, True return last, False # -------------------------------------------------------------------------- # Orchestrator # -------------------------------------------------------------------------- def validate_idea(obj): if not isinstance(obj, dict): return False return all(k in obj for k in ("name", "domain", "pitch", "mechanism", "differentiator")) \ and obj["domain"] in DOMAINS def generate_round(llm, memory, rnd): ideas = [] for gen in GENERATORS: hint = memory.generator_hint() if hint: print(f"[mem] generator hint (round {rnd}): {hint.strip()[:160]}") user = GEN_USER.format( round=rnd, theme=THEME, existing=memory.existing_names()[-15:] or "none yet", memory_hint=hint, n=IDEAS_PER_GEN, domains=", ".join(DOMAINS), ) parsed, ok = ask( llm, GEN_SYSTEM.format(name=gen["name"], brief=gen["brief"]), user, temperature=gen["temperature"], ) if not ok: print(f"[gen:{gen['id']}] WARN unparseable output, skipping") continue for obj in parsed: if not validate_idea(obj): print(f"[gen:{gen['id']}] WARN malformed idea, skipping: {str(obj)[:80]}") continue obj = dict(obj) obj["idea_id"] = uuid.uuid4().hex[:8] obj["agent"] = gen["id"] obj["round"] = rnd ideas.append(obj) if llm.calls > CALL_LIMIT: print("WARN call limit reached, stopping generation") break return ideas def critique_round(llm, memory, rnd, ideas, batch_size=None): chunks = [ideas] if batch_size and len(ideas) > batch_size: chunks = [ideas[i:i + batch_size] for i in range(0, len(ideas), batch_size)] for chunk in chunks: idea_block = json.dumps( [ {"id": i["idea_id"], "name": i["name"], "domain": i["domain"], "pitch": i["pitch"], "mechanism": i["mechanism"]} for i in chunk ], indent=1, ) for critic in CRITICS: parsed, ok = ask( llm, CRITIC_SYSTEM.format(name=critic["name"], rubric=critic["rubric"]), CRITIC_USER.format(round=rnd, ideas=idea_block, key=critic["key"], memory_hint=memory.critic_hint()), temperature=0.3, ) if not ok: print(f"[critic:{critic['id']}] WARN unparseable output") continue by_id = {} for entry in parsed: if isinstance(entry, dict) and "id" in entry: by_id[str(entry["id"])] = entry for idea in chunk: entry = by_id.get(idea["idea_id"]) if not entry: continue raw = entry.get(critic["key"], entry.get("score")) try: idea.setdefault("scores", {})[critic["id"]] = max(1, min(10, int(raw))) except (TypeError, ValueError): pass if entry.get("weakness"): idea.setdefault("weaknesses", {})[critic["id"]] = entry["weakness"] if llm.calls > CALL_LIMIT: print("WARN call limit reached, stopping critique") break return ideas def aggregate(idea): num, den = 0.0, 0.0 for critic in CRITICS: s = idea.get("scores", {}).get(critic["id"]) if isinstance(s, (int, float)): num += critic["weight"] * s den += critic["weight"] idea["aggregate"] = round(num / den, 2) if den else None return idea["aggregate"] def decide(idea): red = idea.get("scores", {}).get("redteam") if idea.get("aggregate") is None: idea["decision"] = "unscored" elif idea["aggregate"] >= ACCEPT_THRESHOLD and red is not None and red >= REDTEAM_MIN: idea["decision"] = "accepted" else: idea["decision"] = "rejected" return idea["decision"] def refine_idea(llm, memory, idea): gen = next(g for g in GENERATORS if g["id"] == idea["agent"]) weaknesses = json.dumps(idea.get("weaknesses", {})) user = ( f"Your idea was attacked by the critic council. Fix the weaknesses " f"without changing the core concept.\n\n" f"Idea: {json.dumps({k: idea[k] for k in ('name', 'domain', 'pitch', 'mechanism', 'differentiator')})}\n" f"Critic weaknesses: {weaknesses}\n\n" "Return the revised idea as ONE JSON object with exactly the same five keys " "(name, domain, pitch, mechanism, differentiator). Output ONLY the JSON." ) parsed, ok = ask( llm, GEN_SYSTEM.format(name=gen["name"], brief=gen["brief"]), user, temperature=0.6, ) if ok and parsed and validate_idea(parsed[0]): revised = dict(parsed[0]) revised["idea_id"] = idea["idea_id"] + "-r" revised["agent"] = idea["agent"] revised["round"] = idea["round"] revised["refined_from"] = idea["idea_id"] return revised print("[refine] WARN unparseable, keeping original") return None def main(): ap = argparse.ArgumentParser() ap.add_argument("--dry-run", action="store_true", help="mock LLM, no GPU") ap.add_argument("--smoke", action="store_true", help="1 tiny round on real hardware") ap.add_argument("--enhance", action="store_true", help="enhance the think-tank seed concepts via the council") args = ap.parse_args() # job runners start scripts without CLI args; allow ENGINE_MODE=smoke|dry|enhance mode_env = os.environ.get("ENGINE_MODE", "") if mode_env == "smoke" and not args.smoke and not args.dry_run and not args.enhance: args.smoke = True elif mode_env == "dry" and not args.smoke and not args.dry_run and not args.enhance: args.dry_run = True elif mode_env == "enhance" and not args.smoke and not args.dry_run: args.enhance = True t0 = time.time() memory = MemoryPlane() llm = MockLLM() if args.dry_run else LLM() if args.dry_run: print("[dry-run] using MockLLM") if args.smoke: global ROUNDS, IDEAS_PER_GEN, REFINE_TOP_K ROUNDS, IDEAS_PER_GEN, REFINE_TOP_K = 1, 1, 0 print("[smoke] 1 round, 1 idea/generator, no refinement") try: import trackio if TRACKIO_SPACE_ID: trackio.init(project="autonomous-innovation-engine", space_id=TRACKIO_SPACE_ID) print(f"[trackio] logging to {TRACKIO_SPACE_ID}") else: trackio = None print("[trackio] TRACKIO_SPACE_ID not set, metrics to stdout only") except Exception as e: trackio = None print(f"[trackio] unavailable ({e}); metrics to stdout only") if args.enhance and not memory.records: # cross-run calibration: pull the memory plane synced by earlier runs try: from huggingface_hub import hf_hub_download p = hf_hub_download( repo_id=MEMORY_DATASET, repo_type="dataset", filename="memory_plane.jsonl", local_dir=".", ) if os.path.abspath(p) != os.path.abspath(memory.path): import shutil shutil.copyfile(p, memory.path) memory.reload() print(f"[memory] seeded {len(memory.records)} records from {MEMORY_DATASET}") except Exception as e: print(f"[memory] no prior memory plane to download ({e}); starting fresh") if args.enhance: run_enhance(llm, memory, trackio) if os.environ.get("ENGINE_PUSH") == "1": push_results(memory, result_files=( ("results/enhanced_ideas.jsonl", "results/enhanced_ideas.jsonl"), ("results/enhanced_summary.json", "results/enhanced_summary.json"), )) if trackio: try: trackio.finish() except Exception as e: print(f"[trackio] finish failed: {e}") return all_records = [] summary_rounds = [] for rnd in range(1, ROUNDS + 1): print(f"\n===== ROUND {rnd}/{ROUNDS} =====") ideas = generate_round(llm, memory, rnd) print(f"[gen] {len(ideas)} ideas from {len(GENERATORS)} generators " f"(llm calls={llm.calls})") if not ideas: print("[gen] no ideas produced, ending run") break ideas = critique_round(llm, memory, rnd, ideas) for idea in ideas: aggregate(idea) decide(idea) # refinement pass on the top-k ideas rankable = [i for i in ideas if i.get("aggregate") is not None] top = sorted(rankable, key=lambda i: -i["aggregate"])[:REFINE_TOP_K] refined = [] for idea in top: rev = refine_idea(llm, memory, idea) if rev is None: continue rev = critique_round(llm, memory, rnd, [rev])[0] aggregate(rev) decide(rev) refined.append(rev) for idea in ideas + refined: memory.add(idea) all_records.extend(ideas + refined) accepted = [i for i in ideas + refined if i["decision"] == "accepted"] scored = [i for i in ideas + refined if i.get("aggregate") is not None] mean_agg = round(sum(i["aggregate"] for i in scored) / len(scored), 2) if scored else 0.0 best_agg = max((i["aggregate"] for i in scored), default=0.0) stats = { "round": rnd, "n_ideas": len(ideas) + len(refined), "n_accepted": len(accepted), "mean_aggregate": mean_agg, "best_aggregate": best_agg, "llm_calls": llm.calls, "output_tokens": llm.output_tokens, } summary_rounds.append(stats) print(f"[eval] mean aggregate={mean_agg} best={best_agg} " f"accepted={len(accepted)} (calls={llm.calls}, out_tokens={llm.output_tokens})") if trackio: try: trackio.log(stats, step=rnd) except Exception as e: print(f"[trackio] log failed: {e}") # ---- final report ----------------------------------------------------- accepted_all = [r for r in all_records if r.get("decision") == "accepted"] print("\n===== ENGINE SUMMARY =====") print(f"total ideas: {len(all_records)} | accepted: {len(accepted_all)} " f"| llm calls: {llm.calls} | output tokens: {llm.output_tokens} " f"| wall: {round(time.time() - t0, 1)}s") for r in sorted(accepted_all, key=lambda x: -(x.get("aggregate") or 0)): print(f" ACCEPTED [{r.get('aggregate')}] {r['name']} ({r['domain']}) — {r['pitch']}") if not accepted_all: print(" (no ideas cleared the council this run)") os.makedirs("results", exist_ok=True) results_path = os.path.join("results", "ideas.jsonl") with open(results_path, "w") as f: for r in all_records: f.write(json.dumps(r, ensure_ascii=False) + "\n") summary = { "run": uuid.uuid4().hex[:8], "config": {"rounds": ROUNDS, "ideas_per_gen": IDEAS_PER_GEN, "model": "mock" if args.dry_run else MODEL_ID}, "rounds": summary_rounds, "total_ideas": len(all_records), "total_accepted": len(accepted_all), "llm_calls": llm.calls, "output_tokens": llm.output_tokens, "wall_seconds": round(time.time() - t0, 1), "accepted": [ {"name": r["name"], "domain": r["domain"], "aggregate": r.get("aggregate"), "pitch": r["pitch"]} for r in accepted_all ], } with open(os.path.join("results", "summary.json"), "w") as f: json.dump(summary, f, indent=2) print(f"[out] wrote {results_path} and results/summary.json; " f"memory plane at {memory.path} ({len(memory.records)} records)") if os.environ.get("ENGINE_PUSH") == "1": push_results(memory) if trackio: try: trackio.finish() except Exception as e: print(f"[trackio] finish failed: {e}") def run_enhance(llm, memory, trackio): """Enhance the think-tank seed concepts: generation layer, then council, then refinement of the top ideas. Results recorded in the memory plane.""" print(f"[enhance] {len(SEED_CONCEPTS)} think-tank seed concepts") # memory plane makes reruns idempotent: skip seeds already enhanced done_seeds = {r.get("seed_name") for r in memory.records if r.get("seed_name")} pending = [s for s in SEED_CONCEPTS if s["name"] not in done_seeds] if len(pending) < len(SEED_CONCEPTS): print(f"[enhance] {len(SEED_CONCEPTS) - len(pending)} seeds already in memory plane, " f"enhancing the remaining {len(pending)}") # one call per seed (small models drop items from large batched JSON), # rotating across generator personas enhanced = [] for idx, seed in enumerate(pending): gen = GENERATORS[idx % len(GENERATORS)] user = ENHANCE_USER.format( seeds=json.dumps( [{k: seed[k] for k in ("name", "domain", "pitch", "mechanism")}], indent=1, ), domains=", ".join(DOMAINS), memory_hint=memory.generator_hint(), ) parsed, ok = ask( llm, GEN_SYSTEM.format(name=gen["name"], brief=gen["brief"]), user, temperature=0.8, ) if not ok or not parsed: print(f"[enhance:{gen['id']}] WARN unparseable output for seed '{seed['name']}'") continue got = 0 for obj in parsed: if not validate_enhanced(obj): print(f"[enhance:{gen['id']}] WARN malformed enhancement: {str(obj)[:80]}") continue obj = dict(obj) obj["idea_id"] = uuid.uuid4().hex[:8] obj["agent"] = gen["id"] obj["round"] = "enhance" enhanced.append(obj) got += 1 if got: print(f"[enhance:{gen['id']}] '{seed['name']}' -> enhanced ({got} object(s))") if not enhanced: print("[enhance] no enhancements produced, aborting") return [] enhanced = critique_round(llm, memory, "enhance", enhanced, batch_size=7) for idea in enhanced: aggregate(idea) decide(idea) rankable = [i for i in enhanced if i.get("aggregate") is not None] top = sorted(rankable, key=lambda i: -i["aggregate"])[:3] refined = [] for idea in top: rev = refine_idea(llm, memory, idea) if rev is None: continue rev = critique_round(llm, memory, "enhance", [rev])[0] aggregate(rev) decide(rev) refined.append(rev) for idea in enhanced + refined: memory.add(idea) accepted = [r for r in enhanced + refined if r.get("decision") == "accepted"] scored = [r for r in enhanced + refined if r.get("aggregate") is not None] mean_agg = round(sum(r["aggregate"] for r in scored) / len(scored), 2) if scored else 0.0 print("\n===== ENHANCEMENT SUMMARY =====") print(f"seeds: {len(SEED_CONCEPTS)} | enhanced: {len(enhanced)} | refined: {len(refined)} " f"| accepted: {len(accepted)} | llm calls: {llm.calls} " f"| output tokens: {llm.output_tokens}") for r in sorted(enhanced + refined, key=lambda x: -(x.get("aggregate") or 0)): tag = r.get("decision", "?")[:3].upper() print(f" [{tag} {r.get('aggregate')}] {r['name']} ({r['domain']}) - {r['pitch']}") if r.get("extension"): print(f" extension: {r['extension']}") os.makedirs("results", exist_ok=True) with open(os.path.join("results", "enhanced_ideas.jsonl"), "w") as f: for r in enhanced + refined: f.write(json.dumps(r, ensure_ascii=False) + "\n") summary = { "mode": "enhance", "n_seeds": len(SEED_CONCEPTS), "n_enhanced": len(enhanced), "n_refined": len(refined), "n_accepted": len(accepted), "mean_aggregate": mean_agg, "llm_calls": llm.calls, "output_tokens": llm.output_tokens, "memory_records": len(memory.records), "results": [ {"name": r["name"], "seed_name": r.get("seed_name"), "domain": r["domain"], "aggregate": r.get("aggregate"), "decision": r.get("decision"), "pitch": r["pitch"], "mechanism": r["mechanism"], "differentiator": r.get("differentiator"), "extension": r.get("extension"), "weaknesses": r.get("weaknesses", {})} for r in sorted(enhanced + refined, key=lambda x: -(x.get("aggregate") or 0)) ], } with open(os.path.join("results", "enhanced_summary.json"), "w") as f: json.dump(summary, f, indent=2) print("[out] wrote results/enhanced_ideas.jsonl and results/enhanced_summary.json; " f"memory plane now {len(memory.records)} records") if trackio: try: trackio.log({"mode": 2, "n_enhanced": len(enhanced), "n_accepted": len(accepted), "mean_aggregate": mean_agg, "llm_calls": llm.calls, "output_tokens": llm.output_tokens}, step=1) except Exception as e: print(f"[trackio] log failed: {e}") return enhanced + refined def push_results(memory, result_files=( ("results/ideas.jsonl", "results/ideas.jsonl"), ("results/summary.json", "results/summary.json"))): """Upload memory plane + run results to the Hub (ENGINE_PUSH=1).""" from huggingface_hub import HfApi api = HfApi() api.upload_file( path_or_fileobj=memory.path, path_in_repo="memory_plane.jsonl", repo_id=MEMORY_DATASET, repo_type="dataset", commit_message=f"Memory plane sync: {len(memory.records)} ideas", ) for local, remote in result_files: api.upload_file( path_or_fileobj=local, path_in_repo=remote, repo_id=RESULTS_REPO, repo_type="model", commit_message="Engine run results", ) print(f"[hub] pushed memory to {MEMORY_DATASET} and results to {RESULTS_REPO}") if __name__ == "__main__": main()