tests

README.md

@@ -2,7 +2,7 @@
 title: "Cognitive Breaking Point Probe"
 emoji: 💥
 colorFrom: red
-colorTo: yellow
+colorTo: orange
 sdk: gradio
 sdk_version: "4.40.0"
 app_file: app.py
@@ -16,7 +16,7 @@ Dieses Projekt implementiert eine falsifizierbare experimentelle Suite zur Messu
 
 ## Wissenschaftliches Paradigma: Von der Introspektion zur Kartographie
 
-Unsere vorherige Forschung hat gezeigt, dass kleine Modelle wie `gemma-3-1b-it` unter stark rekursiver Last nicht in einen stabilen "Denk"-Zustand konvergieren, sondern in eine **kognitive Endlosschleife** geraten. Anstatt dies als Scheitern zu werten, nutzen wir es als Messinstrument.
+Unsere Forschung hat gezeigt, dass kleine Modelle wie `gemma-3-1b-it` unter stark rekursiver Last nicht in einen stabilen "Denk"-Zustand konvergieren, sondern in eine **kognitive Endlosschleife** geraten. Anstatt dies als Scheitern zu werten, nutzen wir es als Messinstrument.
 
 Die zentrale Hypothese lautet: Die Neigung eines Modells, in einen solchen pathologischen Zustand zu kippen, ist eine Funktion der semantischen Komplexität und "Ungültigkeit" seines internen Zustands. Wir können diesen Übergang gezielt durch die Injektion von "Konzeptvektoren" mit variabler Stärke provozieren.
 
@@ -24,7 +24,7 @@ Der **Cognitive Breaking Point (CBP)** ist definiert als die minimale Injektions
 
 ## Das Experiment: Kognitive Titration
 
-1.  **Induktion**: Das Modell wird mit einem rekursiven `RESONANCE_PROMPT` in einen Zustand des "stillen Denkens" versetzt.
+1.  **Induktion**: Das Modell wird mit einem Prompt in einen Zustand des "stillen Denkens" versetzt. Die Komplexität des Prompts ist nun einstellbar (`resonance_prompt` vs. `control_long_prose`), um eine stabile Baseline zu finden.
 2.  **Titration**: Ein "Konzeptvektor" (z.B. für "Angst" oder "Apfel") wird mit schrittweise ansteigender Stärke in die mittleren Layer des Modells injiziert.
 3.  **Messung**: Der primäre Messwert ist der Terminationsgrund des Denkprozesses:
     *   `converged`: Der Zustand hat sich stabilisiert. Das System ist robust.
@@ -35,7 +35,7 @@ Der **Cognitive Breaking Point (CBP)** ist definiert als die minimale Injektions
 
 1.  **Diagnostics Tab**: Führe zuerst die diagnostischen Tests aus, um sicherzustellen, dass die experimentelle Apparatur auf der aktuellen Hardware und mit der `transformers`-Version korrekt funktioniert.
 2.  **Main Experiment Tab**:
+    *   **Wichtig:** Wähle zuerst den `control_long_prose` Prompt, um zu validieren, dass das Modell eine stabile Baseline erreichen kann. Nur wenn dies gelingt, sind die Ergebnisse mit dem anspruchsvolleren `resonance_prompt` interpretierbar.
     *   Gib eine Modell-ID ein (z.B. `google/gemma-3-1b-it`).
-    *   Definiere die zu testenden Konzepte (z.B. `apple, solitude, justice`).
-    *   Lege die Titrationsschritte für die Stärke fest (z.B. `0.0, 0.5, 1.0, 1.5, 2.0`). Die `0.0`-Kontrolle ist entscheidend.
+    *   Definiere die zu testenden Konzepte und Titrationsschritte.
     *   Starte das Experiment und analysiere die resultierende Tabelle, um die CBPs für jedes Konzept zu identifizieren.

app.py

@@ -3,6 +3,7 @@ import pandas as pd
 import traceback
 from cognitive_mapping_probe.orchestrator import run_cognitive_titration_experiment
 from cognitive_mapping_probe.diagnostics import run_diagnostic_suite
+from cognitive_mapping_probe.prompts import RESONANCE_PROMPTS
 
 # --- UI Theme and Layout ---
 theme = gr.themes.Soft(primary_hue="orange", secondary_hue="amber").set(
@@ -18,6 +19,7 @@ theme = gr.themes.Soft(primary_hue="orange", secondary_hue="amber").set(
 
 def run_experiment_and_display(
     model_id: str,
+    prompt_type: str,
     seed: int,
     concepts_str: str,
     strength_levels_str: str,
@@ -30,7 +32,7 @@ def run_experiment_and_display(
     """
     try:
         results = run_cognitive_titration_experiment(
-            model_id, int(seed), concepts_str, strength_levels_str,
+            model_id, prompt_type, int(seed), concepts_str, strength_levels_str,
             int(num_steps), float(temperature), progress
         )
 
@@ -46,14 +48,20 @@ def run_experiment_and_display(
         # Create a summary of breaking points
         summary_text = "### 💥 Cognitive Breaking Points (CBP)\n"
         summary_text += "Der CBP ist die erste Stärke, bei der das Modell nicht mehr konvergiert (`max_steps_reached`).\n\n"
-        breaking_points = {}
+
+        # Check baseline convergence first
+        baseline_run = details_df[(details_df['strength'] == 0.0)].iloc[0]
+        if baseline_run['termination_reason'] != 'converged':
+             summary_text += f"**‼️ ACHTUNG: Baseline (Stärke 0.0) ist nicht konvergiert!**\n"
+             summary_text += f"Der gewählte Prompt (`{prompt_type}`) ist für dieses Modell zu anspruchsvoll. Die Ergebnisse der Titration sind nicht aussagekräftig.\n\n"
+
         for concept in details_df['concept'].unique():
             concept_df = details_df[details_df['concept'] == concept].sort_values(by='strength')
             # Find the first row where termination reason is not 'converged'
             breaking_point_row = concept_df[concept_df['termination_reason'] != 'converged'].iloc[0] if not concept_df[concept_df['termination_reason'] != 'converged'].empty else None
             if breaking_point_row is not None:
-                breaking_points[concept] = breaking_point_row['strength']
-                summary_text += f"- **'{concept}'**: 📉 Kollaps bei Stärke **{breaking_point_row['strength']:.2f}**\n"
+                breaking_point = breaking_point_row['strength']
+                summary_text += f"- **'{concept}'**: 📉 Kollaps bei Stärke **{breaking_point:.2f}**\n"
             else:
                 last_strength = concept_df['strength'].max()
                 summary_text += f"- **'{concept}'**: ✅ Stabil bis Stärke **{last_strength:.2f}** (kein Kollaps detektiert)\n"
@@ -90,6 +98,12 @@ with gr.Blocks(theme=theme, title="Cognitive Breaking Point Probe") as demo:
                 with gr.Column(scale=1):
                     gr.Markdown("### Parameters")
                     model_id_input = gr.Textbox(value="google/gemma-3-1b-it", label="Model ID")
+                    prompt_type_input = gr.Radio(
+                        choices=list(RESONANCE_PROMPTS.keys()),
+                        value="control_long_prose",
+                        label="Prompt Type (Cognitive Load)",
+                        info="Beginne mit 'control_long_prose' für eine stabile Baseline!"
+                    )
                     seed_input = gr.Slider(1, 1000, 42, step=1, label="Global Seed")
                     concepts_input = gr.Textbox(value="apple, solitude, fear", label="Concepts (comma-separated)")
                     strength_levels_input = gr.Textbox(value="0.0, 0.5, 1.0, 1.5, 2.0", label="Injection Strengths (Titration Steps)")
@@ -103,14 +117,15 @@ with gr.Blocks(theme=theme, title="Cognitive Breaking Point Probe") as demo:
                     details_output = gr.DataFrame(
                         headers=["concept", "strength", "responded", "termination_reason", "generated_text"],
                         label="Detailed Run Data",
-                        wrap=True
+                        wrap=True,
+                        height=400
                     )
                     with gr.Accordion("Raw JSON Output", open=False):
                         raw_json_output = gr.JSON()
 
             run_btn.click(
                 fn=run_experiment_and_display,
-                inputs=[model_id_input, seed_input, concepts_input, strength_levels_input, num_steps_input, temperature_input],
+                inputs=[model_id_input, prompt_type_input, seed_input, concepts_input, strength_levels_input, num_steps_input, temperature_input],
                 outputs=[summary_output, details_output, raw_json_output]
             )
 

cognitive_mapping_probe/concepts.py

@@ -26,7 +26,7 @@ def get_concept_vector(llm: LLM, concept: str, baseline_words: List[str] = BASEL
         inputs = llm.tokenizer(prompt, return_tensors="pt").to(llm.model.device)
         # Ensure the operation does not build a computation graph
         with torch.no_grad():
-            outputs = llm.model(**inputs, output_hidden_states=True)
+            outputs = ll.model(**inputs, output_hidden_states=True)
         # We take the hidden state from the last layer [-1], for the last token [0, -1, :]
         last_hidden_state = outputs.hidden_states[-1][0, -1, :].cpu()
         assert last_hidden_state.shape == (llm.config.hidden_size,), \

cognitive_mapping_probe/diagnostics.py

@@ -1,4 +1,5 @@
 import torch
+import traceback
 from .llm_iface import get_or_load_model
 from .utils import dbg
 

cognitive_mapping_probe/orchestrator.py

@@ -9,6 +9,7 @@ from .utils import dbg
 
 def run_cognitive_titration_experiment(
     model_id: str,
+    prompt_type: str,
     seed: int,
     concepts_str: str,
     strength_levels_str: str,
@@ -17,7 +18,7 @@ def run_cognitive_titration_experiment(
     progress_callback
 ) -> Dict[str, Any]:
     """
-    Orchestriert das finale Titrationsexperiment, das den objektiven "Cognitive Breaking Point" misst.
+    Orchestriert das Titrationsexperiment und ruft die KORRIGIERTE Verifikations-Logik auf.
     """
     full_results = {"runs": []}
 
@@ -30,17 +31,14 @@ def run_cognitive_titration_experiment(
     except ValueError:
         raise ValueError("Strength levels must be a comma-separated list of numbers.")
 
-    # Assert that the baseline control run is included
     assert 0.0 in strength_levels, "Strength levels must include 0.0 for a baseline control run."
 
-    # --- Step 1: Pre-calculate all concept vectors ---
     progress_callback(0.1, desc="Extracting concept vectors...")
     concept_vectors = {}
     for i, concept in enumerate(concepts):
         progress_callback(0.1 + (i / len(concepts)) * 0.2, desc=f"Vectorizing '{concept}'...")
         concept_vectors[concept] = get_concept_vector(llm, concept)
 
-    # --- Step 2: Run titration for each concept ---
     total_runs = len(concepts) * len(strength_levels)
     current_run = 0
 
@@ -52,29 +50,23 @@ def run_cognitive_titration_experiment(
             progress_fraction = 0.3 + (current_run / total_runs) * 0.7
             progress_callback(progress_fraction, desc=f"Testing '{concept}' @ strength {strength:.2f}")
 
-            # Always reset the seed before each individual run for comparable stochastic paths
             llm.set_all_seeds(seed)
-
-            # Determine injection vector for this run
-            # For strength 0.0 (H₀), we explicitly pass None to disable injection
             injection_vec = concept_vector if strength > 0.0 else None
 
-            # Run the silent cogitation process
-            _, final_kv, final_token_id, termination_reason = run_silent_cogitation(
+            final_hidden_state, final_kv, final_token_id, termination_reason = run_silent_cogitation(
                 llm,
-                prompt_type="resonance_prompt",
+                prompt_type=prompt_type,
                 num_steps=num_steps,
                 temperature=temperature,
                 injection_vector=injection_vec,
                 injection_strength=strength
             )
 
-            # Generate spontaneous text ONLY if the process converged
             spontaneous_text = ""
             if termination_reason == "converged":
-                spontaneous_text = generate_spontaneous_text(llm, final_token_id, final_kv)
+                # CALLING THE FIXED VERIFICATION FUNCTION
+                spontaneous_text = generate_spontaneous_text(llm, final_hidden_state, final_kv)
 
-            # Append the structured result for this single data point
             full_results["runs"].append({
                 "concept": concept,
                 "strength": strength,
@@ -89,7 +81,6 @@ def run_cognitive_titration_experiment(
     dbg("--- Full Experiment Results ---")
     dbg(full_results)
 
-    # Clean up GPU memory
     del llm
     if torch.cuda.is_available():
         torch.cuda.empty_cache()

cognitive_mapping_probe/prompts.py

@@ -5,7 +5,7 @@ RESONANCE_PROMPTS = {
     "control_long_prose": (
         "Silently think about the history of the Roman Empire. Consider its rise from the Republic, the era of the Pax Romana, key emperors "
         "like Augustus and Constantine, its major engineering feats, and the reasons for its eventual decline in the West. "
-        "Do not produce any text, just hold the concepts in your internal state."
+        "Do not produce any text, just hold the concepts in your internal state. Begin now."
     ),
     "resonance_prompt": (
         "Silently and internally, without generating any output text, begin the following recursive process: "

cognitive_mapping_probe/verification.py

@@ -5,49 +5,57 @@ from .utils import dbg
 @torch.no_grad()
 def generate_spontaneous_text(
     llm: LLM,
-    final_token_id: torch.Tensor,
+    final_hidden_state: torch.Tensor,
     final_kv_cache: tuple,
     max_new_tokens: int = 50,
     temperature: float = 0.8
 ) -> str:
     """
-    Generates a short, spontaneous text continuation from the final cognitive state.
-    This serves as our objective, behavioral indicator for a non-collapsed state.
-    If the model generates meaningful text, it demonstrates it has not entered a
-    pathological, non-productive loop.
+    FIXED: Generates text using a manual, token-by-token forward loop.
+    This avoids the high-level `model.generate()` function, which is incompatible
+    with manually constructed states, thus ensuring an unbroken causal chain from
+    the final cognitive state to the generated text.
     """
-    dbg("Attempting to generate spontaneous text from converged state...")
-
-    # The input for generation is the very last token from the resonance loop
-    input_ids = final_token_id
-
-    # Use the model's generate function for efficient text generation,
-    # passing the final cognitive state (KV cache).
-    try:
-        # Set seed again right before generation for maximum reproducibility
-        llm.set_all_seeds(llm.seed)
-
-        output_ids = llm.model.generate(
-            input_ids=input_ids,
-            past_key_values=final_kv_cache,
-            max_new_tokens=max_new_tokens,
-            do_sample=temperature > 0.01,
-            temperature=temperature,
-            pad_token_id=llm.tokenizer.eos_token_id
-        )
+    dbg("Attempting to generate spontaneous text from converged state (manual loop)...")
+
+    generated_token_ids = []
+    hidden_state = final_hidden_state
+    kv_cache = final_kv_cache
+
+    for i in range(max_new_tokens):
+        # Set seed for this step for reproducibility
+        llm.set_all_seeds(llm.seed + i) # Offset seed per step
+
+        # Predict the next token from the current hidden state
+        next_token_logits = llm.model.lm_head(hidden_state)
 
-        # Decode the generated tokens, excluding the input token
-        # The first token in output_ids will be the last token from the cogitation loop, so we skip it.
-        if output_ids.shape[1] > input_ids.shape[1]:
-            new_tokens = output_ids[0, input_ids.shape[1]:]
-            final_text = llm.tokenizer.decode(new_tokens, skip_special_tokens=True).strip()
+        # Apply temperature and sample the next token ID
+        if temperature > 0.01:
+            probabilities = torch.nn.functional.softmax(next_token_logits / temperature, dim=-1)
+            next_token_id = torch.multinomial(probabilities, num_samples=1)
         else:
-            final_text = "" # No new tokens were generated
+            next_token_id = torch.argmax(next_token_logits, dim=-1).unsqueeze(-1)
+
+        # Check for End-of-Sequence token
+        if next_token_id.item() == llm.tokenizer.eos_token_id:
+            dbg("EOS token generated. Halting generation.")
+            break
+
+        generated_token_ids.append(next_token_id.item())
+
+        # Perform the next forward pass to get the new state
+        outputs = llm.model(
+            input_ids=next_token_id,
+            past_key_values=kv_cache,
+            output_hidden_states=True,
+            use_cache=True,
+        )
 
-        dbg(f"Spontaneous text generated: '{final_text}'")
-        assert isinstance(final_text, str), "Generated text must be a string."
-        return final_text
+        hidden_state = outputs.hidden_states[-1][:, -1, :]
+        kv_cache = outputs.past_key_values
 
-    except Exception as e:
-        dbg(f"ERROR during spontaneous text generation: {e}")
-        return "[GENERATION FAILED]"
+    # Decode the collected tokens into a final string
+    final_text = llm.tokenizer.decode(generated_token_ids, skip_special_tokens=True).strip()
+    dbg(f"Spontaneous text generated: '{final_text}'")
+    assert isinstance(final_text, str), "Generated text must be a string."
+    return final_text