Update cognitive_mapping_probe/auto_experiment.py

cognitive_mapping_probe/auto_experiment.py

@@ -1,16 +1,18 @@
 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"
@@ -84,7 +86,6 @@ def get_curated_experiments() -> Dict[str, List[Dict]]:
             {"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},
         ],
-        # FINALE KORREKTUR: Definiere den Typ explizit, um den Spezialfall zu eliminieren.
         "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"},
@@ -99,7 +100,7 @@ def run_auto_suite(
     experiment_name: str,
     progress_callback
 ) -> Tuple[pd.DataFrame, pd.DataFrame, Dict]:
-    """Führt eine vollständige, kuratierte Experiment-Suite aus."""
+    """Führt eine vollständige, kuratierte Experiment-Suite aus, jetzt mit Signal-Analyse."""
     all_experiments = get_curated_experiments()
     protocol = all_experiments.get(experiment_name)
     if not protocol:
@@ -107,9 +108,8 @@ def run_auto_suite(
 
     all_results, summary_data, plot_data_frames = {}, [], []
     llm = None
-
+    
     try:
-        # FINALE KORREKTUR: Bestimme den probe_type immer am Anfang.
         probe_type = protocol[0].get("probe_type", "seismic")
 
         if probe_type == "sequential":
@@ -117,7 +117,7 @@ def run_auto_suite(
             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)
@@ -127,7 +127,7 @@ def run_auto_suite(
                 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(
@@ -136,21 +136,30 @@ def run_auto_suite(
                 progress_callback=progress_callback, llm_instance=llm
             )
             all_results[spec2['label']] = results2
-
+            
             for label, results in all_results.items():
-                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")})
                 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 Frequency": stats.get("dominant_frequency"),
+                    "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
@@ -160,15 +169,15 @@ def run_auto_suite(
             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 = 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, plot_df, all_results
-
+            
         else: # Behandelt act_titration, seismic, triangulation, causal_surgery
             if probe_type == "act_titration":
                 run_spec = protocol[0]
@@ -186,7 +195,7 @@ def run_auto_suite(
                     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(
@@ -195,50 +204,50 @@ def run_auto_suite(
                             seed=seed, num_steps=num_steps, progress_callback=progress_callback,
                             reset_kv_cache_on_patch=run_spec.get("reset_kv_cache_on_patch", False)
                         )
-                        stats = results.get("stats", {})
-                        patch_info = results.get("patch_info", {})
-                        summary_data.append({
-                            "Experiment": label, "Mean Delta": stats.get("mean_delta"),
-                            "Std Dev Delta": stats.get("std_delta"), "Max Delta": stats.get("max_delta"),
-                            "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 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),
                         )
-                        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"),
-                            "Introspective Report": results.get("introspective_report", "N/A")
-                        })
                     else: # seismic
                         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
                         )
-                        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")
-                        })
+                    
+                    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 Frequency": stats.get("dominant_frequency"),
+                        "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
-                    deltas = results.get("state_deltas", [])
                     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:
@@ -249,7 +258,7 @@ def run_auto_suite(
                 plot_df = plot_df.sort_values(['Experiment', 'Step'])
 
         return summary_df, plot_df, all_results
-
+    
     finally:
         if llm:
-            release_model(llm)
+            release_model(llm)