current chat

tui.py

@@ -0,0 +1,174 @@
+#!/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()