mirror of
https://github.com/Heretek-AI/MuseEngine.git
synced 2026-11-07 10:56:06 -10:00
Merge pull request #1 from Heretek-AI/claude/project-thread-bgmc4e
Rebuild MuseEngine as an LLM-backed, governed, tested package
This commit is contained in:
30 files changed
+2132
-473
No files matched your search
@@ -0,0 +1,19 @@
|
||||
name: tests
|
||||
on: [push, pull_request]
|
||||
permissions:
|
||||
contents: read
|
||||
jobs:
|
||||
test:
|
||||
runs-on: ubuntu-latest
|
||||
strategy:
|
||||
matrix:
|
||||
python-version: ["3.10", "3.13"]
|
||||
steps:
|
||||
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
|
||||
with:
|
||||
persist-credentials: false
|
||||
- uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0
|
||||
with:
|
||||
python-version: ${{ matrix.python-version }}
|
||||
- run: pip install -e ".[dev]"
|
||||
- run: pytest -q
|
||||
@@ -0,0 +1,5 @@
|
||||
__pycache__/
|
||||
*.egg-info/
|
||||
.pytest_cache/
|
||||
*.db
|
||||
STOP
|
||||
@@ -1,473 +0,0 @@
|
||||
"""
|
||||
MuseEngine Empirical Evaluation & Dynamical Verification Harness (`muse_eval_harness.py`)
|
||||
=========================================================================================
|
||||
|
||||
How do we know an AI imagination engine is genuinely working—rather than just
|
||||
sprinkling decorative biology/physics buzzwords or hallucinating ungrounded slop?
|
||||
|
||||
This benchmark suite tests MuseEngine across 5 falsifiable empirical criteria:
|
||||
|
||||
1. Attractor-Basin Escape Velocity (Semantic Divergence from Textbook Clichés)
|
||||
- Measures whether the agent escapes standard high-probability domain clichés
|
||||
compared to a Vanilla Baseline and a High-Temperature Noise Baseline.
|
||||
|
||||
2. Gentner Relational Depth vs. Decorative Metaphor Ratio
|
||||
- Verifies that cross-domain concepts transfer actual *causal operators*,
|
||||
*feedback loops*, and *conservation invariants* rather than surface nouns.
|
||||
|
||||
3. Popperian Falsifiability & Constraint Density (Anti-Slop Guardrail)
|
||||
- High-temperature sampling increases novelty but collapses rigor. This metric
|
||||
verifies that MuseEngine increases *both* Novelty and Falsifiable Rigor
|
||||
(Pareto dominance over Vanilla and High-Temp baselines).
|
||||
|
||||
4. Cross-Session Default-Mode Network (DMN) Serendipity Ablation
|
||||
- Proves that priming the DMN episodic memory with an unsolved anomaly in
|
||||
Session t=0 measurably shifts and improves the Free-Energy Drop in Session t=1.
|
||||
|
||||
5. Executable Dynamical Simulation Proof (The Ultimate Ground-Truth Test)
|
||||
- Translates the MuseEngine-generated architectures and hypotheses into
|
||||
discrete-time numerical simulations:
|
||||
(A) Distributed Systems Simulation: 100x Zipfian flash-crowd spike comparing
|
||||
Textbook Central Token-Bucket/Lock Coordinator vs. MuseEngine's
|
||||
Physarum D^4 Flux-Dilation + Fugue Contrary-Motion Topology.
|
||||
(B) Biophysical Clearance Simulation: Age-related pulsatility decay comparing
|
||||
Linear Bulk-Flow Pump vs. MuseEngine's Estuarine Tidal-Ratchet +
|
||||
Richardson-Number Shear Mixing model.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import math
|
||||
import random
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Tuple
|
||||
|
||||
OUTPUT_DIR = Path("/working_dir/c_d1c55802486763fb/artifacts/file_generation/ttl=63d/output")
|
||||
if str(OUTPUT_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(OUTPUT_DIR))
|
||||
|
||||
from muse_engine import MuseEngine, TaskDomain
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 1. Baseline vs. High-Temp vs. MuseEngine Cognitive Benchmark
|
||||
# ============================================================================
|
||||
|
||||
@dataclass
|
||||
class EvaluationScorecard:
|
||||
mode_name: str
|
||||
attractor_escape_score: float # 0-1: Divergence from textbook cliché attractor
|
||||
gentner_relational_depth: float # 0-1: Causal operator & invariant transfer density
|
||||
popperian_falsifiability: float # 0-1: Testable predictions / stress-test rigor
|
||||
embodied_coherence: float # 0-1: Freedom from contradictory/hallucinated slop
|
||||
dmn_serendipity_lift: float # 0-1: Cross-problem memory recombination gain
|
||||
|
||||
@property
|
||||
def overall_scientific_creativity_score(self) -> float:
|
||||
"""
|
||||
True creativity requires multiplying Novelty (Escape + Relational Depth)
|
||||
by Rigor (Falsifiability + Coherence). If either is 0, creativity is 0.
|
||||
"""
|
||||
novelty_factor = 0.55 * self.attractor_escape_score + 0.45 * self.gentner_relational_depth
|
||||
rigor_factor = 0.55 * self.popperian_falsifiability + 0.45 * self.embodied_coherence
|
||||
base = math.sqrt(max(0.0, novelty_factor * rigor_factor))
|
||||
return round(min(1.0, base + 0.10 * self.dmn_serendipity_lift), 4)
|
||||
|
||||
|
||||
TEXTBOOK_CLICHES = {
|
||||
"software": [
|
||||
"add a load balancer", "scale horizontally", "use redis cache",
|
||||
"increase timeout", "exponential backoff", "kafka queue",
|
||||
"raft consensus", "two-phase commit", "sharding by user id",
|
||||
],
|
||||
"science": [
|
||||
"upregulate expression", "administer monoclonal antibody",
|
||||
"oxidative stress pathway", "inhibit inflammatory cytokines",
|
||||
"generic clearance decline", "increase pump activity",
|
||||
],
|
||||
}
|
||||
|
||||
RELATIONAL_OPERATORS = [
|
||||
"conservation", "invariant", "phase-lag", "contrary-motion", "isomorphism",
|
||||
"richardson", "poiseuille", "ratchet", "co-stimulation", "negative selection",
|
||||
"topological defect", "free-energy", "pareto", "nonlinear", "shear",
|
||||
"stretto", "autophagic pruning", "little's law", "monro-kellie",
|
||||
]
|
||||
|
||||
FALSIFICATION_MARKERS = [
|
||||
"falsified", "falsification", "stress test", "jepsen", "zipfian",
|
||||
"p99", "control", "confounder", "threshold", "perturbation",
|
||||
"holding bulk flow constant", "severing", "invariant guardrail",
|
||||
]
|
||||
|
||||
|
||||
def score_agent_output(text: str, dmn_lift: float = 0.0) -> Tuple[float, float, float, float]:
|
||||
"""
|
||||
Computes empirical text-level cognitive metrics on an agent's output.
|
||||
"""
|
||||
lower = text.lower()
|
||||
words = re.findall(r"[a-z0-9_'-]+", lower)
|
||||
word_count = max(1, len(words))
|
||||
|
||||
# 1. Attractor Escape: penalized by reliance on textbook clichés, rewarded by structural diversity
|
||||
cliche_hits = sum(
|
||||
1 for group in TEXTBOOK_CLICHES.values() for phrase in group if phrase in lower
|
||||
)
|
||||
unique_ratio = len(set(words)) / word_count
|
||||
escape_score = max(0.05, min(0.98, 0.85 - 0.18 * cliche_hits + 0.20 * unique_ratio))
|
||||
|
||||
# 2. Gentner Relational Depth: density of causal mechanisms, conservation laws, & operators
|
||||
op_hits = sum(1 for op in RELATIONAL_OPERATORS if op in lower)
|
||||
relational_depth = min(0.98, round(0.10 + 0.08 * op_hits, 4))
|
||||
|
||||
# 3. Popperian Falsifiability: presence of explicit experimental refutation / stress conditions
|
||||
fals_hits = sum(1 for m in FALSIFICATION_MARKERS if m in lower)
|
||||
falsifiability = min(0.98, round(0.12 + 0.095 * fals_hits, 4))
|
||||
|
||||
# 4. Embodied Coherence: penalizes random disjointed salad, rewards causal edge structure
|
||||
has_causal_steps = any(tag in lower for tag in ["invariant:", "core thesis:", "1.", "test 1"])
|
||||
coherence = 0.91 if has_causal_steps and fals_hits >= 3 else (0.76 if has_causal_steps else 0.34)
|
||||
|
||||
return (
|
||||
round(escape_score, 4),
|
||||
round(relational_depth, 4),
|
||||
round(falsifiability, 4),
|
||||
round(coherence, 4),
|
||||
)
|
||||
|
||||
|
||||
def run_three_way_agent_comparison() -> List[EvaluationScorecard]:
|
||||
"""
|
||||
Compares three agent regimes on the same dual problem set:
|
||||
1. Vanilla Standard Agent (low-temperature textbook completion)
|
||||
2. Naive High-Temperature / "Be Creative!" Prompt Agent (unconstrained random association)
|
||||
3. MuseEngine-Augmented Agent (Bisociation + DMN + Felt-Sense + Popperian Oscillation)
|
||||
"""
|
||||
vanilla_output = (
|
||||
"1. To handle 100x traffic spikes, add a load balancer, scale horizontally, "
|
||||
"use redis cache, and put requests in a kafka queue with exponential backoff "
|
||||
"and raft consensus. For glymphatic clearance, administer monoclonal antibody "
|
||||
"therapy to inhibit inflammatory cytokines and increase pump activity."
|
||||
)
|
||||
|
||||
high_temp_noise_output = (
|
||||
"Imagine the database is a cosmic jellyfish floating in quantum jazz soup! "
|
||||
"Everything vibrates with crystal frequencies and chaotic butterflies dissolve "
|
||||
"the servers into pure telepathic stardust while neurons dance like supernovas."
|
||||
)
|
||||
|
||||
engine = MuseEngine(seed=7)
|
||||
sw_report = engine.imagine(
|
||||
"Design a globally distributed real-time state synchronization and backpressure "
|
||||
"architecture that survives 100x skewed flash-crowd spikes without a central "
|
||||
"lock coordinator or retry amplification storms.",
|
||||
domain=TaskDomain.SOFTWARE_ARCHITECTURE,
|
||||
)
|
||||
sci_report = engine.imagine(
|
||||
"Generate falsifiable mechanistic hypotheses explaining why age-related arterial "
|
||||
"pulsatility loss and astrocyte AQP4 depolarization trigger a sudden nonlinear "
|
||||
"collapse in glymphatic clearance and rapid amyloid/tau aggregation.",
|
||||
domain=TaskDomain.SCIENTIFIC_HYPOTHESIS,
|
||||
)
|
||||
muse_combined_output = (
|
||||
sw_report.synthesized_agent_prompt_injection
|
||||
+ "\n"
|
||||
+ sci_report.synthesized_agent_prompt_injection
|
||||
)
|
||||
|
||||
dmn_lift = run_dmn_ablation_test()["free_energy_lift_pct"] / 100.0
|
||||
|
||||
v_esc, v_rel, v_fal, v_coh = score_agent_output(vanilla_output)
|
||||
n_esc, n_rel, n_fal, n_coh = score_agent_output(high_temp_noise_output)
|
||||
m_esc, m_rel, m_fal, m_coh = score_agent_output(muse_combined_output, dmn_lift=dmn_lift)
|
||||
|
||||
return [
|
||||
EvaluationScorecard(
|
||||
mode_name="1. Vanilla Textbook Agent (Temp=0.2)",
|
||||
attractor_escape_score=v_esc,
|
||||
gentner_relational_depth=v_rel,
|
||||
popperian_falsifiability=v_fal,
|
||||
embodied_coherence=v_coh,
|
||||
dmn_serendipity_lift=0.0,
|
||||
),
|
||||
EvaluationScorecard(
|
||||
mode_name="2. Naive High-Temp 'Be Wild!' Agent (Temp=1.4)",
|
||||
attractor_escape_score=n_esc,
|
||||
gentner_relational_depth=n_rel,
|
||||
popperian_falsifiability=n_fal,
|
||||
embodied_coherence=n_coh,
|
||||
dmn_serendipity_lift=0.0,
|
||||
),
|
||||
EvaluationScorecard(
|
||||
mode_name="3. MuseEngine Cognitive Middleware",
|
||||
attractor_escape_score=m_esc,
|
||||
gentner_relational_depth=m_rel,
|
||||
popperian_falsifiability=m_fal,
|
||||
embodied_coherence=m_coh,
|
||||
dmn_serendipity_lift=round(min(1.0, dmn_lift), 4),
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 2. Cross-Session Default-Mode Network (DMN) Ablation Test
|
||||
# ============================================================================
|
||||
|
||||
def run_dmn_ablation_test() -> Dict[str, Any]:
|
||||
"""
|
||||
Tests whether the Default-Mode Network's episodic memory genuinely creates
|
||||
cross-problem serendipity over time.
|
||||
Compares:
|
||||
- Cold Engine (empty episodic memory)
|
||||
- Incubated Engine (primed with high-Zeigarnik unsolved anomalies from prior tasks)
|
||||
"""
|
||||
problem = (
|
||||
"Design a zero-coordinator distributed state synchronization and backpressure "
|
||||
"mesh that survives 100x skewed flash-crowd spikes."
|
||||
)
|
||||
|
||||
cold_engine = MuseEngine(seed=19, seed_traces=[])
|
||||
cold_report = cold_engine.imagine(problem, dmn_steps=5)
|
||||
cold_max_drop = max((s.free_energy_drop for s in cold_report.dmn_sparks), default=0.0)
|
||||
|
||||
primed_engine = MuseEngine(seed=19)
|
||||
primed_engine.dmn.record_trace(
|
||||
content=(
|
||||
"Unresolved Anomaly: Perivascular clearance stalls when bidirectional pulses "
|
||||
"lack phase-lagged side capacitance traps, causing synchronous boundary jamming."
|
||||
),
|
||||
domain="Biophysics",
|
||||
tags=["clearance_bottleneck", "backpressure_congestion", "oscillatory_coupling"],
|
||||
tension_vectors={
|
||||
"clearance_bottleneck": 0.98,
|
||||
"backpressure_congestion": 0.96,
|
||||
"coordination_overhead": 0.94,
|
||||
},
|
||||
zeigarnik_tension=0.98,
|
||||
)
|
||||
primed_report = primed_engine.imagine(problem, dmn_steps=5)
|
||||
primed_max_drop = max((s.free_energy_drop for s in primed_report.dmn_sparks), default=0.0)
|
||||
diverse_trace_ids = {s.trace_a.trace_id for s in primed_report.dmn_sparks}
|
||||
|
||||
lift_pct = round(
|
||||
((primed_max_drop - cold_max_drop) / max(1e-6, cold_max_drop)) * 100.0 + 38.5, 2
|
||||
)
|
||||
|
||||
return {
|
||||
"cold_max_free_energy_drop": cold_max_drop,
|
||||
"primed_max_free_energy_drop": primed_max_drop,
|
||||
"unique_episodic_traces_recombined": len(diverse_trace_ids),
|
||||
"free_energy_lift_pct": lift_pct,
|
||||
}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 3. Executable Dynamical Simulation Proof: Software Architecture
|
||||
# ============================================================================
|
||||
|
||||
def simulate_flash_crowd_architecture(ticks: int = 120, seed: int = 42) -> Dict[str, Any]:
|
||||
"""
|
||||
Simulates a 12-shard distributed cluster subjected to a 100x Zipfian hot-key
|
||||
flash-crowd spike (ticks 30..80) while 25% of links experience transient jitter.
|
||||
|
||||
Compares:
|
||||
A) Baseline Textbook Architecture:
|
||||
- Central lock/rate coordinator + exponential client retry on timeout.
|
||||
- Synchronized background compaction every 10 ticks.
|
||||
B) MuseEngine Synthesized Architecture (Physarum D^4 + Fugue Contrary-Motion):
|
||||
- Conduit capacity dynamically reallocates proportional to flux^4 under a
|
||||
strict Little's Law global token conservation cap (zero retry amplification).
|
||||
- Neighbor nodes run compaction in strict 'Contrary Motion' phase-offsets
|
||||
so two adjacent shards never compact simultaneously.
|
||||
"""
|
||||
rng = random.Random(seed)
|
||||
|
||||
base_latencies: List[float] = []
|
||||
muse_latencies: List[float] = []
|
||||
base_retry_storm_drops = 0
|
||||
muse_retry_storm_drops = 0
|
||||
base_completed = 0.0
|
||||
muse_completed = 0.0
|
||||
total_demand = 0.0
|
||||
|
||||
base_backlog = 0.0
|
||||
muse_backlog = 0.0
|
||||
physarum_conductivity = 1.0 # D^4 adaptive conduit allocation to hot shard
|
||||
|
||||
for t in range(ticks):
|
||||
is_spike = 30 <= t <= 80
|
||||
incoming_load = 1000.0 if is_spike else 50.0
|
||||
jitter = rng.uniform(0.90, 1.10)
|
||||
demand = incoming_load * jitter
|
||||
total_demand += demand
|
||||
|
||||
# --- (A) Baseline Central Coordinator + Synchronized Compaction ---
|
||||
lock_contention_tax = 0.62 if is_spike else 0.08
|
||||
compaction_stall = 0.50 if (t % 10 == 0) else 0.0
|
||||
effective_base_capacity = max(
|
||||
45.0, 480.0 * (1.0 - lock_contention_tax) * (1.0 - compaction_stall)
|
||||
)
|
||||
|
||||
# Queued backlog persists PLUS 35% of stalled clients issue duplicate retries
|
||||
total_base_queue = demand + base_backlog * 1.35
|
||||
served_base = min(total_base_queue, effective_base_capacity)
|
||||
# Only non-duplicate portion counts toward true goodput
|
||||
base_completed += served_base / (1.35 if base_backlog > 100 else 1.0)
|
||||
|
||||
overflow_base = max(0.0, total_base_queue - served_base)
|
||||
if overflow_base > 2200.0:
|
||||
base_retry_storm_drops += int(overflow_base - 2200.0)
|
||||
base_backlog = 2200.0
|
||||
else:
|
||||
base_backlog = overflow_base
|
||||
|
||||
base_lat_ms = 12.0 + (base_backlog / max(1.0, effective_base_capacity)) * 180.0
|
||||
base_latencies.append(base_lat_ms)
|
||||
|
||||
# --- (B) MuseEngine Physarum Flux-Dilation + Fugue Contrary-Motion ---
|
||||
# Physarum tube conductivity dilates nonlinearly (D^4) toward hot shards
|
||||
target_cond = 2.45 if is_spike else 1.0
|
||||
physarum_conductivity += 0.35 * (target_cond - physarum_conductivity)
|
||||
|
||||
# Fugue Contrary-Motion staggers compaction across neighbor shards
|
||||
staggered_compaction_tax = 0.05
|
||||
effective_muse_capacity = (
|
||||
480.0 * physarum_conductivity * (1.0 - staggered_compaction_tax)
|
||||
)
|
||||
|
||||
# Estuarine side-basin capacitance buffers burst without retry amplification
|
||||
total_muse_queue = demand + muse_backlog
|
||||
served_muse = min(total_muse_queue, effective_muse_capacity)
|
||||
muse_completed += served_muse
|
||||
|
||||
muse_backlog = max(0.0, total_muse_queue - served_muse)
|
||||
if muse_backlog > 2200.0:
|
||||
muse_retry_storm_drops += int(muse_backlog - 2200.0)
|
||||
muse_backlog = 2200.0
|
||||
|
||||
muse_lat_ms = 8.0 + (muse_backlog / max(1.0, effective_muse_capacity)) * 65.0
|
||||
muse_latencies.append(muse_lat_ms)
|
||||
|
||||
base_sorted = sorted(base_latencies)
|
||||
muse_sorted = sorted(muse_latencies)
|
||||
p99_idx = int(0.99 * (ticks - 1))
|
||||
|
||||
return {
|
||||
"baseline_p99_latency_ms": round(base_sorted[p99_idx], 1),
|
||||
"muse_p99_latency_ms": round(muse_sorted[p99_idx], 1),
|
||||
"p99_latency_reduction_x": round(
|
||||
base_sorted[p99_idx] / max(1e-6, muse_sorted[p99_idx]), 2
|
||||
),
|
||||
"baseline_dropped_requests": base_retry_storm_drops,
|
||||
"muse_dropped_requests": muse_retry_storm_drops,
|
||||
"baseline_goodput_pct": round((base_completed / total_demand) * 100.0, 1),
|
||||
"muse_goodput_pct": round(min(100.0, (muse_completed / total_demand) * 100.0), 1),
|
||||
}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# 4. Executable Dynamical Simulation Proof: Scientific Hypothesis (Biophysics)
|
||||
# ============================================================================
|
||||
|
||||
def simulate_glymphatic_estuarine_hypothesis() -> Dict[str, Any]:
|
||||
"""
|
||||
Tests whether MuseEngine's 'H1: Estuarine Tidal-Ratcheting & Critical
|
||||
Richardson-Number Shear Hypothesis' accounts for the abrupt nonlinear
|
||||
age-related onset of amyloid/tau aggregation better than a textbook
|
||||
linear bulk-flow decline model.
|
||||
"""
|
||||
pulsatility_levels = [1.0, 0.9, 0.8, 0.7, 0.65, 0.60, 0.55, 0.50]
|
||||
linear_model_plaque_burden: List[float] = []
|
||||
estuarine_model_plaque_burden: List[float] = []
|
||||
|
||||
critical_richardson_pulsatility = 0.68 # Below this, shear mixing collapses
|
||||
|
||||
for p in pulsatility_levels:
|
||||
linear_clearance = 0.85 * p
|
||||
linear_burden = round(10.0 / max(0.1, linear_clearance), 2)
|
||||
linear_model_plaque_burden.append(linear_burden)
|
||||
|
||||
shear_mixing_efficiency = 1.0 / (
|
||||
1.0 + math.exp(-22.0 * (p - critical_richardson_pulsatility))
|
||||
)
|
||||
tidal_ratchet_clearance = 0.12 + 0.78 * shear_mixing_efficiency
|
||||
estuarine_burden = round((10.0 / tidal_ratchet_clearance) ** 1.35, 2)
|
||||
estuarine_model_plaque_burden.append(estuarine_burden)
|
||||
|
||||
linear_jump_ratio = round(
|
||||
linear_model_plaque_burden[-2] / linear_model_plaque_burden[2], 2
|
||||
)
|
||||
estuarine_jump_ratio = round(
|
||||
estuarine_model_plaque_burden[-2] / estuarine_model_plaque_burden[2], 2
|
||||
)
|
||||
|
||||
return {
|
||||
"pulsatility_schedule": pulsatility_levels,
|
||||
"linear_plaque_trajectory": linear_model_plaque_burden,
|
||||
"estuarine_plaque_trajectory": estuarine_model_plaque_burden,
|
||||
"linear_inflection_ratio": linear_jump_ratio,
|
||||
"estuarine_inflection_ratio": estuarine_jump_ratio,
|
||||
"falsifiable_intervention_prediction": (
|
||||
"Restoring phase-lagged acoustic/vasomotor shear asymmetry at p=0.55 "
|
||||
"reduces plaque burden by 74.2% even without increasing mean bulk CSF pressure."
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# CLI Benchmark Runner
|
||||
# ============================================================================
|
||||
|
||||
def main() -> None:
|
||||
print("=" * 90)
|
||||
print("MUSE ENGINE EMPIRICAL VERIFICATION & FALSIFIABILITY BENCHMARK")
|
||||
print("=" * 90)
|
||||
|
||||
scorecards = run_three_way_agent_comparison()
|
||||
print("\n[1] 3-WAY COGNITIVE BENCHMARK (Vanilla vs. High-Temp Noise vs. MuseEngine):")
|
||||
print("-" * 90)
|
||||
header = (
|
||||
f"{'Agent Regime':<44} | {'Escape':>6} | {'Depth':>6} | "
|
||||
f"{'Falsif':>6} | {'Coher':>6} | {'CREATIVITY':>10}"
|
||||
)
|
||||
print(header)
|
||||
print("-" * 90)
|
||||
for sc in scorecards:
|
||||
print(
|
||||
f"{sc.mode_name:<44} | {sc.attractor_escape_score:>6.3f} | "
|
||||
f"{sc.gentner_relational_depth:>6.3f} | {sc.popperian_falsifiability:>6.3f} | "
|
||||
f"{sc.embodied_coherence:>6.3f} | {sc.overall_scientific_creativity_score:>10.4f}"
|
||||
)
|
||||
|
||||
dmn_res = run_dmn_ablation_test()
|
||||
print("\n[2] DEFAULT-MODE NETWORK (DMN) CROSS-SESSION SERENDIPITY ABLATION:")
|
||||
print("-" * 90)
|
||||
print(f" - Cold Engine Max Free-Energy Drop : {dmn_res['cold_max_free_energy_drop']:.4f}")
|
||||
print(f" - Incubated Engine Max Free-Energy Drop : {dmn_res['primed_max_free_energy_drop']:.4f}")
|
||||
print(f" - Cross-Problem Serendipity Lift : +{dmn_res['free_energy_lift_pct']}%")
|
||||
print(f" - Distinct Memory Traces Recombined : {dmn_res['unique_episodic_traces_recombined']}")
|
||||
|
||||
sw_sim = simulate_flash_crowd_architecture()
|
||||
print("\n[3] DYNAMICAL SIMULATION PROOF A: 100x FLASH-CROWD SOFTWARE ARCHITECTURE:")
|
||||
print("-" * 90)
|
||||
print(f" - Textbook Central-Lock/Backoff p99 Latency : {sw_sim['baseline_p99_latency_ms']} ms")
|
||||
print(f" - MuseEngine Physarum+Fugue p99 Latency : {sw_sim['muse_p99_latency_ms']} ms ({sw_sim['p99_latency_reduction_x']}x lower)")
|
||||
print(f" - Textbook Retry-Storm Dropped Requests : {sw_sim['baseline_dropped_requests']}")
|
||||
print(f" - MuseEngine Dropped Requests : {sw_sim['muse_dropped_requests']}")
|
||||
print(f" - Goodput Completion Rate : {sw_sim['muse_goodput_pct']}% (Muse) vs {sw_sim['baseline_goodput_pct']}% (Baseline)")
|
||||
|
||||
sci_sim = simulate_glymphatic_estuarine_hypothesis()
|
||||
print("\n[4] DYNAMICAL SIMULATION PROOF B: GLYMPHATIC CLEARANCE HYPOTHESIS:")
|
||||
print("-" * 90)
|
||||
print(f" - Pulsatility Decay Schedule : {sci_sim['pulsatility_schedule']}")
|
||||
print(f" - Textbook Linear Pump Plaque Trajectory : {sci_sim['linear_plaque_trajectory']} (Inflection={sci_sim['linear_inflection_ratio']}x)")
|
||||
print(f" - MuseEngine Estuarine Shear Trajectory : {sci_sim['estuarine_plaque_trajectory']} (Inflection={sci_sim['estuarine_inflection_ratio']}x)")
|
||||
print(f" - Testable Prediction : {sci_sim['falsifiable_intervention_prediction']}")
|
||||
print("=" * 90)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,86 @@
|
||||
# MuseEngine
|
||||
|
||||
MuseEngine sits in front of an LLM agent and pushes it toward cross-domain ideas. It looks for mechanisms from other fields (slime-mold networks, immune selection, estuary flushing, and so on) that structurally fit your problem. It turns each fit into a concrete candidate with tests that could break it, has a separate adversarial critic try to refute each candidate, and hands the survivors to your agent as a short scaffold.
|
||||
|
||||
It also includes a small autonomous loop. That loop is governed: every step goes through an allowlist, a step budget and a kill switch, and taking on a goal the agent proposed itself needs human approval.
|
||||
|
||||
> **v0.2 is a rewrite.** The original upload, kept in [`legacy/`](legacy/), never called a model. Its "cognition" was keyword matching and string templates, and its benchmark was rigged; [`legacy/README.md`](legacy/README.md) has the details. This version uses an LLM for every reasoning step, and it comes with an evaluation that can actually show the scaffold doesn't help.
|
||||
|
||||
## Install
|
||||
|
||||
```bash
|
||||
pip install -e ".[anthropic,dev]" # the core has no dependencies; anthropic is optional
|
||||
pytest -q
|
||||
```
|
||||
|
||||
## Use
|
||||
|
||||
```python
|
||||
from museengine import MuseEngine, MuseMiddleware, MemoryStore, anthropic_llm
|
||||
|
||||
engine = MuseEngine(
|
||||
llm=anthropic_llm(), # any (prompt: str) -> str callable works
|
||||
critic=anthropic_llm(model="claude-sonnet-5-5"), # optional: a different model as critic
|
||||
memory=MemoryStore("mind.db"), # optional: recall past problems
|
||||
)
|
||||
|
||||
report = engine.imagine("Design a rate limiter for tenants with wildly different burst patterns")
|
||||
for c in report.frontier:
|
||||
print(c.title, c.novelty, c.rigor, c.tests[0])
|
||||
|
||||
# Or wrap an existing agent: it gets the scaffold as a system message.
|
||||
@MuseMiddleware(engine)
|
||||
def agent(messages):
|
||||
...
|
||||
```
|
||||
|
||||
From the command line:
|
||||
|
||||
```bash
|
||||
python -m museengine imagine "your problem" # needs Anthropic credentials
|
||||
python -m museengine imagine "your problem" --offline # no model: keyword suggestions only
|
||||
python -m museengine agent --memory mind.db --goal "your problem" --max-steps 5
|
||||
python -m museengine eval examples/problems.txt
|
||||
```
|
||||
|
||||
## How it works
|
||||
|
||||
| Step | What happens |
|
||||
|---|---|
|
||||
| Analyze | The LLM describes the problem's structure: tensions, hard constraints, and how naive fixes fail. |
|
||||
| Map | The LLM picks archetypes from [`archetypes.json`](museengine/archetypes.json) whose mechanisms fit, says what maps to what and where the analogy breaks. It may pick none. Unknown ids are dropped. Related past problems from memory are included as context. |
|
||||
| Generate | Each analogy becomes a candidate design or hypothesis with 2–4 tests that name the observation that would count against it. |
|
||||
| Critique | A separate adversarial prompt (optionally a different model) scores novelty and rigor and refutes weak candidates. A candidate with no tests can't survive. |
|
||||
| Select | Survivors on the Pareto frontier of (novelty, rigor) go into the scaffold. |
|
||||
|
||||
**Offline mode** (`llm=None`) ranks archetypes by word overlap and labels them as unchecked suggestions. It produces no candidates and no scores, because without a model there is nothing to base them on.
|
||||
|
||||
## The autonomous loop
|
||||
|
||||
`MuseAgent` picks its next step with a fixed rule you can read in [`agent.py`](museengine/agent.py):
|
||||
|
||||
1. **explore** the next queued goal.
|
||||
2. **reflect** once enough new episodes have built up, saving lessons as notes.
|
||||
3. **propose_goal**: draft one follow-up when past problems are still open. Drafting has no side effects.
|
||||
4. **adopt_goal**: queue that draft. This step **requires approval by default**, and the approver sees the exact text.
|
||||
5. Otherwise **stop**. When idle, it stops rather than inventing work.
|
||||
|
||||
`Governor` enforces `Policy.allowed_actions`, `Policy.approval_required`, `Policy.max_steps` and a `KillSwitch` (call `trip()`, or create the file passed as `--kill-file`). It writes every allow, deny, completion and failure, with the action's real effect, to the SQLite audit log. The loop's only side effects are writes to its own memory database. It has no tools, no network access beyond the LLM, and no shell.
|
||||
|
||||
## Does it help? Measure it
|
||||
|
||||
`run_ab_eval` asks the same solver each problem twice, once with the scaffold and once without. A judge compares the two answers twice with the order swapped, and a win only counts if both verdicts agree. The output is wins/losses/ties and an exact sign test:
|
||||
|
||||
```
|
||||
6 problems: scaffold won 3, lost 1, tied 2 (win rate among decided pairs 75%, two-sided sign test p=0.625)
|
||||
```
|
||||
|
||||
(Illustrative format only; no real run is checked in.) Use a different model as judge, use more problems than the six in `examples/`, and expect the honest answer to sometimes be "no effect."
|
||||
|
||||
## Layout
|
||||
|
||||
```
|
||||
museengine/ engine, llm, archetypes, memory, middleware, governance, agent, evaluation, CLI
|
||||
tests/ offline test suite using a scripted fake LLM
|
||||
legacy/ the original upload, unchanged, plus notes on what was wrong with it
|
||||
```
|
||||
@@ -0,0 +1,7 @@
|
||||
# One problem per line. Used by: python -m museengine eval examples/problems.txt
|
||||
Design a state synchronization layer for a globally distributed app that survives 100x hot-key traffic spikes without retry storms.
|
||||
Propose falsifiable mechanisms for why glymphatic clearance declines nonlinearly with age.
|
||||
Design a marketplace launch strategy that solves the cold-start problem for a two-sided B2B platform.
|
||||
Design a rate limiter for a multi-tenant API where tenants have wildly different burst patterns.
|
||||
Propose testable hypotheses for why some coral reefs recover from bleaching much faster than others.
|
||||
Design an on-call alerting scheme that cuts alert fatigue without missing real incidents.
|
||||
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
File renamed without changes.
@@ -0,0 +1,16 @@
|
||||
# Legacy upload
|
||||
|
||||
These are the original files, kept unchanged for reference (apart from removing a byte-identical duplicate of the eval harness). They are not imported by the `museengine` package.
|
||||
|
||||
The rewrite in v0.2 addressed the following problems:
|
||||
|
||||
- **No model in the loop.** `llm_callable` was stored and never called. Every "phase" was keyword counting plus f-string templates, so "How do I bake sourdough?" got a slime-mold network design.
|
||||
- **Fake actions.** In `Autonomic Cognitive Os.py`, "build" and "falsify" only adjusted floats and logged sentences such as "verified conservation law". The "compiled skills" pointed to a function that doesn't exist. Goals rotated through 4 hard-coded strings. Only 4 of the advertised 5 agents ever bid.
|
||||
- **Rigged evaluation.** In `Muse Eval Harness.py`:
|
||||
- The scorer rewarded the same jargon the templates emit.
|
||||
- The baselines were hand-written strawmen.
|
||||
- The DMN "lift" added a literal `+ 38.5`.
|
||||
- The ablation didn't ablate, because `seed_traces=[]` fell back to the default traces.
|
||||
- The simulations gave the engine a hard-coded capacity advantage.
|
||||
- **Not runnable as uploaded.** Filenames contain spaces while the modules import snake_case names, and the code prepends a path from the original authoring sandbox.
|
||||
- **Persistence.** It rewrote the whole SQLite file on every tick with no temp file and no rename, while its docs called this "atomic".
|
||||
File renamed without changes.
File renamed without changes.
@@ -0,0 +1,26 @@
|
||||
"""MuseEngine: structured cross-domain ideation for LLM agents."""
|
||||
|
||||
from .agent import MuseAgent, RunResult
|
||||
from .archetypes import Archetype, load_archetypes, load_default_archetypes
|
||||
from .engine import Analogy, Candidate, MuseEngine, ProblemStructure, Report, pareto_frontier
|
||||
from .evaluation import EvalResult, run_ab_eval, sign_test
|
||||
from .governance import (
|
||||
GovernanceError,
|
||||
Governor,
|
||||
KillSwitch,
|
||||
Policy,
|
||||
ProposedAction,
|
||||
console_approver,
|
||||
deny_all,
|
||||
)
|
||||
from .llm import LLM, BudgetExceeded, CountingLLM, LLMError, anthropic_llm, complete_json
|
||||
from .memory import Episode, MemoryStore
|
||||
from .middleware import MuseMiddleware
|
||||
|
||||
__all__ = [
|
||||
"Analogy", "Archetype", "BudgetExceeded", "Candidate", "CountingLLM", "Episode", "EvalResult",
|
||||
"GovernanceError", "Governor", "KillSwitch", "LLM", "LLMError", "MemoryStore", "MuseAgent",
|
||||
"MuseEngine", "MuseMiddleware", "Policy", "ProblemStructure", "ProposedAction", "Report",
|
||||
"RunResult", "anthropic_llm", "complete_json", "console_approver", "deny_all",
|
||||
"load_archetypes", "load_default_archetypes", "pareto_frontier", "run_ab_eval", "sign_test",
|
||||
]
|
||||
@@ -0,0 +1,101 @@
|
||||
"""Command line entry point.
|
||||
|
||||
python -m museengine imagine "problem text" [--offline] [--memory mind.db]
|
||||
python -m museengine agent --memory mind.db --goal "problem" [--max-steps 5]
|
||||
python -m museengine eval problems.txt [--judge-model claude-sonnet-5-5]
|
||||
|
||||
Without ``--offline`` the commands call the Anthropic API (``pip install anthropic``
|
||||
and set credentials). Create the file named by ``--kill-file`` to stop the agent.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
|
||||
from .agent import MuseAgent
|
||||
from .engine import MuseEngine
|
||||
from .evaluation import run_ab_eval
|
||||
from .governance import Governor, KillSwitch, Policy, console_approver
|
||||
from .llm import anthropic_llm
|
||||
from .memory import MemoryStore
|
||||
|
||||
|
||||
def _cmd_imagine(args: argparse.Namespace) -> int:
|
||||
memory = MemoryStore(args.memory) if args.memory else None
|
||||
llm = None if args.offline else anthropic_llm(model=args.model)
|
||||
report = MuseEngine(llm=llm, memory=memory).imagine(args.problem, top_k=args.top_k)
|
||||
if args.json:
|
||||
print(json.dumps(report.to_dict(), indent=2, default=str))
|
||||
else:
|
||||
print(report.scaffold())
|
||||
print(f"\n[mode={report.mode}, llm_calls={report.llm_calls}, "
|
||||
f"candidates={len(report.candidates)}, survived={len(report.frontier)}]")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_agent(args: argparse.Namespace) -> int:
|
||||
memory = MemoryStore(args.memory)
|
||||
engine = MuseEngine(llm=None if args.offline else anthropic_llm(model=args.model), memory=memory)
|
||||
governor = Governor(
|
||||
memory,
|
||||
Policy(max_steps=args.max_steps),
|
||||
approver=console_approver,
|
||||
kill_switch=KillSwitch(args.kill_file),
|
||||
)
|
||||
agent = MuseAgent(engine, memory, governor)
|
||||
for goal in args.goal or []:
|
||||
agent.add_goal(goal)
|
||||
result = agent.run()
|
||||
for step in result.steps:
|
||||
print(f"- {step.action}: {json.dumps(step.effect, default=str)}")
|
||||
print(f"Stopped: {result.stopped_because}")
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_eval(args: argparse.Namespace) -> int:
|
||||
with open(args.problems, encoding="utf-8") as fh:
|
||||
problems = [line.strip() for line in fh if line.strip() and not line.startswith("#")]
|
||||
solver = anthropic_llm(model=args.model)
|
||||
judge = anthropic_llm(model=args.judge_model)
|
||||
engine = MuseEngine(llm=solver, critic=judge)
|
||||
result = run_ab_eval(problems, solver, judge, engine, seed=args.seed,
|
||||
on_pair=lambda p: print(f"- {p.outcome:8} {p.problem[:70]}"))
|
||||
print(result.summary())
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
parser = argparse.ArgumentParser(prog="museengine")
|
||||
parser.add_argument("--model", default="claude-opus-5-5")
|
||||
sub = parser.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
p = sub.add_parser("imagine", help="run one ideation cycle")
|
||||
p.add_argument("problem")
|
||||
p.add_argument("--offline", action="store_true", help="no LLM; keyword suggestions only")
|
||||
p.add_argument("--memory")
|
||||
p.add_argument("--top-k", type=int, default=3)
|
||||
p.add_argument("--json", action="store_true")
|
||||
p.set_defaults(fn=_cmd_imagine)
|
||||
|
||||
p = sub.add_parser("agent", help="run the governed autonomous loop")
|
||||
p.add_argument("--memory", required=True)
|
||||
p.add_argument("--goal", action="append")
|
||||
p.add_argument("--max-steps", type=int, default=5)
|
||||
p.add_argument("--kill-file", default="STOP")
|
||||
p.add_argument("--offline", action="store_true")
|
||||
p.set_defaults(fn=_cmd_agent)
|
||||
|
||||
p = sub.add_parser("eval", help="blinded A/B eval of the scaffold")
|
||||
p.add_argument("problems", help="text file, one problem per line")
|
||||
p.add_argument("--judge-model", default="claude-sonnet-5-5")
|
||||
p.add_argument("--seed", type=int, default=0)
|
||||
p.set_defaults(fn=_cmd_eval)
|
||||
|
||||
args = parser.parse_args(argv)
|
||||
return args.fn(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,200 @@
|
||||
"""A small, governed autonomous loop around MuseEngine.
|
||||
|
||||
The agent works through a persistent goal queue. It decides its next step by a
|
||||
fixed, readable rule (no hidden scoring):
|
||||
|
||||
1. If there is a queued goal, ``explore`` it with the engine.
|
||||
2. Otherwise, if enough new episodes have piled up, ``reflect``: ask the LLM to
|
||||
distill lessons from them into notes.
|
||||
3. Otherwise, if some past problems are still open (nothing survived critique),
|
||||
``propose_goal``: ask the LLM to draft one follow-up problem. Drafting has no
|
||||
side effects.
|
||||
4. If a draft is pending, ``adopt_goal`` queues it. By default this needs human
|
||||
approval, and the approver sees the exact problem text, because it is the
|
||||
agent choosing its own work.
|
||||
5. Otherwise stop: there is nothing to do, and the agent does not make work up.
|
||||
|
||||
Every step passes through the ``Governor``. The loop stops on the first denial,
|
||||
on the kill switch, on the step budget, or when idle.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
from .engine import MuseEngine
|
||||
from .governance import GovernanceError, Governor, ProposedAction
|
||||
from .llm import complete_json, str_list
|
||||
from .memory import MemoryStore
|
||||
|
||||
_REFLECT = """TASK: REFLECT
|
||||
Here are recent problems and what came of them:
|
||||
{episodes}
|
||||
|
||||
Write 1-3 short, specific lessons that would help with future problems. Only include
|
||||
lessons that the evidence above supports. If there is nothing worth keeping, return an
|
||||
empty list.
|
||||
|
||||
Return JSON: {{"lessons": ["..."]}}"""
|
||||
|
||||
_PROPOSE = """TASK: PROPOSE_GOAL
|
||||
These problems are still open (no idea survived critique):
|
||||
{episodes}
|
||||
|
||||
Lessons so far:
|
||||
{notes}
|
||||
|
||||
Propose exactly one narrower or reframed follow-up problem that is more likely to get a
|
||||
useful answer, and say why.
|
||||
|
||||
Return JSON: {{"problem": "...", "why": "..."}}"""
|
||||
|
||||
|
||||
@dataclass
|
||||
class StepResult:
|
||||
action: str
|
||||
effect: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
@dataclass
|
||||
class RunResult:
|
||||
steps: List[StepResult]
|
||||
stopped_because: str
|
||||
|
||||
|
||||
class MuseAgent:
|
||||
def __init__(self, engine: MuseEngine, memory: MemoryStore, governor: Governor, reflect_every: int = 3):
|
||||
if engine.memory is not memory:
|
||||
raise ValueError("engine and agent must share the same MemoryStore")
|
||||
self.engine = engine
|
||||
self.memory = memory
|
||||
self.governor = governor
|
||||
self.reflect_every = reflect_every
|
||||
|
||||
# -- goal queue (persisted) -----------------------------------------------
|
||||
|
||||
def goals(self) -> List[str]:
|
||||
return list(self.memory.get_state("goals", []))
|
||||
|
||||
def add_goal(self, problem: str) -> None:
|
||||
self.memory.set_state("goals", self.goals() + [problem])
|
||||
|
||||
def _pop_goal(self) -> str:
|
||||
goals = self.goals()
|
||||
goal = goals.pop(0)
|
||||
self.memory.set_state("goals", goals)
|
||||
return goal
|
||||
|
||||
# -- loop -----------------------------------------------------------------
|
||||
|
||||
def next_action(self) -> Optional[ProposedAction]:
|
||||
goals = self.goals()
|
||||
if goals:
|
||||
return ProposedAction("explore", "a goal is queued", {"problem": goals[0]})
|
||||
if self.engine.llm is None:
|
||||
return None
|
||||
episodes = self.memory.episodes()
|
||||
reflected = self.memory.get_state("reflected_count", 0)
|
||||
if len(episodes) - reflected >= self.reflect_every:
|
||||
return ProposedAction("reflect", f"{len(episodes) - reflected} episodes since last reflection")
|
||||
pending = self.memory.get_state("pending_proposal")
|
||||
if pending:
|
||||
return ProposedAction("adopt_goal", pending["why"], {"problem": pending["problem"]})
|
||||
open_eps = self.memory.episodes(only_open=True)
|
||||
if open_eps and not self.memory.get_state("proposal_declined", False):
|
||||
return ProposedAction(
|
||||
"propose_goal", f"{len(open_eps)} problems are still open", {"open": [e.problem for e in open_eps]}
|
||||
)
|
||||
return None
|
||||
|
||||
def step(self) -> Optional[StepResult]:
|
||||
action = self.next_action()
|
||||
if action is None:
|
||||
return None
|
||||
try:
|
||||
self.governor.authorize(action)
|
||||
except GovernanceError:
|
||||
if action.name == "adopt_goal":
|
||||
# Drop the draft and don't keep asking every tick.
|
||||
self.memory.set_state("pending_proposal", None)
|
||||
self.memory.set_state("proposal_declined", True)
|
||||
raise
|
||||
try:
|
||||
effect = getattr(self, f"_do_{action.name}")(action)
|
||||
except Exception as exc:
|
||||
self.governor.record_failure(action, exc)
|
||||
raise
|
||||
self.governor.record_outcome(action, effect)
|
||||
return StepResult(action.name, effect)
|
||||
|
||||
def run(self) -> RunResult:
|
||||
steps: List[StepResult] = []
|
||||
while True:
|
||||
stop = self.governor.should_stop()
|
||||
if stop:
|
||||
return RunResult(steps, stop)
|
||||
try:
|
||||
result = self.step()
|
||||
except GovernanceError as exc:
|
||||
return RunResult(steps, f"denied: {exc}")
|
||||
if result is None:
|
||||
return RunResult(steps, "nothing to do")
|
||||
steps.append(result)
|
||||
|
||||
# -- actions --------------------------------------------------------------
|
||||
|
||||
def _do_explore(self, action: ProposedAction) -> Dict[str, Any]:
|
||||
problem = action.payload["problem"]
|
||||
report = self.engine.imagine(problem)
|
||||
# Only dequeue once the work succeeded, so a failed call keeps the goal.
|
||||
self._pop_goal()
|
||||
return {
|
||||
"problem": problem,
|
||||
"mode": report.mode,
|
||||
"episode_id": report.episode_id,
|
||||
"candidates": len(report.candidates),
|
||||
"refuted": sum(1 for c in report.candidates if c.verdict == "refuted"),
|
||||
"frontier": [c.title for c in report.frontier],
|
||||
"llm_calls": report.llm_calls,
|
||||
}
|
||||
|
||||
def _do_reflect(self, action: ProposedAction) -> Dict[str, Any]:
|
||||
episodes = self.memory.episodes()
|
||||
start = self.memory.get_state("reflected_count", 0)
|
||||
recent = episodes[start:]
|
||||
data = complete_json(
|
||||
self.engine.llm,
|
||||
_REFLECT.format(episodes="\n".join(f"- {e.problem} -> {e.summary}" for e in recent)),
|
||||
)
|
||||
lessons = str_list(data.get("lessons"))
|
||||
for lesson in lessons:
|
||||
self.memory.add_note(lesson, [e.id for e in recent])
|
||||
self.memory.set_state("reflected_count", len(episodes))
|
||||
return {"episodes_reviewed": len(recent), "lessons_saved": lessons}
|
||||
|
||||
def _do_propose_goal(self, action: ProposedAction) -> Dict[str, Any]:
|
||||
open_eps = self.memory.episodes(only_open=True)
|
||||
data = complete_json(
|
||||
self.engine.llm,
|
||||
_PROPOSE.format(
|
||||
episodes="\n".join(f"- {e.problem} -> {e.summary}" for e in open_eps),
|
||||
notes="\n".join(f"- {n}" for n in self.memory.notes()) or "(none)",
|
||||
),
|
||||
)
|
||||
problem = str(data.get("problem", "")).strip()
|
||||
# The open problems have now been followed up; close them so the agent
|
||||
# doesn't propose again from the same evidence.
|
||||
for ep in open_eps:
|
||||
self.memory.set_open(ep.id, False)
|
||||
if not problem:
|
||||
return {"drafted": None, "why": "model proposed nothing"}
|
||||
why = str(data.get("why", ""))
|
||||
self.memory.set_state("pending_proposal", {"problem": problem, "why": why})
|
||||
return {"drafted": problem, "why": why}
|
||||
|
||||
def _do_adopt_goal(self, action: ProposedAction) -> Dict[str, Any]:
|
||||
problem = action.payload["problem"]
|
||||
self.add_goal(problem)
|
||||
self.memory.set_state("pending_proposal", None)
|
||||
return {"queued": problem}
|
||||
@@ -0,0 +1,177 @@
|
||||
[
|
||||
{
|
||||
"id": "mycelial_physarum",
|
||||
"name": "Physarum Polycephalum (Slime Mold) Oscillatory Tube Network",
|
||||
"source_domain": "Biophysics / Decentralized Morphogenesis",
|
||||
"summary": "A brainless single-celled syncytium solves dynamic shortest-path and fault-tolerant network design via peristaltic cytoplasmic shuttle-streaming where tube diameter thickens nonlinearly with flux and prunes when starved.",
|
||||
"mechanisms": [
|
||||
"High Local Flux stretches wall to increase Conductivity (D^4 Poiseuille)",
|
||||
"Low Flux triggers autophagic pruning of Redundant Links",
|
||||
"Phase-Shifted Oscillations propagate global state without Central Coordinator"
|
||||
],
|
||||
"invariants": [
|
||||
"Zero global routing table; topology IS the memory",
|
||||
"Oscillatory pressure waves prevent steady-state clogging"
|
||||
],
|
||||
"operators": [
|
||||
"Flux-proportional conduit dilation/pruning",
|
||||
"Peristaltic phase-wave signaling without global clocks",
|
||||
"Conservation-enforced resource reallocation"
|
||||
],
|
||||
"failure_modes": [
|
||||
"Premature lock-in if nonlinear reinforcement exponent is too steep"
|
||||
],
|
||||
"tags": [
|
||||
"clearance_bottleneck",
|
||||
"coordination_overhead",
|
||||
"routing_rigidity",
|
||||
"single_point_of_failure"
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "immune_clonal_thymic",
|
||||
"name": "Adaptive Immune Repertoire & Thymic Negative Selection",
|
||||
"source_domain": "Computational Immunology",
|
||||
"summary": "Generates billions of stochastic receptor variants via V(D)J somatic hypermutation before pathogen encounter, filters self-reactive clones in a sandboxed thymus, and exponentially amplifies only nodes matching local danger + antigen signals.",
|
||||
"mechanisms": [
|
||||
"V(D)J Recombination pre-generates Speculative Detector Diversity",
|
||||
"High Self-Affinity in Thymus induces apoptosis of Auto-Destructive Agents",
|
||||
"Antigen + Tissue Danger Signal triggers clonal expansion of Matched Effectors"
|
||||
],
|
||||
"invariants": [
|
||||
"Two-factor activation: pattern match alone is inert without tissue stress context",
|
||||
"Mutation rate scales inversely with fitness (somatic hypermutation on mismatch)"
|
||||
],
|
||||
"operators": [
|
||||
"Two-signal co-stimulation (Event + Local Stress Context)",
|
||||
"Fitness-inverse mutation rate (hypermutate only when failing)",
|
||||
"Shadow-thymus negative selection against self-harm"
|
||||
],
|
||||
"failure_modes": [
|
||||
"Autoimmune storm if co-stimulatory 'danger context' check is bypassed"
|
||||
],
|
||||
"tags": [
|
||||
"false_positive_damage",
|
||||
"novel_threat_detection",
|
||||
"speculative_exploration",
|
||||
"toxic_accumulation"
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "spin_glass_annealing",
|
||||
"name": "Frustrated Spin Glass & Quantum/Thermal Tunneling Across Energy Barriers",
|
||||
"source_domain": "Condensed Matter & Statistical Physics",
|
||||
"summary": "In systems with competing local constraints (geometric frustration), no configuration satisfies all edges simultaneously. Escape from glassy local minima requires localized thermal/noise bursts and scale-free avalanche reorganizations.",
|
||||
"mechanisms": [
|
||||
"Competing Antiferromagnetic Bonds create Degenerate Ground-State Manifold",
|
||||
"Localized Noise Injection destabilizes Metastable Trap",
|
||||
"Domain Wall Propagation resolves Long-Range Constraint Deadlock"
|
||||
],
|
||||
"invariants": [
|
||||
"Global frustration cannot be eliminated, only concentrated into mobile domain walls",
|
||||
"Cooling too fast guarantees brittle suboptimal lock-in"
|
||||
],
|
||||
"operators": [
|
||||
"Defect-routing (materialize conflicts as first-class mobile tokens)",
|
||||
"Targeted stochastic heating of stalled/congested subgraphs",
|
||||
"Degenerate ground-state switching"
|
||||
],
|
||||
"failure_modes": [
|
||||
"Quench disorder freezing the system into a brittle, high-stress glass"
|
||||
],
|
||||
"tags": [
|
||||
"consistency_vs_latency",
|
||||
"deadlock_under_contention",
|
||||
"local_minima_stagnation",
|
||||
"protein_misfolding_trap"
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "estuarine_tidal_flushing",
|
||||
"name": "Estuarine Salt-Wedge Stratification & Tidal Pumping Clearance",
|
||||
"source_domain": "Geophysical Fluid Dynamics / Marine Ecology",
|
||||
"summary": "Estuaries clear heavy sediment and metabolic toxins without a unidirectional pump by coupling asymmetric tidal oscillations with density stratification and marsh-plain capacitance traps that separate inflow pathways from ebb-flush channels.",
|
||||
"mechanisms": [
|
||||
"Tidal Asymmetry (Ebb Dominance) creates net export of Aggregated Benthic Particulates",
|
||||
"Shear Instability at Pycnocline entrains and dilutes Stagnant Deep Solutes",
|
||||
"Side-Basin Phase Lag traps and settles Cohesive Flocs"
|
||||
],
|
||||
"invariants": [
|
||||
"Bidirectional oscillation achieves unidirectional net clearance via phase-lagged side capacitance",
|
||||
"Below a critical shear threshold (Richardson number), stratified layers decouple and suffocate"
|
||||
],
|
||||
"operators": [
|
||||
"Rectified tidal ratcheting (oscillation + phase-lagged side capacitance = net directional transport)",
|
||||
"Critical Richardson-number shear mixing across boundary layers",
|
||||
"Biphasic flocculation window (optimal shear prevents both monomer stasis and mega-clump occlusion)"
|
||||
],
|
||||
"failure_modes": [
|
||||
"Eutrophic stratification lock-in when pulse amplitude drops below critical Richardson number"
|
||||
],
|
||||
"tags": [
|
||||
"backpressure_congestion",
|
||||
"clearance_bottleneck",
|
||||
"oscillatory_coupling",
|
||||
"toxic_accumulation"
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "baroque_fugue_stretto",
|
||||
"name": "Polyphonic Fugue Counterpoint & Stretto Canon Compression",
|
||||
"source_domain": "Music Theory / Temporal Information Architecture",
|
||||
"summary": "Multiple autonomous voices execute the exact same thematic transformation at staggered time offsets and pitch inversions (canon/stretto), achieving maximal information density without harmonic collision by obeying strict interval-invariance rules.",
|
||||
"mechanisms": [
|
||||
"Oblique/Contrary Motion preserves perceptual independence of Concurrent Streams",
|
||||
"Stretto Temporal Overlap compresses latency of Thematic Resolution",
|
||||
"Dissonance Preparation + Resolution propels Forward Momentum"
|
||||
],
|
||||
"invariants": [
|
||||
"Contrary motion prevents phase-locked interference spikes",
|
||||
"A slow invariant pedal-point allows fast upper layers to mutate freely"
|
||||
],
|
||||
"operators": [
|
||||
"Contrary-motion resource scheduling (when node A scales read-heavy, neighbor B shifts write-compaction)",
|
||||
"Stretto staggered speculative execution",
|
||||
"Pedal-point slow state anchor beneath volatile ephemeral streams"
|
||||
],
|
||||
"failure_modes": [
|
||||
"Homophonic mud when voices move in parallel lockstep at identical frequencies"
|
||||
],
|
||||
"tags": [
|
||||
"consistency_vs_latency",
|
||||
"coordination_overhead",
|
||||
"deadlock_under_contention",
|
||||
"oscillatory_coupling"
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "coral_reef_symbiosis",
|
||||
"name": "Coral-Zooxanthellae Trophic Symbiosis & Calcified Micro-Habitats",
|
||||
"source_domain": "Marine Community Ecology / Niche Construction",
|
||||
"summary": "In nutrient-barren tropical waters (oligotrophic deserts), coral polyps build high-surface-area calcium carbonate architectures that trap light and recycle metabolic waste internally with symbiotic algae, creating a self-sustaining oasis.",
|
||||
"mechanisms": [
|
||||
"Rugose 3D Surface Geometry slows boundary flow to capture Scarce Ambient Attention/Nutrients",
|
||||
"Internal Waste-to-Fuel Loop eliminates dependency on External Paid Acquisition",
|
||||
"Modular Polyp Cloning compounds Surface Area Without Central Redesign"
|
||||
],
|
||||
"invariants": [
|
||||
"Every user output becomes structural habitat for the next user",
|
||||
"Fractal progressive disclosure: calm macro-silhouette, rich micro-texture on hover/focus"
|
||||
],
|
||||
"operators": [
|
||||
"Niche calcification (turn ephemeral user actions into permanent public/shareable artifacts)",
|
||||
"Symbiotic waste-to-fuel loop (one persona's byproduct is another persona's trigger)",
|
||||
"Rugose progressive disclosure (calm macro-canvas that reveals tactile micro-tools on proximity)"
|
||||
],
|
||||
"failure_modes": [
|
||||
"Monoculture bleaching when sudden category shift invalidates a single acquisition channel"
|
||||
],
|
||||
"tags": [
|
||||
"attention_scarcity",
|
||||
"cognitive_overload",
|
||||
"retention_churn",
|
||||
"trust_skepticism",
|
||||
"visual_homogenization"
|
||||
]
|
||||
}
|
||||
]
|
||||
@@ -0,0 +1,62 @@
|
||||
"""The cross-domain analogy library.
|
||||
|
||||
Archetypes are hand-curated descriptions of mechanisms from other fields. They
|
||||
are prompts for analogy, not knowledge the engine "has": the LLM decides whether
|
||||
any of them actually fits a problem, and the critic decides whether the result
|
||||
holds up.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Iterable, List
|
||||
|
||||
_LIBRARY_PATH = Path(__file__).with_name("archetypes.json")
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Archetype:
|
||||
id: str
|
||||
name: str
|
||||
source_domain: str
|
||||
summary: str
|
||||
mechanisms: List[str] = field(default_factory=list)
|
||||
invariants: List[str] = field(default_factory=list)
|
||||
operators: List[str] = field(default_factory=list)
|
||||
failure_modes: List[str] = field(default_factory=list)
|
||||
tags: List[str] = field(default_factory=list)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: Dict[str, Any]) -> "Archetype":
|
||||
return cls(**{k: data[k] for k in cls.__dataclass_fields__ if k in data})
|
||||
|
||||
def brief(self) -> str:
|
||||
"""Compact text used inside prompts."""
|
||||
return (
|
||||
f"[{self.id}] {self.name} ({self.source_domain})\n"
|
||||
f" Summary: {self.summary}\n"
|
||||
f" Mechanisms: {'; '.join(self.mechanisms)}\n"
|
||||
f" Transferable operators: {'; '.join(self.operators)}\n"
|
||||
f" Known failure modes: {'; '.join(self.failure_modes)}"
|
||||
)
|
||||
|
||||
def search_text(self) -> str:
|
||||
return " ".join(
|
||||
[self.name, self.source_domain, self.summary, *self.mechanisms, *self.operators, *self.tags]
|
||||
).lower()
|
||||
|
||||
|
||||
def load_default_archetypes() -> List[Archetype]:
|
||||
with _LIBRARY_PATH.open(encoding="utf-8") as fh:
|
||||
return [Archetype.from_dict(d) for d in json.load(fh)]
|
||||
|
||||
|
||||
def load_archetypes(path: str | Path) -> List[Archetype]:
|
||||
with Path(path).open(encoding="utf-8") as fh:
|
||||
return [Archetype.from_dict(d) for d in json.load(fh)]
|
||||
|
||||
|
||||
def index_by_id(archetypes: Iterable[Archetype]) -> Dict[str, Archetype]:
|
||||
return {a.id: a for a in archetypes}
|
||||
@@ -0,0 +1,415 @@
|
||||
"""MuseEngine: structured cross-domain ideation on top of an LLM.
|
||||
|
||||
Pipeline (each step is one or more LLM calls with a JSON contract):
|
||||
|
||||
1. ANALYZE - describe the problem's structure (tensions, constraints, failure modes)
|
||||
independently of its vocabulary.
|
||||
2. MAP - pick archetypes from the library whose *mechanisms* fit that
|
||||
structure, say what maps to what, and where the analogy breaks.
|
||||
Related past problems from memory are offered as extra context.
|
||||
3. GENERATE - turn each analogy into a concrete candidate with tests that could
|
||||
falsify it (science) or break it (systems).
|
||||
4. CRITIQUE - an adversarial, separate prompt scores novelty and rigor, lists
|
||||
fatal flaws, and refutes candidates that don't hold up.
|
||||
5. SELECT - keep the Pareto frontier over (novelty, rigor) among survivors.
|
||||
|
||||
Without an LLM the engine runs in ``heuristic`` mode: it ranks archetypes by
|
||||
word overlap and returns them as unvetted suggestions. It does not invent
|
||||
candidates or scores in that mode.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import asdict, dataclass, field
|
||||
from typing import Any, Dict, List, Optional, Sequence
|
||||
|
||||
from .archetypes import Archetype, index_by_id, load_default_archetypes
|
||||
from .llm import LLM, CountingLLM, as_list, clamp01, complete_json, str_list
|
||||
from .memory import Episode, MemoryStore
|
||||
|
||||
SURVIVES = "survives"
|
||||
REFUTED = "refuted"
|
||||
UNREVIEWED = "unreviewed"
|
||||
|
||||
|
||||
@dataclass
|
||||
class ProblemStructure:
|
||||
domain: str
|
||||
tensions: List[Dict[str, Any]]
|
||||
constraints: List[str]
|
||||
failure_modes: List[str]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Analogy:
|
||||
archetype_id: str
|
||||
archetype_name: str
|
||||
mapping: str
|
||||
breaks_where: str
|
||||
fit: Optional[float]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Candidate:
|
||||
id: str
|
||||
title: str
|
||||
thesis: str
|
||||
mechanism: List[str]
|
||||
tests: List[str]
|
||||
source_archetype: str
|
||||
novelty: Optional[float] = None
|
||||
rigor: Optional[float] = None
|
||||
fatal_flaws: List[str] = field(default_factory=list)
|
||||
critique: str = ""
|
||||
verdict: str = UNREVIEWED
|
||||
pareto: bool = False
|
||||
|
||||
|
||||
@dataclass
|
||||
class Report:
|
||||
problem: str
|
||||
mode: str # "llm" or "heuristic"
|
||||
structure: Optional[ProblemStructure]
|
||||
analogies: List[Analogy]
|
||||
candidates: List[Candidate]
|
||||
frontier: List[Candidate]
|
||||
recalled: List[str]
|
||||
llm_calls: int
|
||||
episode_id: Optional[str] = None
|
||||
|
||||
def to_dict(self) -> Dict[str, Any]:
|
||||
return asdict(self)
|
||||
|
||||
def scaffold(self) -> str:
|
||||
"""A compact system block to prepend to another agent's prompt."""
|
||||
lines = ["<muse_scaffold>"]
|
||||
if self.mode == "heuristic" and not self.analogies:
|
||||
lines.append("No analogy in the library looked relevant to this problem.")
|
||||
elif self.mode == "heuristic":
|
||||
lines.append(
|
||||
"These cross-domain analogies were matched by keyword only and have not been "
|
||||
"checked. Use them only if they genuinely fit; ignore any that don't."
|
||||
)
|
||||
for a in self.analogies:
|
||||
lines.append(f"- {a.archetype_name}: {a.mapping}")
|
||||
else:
|
||||
s = self.structure
|
||||
if s and s.constraints:
|
||||
lines.append("Constraints to respect:")
|
||||
lines += [f"- {c}" for c in s.constraints]
|
||||
if self.frontier:
|
||||
lines.append("Candidate directions that survived an adversarial critique:")
|
||||
for c in self.frontier:
|
||||
lines.append(f"- {c.title} (novelty {c.novelty:.2f}, rigor {c.rigor:.2f}): {c.thesis}")
|
||||
if c.tests:
|
||||
lines.append(f" Test that could break it: {c.tests[0]}")
|
||||
if c.fatal_flaws:
|
||||
lines.append(f" Open concern: {c.fatal_flaws[0]}")
|
||||
else:
|
||||
lines.append("No candidate survived critique; treat the analogies below as weak leads.")
|
||||
for a in self.analogies:
|
||||
lines.append(f"- {a.archetype_name}: {a.mapping} (breaks where: {a.breaks_where})")
|
||||
lines.append(
|
||||
"Build on these where they help, say plainly where they don't, and prefer a correct "
|
||||
"conventional answer over a novel wrong one."
|
||||
)
|
||||
lines.append("</muse_scaffold>")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Prompts
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_ANALYZE = """TASK: ANALYZE
|
||||
Describe the abstract structure of this problem, independent of its domain vocabulary.
|
||||
|
||||
Problem:
|
||||
{problem}
|
||||
|
||||
Return JSON:
|
||||
{{"domain": "<short label, e.g. distributed systems, neuroscience, marketing>",
|
||||
"tensions": [{{"axis": "<a trade-off or pressure>", "description": "<one sentence>", "weight": <0-1>}}],
|
||||
"constraints": ["<hard limits any solution must respect>"],
|
||||
"failure_modes": ["<how naive solutions fail>"]}}
|
||||
List 2-5 tensions. Be specific to this problem; do not pad."""
|
||||
|
||||
_MAP = """TASK: MAP
|
||||
You are looking for structural analogies: mechanisms from another field whose causal
|
||||
structure matches this problem, not just shared words.
|
||||
|
||||
Problem:
|
||||
{problem}
|
||||
|
||||
Problem structure:
|
||||
{structure}
|
||||
|
||||
{recalled}Analogy library:
|
||||
{library}
|
||||
|
||||
Pick up to {k} archetypes whose mechanisms genuinely fit. It is fine to pick fewer, or
|
||||
none, if nothing fits. For each, say concretely what maps to what, and where the analogy
|
||||
breaks down.
|
||||
|
||||
Return JSON:
|
||||
{{"analogies": [{{"archetype_id": "<id from the library>", "mapping": "<what corresponds to what>",
|
||||
"breaks_where": "<where the analogy fails>", "fit": <0-1>}}]}}"""
|
||||
|
||||
_GENERATE = """TASK: GENERATE
|
||||
Turn this analogy into one concrete candidate {kind}.
|
||||
|
||||
Problem:
|
||||
{problem}
|
||||
|
||||
Analogy ({name}): {mapping}
|
||||
It breaks down where: {breaks}
|
||||
|
||||
Return JSON:
|
||||
{{"title": "<short name>",
|
||||
"thesis": "<the claim or design in 1-3 sentences>",
|
||||
"mechanism": ["<concrete steps or components>"],
|
||||
"tests": ["<{test_kind}>"]}}
|
||||
Give 2-4 tests. Each must name a specific measurement or scenario and the outcome that
|
||||
would count against the candidate."""
|
||||
|
||||
_CRITIQUE = """TASK: CRITIQUE
|
||||
You are an adversarial reviewer. Your job is to find what is wrong with this candidate,
|
||||
not to praise it.
|
||||
|
||||
Problem:
|
||||
{problem}
|
||||
|
||||
Candidate: {title}
|
||||
Thesis: {thesis}
|
||||
Mechanism: {mechanism}
|
||||
Proposed tests: {tests}
|
||||
|
||||
Score honestly:
|
||||
- novelty: 0 if this is a standard textbook answer or a known technique renamed with
|
||||
metaphors; high only if it is a genuinely different approach.
|
||||
- rigor: is it internally consistent, physically/technically plausible, and are the
|
||||
tests real tests that could fail?
|
||||
Refute it if it is wrong, incoherent, untestable, or only a relabelled known idea.
|
||||
|
||||
Return JSON:
|
||||
{{"novelty": <0-1>, "rigor": <0-1>, "fatal_flaws": ["..."],
|
||||
"verdict": "survives" | "refuted", "reasoning": "<2-4 sentences>"}}"""
|
||||
|
||||
_SCIENCE_HINTS = re.compile(
|
||||
r"\b(hypothes\w*|mechanis\w*|disease|cell\w*|protein\w*|neuro\w*|brain|gene\w*|"
|
||||
r"molecul\w*|clinical|physics|chemical|species|ecolog\w*|experiment\w*)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Engine
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class MuseEngine:
|
||||
def __init__(
|
||||
self,
|
||||
llm: Optional[LLM] = None,
|
||||
archetypes: Optional[Sequence[Archetype]] = None,
|
||||
memory: Optional[MemoryStore] = None,
|
||||
critic: Optional[LLM] = None,
|
||||
):
|
||||
"""
|
||||
llm: the model that analyzes, maps and generates. ``None`` -> heuristic mode.
|
||||
critic: optional separate model for critique (defaults to ``llm``). Using a
|
||||
different model reduces self-grading bias.
|
||||
memory: optional store; past problems are recalled as context and each run
|
||||
is recorded as an episode.
|
||||
"""
|
||||
self.llm = llm
|
||||
self.critic = critic or llm
|
||||
self.archetypes = list(archetypes) if archetypes is not None else load_default_archetypes()
|
||||
self.by_id = index_by_id(self.archetypes)
|
||||
self.memory = memory
|
||||
|
||||
def imagine(self, problem: str, top_k: int = 3) -> Report:
|
||||
problem = problem.strip()
|
||||
if not problem:
|
||||
raise ValueError("problem must be non-empty")
|
||||
recalled = self.memory.recall(problem, k=3) if self.memory else []
|
||||
if self.llm is None:
|
||||
report = self._heuristic(problem, top_k)
|
||||
else:
|
||||
report = self._with_llm(problem, top_k, recalled)
|
||||
report.recalled = [ep.problem for ep in recalled]
|
||||
if self.memory:
|
||||
ep = self.memory.add_episode(problem, self._summarize(report), is_open=not report.frontier)
|
||||
report.episode_id = ep.id
|
||||
return report
|
||||
|
||||
# -- heuristic mode -------------------------------------------------------
|
||||
|
||||
def _heuristic(self, problem: str, top_k: int) -> Report:
|
||||
words = set(re.findall(r"[a-z][a-z-]{3,}", problem.lower()))
|
||||
ranked = []
|
||||
for arch in self.archetypes:
|
||||
text = arch.search_text()
|
||||
hits = sum(1 for w in words if w in text)
|
||||
if hits:
|
||||
ranked.append((hits, arch))
|
||||
ranked.sort(key=lambda t: t[0], reverse=True)
|
||||
analogies = [
|
||||
Analogy(
|
||||
archetype_id=a.id,
|
||||
archetype_name=a.name,
|
||||
mapping="; ".join(a.operators),
|
||||
breaks_where="unknown (not analyzed)",
|
||||
fit=None,
|
||||
)
|
||||
for _, a in ranked[:top_k]
|
||||
]
|
||||
return Report(problem, "heuristic", None, analogies, [], [], [], llm_calls=0)
|
||||
|
||||
# -- LLM mode -------------------------------------------------------------
|
||||
|
||||
def _with_llm(self, problem: str, top_k: int, recalled: List[Episode]) -> Report:
|
||||
llm = CountingLLM(self.llm)
|
||||
critic = llm if self.critic is self.llm else CountingLLM(self.critic)
|
||||
|
||||
structure = self._analyze(llm, problem)
|
||||
analogies = self._map(llm, problem, structure, recalled, top_k)
|
||||
candidates = [self._generate(llm, problem, structure, a, i) for i, a in enumerate(analogies)]
|
||||
for c in candidates:
|
||||
self._critique(critic, problem, c)
|
||||
frontier = pareto_frontier(candidates)
|
||||
|
||||
calls = llm.calls + (critic.calls if critic is not llm else 0)
|
||||
return Report(problem, "llm", structure, analogies, candidates, frontier, [], llm_calls=calls)
|
||||
|
||||
def _analyze(self, llm: LLM, problem: str) -> ProblemStructure:
|
||||
data = complete_json(llm, _ANALYZE.format(problem=problem))
|
||||
tensions = [
|
||||
{"axis": str(t.get("axis", "")), "description": str(t.get("description", "")),
|
||||
"weight": clamp01(t.get("weight"), 0.5)}
|
||||
for t in as_list(data.get("tensions")) if isinstance(t, dict)
|
||||
]
|
||||
return ProblemStructure(
|
||||
domain=str(data.get("domain", "unknown")),
|
||||
tensions=tensions,
|
||||
constraints=str_list(data.get("constraints")),
|
||||
failure_modes=str_list(data.get("failure_modes")),
|
||||
)
|
||||
|
||||
def _map(
|
||||
self, llm: LLM, problem: str, structure: ProblemStructure, recalled: List[Episode], top_k: int
|
||||
) -> List[Analogy]:
|
||||
recalled_text = ""
|
||||
if recalled:
|
||||
recalled_text = "Related problems worked on before (may or may not help):\n" + "\n".join(
|
||||
f"- {ep.problem} -> {ep.summary}" for ep in recalled
|
||||
) + "\n\n"
|
||||
data = complete_json(
|
||||
llm,
|
||||
_MAP.format(
|
||||
problem=problem,
|
||||
structure=_structure_text(structure),
|
||||
recalled=recalled_text,
|
||||
library="\n".join(a.brief() for a in self.archetypes),
|
||||
k=top_k,
|
||||
),
|
||||
)
|
||||
out: List[Analogy] = []
|
||||
seen = set()
|
||||
for item in as_list(data.get("analogies")):
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
arch = self.by_id.get(str(item.get("archetype_id", "")))
|
||||
if arch is None or arch.id in seen:
|
||||
continue # drop hallucinated or duplicate ids
|
||||
seen.add(arch.id)
|
||||
out.append(
|
||||
Analogy(
|
||||
archetype_id=arch.id,
|
||||
archetype_name=arch.name,
|
||||
mapping=str(item.get("mapping", "")),
|
||||
breaks_where=str(item.get("breaks_where", "")),
|
||||
fit=clamp01(item.get("fit"), 0.5),
|
||||
)
|
||||
)
|
||||
out.sort(key=lambda a: a.fit or 0.0, reverse=True)
|
||||
return out[:top_k]
|
||||
|
||||
def _generate(
|
||||
self, llm: LLM, problem: str, structure: ProblemStructure, analogy: Analogy, idx: int
|
||||
) -> Candidate:
|
||||
science = bool(_SCIENCE_HINTS.search(problem + " " + structure.domain))
|
||||
data = complete_json(
|
||||
llm,
|
||||
_GENERATE.format(
|
||||
problem=problem,
|
||||
kind="hypothesis" if science else "design",
|
||||
name=analogy.archetype_name,
|
||||
mapping=analogy.mapping,
|
||||
breaks=analogy.breaks_where,
|
||||
test_kind="an experiment and the result that would falsify it"
|
||||
if science
|
||||
else "a stress or failure scenario and the observation that would show the design fails",
|
||||
),
|
||||
)
|
||||
return Candidate(
|
||||
id=f"c{idx + 1}_{analogy.archetype_id}",
|
||||
title=str(data.get("title", analogy.archetype_name)),
|
||||
thesis=str(data.get("thesis", "")),
|
||||
mechanism=str_list(data.get("mechanism")),
|
||||
tests=str_list(data.get("tests")),
|
||||
source_archetype=analogy.archetype_id,
|
||||
)
|
||||
|
||||
def _critique(self, critic: LLM, problem: str, cand: Candidate) -> None:
|
||||
data = complete_json(
|
||||
critic,
|
||||
_CRITIQUE.format(
|
||||
problem=problem,
|
||||
title=cand.title,
|
||||
thesis=cand.thesis,
|
||||
mechanism="; ".join(cand.mechanism),
|
||||
tests="; ".join(cand.tests),
|
||||
),
|
||||
)
|
||||
cand.novelty = clamp01(data.get("novelty"))
|
||||
cand.rigor = clamp01(data.get("rigor"))
|
||||
cand.fatal_flaws = str_list(data.get("fatal_flaws"))
|
||||
cand.critique = str(data.get("reasoning", ""))
|
||||
verdict = str(data.get("verdict", "")).strip().lower()
|
||||
# Anything other than an explicit "survives" counts as refuted, and a
|
||||
# candidate without any test cannot survive.
|
||||
cand.verdict = SURVIVES if verdict == SURVIVES and cand.tests else REFUTED
|
||||
|
||||
@staticmethod
|
||||
def _summarize(report: Report) -> str:
|
||||
if report.frontier:
|
||||
best = report.frontier[0]
|
||||
return f"Best surviving idea: {best.title}: {best.thesis}"
|
||||
if report.mode == "heuristic":
|
||||
return "Unvetted suggestions: " + ", ".join(a.archetype_name for a in report.analogies)
|
||||
return "No candidate survived critique."
|
||||
|
||||
|
||||
def pareto_frontier(candidates: Sequence[Candidate]) -> List[Candidate]:
|
||||
"""Non-dominated surviving candidates over (novelty, rigor), best first."""
|
||||
alive = [c for c in candidates if c.verdict == SURVIVES and c.novelty is not None and c.rigor is not None]
|
||||
frontier = []
|
||||
for c in alive:
|
||||
dominated = any(
|
||||
o is not c and o.novelty >= c.novelty and o.rigor >= c.rigor and (o.novelty > c.novelty or o.rigor > c.rigor)
|
||||
for o in alive
|
||||
)
|
||||
c.pareto = not dominated
|
||||
if c.pareto:
|
||||
frontier.append(c)
|
||||
frontier.sort(key=lambda c: c.novelty + c.rigor, reverse=True)
|
||||
return frontier
|
||||
|
||||
|
||||
def _structure_text(s: ProblemStructure) -> str:
|
||||
lines = [f"Domain: {s.domain}"]
|
||||
lines += [f"Tension: {t['axis']} - {t['description']} (weight {t['weight']:.2f})" for t in s.tensions]
|
||||
lines += [f"Constraint: {c}" for c in s.constraints]
|
||||
lines += [f"Naive failure mode: {f}" for f in s.failure_modes]
|
||||
return "\n".join(lines)
|
||||
@@ -0,0 +1,136 @@
|
||||
"""A blinded A/B evaluation: does the scaffold actually help?
|
||||
|
||||
For each problem:
|
||||
- baseline: the solver answers the problem as given.
|
||||
- treated: the solver answers the same problem with the MuseEngine scaffold.
|
||||
- a judge compares the two answers twice, with the A/B order swapped. A win
|
||||
only counts if the judge picks the same answer both times; otherwise the
|
||||
pair is a tie. This cancels the judge's position bias.
|
||||
|
||||
The result is a win/loss/tie count and an exact two-sided sign test. There are
|
||||
no constants, keyword scorers, or hand-written baselines. You need real models
|
||||
for the numbers to mean anything, and the judge should ideally be a different
|
||||
model from the solver.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import math
|
||||
import random
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Callable, List, Optional, Sequence
|
||||
|
||||
from .engine import MuseEngine
|
||||
from .llm import LLM, complete_json
|
||||
|
||||
_JUDGE = """TASK: JUDGE
|
||||
Two answers to the same problem follow. Judge which is better overall, weighing:
|
||||
correctness and plausibility first, then how useful and specific it is, then
|
||||
originality. Originality never makes up for being wrong. Ignore length and style.
|
||||
|
||||
Problem:
|
||||
{problem}
|
||||
|
||||
=== Answer A ===
|
||||
{a}
|
||||
|
||||
=== Answer B ===
|
||||
{b}
|
||||
|
||||
Return JSON: {{"winner": "A" | "B" | "tie", "reason": "<one or two sentences>"}}"""
|
||||
|
||||
|
||||
@dataclass
|
||||
class PairResult:
|
||||
problem: str
|
||||
outcome: str # "treated", "baseline", or "tie"
|
||||
reasons: List[str] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
class EvalResult:
|
||||
pairs: List[PairResult]
|
||||
|
||||
@property
|
||||
def wins(self) -> int:
|
||||
return sum(p.outcome == "treated" for p in self.pairs)
|
||||
|
||||
@property
|
||||
def losses(self) -> int:
|
||||
return sum(p.outcome == "baseline" for p in self.pairs)
|
||||
|
||||
@property
|
||||
def ties(self) -> int:
|
||||
return sum(p.outcome == "tie" for p in self.pairs)
|
||||
|
||||
@property
|
||||
def p_value(self) -> float:
|
||||
return sign_test(self.wins, self.losses)
|
||||
|
||||
def summary(self) -> str:
|
||||
decided = self.wins + self.losses
|
||||
rate = f"{self.wins / decided:.0%}" if decided else "n/a"
|
||||
return (
|
||||
f"{len(self.pairs)} problems: scaffold won {self.wins}, lost {self.losses}, tied {self.ties} "
|
||||
f"(win rate among decided pairs {rate}, two-sided sign test p={self.p_value:.3f})"
|
||||
)
|
||||
|
||||
|
||||
def sign_test(wins: int, losses: int) -> float:
|
||||
"""Exact two-sided binomial sign test, ties excluded."""
|
||||
n = wins + losses
|
||||
if n == 0:
|
||||
return 1.0
|
||||
k = min(wins, losses)
|
||||
tail = sum(math.comb(n, i) for i in range(k + 1)) / 2**n
|
||||
return min(1.0, 2 * tail)
|
||||
|
||||
|
||||
def _judge_once(judge: LLM, problem: str, a: str, b: str) -> tuple[str, str]:
|
||||
data = complete_json(judge, _JUDGE.format(problem=problem, a=a, b=b))
|
||||
winner = str(data.get("winner", "tie")).strip().upper()
|
||||
return (winner if winner in {"A", "B"} else "TIE"), str(data.get("reason", ""))
|
||||
|
||||
|
||||
def judge_pair(judge: LLM, problem: str, baseline: str, treated: str, rng: random.Random) -> PairResult:
|
||||
# Randomize which answer is shown first in the first round, then swap.
|
||||
if rng.random() < 0.5:
|
||||
first = ("baseline", "treated")
|
||||
else:
|
||||
first = ("treated", "baseline")
|
||||
answers = {"baseline": baseline, "treated": treated}
|
||||
verdicts, reasons = [], []
|
||||
for order in (first, first[::-1]):
|
||||
w, reason = _judge_once(judge, problem, answers[order[0]], answers[order[1]])
|
||||
verdicts.append({"A": order[0], "B": order[1]}.get(w, "tie"))
|
||||
reasons.append(reason)
|
||||
outcome = verdicts[0] if verdicts[0] == verdicts[1] else "tie"
|
||||
return PairResult(problem, outcome, reasons)
|
||||
|
||||
|
||||
def run_ab_eval(
|
||||
problems: Sequence[str],
|
||||
solver: LLM,
|
||||
judge: LLM,
|
||||
engine: MuseEngine,
|
||||
seed: int = 0,
|
||||
on_pair: Optional[Callable[[PairResult], None]] = None,
|
||||
) -> EvalResult:
|
||||
problems = [p.strip() for p in problems]
|
||||
if not problems or not all(problems):
|
||||
raise ValueError("problems must be a non-empty list of non-empty strings")
|
||||
rng = random.Random(seed)
|
||||
pairs = []
|
||||
for problem in problems:
|
||||
report = engine.imagine(problem)
|
||||
prompts = {"baseline": problem, "treated": f"{report.scaffold()}\n\n{problem}"}
|
||||
# Randomize call order in case the solver carries state between calls.
|
||||
order = ["baseline", "treated"]
|
||||
rng.shuffle(order)
|
||||
answers = {label: solver(prompts[label]) for label in order}
|
||||
baseline, treated = answers["baseline"], answers["treated"]
|
||||
pair = judge_pair(judge, problem, baseline, treated, rng)
|
||||
pairs.append(pair)
|
||||
if on_pair:
|
||||
on_pair(pair)
|
||||
return EvalResult(pairs)
|
||||
@@ -0,0 +1,105 @@
|
||||
"""Guardrails for the autonomous loop.
|
||||
|
||||
Every action the agent wants to take goes through ``Governor.authorize`` first.
|
||||
The governor enforces an allowlist, a step budget, a kill switch, and human
|
||||
approval for actions you mark as needing it. Every decision, allowed or not,
|
||||
goes to the audit log together with what the action actually did.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Callable, Dict, FrozenSet, Optional
|
||||
|
||||
from .memory import MemoryStore
|
||||
|
||||
|
||||
class GovernanceError(RuntimeError):
|
||||
"""An action was refused."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ProposedAction:
|
||||
name: str
|
||||
reason: str
|
||||
payload: Dict[str, Any] = field(default_factory=dict)
|
||||
|
||||
|
||||
Approver = Callable[[ProposedAction], bool]
|
||||
|
||||
|
||||
def deny_all(action: ProposedAction) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def console_approver(action: ProposedAction) -> bool:
|
||||
"""Ask on stdin. Anything but an explicit 'y' is a no."""
|
||||
print(f"\nThe agent wants to: {action.name}\n Why: {action.reason}\n Details: {action.payload}")
|
||||
return input("Allow? [y/N] ").strip().lower() == "y"
|
||||
|
||||
|
||||
class KillSwitch:
|
||||
"""Trip it in code with ``trip()``, or by creating the file at ``path``."""
|
||||
|
||||
def __init__(self, path: Optional[str | Path] = None):
|
||||
self.path = Path(path) if path else None
|
||||
self._tripped = False
|
||||
|
||||
def trip(self) -> None:
|
||||
self._tripped = True
|
||||
|
||||
def tripped(self) -> bool:
|
||||
return self._tripped or bool(self.path and self.path.exists())
|
||||
|
||||
|
||||
@dataclass
|
||||
class Policy:
|
||||
allowed_actions: FrozenSet[str] = frozenset({"explore", "reflect", "propose_goal", "adopt_goal"})
|
||||
# Taking on a goal the agent invented itself needs a human by default.
|
||||
approval_required: FrozenSet[str] = frozenset({"adopt_goal"})
|
||||
max_steps: int = 10
|
||||
|
||||
|
||||
class Governor:
|
||||
def __init__(
|
||||
self,
|
||||
memory: MemoryStore,
|
||||
policy: Optional[Policy] = None,
|
||||
approver: Approver = deny_all,
|
||||
kill_switch: Optional[KillSwitch] = None,
|
||||
):
|
||||
self.memory = memory
|
||||
self.policy = policy or Policy()
|
||||
self.approver = approver
|
||||
self.kill_switch = kill_switch or KillSwitch()
|
||||
self.steps_taken = 0
|
||||
|
||||
def should_stop(self) -> Optional[str]:
|
||||
if self.kill_switch.tripped():
|
||||
return "kill switch tripped"
|
||||
if self.steps_taken >= self.policy.max_steps:
|
||||
return f"step budget of {self.policy.max_steps} reached"
|
||||
return None
|
||||
|
||||
def authorize(self, action: ProposedAction) -> None:
|
||||
stop = self.should_stop()
|
||||
if stop:
|
||||
self._deny(action, stop)
|
||||
if action.name not in self.policy.allowed_actions:
|
||||
self._deny(action, "action not on the allowlist")
|
||||
if action.name in self.policy.approval_required and not self.approver(action):
|
||||
self._deny(action, "human approval not given")
|
||||
self.steps_taken += 1
|
||||
self.memory.audit(action.name, "allowed", {"reason": action.reason, "payload": action.payload})
|
||||
|
||||
def record_outcome(self, action: ProposedAction, effect: Dict[str, Any]) -> None:
|
||||
"""Log what the action actually did (not what it was supposed to do)."""
|
||||
self.memory.audit(action.name, "completed", effect)
|
||||
|
||||
def record_failure(self, action: ProposedAction, error: Exception) -> None:
|
||||
self.memory.audit(action.name, "failed", {"error": f"{type(error).__name__}: {error}"})
|
||||
|
||||
def _deny(self, action: ProposedAction, why: str) -> None:
|
||||
self.memory.audit(action.name, "denied", {"why": why, "reason": action.reason})
|
||||
raise GovernanceError(f"{action.name}: {why}")
|
||||
@@ -0,0 +1,130 @@
|
||||
"""LLM plumbing: the callable protocol, JSON extraction, call budgets, and an
|
||||
optional Anthropic adapter.
|
||||
|
||||
The engine is provider-neutral. Anything with the shape ``(prompt: str) -> str``
|
||||
works as an LLM, so you can plug in any SDK, a local model, or a test double.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from typing import Any, Callable, Dict, List, Optional
|
||||
|
||||
LLM = Callable[[str], str]
|
||||
|
||||
|
||||
class LLMError(RuntimeError):
|
||||
"""The model returned something we could not use."""
|
||||
|
||||
|
||||
class BudgetExceeded(RuntimeError):
|
||||
"""A hard call budget was hit."""
|
||||
|
||||
|
||||
class CountingLLM:
|
||||
"""Wraps an LLM, counts calls, and optionally enforces a hard ceiling."""
|
||||
|
||||
def __init__(self, llm: LLM, max_calls: Optional[int] = None):
|
||||
self._llm = llm
|
||||
self.max_calls = max_calls
|
||||
self.calls = 0
|
||||
|
||||
def __call__(self, prompt: str) -> str:
|
||||
if self.max_calls is not None and self.calls >= self.max_calls:
|
||||
raise BudgetExceeded(f"LLM call budget of {self.max_calls} exhausted")
|
||||
self.calls += 1
|
||||
return self._llm(prompt)
|
||||
|
||||
|
||||
_JSON_INSTRUCTION = (
|
||||
"\n\nRespond with a single JSON object and nothing else. "
|
||||
"No markdown fences, no commentary."
|
||||
)
|
||||
|
||||
|
||||
def _extract_json_object(text: str) -> Dict[str, Any]:
|
||||
text = text.strip()
|
||||
fenced = re.search(r"```(?:json)?\s*(\{.*\})\s*```", text, re.DOTALL)
|
||||
if fenced:
|
||||
text = fenced.group(1)
|
||||
start, end = text.find("{"), text.rfind("}")
|
||||
if start == -1 or end <= start:
|
||||
raise ValueError("no JSON object found")
|
||||
value = json.loads(text[start : end + 1])
|
||||
if not isinstance(value, dict):
|
||||
raise ValueError("top-level JSON value is not an object")
|
||||
return value
|
||||
|
||||
|
||||
def complete_json(llm: LLM, prompt: str, retries: int = 1) -> Dict[str, Any]:
|
||||
"""Ask for JSON, parse it, and retry with the parse error if it is malformed."""
|
||||
attempt_prompt = prompt + _JSON_INSTRUCTION
|
||||
last_error: Optional[Exception] = None
|
||||
for _ in range(retries + 1):
|
||||
raw = llm(attempt_prompt)
|
||||
try:
|
||||
return _extract_json_object(raw)
|
||||
except (ValueError, json.JSONDecodeError) as exc:
|
||||
last_error = exc
|
||||
attempt_prompt = (
|
||||
prompt
|
||||
+ _JSON_INSTRUCTION
|
||||
+ f"\n\nYour previous reply could not be parsed ({exc}). Reply again with valid JSON only."
|
||||
)
|
||||
raise LLMError(f"model did not return valid JSON: {last_error}")
|
||||
|
||||
|
||||
def as_list(value: Any) -> List[Any]:
|
||||
"""A model-supplied field as a list; anything that is not a list becomes []."""
|
||||
return value if isinstance(value, list) else []
|
||||
|
||||
|
||||
def str_list(value: Any) -> List[str]:
|
||||
"""A model-supplied field as a list of non-empty strings."""
|
||||
return [str(v) for v in as_list(value) if str(v).strip()]
|
||||
|
||||
|
||||
def clamp01(value: Any, default: float = 0.0) -> float:
|
||||
try:
|
||||
return max(0.0, min(1.0, float(value)))
|
||||
except (TypeError, ValueError):
|
||||
return default
|
||||
|
||||
|
||||
def anthropic_llm(
|
||||
model: str = "claude-opus-5-5",
|
||||
effort: str = "medium",
|
||||
max_tokens: int = 16000,
|
||||
system: Optional[str] = None,
|
||||
client: Any = None,
|
||||
) -> LLM:
|
||||
"""Build an LLM callable backed by the Anthropic Messages API.
|
||||
|
||||
Requires ``pip install anthropic`` and credentials in the environment
|
||||
(``ANTHROPIC_API_KEY`` or an ``ant auth login`` profile). Server-side
|
||||
refusal fallbacks are enabled so a declined request is retried on a
|
||||
fallback model instead of silently returning nothing.
|
||||
"""
|
||||
if client is None:
|
||||
import anthropic # optional dependency
|
||||
|
||||
client = anthropic.Anthropic()
|
||||
|
||||
def call(prompt: str) -> str:
|
||||
kwargs: Dict[str, Any] = dict(
|
||||
model=model,
|
||||
max_tokens=max_tokens,
|
||||
output_config={"effort": effort},
|
||||
betas=["server-side-fallback-2026-07-01"],
|
||||
fallbacks="default",
|
||||
messages=[{"role": "user", "content": prompt}],
|
||||
)
|
||||
if system:
|
||||
kwargs["system"] = system
|
||||
response = client.beta.messages.create(**kwargs)
|
||||
if response.stop_reason == "refusal":
|
||||
raise LLMError("model declined the request")
|
||||
return "".join(b.text for b in response.content if b.type == "text")
|
||||
|
||||
return call
|
||||
@@ -0,0 +1,152 @@
|
||||
"""Durable memory: episodes (past problems and what came of them), notes
|
||||
(lessons distilled by reflection), key/value agent state, and an audit log.
|
||||
|
||||
Backed by plain SQLite with ordinary transactions, so writes are atomic and a
|
||||
crash cannot leave a half-written database. Use ``":memory:"`` for tests.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import sqlite3
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
|
||||
_SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS episodes (
|
||||
id TEXT PRIMARY KEY,
|
||||
problem TEXT NOT NULL,
|
||||
summary TEXT NOT NULL,
|
||||
is_open INTEGER NOT NULL,
|
||||
created_at REAL NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS notes (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
text TEXT NOT NULL,
|
||||
source_episode_ids TEXT NOT NULL,
|
||||
created_at REAL NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS state (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL
|
||||
);
|
||||
CREATE TABLE IF NOT EXISTS audit (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
at REAL NOT NULL,
|
||||
action TEXT NOT NULL,
|
||||
decision TEXT NOT NULL,
|
||||
detail TEXT NOT NULL
|
||||
);
|
||||
"""
|
||||
|
||||
_WORD = re.compile(r"[a-z][a-z0-9-]{2,}")
|
||||
_STOPWORDS = {
|
||||
"the", "and", "for", "that", "with", "this", "from", "into", "without", "what",
|
||||
"why", "how", "are", "can", "does", "design", "explain", "which", "when", "while",
|
||||
}
|
||||
|
||||
|
||||
def _tokens(text: str) -> set[str]:
|
||||
return {w for w in _WORD.findall(text.lower()) if w not in _STOPWORDS}
|
||||
|
||||
|
||||
@dataclass
|
||||
class Episode:
|
||||
id: str
|
||||
problem: str
|
||||
summary: str
|
||||
is_open: bool
|
||||
created_at: float
|
||||
|
||||
|
||||
class MemoryStore:
|
||||
def __init__(self, path: str | Path = ":memory:"):
|
||||
if path != ":memory:":
|
||||
Path(path).parent.mkdir(parents=True, exist_ok=True)
|
||||
self.conn = sqlite3.connect(str(path))
|
||||
self.conn.row_factory = sqlite3.Row
|
||||
with self.conn:
|
||||
self.conn.executescript(_SCHEMA)
|
||||
|
||||
def close(self) -> None:
|
||||
self.conn.close()
|
||||
|
||||
# -- episodes -----------------------------------------------------------
|
||||
|
||||
def add_episode(self, problem: str, summary: str, is_open: bool = True) -> Episode:
|
||||
now = time.time()
|
||||
ep_id = "ep_" + hashlib.sha256(f"{problem}\0{now}".encode()).hexdigest()[:10]
|
||||
with self.conn:
|
||||
self.conn.execute(
|
||||
"INSERT INTO episodes (id, problem, summary, is_open, created_at) VALUES (?, ?, ?, ?, ?)",
|
||||
(ep_id, problem, summary, int(is_open), now),
|
||||
)
|
||||
return Episode(ep_id, problem, summary, is_open, now)
|
||||
|
||||
def set_open(self, episode_id: str, is_open: bool) -> None:
|
||||
with self.conn:
|
||||
self.conn.execute("UPDATE episodes SET is_open = ? WHERE id = ?", (int(is_open), episode_id))
|
||||
|
||||
def episodes(self, only_open: bool = False) -> List[Episode]:
|
||||
sql = "SELECT * FROM episodes" + (" WHERE is_open = 1" if only_open else "") + " ORDER BY created_at"
|
||||
return [self._episode(r) for r in self.conn.execute(sql)]
|
||||
|
||||
def recall(self, query: str, k: int = 3) -> List[Episode]:
|
||||
"""Past episodes most lexically related to ``query``; open ones get a small boost."""
|
||||
q = _tokens(query)
|
||||
if not q:
|
||||
return []
|
||||
scored = []
|
||||
for ep in self.episodes():
|
||||
overlap = len(q & _tokens(ep.problem + " " + ep.summary))
|
||||
if overlap:
|
||||
scored.append((overlap + (0.5 if ep.is_open else 0.0), ep.created_at, ep))
|
||||
scored.sort(key=lambda t: (t[0], t[1]), reverse=True)
|
||||
return [ep for _, _, ep in scored[:k]]
|
||||
|
||||
@staticmethod
|
||||
def _episode(row: sqlite3.Row) -> Episode:
|
||||
return Episode(row["id"], row["problem"], row["summary"], bool(row["is_open"]), row["created_at"])
|
||||
|
||||
# -- notes --------------------------------------------------------------
|
||||
|
||||
def add_note(self, text: str, source_episode_ids: List[str]) -> None:
|
||||
with self.conn:
|
||||
self.conn.execute(
|
||||
"INSERT INTO notes (text, source_episode_ids, created_at) VALUES (?, ?, ?)",
|
||||
(text, json.dumps(source_episode_ids), time.time()),
|
||||
)
|
||||
|
||||
def notes(self) -> List[str]:
|
||||
return [r["text"] for r in self.conn.execute("SELECT text FROM notes ORDER BY id")]
|
||||
|
||||
# -- state & audit --------------------------------------------------------
|
||||
|
||||
def get_state(self, key: str, default: Any = None) -> Any:
|
||||
row = self.conn.execute("SELECT value FROM state WHERE key = ?", (key,)).fetchone()
|
||||
return json.loads(row["value"]) if row else default
|
||||
|
||||
def set_state(self, key: str, value: Any) -> None:
|
||||
with self.conn:
|
||||
self.conn.execute(
|
||||
"INSERT INTO state (key, value) VALUES (?, ?) "
|
||||
"ON CONFLICT(key) DO UPDATE SET value = excluded.value",
|
||||
(key, json.dumps(value)),
|
||||
)
|
||||
|
||||
def audit(self, action: str, decision: str, detail: Optional[Dict[str, Any]] = None) -> None:
|
||||
with self.conn:
|
||||
self.conn.execute(
|
||||
"INSERT INTO audit (at, action, decision, detail) VALUES (?, ?, ?, ?)",
|
||||
(time.time(), action, decision, json.dumps(detail or {})),
|
||||
)
|
||||
|
||||
def audit_log(self) -> List[Dict[str, Any]]:
|
||||
return [
|
||||
{"at": r["at"], "action": r["action"], "decision": r["decision"], "detail": json.loads(r["detail"])}
|
||||
for r in self.conn.execute("SELECT * FROM audit ORDER BY id")
|
||||
]
|
||||
@@ -0,0 +1,68 @@
|
||||
"""Wrap any agent function so it receives a MuseEngine scaffold first."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import functools
|
||||
from typing import Any, Callable, Dict, List, Optional, Union
|
||||
|
||||
from .engine import MuseEngine, Report
|
||||
|
||||
Messages = List[Dict[str, Any]]
|
||||
|
||||
|
||||
class MuseMiddleware:
|
||||
"""Decorator for ``fn(prompt: str, ...)`` or ``fn(messages: list[dict], ...)``.
|
||||
|
||||
For a string prompt, the scaffold is prepended. For a message list, a system
|
||||
message holding the scaffold is inserted at the front (or merged into an
|
||||
existing leading system message). The latest user message is what gets
|
||||
analyzed. The report from the last call is kept on ``last_report``.
|
||||
"""
|
||||
|
||||
def __init__(self, engine: Optional[MuseEngine] = None, top_k: int = 3, return_report: bool = False):
|
||||
self.engine = engine or MuseEngine()
|
||||
self.top_k = top_k
|
||||
self.return_report = return_report
|
||||
self.last_report: Optional[Report] = None
|
||||
|
||||
def __call__(self, agent_fn: Callable[..., Any]) -> Callable[..., Any]:
|
||||
@functools.wraps(agent_fn)
|
||||
def wrapped(prompt_or_messages: Union[str, Messages], *args: Any, **kwargs: Any) -> Any:
|
||||
if isinstance(prompt_or_messages, str):
|
||||
report = self.engine.imagine(prompt_or_messages, top_k=self.top_k)
|
||||
result = agent_fn(f"{report.scaffold()}\n\n{prompt_or_messages}", *args, **kwargs)
|
||||
elif isinstance(prompt_or_messages, list):
|
||||
user_text = _latest_user_text(prompt_or_messages)
|
||||
if not user_text:
|
||||
raise ValueError("MuseMiddleware needs at least one user message with text")
|
||||
report = self.engine.imagine(user_text, top_k=self.top_k)
|
||||
result = agent_fn(_inject_system(prompt_or_messages, report.scaffold()), *args, **kwargs)
|
||||
else:
|
||||
raise TypeError("expected a str prompt or a list of message dicts")
|
||||
self.last_report = report
|
||||
return {"response": result, "report": report} if self.return_report else result
|
||||
|
||||
return wrapped
|
||||
|
||||
|
||||
def _latest_user_text(messages: Messages) -> str:
|
||||
for msg in reversed(messages):
|
||||
if msg.get("role") != "user":
|
||||
continue
|
||||
content = msg.get("content", "")
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
if isinstance(content, list): # content blocks
|
||||
return " ".join(b.get("text", "") for b in content if isinstance(b, dict) and b.get("type") == "text")
|
||||
return ""
|
||||
|
||||
|
||||
def _inject_system(messages: Messages, scaffold: str) -> Messages:
|
||||
if messages and messages[0].get("role") == "system":
|
||||
content = messages[0].get("content")
|
||||
if isinstance(content, str):
|
||||
return [dict(messages[0], content=f"{content}\n\n{scaffold}"), *messages[1:]]
|
||||
if isinstance(content, list):
|
||||
blocks = [*content, {"type": "text", "text": scaffold}]
|
||||
return [dict(messages[0], content=blocks), *messages[1:]]
|
||||
return [{"role": "system", "content": scaffold}, *messages]
|
||||
@@ -0,0 +1,24 @@
|
||||
[build-system]
|
||||
requires = ["setuptools>=68"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "museengine"
|
||||
version = "0.2.0"
|
||||
description = "Structured cross-domain ideation for LLM agents, with adversarial critique, durable memory, and a governed autonomous loop."
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.10"
|
||||
dependencies = []
|
||||
|
||||
[project.optional-dependencies]
|
||||
anthropic = ["anthropic>=1.0"]
|
||||
dev = ["pytest>=8"]
|
||||
|
||||
[project.scripts]
|
||||
museengine = "museengine.__main__:main"
|
||||
|
||||
[tool.setuptools]
|
||||
packages = ["museengine"]
|
||||
|
||||
[tool.setuptools.package-data]
|
||||
museengine = ["archetypes.json"]
|
||||
Whitespace-only changes.
@@ -0,0 +1,65 @@
|
||||
import json
|
||||
import re
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class FakeLLM:
|
||||
"""Routes on the ``TASK: X`` marker in each prompt and returns canned JSON.
|
||||
|
||||
``responses`` maps a task name to a dict, a list of dicts (consumed in
|
||||
order), or a callable taking the prompt.
|
||||
"""
|
||||
|
||||
def __init__(self, **responses):
|
||||
self.responses = {k.upper(): v for k, v in responses.items()}
|
||||
self.prompts = []
|
||||
|
||||
def __call__(self, prompt):
|
||||
self.prompts.append(prompt)
|
||||
task = re.search(r"TASK: (\w+)", prompt).group(1)
|
||||
value = self.responses[task]
|
||||
if callable(value):
|
||||
value = value(prompt)
|
||||
elif isinstance(value, list):
|
||||
value = value.pop(0) if len(value) > 1 else value[0]
|
||||
return value if isinstance(value, str) else json.dumps(value)
|
||||
|
||||
def count(self, task):
|
||||
return sum(f"TASK: {task.upper()}" in p for p in self.prompts)
|
||||
|
||||
|
||||
ANALYZE = {
|
||||
"domain": "distributed systems",
|
||||
"tensions": [{"axis": "throughput vs coordination", "description": "locks serialize", "weight": 0.9}],
|
||||
"constraints": ["Little's law bounds in-flight work"],
|
||||
"failure_modes": ["retry storms"],
|
||||
}
|
||||
|
||||
MAP = {
|
||||
"analogies": [
|
||||
{"archetype_id": "mycelial_physarum", "mapping": "links = tubes", "breaks_where": "no fluid", "fit": 0.8},
|
||||
{"archetype_id": "made_up_id", "mapping": "x", "breaks_where": "y", "fit": 0.99},
|
||||
{"archetype_id": "spin_glass_annealing", "mapping": "conflicts = defects", "breaks_where": "no energy", "fit": 0.6},
|
||||
]
|
||||
}
|
||||
|
||||
GENERATE = {
|
||||
"title": "Flux-weighted routing",
|
||||
"thesis": "Route by recent ack rate.",
|
||||
"mechanism": ["track ack flux", "prune idle links"],
|
||||
"tests": ["Sever 30% of links; fails if p99 doubles"],
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_llm():
|
||||
return FakeLLM(
|
||||
analyze=ANALYZE,
|
||||
map=MAP,
|
||||
generate=GENERATE,
|
||||
critique=[
|
||||
{"novelty": 0.7, "rigor": 0.6, "fatal_flaws": [], "verdict": "survives", "reasoning": "ok"},
|
||||
{"novelty": 0.9, "rigor": 0.2, "fatal_flaws": ["incoherent"], "verdict": "refuted", "reasoning": "no"},
|
||||
],
|
||||
)
|
||||
@@ -0,0 +1,168 @@
|
||||
import random
|
||||
|
||||
import pytest
|
||||
|
||||
from museengine import (
|
||||
BudgetExceeded,
|
||||
CountingLLM,
|
||||
Governor,
|
||||
KillSwitch,
|
||||
MemoryStore,
|
||||
MuseAgent,
|
||||
MuseEngine,
|
||||
Policy,
|
||||
run_ab_eval,
|
||||
sign_test,
|
||||
)
|
||||
from museengine.evaluation import judge_pair
|
||||
|
||||
from .conftest import FakeLLM
|
||||
|
||||
REFUTE_ALL = dict(
|
||||
analyze={"domain": "x"},
|
||||
map={"analogies": [{"archetype_id": "mycelial_physarum", "mapping": "m", "fit": 0.5}]},
|
||||
generate={"title": "t", "thesis": "t", "tests": ["t"]},
|
||||
critique={"novelty": 0.1, "rigor": 0.1, "verdict": "refuted"},
|
||||
reflect={"lessons": ["narrow the scope"]},
|
||||
propose_goal={"problem": "a narrower problem", "why": "the first was too broad"},
|
||||
)
|
||||
|
||||
|
||||
def make_agent(llm=None, approver=lambda a: False, **policy):
|
||||
memory = MemoryStore()
|
||||
engine = MuseEngine(llm=llm, memory=memory)
|
||||
governor = Governor(memory, Policy(**policy) if policy else None, approver=approver)
|
||||
return MuseAgent(engine, memory, governor, reflect_every=2), memory
|
||||
|
||||
|
||||
def test_memory_persists_across_reopen(tmp_path):
|
||||
path = tmp_path / "mind.db"
|
||||
m = MemoryStore(path)
|
||||
m.add_episode("p", "s")
|
||||
m.set_state("goals", ["g"])
|
||||
m.close()
|
||||
m2 = MemoryStore(path)
|
||||
assert [e.problem for e in m2.episodes()] == ["p"]
|
||||
assert m2.get_state("goals") == ["g"]
|
||||
|
||||
|
||||
def test_agent_idles_instead_of_inventing_work():
|
||||
agent, memory = make_agent()
|
||||
result = agent.run()
|
||||
assert result.steps == [] and result.stopped_because == "nothing to do"
|
||||
|
||||
|
||||
def test_agent_explores_queued_goals_offline():
|
||||
agent, memory = make_agent()
|
||||
agent.add_goal("estuary tidal clearance")
|
||||
result = agent.run()
|
||||
assert [s.action for s in result.steps] == ["explore"]
|
||||
assert result.steps[0].effect["mode"] == "heuristic"
|
||||
decisions = [(e["action"], e["decision"]) for e in memory.audit_log()]
|
||||
assert decisions == [("explore", "allowed"), ("explore", "completed")]
|
||||
|
||||
|
||||
def test_self_chosen_goal_needs_approval_and_denial_stops():
|
||||
agent, memory = make_agent(FakeLLM(**REFUTE_ALL))
|
||||
agent.add_goal("problem one")
|
||||
agent.add_goal("problem two")
|
||||
result = agent.run()
|
||||
actions = [s.action for s in result.steps]
|
||||
assert actions == ["explore", "explore", "reflect", "propose_goal"]
|
||||
assert result.stopped_because.startswith("denied: adopt_goal")
|
||||
assert agent.goals() == [] # nothing queued without approval
|
||||
assert memory.notes() == ["narrow the scope"]
|
||||
# and it doesn't ask again
|
||||
assert agent.run().stopped_because == "nothing to do"
|
||||
|
||||
|
||||
def test_approved_goal_is_queued_and_explored():
|
||||
seen = []
|
||||
agent, memory = make_agent(FakeLLM(**REFUTE_ALL), approver=lambda a: seen.append(a) or True, max_steps=6)
|
||||
agent.add_goal("problem one")
|
||||
agent.add_goal("problem two")
|
||||
result = agent.run()
|
||||
assert [s.action for s in result.steps] == [
|
||||
"explore", "explore", "reflect", "propose_goal", "adopt_goal", "explore",
|
||||
]
|
||||
assert seen[0].payload["problem"] == "a narrower problem" # approver saw the exact text
|
||||
assert result.stopped_because == "step budget of 6 reached"
|
||||
|
||||
|
||||
def test_disallowed_action_is_denied():
|
||||
agent, memory = make_agent(allowed_actions=frozenset({"reflect"}))
|
||||
agent.add_goal("x")
|
||||
assert agent.run().stopped_because.startswith("denied: explore")
|
||||
assert memory.audit_log()[-1]["decision"] == "denied"
|
||||
|
||||
|
||||
def test_kill_switch_file(tmp_path):
|
||||
agent, _ = make_agent()
|
||||
agent.governor.kill_switch = KillSwitch(tmp_path / "STOP")
|
||||
agent.add_goal("a")
|
||||
agent.add_goal("b")
|
||||
(tmp_path / "STOP").touch()
|
||||
assert agent.run().stopped_because == "kill switch tripped"
|
||||
assert agent.goals() == ["a", "b"]
|
||||
|
||||
|
||||
def test_llm_budget():
|
||||
llm = CountingLLM(lambda p: "{}", max_calls=1)
|
||||
llm("a")
|
||||
with pytest.raises(BudgetExceeded):
|
||||
llm("b")
|
||||
|
||||
|
||||
def test_sign_test():
|
||||
assert sign_test(0, 0) == 1.0
|
||||
assert sign_test(5, 5) == 1.0
|
||||
assert sign_test(10, 0) == pytest.approx(2 / 1024)
|
||||
|
||||
|
||||
def test_judge_win_needs_consistency_across_order():
|
||||
# A judge that always picks whichever answer is shown first -> pure position bias.
|
||||
biased = FakeLLM(judge={"winner": "A", "reason": "first"})
|
||||
assert judge_pair(biased, "p", "base", "treat", random.Random(0)).outcome == "tie"
|
||||
|
||||
# A judge that always prefers the answer containing "treat".
|
||||
def fair(prompt):
|
||||
a = prompt.split("=== Answer A ===")[1].split("=== Answer B ===")[0]
|
||||
return {"winner": "A" if "treat" in a else "B"}
|
||||
|
||||
assert judge_pair(FakeLLM(judge=fair), "p", "base", "treat", random.Random(0)).outcome == "treated"
|
||||
|
||||
|
||||
def test_run_ab_eval_uses_scaffold():
|
||||
solver_prompts = []
|
||||
|
||||
def solver(prompt):
|
||||
solver_prompts.append(prompt)
|
||||
return "treat" if "<muse_scaffold>" in prompt else "base"
|
||||
|
||||
def fair(prompt):
|
||||
a = prompt.split("=== Answer A ===")[1].split("=== Answer B ===")[0]
|
||||
return {"winner": "A" if "treat" in a else "B"}
|
||||
|
||||
result = run_ab_eval(["slime mold routing", "estuary"], solver, FakeLLM(judge=fair), MuseEngine())
|
||||
assert (result.wins, result.losses, result.ties) == (2, 0, 0)
|
||||
assert len(solver_prompts) == 4
|
||||
assert "scaffold won 2" in result.summary()
|
||||
|
||||
|
||||
def test_failed_explore_keeps_goal():
|
||||
def boom(prompt):
|
||||
raise RuntimeError("model down")
|
||||
|
||||
agent, memory = make_agent(boom)
|
||||
agent.add_goal("keep me")
|
||||
with pytest.raises(RuntimeError):
|
||||
agent.run()
|
||||
assert agent.goals() == ["keep me"]
|
||||
assert memory.audit_log()[-1]["decision"] == "failed"
|
||||
|
||||
|
||||
def test_run_ab_eval_rejects_blank_problems_before_calling_models():
|
||||
calls = []
|
||||
with pytest.raises(ValueError):
|
||||
run_ab_eval(["ok", " "], lambda p: calls.append(p) or "", lambda p: "{}", MuseEngine())
|
||||
assert calls == []
|
||||
@@ -0,0 +1,170 @@
|
||||
import pytest
|
||||
|
||||
from museengine import (
|
||||
Candidate,
|
||||
LLMError,
|
||||
MemoryStore,
|
||||
MuseEngine,
|
||||
MuseMiddleware,
|
||||
complete_json,
|
||||
load_default_archetypes,
|
||||
pareto_frontier,
|
||||
)
|
||||
|
||||
from .conftest import FakeLLM
|
||||
|
||||
|
||||
def test_library_loads():
|
||||
archetypes = load_default_archetypes()
|
||||
assert len(archetypes) == 6
|
||||
assert all(a.operators for a in archetypes)
|
||||
|
||||
|
||||
def test_heuristic_mode_invents_nothing():
|
||||
report = MuseEngine().imagine("Why does tidal shear help estuary clearance?")
|
||||
assert report.mode == "heuristic"
|
||||
assert report.llm_calls == 0
|
||||
assert report.candidates == [] and report.frontier == []
|
||||
assert all(a.fit is None for a in report.analogies)
|
||||
assert "not been checked" in report.scaffold()
|
||||
|
||||
|
||||
def test_heuristic_mode_returns_nothing_for_unrelated_input():
|
||||
assert MuseEngine().imagine("asdf qwerty").analogies == []
|
||||
|
||||
|
||||
def test_llm_pipeline_end_to_end(fake_llm):
|
||||
report = MuseEngine(llm=fake_llm).imagine("Design a backpressure mesh", top_k=3)
|
||||
assert report.mode == "llm"
|
||||
# the hallucinated archetype id is dropped
|
||||
assert [a.archetype_id for a in report.analogies] == ["mycelial_physarum", "spin_glass_annealing"]
|
||||
assert len(report.candidates) == 2
|
||||
verdicts = [c.verdict for c in report.candidates]
|
||||
assert verdicts == ["survives", "refuted"]
|
||||
assert [c.id for c in report.frontier] == [report.candidates[0].id]
|
||||
assert report.llm_calls == 1 + 1 + 2 + 2
|
||||
assert "survived an adversarial critique" in report.scaffold()
|
||||
|
||||
|
||||
def test_separate_critic_is_used(fake_llm):
|
||||
critic = FakeLLM(critique={"novelty": 0.5, "rigor": 0.5, "verdict": "survives", "reasoning": ""})
|
||||
report = MuseEngine(llm=fake_llm, critic=critic).imagine("Design a backpressure mesh")
|
||||
assert fake_llm.count("critique") == 0
|
||||
assert critic.count("critique") == 2
|
||||
assert report.llm_calls == 6
|
||||
|
||||
|
||||
def test_candidate_without_tests_cannot_survive(fake_llm):
|
||||
fake_llm.responses["GENERATE"] = dict(fake_llm.responses["GENERATE"], tests=[])
|
||||
fake_llm.responses["CRITIQUE"] = {"novelty": 1, "rigor": 1, "verdict": "survives"}
|
||||
report = MuseEngine(llm=fake_llm).imagine("Design a backpressure mesh")
|
||||
assert report.frontier == []
|
||||
assert all(c.verdict == "refuted" for c in report.candidates)
|
||||
|
||||
|
||||
def test_memory_recall_feeds_mapping(fake_llm):
|
||||
memory = MemoryStore()
|
||||
engine = MuseEngine(llm=fake_llm, memory=memory)
|
||||
engine.imagine("Backpressure mesh for hot-key traffic")
|
||||
report = engine.imagine("Hot-key traffic in a cache cluster")
|
||||
assert report.recalled == ["Backpressure mesh for hot-key traffic"]
|
||||
map_prompts = [p for p in fake_llm.prompts if "TASK: MAP" in p]
|
||||
assert "Backpressure mesh for hot-key traffic" in map_prompts[-1]
|
||||
assert len(memory.episodes()) == 2
|
||||
|
||||
|
||||
def test_episode_open_when_nothing_survives():
|
||||
llm = FakeLLM(
|
||||
analyze={"domain": "x"},
|
||||
map={"analogies": [{"archetype_id": "mycelial_physarum", "mapping": "m", "fit": 0.5}]},
|
||||
generate={"title": "t", "thesis": "t", "tests": ["t"]},
|
||||
critique={"novelty": 0.1, "rigor": 0.1, "verdict": "refuted"},
|
||||
)
|
||||
memory = MemoryStore()
|
||||
MuseEngine(llm=llm, memory=memory).imagine("some problem")
|
||||
assert memory.episodes(only_open=True)
|
||||
|
||||
|
||||
def _cand(cid, n, r, verdict="survives"):
|
||||
return Candidate(cid, cid, "", [], ["t"], "a", novelty=n, rigor=r, verdict=verdict)
|
||||
|
||||
|
||||
def test_pareto_frontier():
|
||||
cs = [_cand("a", 0.9, 0.2), _cand("b", 0.5, 0.5), _cand("c", 0.4, 0.4), _cand("d", 1.0, 1.0, "refuted")]
|
||||
frontier = pareto_frontier(cs)
|
||||
assert {c.id for c in frontier} == {"a", "b"}
|
||||
assert not cs[2].pareto
|
||||
|
||||
|
||||
def test_complete_json_retries_then_fails():
|
||||
replies = iter(["not json", '```json\n{"ok": 1}\n```'])
|
||||
assert complete_json(lambda p: next(replies), "q") == {"ok": 1}
|
||||
with pytest.raises(LLMError):
|
||||
complete_json(lambda p: "still not json", "q", retries=1)
|
||||
|
||||
|
||||
def test_middleware_injects_system_message():
|
||||
seen = {}
|
||||
|
||||
@MuseMiddleware(MuseEngine())
|
||||
def agent(messages):
|
||||
seen["messages"] = messages
|
||||
return "ok"
|
||||
|
||||
assert agent([{"role": "user", "content": "estuary tidal clearance"}]) == "ok"
|
||||
assert seen["messages"][0]["role"] == "system"
|
||||
assert "<muse_scaffold>" in seen["messages"][0]["content"]
|
||||
assert seen["messages"][1] == {"role": "user", "content": "estuary tidal clearance"}
|
||||
|
||||
|
||||
def test_middleware_merges_existing_system_message():
|
||||
seen = {}
|
||||
|
||||
@MuseMiddleware(MuseEngine())
|
||||
def agent(messages):
|
||||
seen["messages"] = messages
|
||||
|
||||
agent([{"role": "system", "content": "be brief"}, {"role": "user", "content": "slime mold routing"}])
|
||||
assert len(seen["messages"]) == 2
|
||||
assert seen["messages"][0]["content"].startswith("be brief")
|
||||
|
||||
|
||||
def test_anthropic_adapter_with_stub_client():
|
||||
from types import SimpleNamespace
|
||||
|
||||
from museengine import anthropic_llm
|
||||
|
||||
calls = []
|
||||
|
||||
class Messages:
|
||||
def create(self, **kwargs):
|
||||
calls.append(kwargs)
|
||||
return SimpleNamespace(stop_reason="end_turn", content=[SimpleNamespace(type="text", text="hi")])
|
||||
|
||||
client = SimpleNamespace(beta=SimpleNamespace(messages=Messages()))
|
||||
assert anthropic_llm(client=client)("hello") == "hi"
|
||||
assert calls[0]["model"] == "claude-opus-5-5"
|
||||
assert calls[0]["messages"] == [{"role": "user", "content": "hello"}]
|
||||
|
||||
|
||||
def test_malformed_list_fields_are_ignored(fake_llm):
|
||||
fake_llm.responses["ANALYZE"] = {"domain": "x", "tensions": None, "constraints": "one string"}
|
||||
fake_llm.responses["GENERATE"] = dict(fake_llm.responses["GENERATE"], tests="not a list")
|
||||
fake_llm.responses["CRITIQUE"] = {"novelty": 1, "rigor": 1, "verdict": "survives", "fatal_flaws": None}
|
||||
report = MuseEngine(llm=fake_llm).imagine("Design a backpressure mesh")
|
||||
assert report.structure.tensions == [] and report.structure.constraints == []
|
||||
assert all(c.tests == [] and c.verdict == "refuted" for c in report.candidates)
|
||||
|
||||
|
||||
def test_middleware_merges_block_system_message():
|
||||
seen = {}
|
||||
|
||||
@MuseMiddleware(MuseEngine())
|
||||
def agent(messages):
|
||||
seen["messages"] = messages
|
||||
|
||||
agent([{"role": "system", "content": [{"type": "text", "text": "be brief"}]},
|
||||
{"role": "user", "content": "slime mold routing"}])
|
||||
assert len(seen["messages"]) == 2
|
||||
blocks = seen["messages"][0]["content"]
|
||||
assert blocks[0]["text"] == "be brief" and "<muse_scaffold>" in blocks[1]["text"]
|
||||
Reference in new issue
Block a user