| 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 | |