|
|
import torch |
|
|
from typing import Optional, List |
|
|
from tqdm import tqdm |
|
|
|
|
|
from .llm_iface import LLM |
|
|
from .prompts import RESONANCE_PROMPTS |
|
|
from .utils import dbg |
|
|
|
|
|
@torch.no_grad() |
|
|
def run_silent_cogitation_seismic( |
|
|
llm: LLM, |
|
|
prompt_type: str, |
|
|
num_steps: int, |
|
|
temperature: float, |
|
|
) -> List[float]: |
|
|
""" |
|
|
NEUE VERSION: Führt den 'silent thought' Prozess aus und gibt die gesamte |
|
|
Zeitreihe der `state_delta`-Werte zurück, anstatt auf Konvergenz zu prüfen. |
|
|
""" |
|
|
prompt = RESONANCE_PROMPTS[prompt_type] |
|
|
inputs = llm.tokenizer(prompt, return_tensors="pt").to(llm.model.device) |
|
|
|
|
|
outputs = llm.model(**inputs, output_hidden_states=True, use_cache=True) |
|
|
|
|
|
hidden_state_2d = outputs.hidden_states[-1][:, -1, :] |
|
|
kv_cache = outputs.past_key_values |
|
|
|
|
|
previous_hidden_state = hidden_state_2d.clone() |
|
|
state_deltas = [] |
|
|
|
|
|
for i in tqdm(range(num_steps), desc=f"Recording Dynamics (Temp {temperature:.2f})", leave=False, bar_format="{l_bar}{bar:10}{r_bar}"): |
|
|
next_token_logits = llm.model.lm_head(hidden_state_2d) |
|
|
|
|
|
|
|
|
probabilities = torch.nn.functional.softmax(next_token_logits / temperature, dim=-1) |
|
|
next_token_id = torch.multinomial(probabilities, num_samples=1) |
|
|
|
|
|
outputs = llm.model( |
|
|
input_ids=next_token_id, |
|
|
past_key_values=kv_cache, |
|
|
output_hidden_states=True, |
|
|
use_cache=True, |
|
|
) |
|
|
|
|
|
hidden_state_2d = outputs.hidden_states[-1][:, -1, :] |
|
|
kv_cache = outputs.past_key_values |
|
|
|
|
|
delta = torch.norm(hidden_state_2d - previous_hidden_state).item() |
|
|
state_deltas.append(delta) |
|
|
|
|
|
previous_hidden_state = hidden_state_2d.clone() |
|
|
|
|
|
dbg(f"Seismic recording finished after {num_steps} steps.") |
|
|
|
|
|
return state_deltas |
|
|
|
