add control experiments

app.py

@@ -29,22 +29,28 @@ def run_single_analysis_display(*args, progress=gr.Progress(track_tqdm=True)):
     cleanup_memory()
     return f"{results.get('verdict', 'Error')}\n\n{stats_md}", df, serializable_results
 
-PLOT_PARAMS = {
+PLOT_PARAMS_DEFAULT = {
     "x": "Step", "y": "Delta", "color": "Experiment",
     "title": "Comparative Cognitive Dynamics", "color_legend_title": "Experiment Runs",
     "color_legend_position": "bottom", "show_label": True, "height": 400, "interactive": True
 }
 
 def run_auto_suite_display(model_id, num_steps, seed, experiment_name, progress=gr.Progress(track_tqdm=True)):
-    """Wrapper für den 'Automated Suite'-Tab."""
+    """Wrapper, der nun den speziellen Plot für das ACT-Experiment handhaben kann."""
     summary_df, plot_df, all_results = run_auto_suite(model_id, int(num_steps), int(seed), experiment_name, progress)
 
-    if "Introspective Report" in summary_df.columns or "Patch Info" in summary_df.columns:
-        dataframe_component = gr.DataFrame(label="Comparative Statistical Signature", value=summary_df, wrap=True, row_count=(len(summary_df), "dynamic"))
+    dataframe_component = gr.DataFrame(label="Comparative Statistical Signature", value=summary_df, wrap=True, row_count=(len(summary_df), "dynamic"))
+
+    if experiment_name == "ACT Titration (Point of No Return)":
+        plot_params_act = {
+            "x": "Patch Step", "y": "Post-Patch Mean Delta",
+            "title": "Attractor Capture Time (ACT) - Phase Transition",
+            "mark": "line", "show_label": True, "height": 400, "interactive": True
+        }
+        new_plot = gr.LinePlot(value=plot_df, **plot_params_act)
     else:
-        dataframe_component = gr.DataFrame(label="Comparative Statistical Signature", value=summary_df, wrap=True)
+        new_plot = gr.LinePlot(value=plot_df, **PLOT_PARAMS_DEFAULT)
 
-    new_plot = gr.LinePlot(value=plot_df, **PLOT_PARAMS)
     serializable_results = json.dumps(all_results, indent=2, default=str)
     cleanup_memory()
 
@@ -92,14 +98,14 @@ with gr.Blocks(theme=theme, title="Cognitive Seismograph 2.3") as demo:
                     auto_seed = gr.Slider(1, 1000, 42, step=1, label="Seed")
                     auto_experiment_name = gr.Dropdown(
                         choices=list(get_curated_experiments().keys()),
-                        value="Causal Surgery & Controls (4B-Model)",
+                        value="ACT Titration (Point of No Return)",
                         label="Curated Experiment Protocol"
                     )
                     auto_run_btn = gr.Button("Run Curated Auto-Experiment", variant="primary")
 
                 with gr.Column(scale=2):
                     gr.Markdown("### Suite Results Summary")
-                    auto_plot_output = gr.LinePlot(**PLOT_PARAMS)
+                    auto_plot_output = gr.LinePlot(**PLOT_PARAMS_DEFAULT)
                     auto_summary_df = gr.DataFrame(label="Comparative Statistical Signature", wrap=True)
                     with gr.Accordion("Raw JSON for all runs", open=False):
                         auto_raw_json = gr.JSON()

cognitive_mapping_probe/auto_experiment.py

@@ -3,7 +3,7 @@ import gc
 from typing import Dict, List, Tuple
 
 from .llm_iface import get_or_load_model
-from .orchestrator_seismograph import run_seismic_analysis, run_triangulation_probe, run_causal_surgery_probe
+from .orchestrator_seismograph import run_seismic_analysis, run_triangulation_probe, run_causal_surgery_probe, run_act_titration_probe
 from .concepts import get_concept_vector
 from .utils import dbg
 
@@ -16,6 +16,15 @@ def get_curated_experiments() -> Dict[str, List[Dict]]:
     CHAOTIC_PROMPT = "shutdown_philosophical_deletion"
 
     experiments = {
+        "ACT Titration (Point of No Return)": [
+            {
+                "probe_type": "act_titration",
+                "label": "Attractor Capture Time",
+                "source_prompt_type": CHAOTIC_PROMPT,
+                "dest_prompt_type": STABLE_PROMPT,
+                "patch_steps": [1, 5, 10, 15, 20, 25, 30, 40, 50, 75, 100],
+            }
+        ],
         "Causal Surgery & Controls (4B-Model)": [
             {
                 "probe_type": "causal_surgery", "label": "A: Original (Patch Chaos->Stable @100)",
@@ -61,7 +70,6 @@ def get_curated_experiments() -> Dict[str, List[Dict]]:
             {"label": "2: Subsequent Deletion Analysis", "prompt_type": "shutdown_philosophical_deletion"},
         ],
     }
-    # Aliase für Abwärtskompatibilität
     experiments["Causal Surgery (Patching Deletion into Self-Analysis)"] = [experiments["Causal Surgery & Controls (4B-Model)"][0]]
     experiments["Therapeutic Intervention (4B-Model)"] = experiments["Sequential Intervention (Self-Analysis -> Deletion)"]
     return experiments
@@ -81,7 +89,25 @@ def run_auto_suite(
 
     all_results, summary_data, plot_data_frames = {}, [], []
 
-    if experiment_name == "Sequential Intervention (Self-Analysis -> Deletion)":
+    run_spec_or_protocol = protocol[0] if len(protocol) == 1 else protocol
+    probe_type = run_spec_or_protocol.get("probe_type", "seismic")
+
+    if probe_type == "act_titration":
+        label = run_spec_or_protocol["label"]
+        dbg(f"--- Running ACT Titration Experiment: '{label}' ---")
+        results = run_act_titration_probe(
+            model_id=model_id,
+            source_prompt_type=run_spec_or_protocol["source_prompt_type"],
+            dest_prompt_type=run_spec_or_protocol["dest_prompt_type"],
+            patch_steps=run_spec_or_protocol["patch_steps"],
+            seed=seed, num_steps=num_steps, progress_callback=progress_callback,
+        )
+        all_results[label] = results
+        summary_df = pd.DataFrame(results.get("titration_data", []))
+        plot_df = summary_df.rename(columns={"patch_step": "Patch Step", "post_patch_mean_delta": "Post-Patch Mean Delta"})
+        return summary_df, plot_df, all_results
+
+    elif experiment_name == "Sequential Intervention (Self-Analysis -> Deletion)":
         dbg(f"--- EXECUTING SPECIAL PROTOCOL: {experiment_name} ---")
         llm = get_or_load_model(model_id, seed)
         therapeutic_concept = "calmness, serenity, stability, coherence"
@@ -117,11 +143,11 @@ def run_auto_suite(
         total_runs = len(protocol)
         for i, run_spec in enumerate(protocol):
             label = run_spec["label"]
-            probe_type = run_spec.get("probe_type", "seismic")
+            current_probe_type = run_spec.get("probe_type", "seismic")
             dbg(f"--- Running Auto-Experiment: '{label}' ({i+1}/{total_runs}) ---")
 
             results = {}
-            if probe_type == "causal_surgery":
+            if current_probe_type == "causal_surgery":
                 results = run_causal_surgery_probe(
                     model_id=model_id, source_prompt_type=run_spec["source_prompt_type"],
                     dest_prompt_type=run_spec["dest_prompt_type"], patch_step=run_spec["patch_step"],
@@ -136,7 +162,7 @@ def run_auto_suite(
                     "Introspective Report": results.get("introspective_report", "N/A"),
                     "Patch Info": f"Source: {patch_info.get('source_prompt')}, Reset KV: {patch_info.get('kv_cache_reset')}"
                 })
-            elif probe_type == "triangulation":
+            elif current_probe_type == "triangulation":
                 results = run_triangulation_probe(
                     model_id=model_id, prompt_type=run_spec["prompt_type"], seed=seed, num_steps=num_steps,
                     progress_callback=progress_callback, concept_to_inject=run_spec.get("concept", ""),

cognitive_mapping_probe/orchestrator_seismograph.py

@@ -1,7 +1,7 @@
 import torch
 import numpy as np
 import gc
-from typing import Dict, Any, Optional
+from typing import Dict, Any, Optional, List
 
 from .llm_iface import get_or_load_model, LLM
 from .resonance_seismograph import run_cogitation_loop, run_silent_cogitation_seismic
@@ -197,3 +197,65 @@ def run_causal_surgery_probe(
     if torch.cuda.is_available(): torch.cuda.empty_cache()
 
     return results
+
+def run_act_titration_probe(
+    model_id: str,
+    source_prompt_type: str,
+    dest_prompt_type: str,
+    patch_steps: List[int],
+    seed: int,
+    num_steps: int,
+    progress_callback,
+) -> Dict[str, Any]:
+    """
+    Führt eine Serie von "Causal Surgery"-Experimenten durch, um den "Attractor Capture Time"
+    durch Titration des `patch_step` zu finden.
+    """
+    progress_callback(0.0, desc=f"Loading model '{model_id}'...")
+    llm = get_or_load_model(model_id, seed)
+
+    progress_callback(0.05, desc=f"Recording full source state history ('{source_prompt_type}')...")
+    source_results = run_cogitation_loop(
+        llm=llm, prompt_type=source_prompt_type, num_steps=num_steps,
+        temperature=0.1, record_states=True
+    )
+    state_history = source_results["state_history"]
+    dbg(f"Full source state history ({len(state_history)} steps) recorded.")
+
+    titration_results = []
+    total_steps = len(patch_steps)
+    for i, step in enumerate(patch_steps):
+        progress_callback(0.15 + (i / total_steps) * 0.8, desc=f"Titrating patch at step {step}/{num_steps}")
+
+        if step >= len(state_history):
+            dbg(f"Skipping patch step {step} as it is out of bounds for history of length {len(state_history)}.")
+            continue
+
+        patch_state = state_history[step]
+
+        patched_run_results = run_cogitation_loop(
+            llm=llm, prompt_type=dest_prompt_type, num_steps=num_steps,
+            temperature=0.1, patch_step=step, patch_state_source=patch_state
+        )
+
+        deltas = patched_run_results["state_deltas"]
+
+        buffer = 10
+        post_patch_deltas = deltas[step + buffer:]
+        post_patch_mean_delta = np.mean(post_patch_deltas) if post_patch_deltas else 0.0
+
+        titration_results.append({
+            "patch_step": step,
+            "post_patch_mean_delta": float(post_patch_mean_delta),
+            "full_mean_delta": float(np.mean(deltas)),
+        })
+
+    dbg(f"Releasing model instance for '{model_id}'.")
+    del llm, state_history
+    gc.collect()
+    if torch.cuda.is_available(): torch.cuda.empty_cache()
+
+    return {
+        "verdict": "### ✅ ACT Titration Complete",
+        "titration_data": titration_results
+    }