""" chat_tab.py — Gradio Chat Tab with ChatInterface & Subjective Mode ================================================================ Integrated chat interface with session management and PX steering. """ import gradio as gr import torch import asyncio import os import json import statistics from typing import Optional, List, Dict, Any from threading import Thread from transformers import TextIteratorStreamer from config import MODEL_REGISTRY from model_manager import ModelManager from sessions import save_session, load_session, get_new_session_id, list_sessions from telemetry import telemetry from gradio_tabs.multimodal_input import ( normalize_multimodal_message, is_empty_message, extract_text_blocks, _normalize_history_for_chatbot, ) # ── Session Handlers ── def _stringify_content(content): """Ensure content is a string for text-only templates.""" if isinstance(content, str): return content if isinstance(content, list): parts = [] for item in content: if isinstance(item, str): parts.append(item) elif isinstance(item, dict): if item.get("type") == "text": parts.append(item.get("text", "")) elif "text" in item and "files" not in item: # Handle some Gradio formats parts.append(item["text"]) return "\n".join(parts) if isinstance(content, dict): return content.get("text", str(content)) return str(content) def _clean_history(history): """Filter empty messages and merge consecutive same-role messages.""" result = [] for msg in (history or []): if not isinstance(msg, dict): continue role = msg.get("role", "") content = msg.get("content", "") if isinstance(content, str): if not content.strip(): continue elif not content: continue if result and result[-1]["role"] == role: prev_content = result[-1]["content"] if isinstance(prev_content, str) and isinstance(content, str): result[-1]["content"] += "\n" + content elif isinstance(prev_content, list) and isinstance(content, list): result[-1]["content"].extend(content) elif isinstance(prev_content, list) and isinstance(content, str): result[-1]["content"].append({"type": "text", "text": content}) elif isinstance(prev_content, str) and isinstance(content, list): result[-1]["content"] = [{"type": "text", "text": prev_content}] + content else: result.append({"role": role, "content": content}) return result def on_load(session_id): """Called when the page loads.""" if session_id is None or session_id == "": session_id = get_new_session_id() data = load_session(session_id) # Normalize history for Gradio 6.15.2 Chatbot (OpenAI blocks + untyped dicts # would otherwise crash _postprocess_content; see multimodal_input.py). history = _normalize_history_for_chatbot(data.get("history", [])) return session_id, history, gr.update(choices=list_sessions()), session_id def handle_new_session(): new_id = get_new_session_id() return new_id, [], gr.update(choices=list_sessions()), new_id def handle_load_saved(session_id): if not session_id: return gr.skip(), [], gr.skip(), gr.skip() data = load_session(session_id) return session_id, _normalize_history_for_chatbot(data.get("history", [])), gr.skip(), session_id def handle_export(session_id, history): if not history: return gr.update(visible=False) path = f"exported_session_{session_id}.json" with open(path, "w") as f: json.dump({"session_id": session_id, "history": history}, f, indent=2) return gr.update(value=path, visible=True) def handle_import(file_obj): if file_obj is None: return gr.skip(), [], gr.skip(), gr.skip() try: with open(file_obj.name, "r") as f: data = json.load(f) new_id = data.get("session_id", get_new_session_id()) history = data.get("history", []) save_session(new_id, history) return new_id, history, gr.update(choices=list_sessions(), value=new_id), new_id except Exception as e: print(f"Import error: {e}") return gr.skip(), [], gr.skip(), gr.skip() def handle_refresh(): return gr.update(choices=list_sessions()) # Plan ui-styling 2026-07-08: Undo-Button für "letzte Nachricht rückgängig" # (User-Feedback: vorher poppte er das ganze (user, assistant)-Paar — User # wollte aber nur 1 Element rückgängig machen, entweder die letzte User- oder # die letzte Agent-Antwort). Nutzt chat_actions.undo_last_entry statt # undo_last_turn. Persistiert die gekürzte History sofort via save_session. def handle_undo(session_id, history): """Click-handler für den Undo-Button. Returns: (updated_history, status_text) - updated_history: gekürzte History (oder skip wenn nichts zu undo) - status_text: "✓ Undone" oder "⚠ Nothing to undo" """ from gradio_tabs.chat_actions import undo_last_entry, can_undo_entry if not can_undo_entry(history): return gr.skip(), "⚠ Nothing to undo" new_history = undo_last_entry(history) if session_id: save_session(session_id, new_history) return new_history, f"✓ Undone (history: {len(new_history)} msgs)" def chat_fn(message, history, model_id, px_preset, temp, tp, mt, rp, gamma, relay_sign, relay_alpha, relay_layer, system_profile, system_prompt_text, session_id, manager: ModelManager): """Core chat logic with history management and model generation. Plan ui-styling 2026-07-06: zwei neue Parameter (system_profile, system_prompt_text) — kommen aus dem Einstellungen-Tab. Vor dem chat_template-apply wird inject_into_messages() aufgerufen, das die System-Message an Index 0 setzt (und alle existing system-Einträge strippt). Bei neutral+leerer Edit: no-op (Original-Liste). """ print(f"DEBUG: history received from Gradio (UI state): {len(history) if history else 0} messages") # verstärkbar Relay-Parameter nur beim RELAY-Preset durchreichen (sonst None # → kein Surprise-Relay auf BASELINE/LEAN/ACTIVE_MANIFOLD; diese verhalten # sich exakt wie vorher). Bei RELAY steuert die UI (Radio/Slider). if px_preset == "ACTIVE_MANIFOLD_RELAY": _rsign, _ralpha, _rlayer = relay_sign, relay_alpha, relay_layer else: _rsign = _ralpha = _rlayer = None # 1. Update config loop = asyncio.new_event_loop() try: model_entry = loop.run_until_complete( manager.get_model( model_id, px_subjective=(px_preset != "BASELINE"), px_gamma=gamma, px_config_preset=px_preset, px_relay_sign=_rsign, px_relay_alpha=_ralpha, px_relay_layer=_rlayer, ) ) finally: loop.close() model = model_entry["model"] tokenizer = model_entry["tokenizer"] # 2. Build history (cleaned) # If history is empty (e.g. after loading a session or if save_history=False), # load it from the session storage to ensure continuity. if (not history or len(history) == 0) and session_id: data = load_session(session_id) history = data.get("history", []) cleaned_history = _clean_history(history) print(f"DEBUG: Initial cleaned_history length: {len(cleaned_history)}") # Plan ui-styling 2026-07-06: System-Prompt injizieren (Frame-Orientierer). # inject_into_messages setzt System-Message an Index 0 und strippt alle # existing system-Einträge. Bei neutral+leerem Edit: no-op (Pin T3). from gradio_tabs.system_prompt import inject_into_messages cleaned_history = inject_into_messages( cleaned_history, system_profile, system_prompt_text, ) # Normalize the Gradio MultimodalTextbox value into chat_fn's expected # message content (plain str, or content-list with text + image blocks). # All shape-handling lives in gradio_tabs.multimodal_input — kept pure # for testability (tests/test_multimodal_input.py). actual_message = normalize_multimodal_message(message) messages = cleaned_history + [{"role": "user", "content": actual_message}] print(f"DEBUG: Combined messages length: {len(messages)}") # SR-61b: Explicitly clear cache to prevent OOM on 12GB cards import torch if torch.cuda.is_available(): torch.cuda.empty_cache() # Phase 63: Proactive Auto-save (save user message before generation) save_session(session_id, messages, model_id=model_id) # 3. Generate with streaming # Robustness: Flatten to strings if no images are present to satisfy text-only templates. # Plan 2026-07-09: Image-Detection erweitert. Vorher: ``type == "image"`` # (altes _file_block-Format). Jetzt: ``type == "file"`` mit mime_type # image/* (Gradio file-Block) ODER legacy ``type == "image"`` (zur # Sicherheit — sollte nicht mehr auftreten, aber defensive Programmierung). has_images = any( isinstance(m.get("content"), list) and any( isinstance(c, dict) and ( c.get("type") == "image" # legacy pre-2026-07-09 or ( c.get("type") == "file" and str((c.get("file") or {}).get("mime_type", "")).startswith("image/") ) ) for c in m["content"] ) for m in messages ) if not has_images: processed_messages = [{"role": m["role"], "content": _stringify_content(m["content"])} for m in messages] else: processed_messages = messages input_text = tokenizer.apply_chat_template(processed_messages, tokenize=False, add_generation_prompt=True) inputs = tokenizer(input_text, return_tensors="pt").to(model.device) streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True) # Plan 2026-07-08: Output-Quality — Greedy Decoding für alle PX-Presets. # BASELINE = vanilla Modell, behält temperature/top_p/sampling wie User # es im Slider einstellt. ACTIVE_MANIFOLD/LEAN/RELAY = deterministisch # (do_sample=False, temperature=1e-10), damit RELAY-Output reproduzierbar # und nachvollziehbar wird. PX-Mechanik selbst (patch.py, relay_inject) # bleibt unangetastet — nur der Decoder wechselt. if px_preset == "BASELINE": _temperature = temp if temp > 0 else 1e-10 _do_sample = temp > 0 else: _temperature = 1e-10 _do_sample = False gen_kwargs = dict( **inputs, streamer=streamer, max_new_tokens=int(mt), temperature=_temperature, top_p=tp, repetition_penalty=rp, do_sample=_do_sample, ) # Inject EOS/EOT and PX-specific kwargs (SR-61b: StopOnEOT criteria) try: from generators import _px_gen_kwargs, _inject_eot_eos, strip_unsupported_model_kwargs gen_kwargs = _inject_eot_eos(gen_kwargs, tokenizer) gen_kwargs = _px_gen_kwargs(model, gen_kwargs) # Plan 7.2 / Live-Crash 2026-06-30: Llama-Pfade (z.B. MiniCPM5-1B) # lehnen `token_type_ids` ab, das der Tokenizer fälschlich setzt. # model.generate() validiert VOR dem ersten forward → muss hier # gestrippt werden, nicht erst im forward. gen_kwargs = strip_unsupported_model_kwargs(model, gen_kwargs) except ImportError: pass def generate_with_lock(): import torch if torch.cuda.is_available(): torch.cuda.empty_cache() manager.lock_model(model_id) try: model.generate(**gen_kwargs) finally: manager.unlock_model(model_id) thread = Thread(target=generate_with_lock) thread.start() partial_text = "" for new_text in streamer: partial_text += new_text if len(partial_text) % 20 == 0: print(f"DEBUG: Yielding partial_text length: {len(partial_text)}") yield partial_text # 4. Record Telemetry px_metrics = manager.get_px_metrics(model_id) telemetry.record( model_id=model_id, prompt_tokens=inputs["input_ids"].shape[1], completion_tokens=len(tokenizer.encode(partial_text)), px_metrics=px_metrics ) # 5. Save session on completion full_history = messages + [{"role": "assistant", "content": partial_text}] save_session(session_id, full_history, model_id=model_id) # ── Sidebar System-Prompt Helper (Plan 2026-07-08) ─────────────────── # Diese Helper werden als click/change-Handler in der Sidebar gemountet. # Sie sind pure-Funktionen (kein Side-Effect außer den Rückgabewerten) # und werden von tests direkt getestet. def on_preset_change_load_profile(preset: str) -> tuple: """px_preset.change-Handler (LEGACY, nicht mehr gemountet seit 2026-07-09). Lädt Profil-Name + Profil-Body passend zum PX-Preset. Wurde am 2026-07-09 entkoppelt: User will, dass px_preset NICHT mehr automatisch citmind/juexin lädt. Bleibt als pure-Funktion erhalten, damit ggf. Tests und Backward-Compat gewahrt sind. """ from gradio_tabs.system_prompt import ( preset_to_profile, load_profile_for_preset, ) profile_name = preset_to_profile(preset) profile_body = load_profile_for_preset(preset) return profile_name, profile_body def on_profile_change_load_body(profile_name: str) -> str: """system_profile.change-Handler: lädt Profil-Body in die Textarea. Plan 2026-07-09: User klickt in der Sidebar auf einen Profil-Eintrag (z.B. "juexin", "citmind", "neutral") → der gerenderte Body wird in das system_prompt_text-Feld geladen. Der User kann danach den Text frei editieren — was auch immer in der Textarea steht, geht in den nächsten Chat (build_system_message nutzt edit_text, wenn vorhanden, sonst den Profil-Body). Returns: profile_body — String für gr.update(value=...) auf der Textarea. """ from gradio_tabs.system_prompt import load_profile_body return load_profile_body(profile_name) def on_reset_prompt_click() -> str: """Reset-Button click-handler: leert die Edit-Textbox. User klickt "↺ Reset auf Profil-Default" → das Edit-Text-Feld wird geleert, das Profil selbst bleibt unverändert. Nach dem Reset wird beim nächsten chat_fn der Profil-Body (statt Edit-Override) verwendet (siehe build_system_message in system_prompt.py). """ return "" def build_chat_tab(manager: ModelManager): """Build and return the Chat tab components.""" # ── Client-side state ── session_id_state = gr.BrowserState(default_value=None, storage_key="px_session_id") model_choices = list(MODEL_REGISTRY.keys()) # Plan ui-styling Task #183: position="right" verschiebt die Sidebar # auf die rechte Seite (Standard in Gradio 6.x ist "left"; rechts ist # moderner für Chat-UIs). width=320 ist Default und explizit gesetzt. # Plan ui-styling Task #184: Status-Pill oben (live-Anwendung von # _styles.pill_style — accent=Hintergrund, pill=Radius, weißer Text). with gr.Sidebar(label="PX Controls", position="right", width=320): from gradio_tabs._styles import pill_style gr.HTML( f"
Engine: PX ✓
" ) gr.Markdown("### Model Selection") model_select = gr.Dropdown( choices=model_choices, value=model_choices[0], label="Current Model", ) px_preset = gr.Dropdown( choices=["BASELINE", "ACTIVE_MANIFOLD", "ACTIVE_MANIFOLD_LEAN", "ACTIVE_MANIFOLD_RELAY"], value="ACTIVE_MANIFOLD_RELAY", label="PX Mode Preset", ) with gr.Accordion("Parameters", open=False): temperature = gr.Slider(0.0, 2.0, value=0.7, step=0.05, label="Temperature") top_p = gr.Slider(0.0, 1.0, value=0.95, step=0.05, label="Top P") max_tokens = gr.Slider(64, 4096, value=1024, step=64, label="Max Tokens") rep_p = gr.Slider(1.0, 2.0, value=1.15, step=0.05, label="Repetition Penalty") px_gamma = gr.Slider(0.0, 0.5, value=0.08, step=0.01, label="PX Gamma") with gr.Accordion("verstärkbar Relay (seite15)", open=False): gr.Markdown( "Re-Injektion der modell-eigenen L16-Zustands-Richtung `d_width` am " "post-recur Layer (Motor unangetastet, forward_hook). Wirksam mit " "**ACTIVE_MANIFOLD_RELAY** (default sign=+1) oder sign≠0 auf jedem " "Preset. Nur gemma3-1b-it hat ein d_width-Artefakt; andere Modelle " "→ relay no-op, LEAN-Engine läuft. siehe scratches/psychomotrik/LESUNG15.md" ) relay_sign = gr.Radio( choices=[("+1 (WIDE / expansiv / aktiv)", 1), (" 0 (relay off)", 0), ("-1 (NARROW / eng / still)", -1)], value=1, label="Relay Richtung (sign)" ) relay_alpha = gr.Slider(0.0, 1.5, value=0.30, step=0.05, label="Relay Alpha (Bruchteil L21-Norm; kohärenter Chat ~0.30, seite15-stark=0.5)") relay_layer = gr.Slider(1, 25, value=21, step=1, label="Relay Injektions-Layer") gr.Markdown("---") # Plan 2026-07-08: System-Prompt prominent in der Sidebar. # User-Wahl: Einstellungs-Tab weg, System-Prompt NUR in Sidebar. # Beim Wechsel des px_preset wird automatisch der passende # System-Prompt in die Textarea geladen (preset_to_profile + # load_profile_for_preset in system_prompt.py). from gradio_tabs.system_prompt import ( list_profiles as _list_profiles, load_profile_for_preset as _load_profile_for_preset, ) _profile_choices = _list_profiles() with gr.Accordion("System-Prompt (Frame-Orientierer)", open=True): gr.Markdown( "Frame = Orientierer, nicht 观-Produzent. Klicke auf " "'Profil' um einen Default in die Textarea zu laden, oder " "schreibe direkt deinen eigenen Text. Der Textarea-Inhalt " "geht direkt in den nächsten Chat." ) system_profile = gr.Dropdown( choices=_profile_choices, value="neutral" if "neutral" in _profile_choices else _profile_choices[0], label="Profil", info="citmind=PX-Frame, juexin=RELAY/Kontemplation, neutral=kein Frame", ) system_prompt_text = gr.Textbox( label="System-Prompt (editierbar)", placeholder=( "Klicke unten auf 'Profil' um einen Default zu laden, oder " "schreibe direkt deinen eigenen Text. Leer = kein System-Prompt." ), lines=6, info=( "Source-of-Truth: was hier steht, geht in den Chat. " "Profil-Auswahl überschreibt das Edit-Feld." ), ) reset_prompt_btn = gr.Button( "↺ Reset auf Profil-Default", size="sm", variant="secondary", ) gr.Markdown("---") gr.Markdown("### Sessions") new_session_btn = gr.Button("New Session", variant="secondary") with gr.Row(): session_dropdown = gr.Dropdown(choices=list_sessions(), label="Saved Sessions", scale=4) refresh_sessions_btn = gr.Button("🔄", scale=1) load_session_btn = gr.Button("Load Selected", size="sm") session_id_display = gr.Textbox(label="Current ID", interactive=False) gr.Markdown("---") export_btn = gr.Button("Download JSON", size="sm") export_file = gr.File(label="Export", visible=False) import_file = gr.File(label="Import JSON", file_types=[".json"]) import_btn = gr.Button("Import & Load", size="sm") # ── Chat Components ── # Plan ui-styling Task #182: Markdown-Renderer + Bubble-Alignment. # render_markdown=True: Assistant-Antworten mit ``` Code-Blöcken werden # gerendert (default). # sanitize_html=True: HTML in Messages wird escaped (default). # line_breaks=True: Newlines in Plain-Text bleiben sichtbar. # group_consecutive_messages=True: aufeinanderfolgende gleiche Rollen # werden zu einer Bubble gruppiert (default in 6.15.2). # avatar_images=(None, None): kein Bild-Avatar (Plan-Default). # Hinweis: gr.Blocks(css=...) wird in app.py gesetzt (siehe _styles.py). chatbot = gr.Chatbot( autoscroll=False, scale=1, render_markdown=True, sanitize_html=True, line_breaks=True, group_consecutive_messages=True, avatar_images=(None, None), height=520, show_label=False, ) with gr.Row(): # Plan 2026-07-08: Bild-/Anhang-Upload im Chat. gr.MultimodalTextbox # gibt {"text": str, "files": [path_or_dict, ...]} zurück — wird # in chat_fn via normalize_multimodal_message() in einen # content-list mit text+image-Blöcken konvertiert. Server-Side # (streaming_bridge._build_image_data_url, generators._extract_images, # schemas) ist bereits da; nur die UI fehlte. msg_input = gr.MultimodalTextbox( placeholder="Type a message, or attach an image…", show_label=False, scale=9, container=False, file_count="multiple", file_types=["image"], ) submit_btn = gr.Button("Send", scale=1, variant="primary") # Plan ui-styling 2026-07-08: Undo-Button umbenannt von "Undo Last Turn" # zu "Undo Last Message" — semantisch korrekt: poppt nur 1 Element # (User oder Agent), nicht das ganze (user, assistant)-Paar. Eigene Zeile # unter dem Input, damit er optisch von der Haupt-Action (Send) getrennt # ist. Status-Markdown zeigt "✓ Undone" / "⚠ Nothing to undo". with gr.Row(): undo_btn = gr.Button("↶ Undo Last Message", size="sm", variant="secondary") undo_status = gr.Markdown("") # ── Logic ── def user_message(message, history): # Plan 2026-07-08: Bild-Upload. MultmodalTextbox liefert dict # {"text": ..., "files": [...]} (oder str bei reiner Text-Eingabe). # Normalisierung in multimodal_input.normalize_multimodal_message # damit der Chat-Eintrag exakt das Format hat das chat_fn erwartet. if is_empty_message(message): return gr.update(), history content = normalize_multimodal_message(message) return gr.update(value=None), history + [{"role": "user", "content": content}] def bot_response(history, model_id, px_preset, temp, tp, mt, rp, gamma, relay_sign, relay_alpha, relay_layer, system_profile, system_prompt_text, session_id): # 2. Call our core chat_fn and yield full updated history # We pass history (which now has the user message) to chat_fn # But chat_fn also does its own history recovery if needed. # To avoid duplication, we ensure chat_fn sees the 'true' state. # Generator for streaming updates generator = chat_fn( message=history[-1]["content"], # Last message is the user message history=history[:-1], # Everything before is the history model_id=model_id, px_preset=px_preset, temp=temp, tp=tp, mt=mt, rp=rp, gamma=gamma, relay_sign=relay_sign, relay_alpha=relay_alpha, relay_layer=relay_layer, system_profile=system_profile, system_prompt_text=system_prompt_text, session_id=session_id, manager=manager ) # Since chat_fn now yields only partial_text (string), # we need to append it to history for the UI current_history = list(history) current_history.append({"role": "assistant", "content": ""}) for partial_text in generator: current_history[-1]["content"] = partial_text yield current_history # Events msg_input.submit( fn=user_message, inputs=[msg_input, chatbot], outputs=[msg_input, chatbot], queue=False ).then( fn=bot_response, inputs=[chatbot, model_select, px_preset, temperature, top_p, max_tokens, rep_p, px_gamma, relay_sign, relay_alpha, relay_layer, system_profile, system_prompt_text, session_id_state], outputs=[chatbot] ) submit_btn.click( fn=user_message, inputs=[msg_input, chatbot], outputs=[msg_input, chatbot], queue=False ).then( fn=bot_response, inputs=[chatbot, model_select, px_preset, temperature, top_p, max_tokens, rep_p, px_gamma, relay_sign, relay_alpha, relay_layer, system_profile, system_prompt_text, session_id_state], outputs=[chatbot] ) # ── Internal connections ── new_session_btn.click( fn=handle_new_session, outputs=[session_id_state, chatbot, session_dropdown, session_id_display] ) load_session_btn.click( fn=handle_load_saved, inputs=[session_dropdown], outputs=[session_id_state, chatbot, session_dropdown, session_id_display] ) export_btn.click(fn=handle_export, inputs=[session_id_state, chatbot], outputs=[export_file]) import_btn.click(fn=handle_import, inputs=[import_file], outputs=[session_id_state, chatbot, session_dropdown, session_id_display]) refresh_sessions_btn.click(fn=handle_refresh, outputs=[session_dropdown]) # Plan ui-styling 2026-07-06: Undo-Button click-handler. # Outputs: chatbot (gekürzte History) + undo_status (Feedback). undo_btn.click( fn=handle_undo, inputs=[session_id_state, chatbot], outputs=[chatbot, undo_status], ) # Plan 2026-07-09: Preset und System-Prompt sind komplett entkoppelt. # User-Entscheidung 2026-07-09: "ich will nicht automatisch citmind # oder juexin paden, default soll der systemprompt leer sein. und # wenn ich auf die systemprompt liste einen eintrag klicke, dann # soll der geladen werden, ins systemprompt feld. und von dort soll # er immer für den chat benutzt werden, also wenn ich da was ändere, # dass es gleich auswirkungen auf den chat danach hat". # # → px_preset wählt NUR den PX-Engine-Modus (BASELINE/ACTIVE_MANIFOLD/ # ACTIVE_MANIFOLD_RELAY), ändert NICHT den System-Prompt. # → system_profile wählt NUR den Profil-Default, lädt dessen Body in # die Textarea (überschreibt aktuelles Edit). # → system_prompt_text ist die Source-of-Truth: was dort steht, geht # in den Chat (build_system_message nutzt edit_text, wenn vorhanden). # → reset_prompt_btn leert die Textarea (Profil-Default wird beim # nächsten system_profile.change() neu geladen). system_profile.change( fn=on_profile_change_load_body, inputs=[system_profile], outputs=[system_prompt_text], ) reset_prompt_btn.click( fn=on_reset_prompt_click, inputs=[], outputs=[system_prompt_text], ) return session_id_state, chatbot, session_dropdown, session_id_display