| """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 |
|
|
| |
| |
| |
|
|
| MODEL_ID = "Qwen/Qwen3-1.7B" |
| ROUNDS = 3 |
| IDEAS_PER_GEN = 2 |
| ACCEPT_THRESHOLD = 7.0 |
| REDTEAM_MIN = 6.0 |
| REFINE_TOP_K = 2 |
| MAX_NEW_TOKENS = 512 |
| CALL_LIMIT = 120 |
|
|
| 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"] |
|
|
| |
| |
| |
|
|
| 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." |
| ) |
|
|
| |
| |
| |
|
|
| 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": <idea id>, "{key}": <int 1-10>, "weakness": "<=15 words>"}}. ' |
| "Output ONLY the JSON array." |
| ) |
|
|
| |
| |
| |
|
|
| 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": <original name>, "name": <possibly improved name>, "domain": <one of: {domains}>, ' |
| '"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 |
|
|
|
|
| |
| |
| |
|
|
| 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 |
|
|
|
|
| |
| |
| |
|
|
| 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) |
|
|
|
|
| |
| |
| |
|
|
| 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: |
| 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) |
| |
| 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 |
|
|
|
|
| |
| |
| |
|
|
| 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() |
| |
| 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: |
| |
| 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) |
|
|
| |
| 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}") |
|
|
| |
| 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") |
| |
| 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)}") |
| |
| |
| 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() |
