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