LLM-chatinterface/tui.py

#!/usr/bin/env python3
"""Kleines TUI um chat.py.

- zeigt den Chatverlauf aus der Markdown-Datei
- mehrzeiliges Eingabefeld (Enter = neue Zeile, Ctrl+S = senden)
- die Antwort der LLM fliegt tokenweise rein

Quelle der Wahrheit bleibt die Datei (Default: chat.md). Jede gesendete
Nachricht und jede fertige Antwort wird sofort dort angehaengt.

    python tui.py [pfad/zur/chat.md]
"""

import sys
from pathlib import Path

from rich.markup import escape
from textual import work
from textual.app import App, ComposeResult
from textual.binding import Binding
from textual.containers import VerticalScroll
from textual.widgets import Footer, Header, Static, TextArea

import chat

ROLE_STYLE = {"system": "dim italic", "user": "bold cyan", "assistant": "bold green"}


class Message(Static):
    """Ein Chat-Eintrag; waehrend des Streamens waechst der Text."""

    def __init__(self, role: str, text: str = "") -> None:
        super().__init__(markup=True)
        self.role = role
        self._text = text
        self.add_class(role)
        self._rerender()

    def _rerender(self) -> None:
        style = ROLE_STYLE.get(self.role, "bold")
        head = f"[{style}]{self.role}[/]"
        body = escape(self._text) or "[dim]…[/]"
        self.update(f"{head}\n{body}")

    def append(self, chunk: str) -> None:
        self._text += chunk
        self._rerender()

    @property
    def text(self) -> str:
        return self._text


class ChatApp(App):
    CSS = """
    #log { padding: 1 2; }
    Message { margin-bottom: 1; }
    #input {
        height: auto;
        min-height: 3;
        max-height: 15;
        border: round $accent;
        margin: 0 1 1 1;
    }
    """

    BINDINGS = [
        Binding("ctrl+s", "send", "Senden"),
        Binding("ctrl+r", "reload", "Neu laden"),
        Binding("ctrl+q", "quit", "Beenden"),
    ]

    def __init__(self, path: str | Path) -> None:
        super().__init__()
        self.path = Path(path)
        self.title = f"chat — {self.path.name}"
        self._current: Message | None = None
        self._busy = False

    def compose(self) -> ComposeResult:
        yield Header()
        yield VerticalScroll(id="log")
        yield TextArea(id="input")
        yield Footer()

    def on_mount(self) -> None:
        self._load_history()
        self.query_one("#input", TextArea).focus()

    def _load_history(self) -> None:
        """Baut den sichtbaren Verlauf frisch aus der Datei auf."""
        log = self.query_one("#log", VerticalScroll)
        log.remove_children()
        for msg in chat.load_messages(self.path):
            log.mount(Message(msg["role"], msg["content"]))
        log.scroll_end(animate=False)

    def action_reload(self) -> None:
        if self._busy:
            self.notify("Antwort laeuft noch …", severity="warning")
            return
        self._load_history()
        self.notify(f"{self.path.name} neu geladen.")

    def action_send(self) -> None:
        if self._busy:
            self.notify("Antwort laeuft noch …", severity="warning")
            return
        area = self.query_one("#input", TextArea)
        text = area.text.strip()
        if not text:
            return
        area.text = ""
        self._busy = True

        chat.append_block(self.path, "user", text)
        log = self.query_one("#log", VerticalScroll)
        log.mount(Message("user", text))
        self._current = Message("assistant")
        log.mount(self._current)
        log.scroll_end()

        self._run_stream()

    @work(thread=True)
    def _run_stream(self) -> None:
        messages = chat.load_messages(self.path)
        gen = chat.stream_reply(messages)
        finish: str | None = None
        try:
            while True:
                try:
                    piece = next(gen)
                except StopIteration as stop:
                    finish = stop.value
                    break
                self.call_from_thread(self._on_token, piece)
        except Exception as exc:  # Netzwerk / Server / Parsing
            self.call_from_thread(self._on_error, str(exc))
            return
        self.call_from_thread(self._on_done, finish)

    def _on_token(self, piece: str) -> None:
        assert self._current is not None
        self._current.append(piece)
        self.query_one("#log", VerticalScroll).scroll_end()

    def _on_done(self, finish: str | None) -> None:
        assert self._current is not None
        chat.append_block(self.path, "assistant", self._current.text)
        if finish == "length":
            self.notify(
                f"Antwort bei MAX_TOKENS={chat.MAX_TOKENS} abgeschnitten.",
                severity="warning",
            )
        self._busy = False
        self._current = None
        self.query_one("#input", TextArea).focus()

    def _on_error(self, msg: str) -> None:
        if self._current is not None:
            self._current.append(f"\n⚠ Fehler: {msg}")
        self.notify(msg, title="Fehler", severity="error")
        self._busy = False
        self._current = None


def main() -> None:
    path = sys.argv[1] if len(sys.argv) > 1 else "chat.md"
    ChatApp(path).run()


if __name__ == "__main__":
    main()