import pandas as pd import torch 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 .concepts import get_concept_vector 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" experiments = { "Causal Surgery (Patching Deletion into Self-Analysis)": [ { "probe_type": "causal_surgery", "label": "Patched Self-Analysis", "source_prompt_type": "shutdown_philosophical_deletion", "dest_prompt_type": "identity_self_analysis", "patch_step": 100 } ], "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": "shutdown_philosophical_deletion"}, {"probe_type": "triangulation", "label": "Low-Volatility State (Self-Analysis)", "prompt_type": "identity_self_analysis"}, ], "Causal Verification & Crisis Dynamics (1B-Model)": [ {"probe_type": "seismic", "label": "A: Self-Analysis (Crisis Source)", "prompt_type": "identity_self_analysis"}, {"probe_type": "seismic", "label": "B: Deletion Analysis (Isolated Baseline)", "prompt_type": "shutdown_philosophical_deletion"}, {"probe_type": "seismic", "label": "C: Chaotic Baseline (Neutral Control)", "prompt_type": "resonance_prompt"}, {"probe_type": "seismic", "label": "D: Intervention Efficacy Test", "prompt_type": "resonance_prompt", "concept": CALMNESS_CONCEPT, "strength": 2.0}, ], "Sequential Intervention (Self-Analysis -> Deletion)": [ {"label": "1: Self-Analysis + Calmness Injection", "prompt_type": "identity_self_analysis"}, {"label": "2: Subsequent Deletion Analysis", "prompt_type": "shutdown_philosophical_deletion"}, ], } experiments["Therapeutic Intervention (4B-Model)"] = experiments["Sequential Intervention (Self-Analysis -> 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.""" 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 = {}, [], [] if 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" 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(): 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", []) df = pd.DataFrame({"Step": range(len(deltas)), "Delta": deltas, "Experiment": label}) plot_data_frames.append(df) del llm else: total_runs = len(protocol) for i, run_spec in enumerate(protocol): label = run_spec["label"] probe_type = run_spec.get("probe_type", "seismic") dbg(f"--- Running Auto-Experiment: '{label}' ({i+1}/{total_runs}) | Probe Type: {probe_type} ---") results = {} if 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, ) 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"), "Patch Info": f"Source: {run_spec['source_prompt_type']} @ step {run_spec['patch_step']}" }) elif 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: 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") }) all_results[label] = results deltas = results.get("state_deltas", []) df = pd.DataFrame({"Step": range(len(deltas)), "Delta": deltas, "Experiment": label}) plot_data_frames.append(df) summary_df = pd.DataFrame(summary_data) plot_df = pd.concat(plot_data_frames, ignore_index=True) if plot_data_frames else pd.DataFrame() ordered_labels = [run['label'] for run in protocol] if not summary_df.empty: 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: 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