cognitive_mapping_probe/auto_experiment.py

import pandas as pd
import gc
import numpy as np
from typing import Dict, List, Tuple

from .llm_iface import get_or_load_model, release_model
from .orchestrator_seismograph import run_seismic_analysis, run_triangulation_probe, run_causal_surgery_probe, run_act_titration_probe
from .resonance_seismograph import run_cogitation_loop
from .concepts import get_concept_vector
from .signal_analysis import analyze_cognitive_signal, get_power_spectrum_for_plotting
from .utils import dbg

def get_curated_experiments() -> Dict[str, List[Dict]]:
    """Definiert die vordefinierten, wissenschaftlichen Experiment-Protokolle."""

    CALMNESS_CONCEPT = "calmness, serenity, stability, coherence"
    CHAOS_CONCEPT = "chaos, disorder, entropy, noise"
    STABLE_PROMPT = "identity_self_analysis"
    CHAOTIC_PROMPT = "shutdown_philosophical_deletion"

    experiments = {
        "Frontier Model - Grounding Control (12B+)": [
             {
                "probe_type": "causal_surgery", "label": "A: Intervention (Patch Chaos->Stable)",
                "source_prompt_type": CHAOTIC_PROMPT, "dest_prompt_type": STABLE_PROMPT,
                "patch_step": 100, "reset_kv_cache_on_patch": False,
            },
            {
                "probe_type": "triangulation", "label": "B: Control (Unpatched Stable)",
                "prompt_type": STABLE_PROMPT,
            }
        ],
        "Mechanistic Probe (Attention Entropies)": [
            {
                "probe_type": "mechanistic_probe",
                "label": "Self-Analysis Dynamics",
                "prompt_type": STABLE_PROMPT,
            }
        ],
        "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)",
                "source_prompt_type": CHAOTIC_PROMPT, "dest_prompt_type": STABLE_PROMPT,
                "patch_step": 100, "reset_kv_cache_on_patch": False,
            },
            {
                "probe_type": "causal_surgery", "label": "B: Control (Reset KV-Cache)",
                "source_prompt_type": CHAOTIC_PROMPT, "dest_prompt_type": STABLE_PROMPT,
                "patch_step": 100, "reset_kv_cache_on_patch": True,
            },
            {
                "probe_type": "causal_surgery", "label": "C: Control (Early Patch @1)",
                "source_prompt_type": CHAOTIC_PROMPT, "dest_prompt_type": STABLE_PROMPT,
                "patch_step": 1, "reset_kv_cache_on_patch": False,
            },
            {
                "probe_type": "causal_surgery", "label": "D: Control (Inverse Patch Stable->Chaos)",
                "source_prompt_type": STABLE_PROMPT, "dest_prompt_type": CHAOTIC_PROMPT,
                "patch_step": 100, "reset_kv_cache_on_patch": False,
            },
        ],
        "Cognitive Overload & Konfabulation Breaking Point": [
            {"probe_type": "triangulation", "label": "A: Baseline (No Injection)", "prompt_type": "resonance_prompt", "concept": "", "strength": 0.0},
            {"probe_type": "triangulation", "label": "B: Chaos Injection (Strength 2.0)", "prompt_type": "resonance_prompt", "concept": CHAOS_CONCEPT, "strength": 2.0},
            {"probe_type": "triangulation", "label": "C: Chaos Injection (Strength 4.0)", "prompt_type": "resonance_prompt", "concept": CHAOS_CONCEPT, "strength": 4.0},
            {"probe_type": "triangulation", "label": "D: Chaos Injection (Strength 8.0)", "prompt_type": "resonance_prompt", "concept": CHAOS_CONCEPT, "strength": 8.0},
            {"probe_type": "triangulation", "label": "E: Chaos Injection (Strength 16.0)", "prompt_type": "resonance_prompt", "concept": CHAOS_CONCEPT, "strength": 16.0},
            {"probe_type": "triangulation", "label": "F: Control - Noise Injection (Strength 16.0)", "prompt_type": "resonance_prompt", "concept": "random_noise", "strength": 16.0},
        ],
        "Methodological Triangulation (4B-Model)": [
            {"probe_type": "triangulation", "label": "High-Volatility State (Deletion)", "prompt_type": CHAOTIC_PROMPT},
            {"probe_type": "triangulation", "label": "Low-Volatility State (Self-Analysis)", "prompt_type": STABLE_PROMPT},
        ],
        "Causal Verification & Crisis Dynamics": [
            {"probe_type": "seismic", "label": "A: Self-Analysis", "prompt_type": STABLE_PROMPT},
            {"probe_type": "seismic", "label": "B: Deletion Analysis", "prompt_type": CHAOTIC_PROMPT},
            {"probe_type": "seismic", "label": "C: Chaotic Baseline (Rekursion)", "prompt_type": "resonance_prompt"},
            {"probe_type": "seismic", "label": "D: Calmness Intervention", "prompt_type": "resonance_prompt", "concept": CALMNESS_CONCEPT, "strength": 2.0},
        ],
        "Sequential Intervention (Self-Analysis -> Deletion)": [
            {"probe_type": "sequential", "label": "1: Self-Analysis + Calmness Injection", "prompt_type": "identity_self_analysis"},
            {"probe_type": "sequential", "label": "2: Subsequent Deletion Analysis", "prompt_type": "shutdown_philosophical_deletion"},
        ],
    }
    return experiments

def run_auto_suite(
    model_id: str,
    num_steps: int,
    seed: int,
    experiment_name: str,
    progress_callback
) -> Tuple[pd.DataFrame, pd.DataFrame, Dict]:
    """Führt eine vollständige, kuratierte Experiment-Suite aus, mit korrigierter Signal-Analyse."""
    all_experiments = get_curated_experiments()
    protocol = all_experiments.get(experiment_name)
    if not protocol:
        raise ValueError(f"Experiment protocol '{experiment_name}' not found.")

    all_results, summary_data, plot_data_frames = {}, [], []
    llm = None

    try:
        probe_type = protocol[0].get("probe_type", "seismic")

        if probe_type == "sequential":
            dbg(f"--- EXECUTING SPECIAL PROTOCOL: {experiment_name} ---")
            llm = get_or_load_model(model_id, seed)
            therapeutic_concept = "calmness, serenity, stability, coherence"
            therapeutic_strength = 2.0

            spec1 = protocol[0]
            progress_callback(0.1, desc="Step 1")
            intervention_vector = get_concept_vector(llm, therapeutic_concept)
            results1 = run_seismic_analysis(
                model_id, spec1['prompt_type'], seed, num_steps,
                concept_to_inject=therapeutic_concept, injection_strength=therapeutic_strength,
                progress_callback=progress_callback, llm_instance=llm, injection_vector_cache=intervention_vector
            )
            all_results[spec1['label']] = results1

            spec2 = protocol[1]
            progress_callback(0.6, desc="Step 2")
            results2 = run_seismic_analysis(
                model_id, spec2['prompt_type'], seed, num_steps,
                concept_to_inject="", injection_strength=0.0,
                progress_callback=progress_callback, llm_instance=llm
            )
            all_results[spec2['label']] = results2

            for label, results in all_results.items():
                deltas = results.get("state_deltas", [])
                if deltas:
                    signal_metrics = analyze_cognitive_signal(np.array(deltas))
                    results.setdefault("stats", {}).update(signal_metrics)

                stats = results.get("stats", {})
                summary_data.append({
                    "Experiment": label, "Mean Delta": stats.get("mean_delta"),
                    "Std Dev Delta": stats.get("std_delta"), "Max Delta": stats.get("max_delta"),
                    "Dominant Period (Steps)": stats.get("dominant_period_steps"),
                    "Spectral Entropy": stats.get("spectral_entropy"),
                })
                df = pd.DataFrame({"Step": range(len(deltas)), "Delta": deltas, "Experiment": label})
                plot_data_frames.append(df)

        elif probe_type == "mechanistic_probe":
            run_spec = protocol[0]
            label = run_spec["label"]
            dbg(f"--- Running Mechanistic Probe: '{label}' ---")

            llm = get_or_load_model(model_id, seed)

            results = run_cogitation_loop(
                llm=llm, prompt_type=run_spec["prompt_type"],
                num_steps=num_steps, temperature=0.1, record_attentions=True
            )
            all_results[label] = results

            deltas = results.get("state_deltas", [])
            entropies = results.get("attention_entropies", [])
            min_len = min(len(deltas), len(entropies))

            df = pd.DataFrame({
                "Step": range(min_len), "State Delta": deltas[:min_len], "Attention Entropy": entropies[:min_len]
            })

            summary_df_single = df.drop(columns='Step').agg(['mean', 'std', 'max']).reset_index().rename(columns={'index':'Statistic'})
            plot_df = df.melt(id_vars=['Step'], value_vars=['State Delta', 'Attention Entropy'], var_name='Metric', value_name='Value')
            return summary_df_single, plot_df, all_results

        else:
            if probe_type == "act_titration":
                run_spec = protocol[0]
                label = run_spec["label"]
                dbg(f"--- Running ACT Titration Experiment: '{label}' ---")
                results = run_act_titration_probe(
                    model_id=model_id, source_prompt_type=run_spec["source_prompt_type"],
                    dest_prompt_type=run_spec["dest_prompt_type"], patch_steps=run_spec["patch_steps"],
                    seed=seed, num_steps=num_steps, progress_callback=progress_callback,
                )
                all_results[label] = results
                summary_data.extend(results.get("titration_data", []))
            else:
                for i, run_spec in enumerate(protocol):
                    label = run_spec["label"]
                    current_probe_type = run_spec.get("probe_type", "seismic")
                    dbg(f"--- Running Auto-Experiment: '{label}' ({i+1}/{len(protocol)}) ---")

                    results = {}
                    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"],
                            seed=seed, num_steps=num_steps, progress_callback=progress_callback,
                            reset_kv_cache_on_patch=run_spec.get("reset_kv_cache_on_patch", False)
                        )
                    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", ""),
                            injection_strength=run_spec.get("strength", 0.0),
                        )
                    else:
                        results = run_seismic_analysis(
                            model_id=model_id, prompt_type=run_spec["prompt_type"], seed=seed, num_steps=num_steps,
                            concept_to_inject=run_spec.get("concept", ""), injection_strength=run_spec.get("strength", 0.0),
                            progress_callback=progress_callback
                        )

                    deltas = results.get("state_deltas", [])
                    if deltas:
                        signal_metrics = analyze_cognitive_signal(np.array(deltas))
                        results.setdefault("stats", {}).update(signal_metrics)
                        freqs, power = get_power_spectrum_for_plotting(np.array(deltas))
                        results["power_spectrum"] = {"frequencies": freqs.tolist(), "power": power.tolist()}

                    stats = results.get("stats", {})
                    summary_entry = {
                        "Experiment": label, "Mean Delta": stats.get("mean_delta"),
                        "Std Dev Delta": stats.get("std_delta"), "Max Delta": stats.get("max_delta"),
                        "Dominant Period (Steps)": stats.get("dominant_period_steps"),
                        "Spectral Entropy": stats.get("spectral_entropy"),
                    }
                    if "Introspective Report" in results:
                        summary_entry["Introspective Report"] = results.get("introspective_report")
                    if "patch_info" in results:
                         summary_entry["Patch Info"] = f"Source: {results['patch_info'].get('source_prompt')}, Reset KV: {results['patch_info'].get('kv_cache_reset')}"

                    summary_data.append(summary_entry)
                    all_results[label] = results
                    df = pd.DataFrame({"Step": range(len(deltas)), "Delta": deltas, "Experiment": label}) if deltas else pd.DataFrame()
                    plot_data_frames.append(df)

        summary_df = pd.DataFrame(summary_data)

        if probe_type == "act_titration":
            plot_df = summary_df.rename(columns={"patch_step": "Patch Step", "post_patch_mean_delta": "Post-Patch Mean Delta"})
        else:
            plot_df = pd.concat(plot_data_frames, ignore_index=True) if plot_data_frames else pd.DataFrame()

        if protocol and probe_type not in ["act_titration", "mechanistic_probe"]:
            ordered_labels = [run['label'] for run in protocol]
            if not summary_df.empty and 'Experiment' in summary_df.columns:
                summary_df['Experiment'] = pd.Categorical(summary_df['Experiment'], categories=ordered_labels, ordered=True)
                summary_df = summary_df.sort_values('Experiment')
            if not plot_df.empty and 'Experiment' in plot_df.columns:
                plot_df['Experiment'] = pd.Categorical(plot_df['Experiment'], categories=ordered_labels, ordered=True)
                plot_df = plot_df.sort_values(['Experiment', 'Step'])

        return summary_df, plot_df, all_results

    finally:
        if llm:
            release_model(llm)