feat: add local chat provider and CLI
This commit is contained in:
@@ -0,0 +1,182 @@
|
||||
"""Command-line interface for the first local Javis text chat."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from collections.abc import Callable, Sequence
|
||||
|
||||
from javis.config.settings import ConfigurationError, Settings
|
||||
from javis.core.chat_service import ChatService, SessionProviderMismatchError
|
||||
from javis.memory.sqlite_store import (
|
||||
ChatSession,
|
||||
SessionNotFoundError,
|
||||
SessionStoreError,
|
||||
SQLiteSessionStore,
|
||||
)
|
||||
from javis.providers.base import ProviderError
|
||||
from javis.providers.ollama import OllamaProvider
|
||||
|
||||
InputFunction = Callable[[str], str]
|
||||
OutputFunction = Callable[[str], None]
|
||||
|
||||
HELP_TEXT = """Befehle:
|
||||
/new neue Sitzung beginnen
|
||||
/sessions gespeicherte Sitzungen auflisten
|
||||
/load <ID> frühere Sitzung laden und fortsetzen
|
||||
/clear Nachrichten der aktiven Sitzung leeren
|
||||
/help diese Hilfe anzeigen
|
||||
/exit Chat ordentlich beenden"""
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(prog="javis", description="Lokaler Javis-Textchat")
|
||||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||
chat_parser = subparsers.add_parser("chat", help="lokalen Textchat starten")
|
||||
chat_parser.add_argument("--session", help="vorhandene Sitzungs-ID laden")
|
||||
return parser
|
||||
|
||||
|
||||
def _show_history(service: ChatService, session_id: str, output: OutputFunction) -> None:
|
||||
loaded = service.load_session(session_id)
|
||||
if not loaded.messages:
|
||||
output("(Sitzung enthält noch keine Nachrichten.)")
|
||||
return
|
||||
for message in loaded.messages:
|
||||
label = "Du" if message.role == "user" else "Javis"
|
||||
output(f"{label}: {message.content}")
|
||||
|
||||
|
||||
def _show_sessions(sessions: list[ChatSession], output: OutputFunction) -> None:
|
||||
if not sessions:
|
||||
output("Keine gespeicherten Sitzungen vorhanden.")
|
||||
return
|
||||
for session in sessions:
|
||||
output(
|
||||
f"{session.id} | {session.updated_at} | "
|
||||
f"{session.provider}/{session.model} | {session.message_count} Nachrichten"
|
||||
)
|
||||
|
||||
|
||||
def run_chat(
|
||||
service: ChatService,
|
||||
*,
|
||||
session_id: str | None = None,
|
||||
input_fn: InputFunction = input,
|
||||
output: OutputFunction = print,
|
||||
) -> int:
|
||||
try:
|
||||
if session_id:
|
||||
active_id = service.load_session(session_id).session.id
|
||||
output(f"Sitzung geladen: {active_id}")
|
||||
_show_history(service, active_id, output)
|
||||
else:
|
||||
active_id = service.new_session().id
|
||||
output(f"Neue Sitzung: {active_id}")
|
||||
except (
|
||||
SessionNotFoundError,
|
||||
SessionProviderMismatchError,
|
||||
SessionStoreError,
|
||||
) as exc:
|
||||
output(f"Fehler: {exc}")
|
||||
return 2
|
||||
|
||||
output("Lokaler Javis-Chat. /help zeigt die Befehle.")
|
||||
|
||||
while True:
|
||||
try:
|
||||
entered = input_fn("Du> ").strip()
|
||||
except (EOFError, KeyboardInterrupt):
|
||||
output("\nChat beendet.")
|
||||
return 0
|
||||
|
||||
if not entered:
|
||||
continue
|
||||
if entered == "/exit":
|
||||
output("Chat beendet.")
|
||||
return 0
|
||||
if entered == "/help":
|
||||
output(HELP_TEXT)
|
||||
continue
|
||||
if entered == "/new":
|
||||
try:
|
||||
active_id = service.new_session().id
|
||||
output(f"Neue Sitzung: {active_id}")
|
||||
except SessionStoreError as exc:
|
||||
output(f"Fehler: {exc}")
|
||||
continue
|
||||
if entered == "/sessions":
|
||||
try:
|
||||
_show_sessions(service.list_sessions(), output)
|
||||
except SessionStoreError as exc:
|
||||
output(f"Fehler: {exc}")
|
||||
continue
|
||||
if entered == "/clear":
|
||||
try:
|
||||
service.clear_session(active_id)
|
||||
output("Aktive Sitzung wurde geleert.")
|
||||
except (SessionNotFoundError, SessionStoreError) as exc:
|
||||
output(f"Fehler: {exc}")
|
||||
continue
|
||||
if entered.startswith("/load"):
|
||||
parts = entered.split(maxsplit=1)
|
||||
if len(parts) != 2 or not parts[1].strip():
|
||||
output("Verwendung: /load <Sitzungs-ID>")
|
||||
continue
|
||||
try:
|
||||
loaded = service.load_session(parts[1].strip())
|
||||
active_id = loaded.session.id
|
||||
output(f"Sitzung geladen: {active_id}")
|
||||
_show_history(service, active_id, output)
|
||||
except (
|
||||
SessionNotFoundError,
|
||||
SessionProviderMismatchError,
|
||||
SessionStoreError,
|
||||
) as exc:
|
||||
output(f"Fehler: {exc}")
|
||||
continue
|
||||
if entered.startswith("/"):
|
||||
output("Unbekannter Befehl. /help zeigt die verfügbaren Befehle.")
|
||||
continue
|
||||
|
||||
try:
|
||||
response = service.send(active_id, entered)
|
||||
output(f"Javis: {response}")
|
||||
except (ProviderError, SessionStoreError, SessionProviderMismatchError) as exc:
|
||||
output(f"Fehler: {exc}")
|
||||
except ValueError as exc:
|
||||
output(f"Fehler: {exc}")
|
||||
|
||||
|
||||
def main(
|
||||
argv: Sequence[str] | None = None,
|
||||
*,
|
||||
input_fn: InputFunction = input,
|
||||
output: OutputFunction = print,
|
||||
) -> int:
|
||||
args = build_parser().parse_args(argv)
|
||||
if args.command != "chat":
|
||||
return 2
|
||||
|
||||
try:
|
||||
settings = Settings.from_env()
|
||||
store = SQLiteSessionStore(settings.database_path)
|
||||
provider = OllamaProvider(
|
||||
settings.model,
|
||||
settings.ollama_base_url,
|
||||
settings.timeout_seconds,
|
||||
)
|
||||
except (ConfigurationError, SessionStoreError) as exc:
|
||||
output(f"Konfigurationsfehler: {exc}")
|
||||
return 2
|
||||
|
||||
return run_chat(
|
||||
ChatService(store, provider),
|
||||
session_id=args.session,
|
||||
input_fn=input_fn,
|
||||
output=output,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user