Build a fully local desktop chatbot with LM Studio

from __future__ import annotations

import http.client
import json
import os
import threading
from collections.abc import Iterator
from urllib.parse import urlsplit


class LmStudioError(RuntimeError):
    pass


class LmStudioClient:
    def __init__(self, base_url: str, token: str, timeout_seconds: float = 60.0) -> None:
        parsed = urlsplit(base_url)
        if parsed.scheme != "http" or parsed.hostname not in {"127.0.0.1", "localhost", "::1"}:
            raise ValueError("LM Studio must use an HTTP loopback address.")
        if parsed.username or parsed.password:
            raise ValueError("Credentials may not appear in the server URL.")
        self.host = parsed.hostname or "127.0.0.1"
        self.port = parsed.port or 1234
        self.token = token
        self.timeout_seconds = timeout_seconds

    @classmethod
    def from_environment(cls) -> "LmStudioClient":
        return cls(
            os.getenv("LM_STUDIO_URL", "http://127.0.0.1:1234"),
            os.environ["LM_STUDIO_TOKEN"],
            float(os.getenv("LM_STUDIO_TIMEOUT_SECONDS", "60")),
        )

    def _headers(self) -> dict[str, str]:
        return {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json",
            "Accept": "application/json",
        }

    def list_models(self) -> list[dict[str, object]]:
        connection = http.client.HTTPConnection(self.host, self.port, timeout=5)
        try:
            connection.request("GET", "/api/v1/models", headers=self._headers())
            response = connection.getresponse()
            body = response.read(1_048_577)
            if len(body) > 1_048_576:
                raise LmStudioError("LM Studio returned an oversized model list.")
            if response.status != 200:
                raise LmStudioError(f"Model listing failed with HTTP {response.status}.")
            document = json.loads(body)
            return list(document.get("models", []))
        except (OSError, ValueError, json.JSONDecodeError) as error:
            raise LmStudioError(f"Could not read the local model list: {error}") from error
        finally:
            connection.close()

    def load_model(self, model: str, context_length: int = 8192) -> str:
        if not model or len(model) > 240:
            raise ValueError("Model identifier is missing or too long.")
        body = json.dumps({
            "model": model,
            "context_length": context_length,
            "echo_load_config": True,
        })
        connection = http.client.HTTPConnection(self.host, self.port, timeout=180)
        try:
            connection.request("POST", "/api/v1/models/load", body, self._headers())
            response = connection.getresponse()
            payload = json.loads(response.read(1_048_576))
            if response.status not in {200, 201}:
                raise LmStudioError(f"Model load failed with HTTP {response.status}.")
            return str(payload["instance_id"])
        finally:
            connection.close()

    def unload_model(self, instance_id: str) -> None:
        body = json.dumps({"instance_id": instance_id})
        connection = http.client.HTTPConnection(self.host, self.port, timeout=30)
        try:
            connection.request("POST", "/api/v1/models/unload", body, self._headers())
            response = connection.getresponse()
            response.read(1_048_576)
            if response.status != 200:
                raise LmStudioError(f"Model unload failed with HTTP {response.status}.")
        finally:
            connection.close()

    def stream_chat(
        self,
        model: str,
        messages: list[dict[str, str]],
        cancelled: threading.Event,
    ) -> Iterator[str]:
        body = json.dumps({
            "model": model,
            "messages": messages[-20:],
            "stream": True,
            "temperature": 0.3,
            "max_tokens": 1200,
        })
        connection = http.client.HTTPConnection(self.host, self.port, timeout=self.timeout_seconds)
        try:
            connection.request("POST", "/v1/chat/completions", body, self._headers())
            response = connection.getresponse()
            if response.status != 200:
                detail = response.read(16_384).decode("utf-8", "replace")
                raise LmStudioError(f"Generation failed with HTTP {response.status}: {detail[:400]}")
            while not cancelled.is_set():
                line = response.readline(65_537)
                if not line:
                    break
                if len(line) > 65_536:
                    raise LmStudioError("A streaming event exceeded the 64 KiB safety limit.")
                text = line.decode("utf-8", "strict").strip()
                if not text.startswith("data:"):
                    continue
                data = text[5:].strip()
                if data == "[DONE]":
                    break
                event = json.loads(data)
                token = event.get("choices", [{}])[0].get("delta", {}).get("content")
                if isinstance(token, str):
                    yield token
        except (OSError, UnicodeError, json.JSONDecodeError) as error:
            raise LmStudioError(f"Local streaming failed: {error}") from error
        finally:
            connection.close()

from __future__ import annotations

import json
import sqlite3
import time
import uuid
from pathlib import Path


class ConversationStore:
    def __init__(self, path: Path) -> None:
        self.path = path
        self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        self.database = sqlite3.connect(path, check_same_thread=False)
        self.database.execute("PRAGMA journal_mode=WAL")
        self.database.execute("PRAGMA foreign_keys=ON")
        self.database.execute("""
          CREATE TABLE IF NOT EXISTS conversations (
            id TEXT PRIMARY KEY,
            created_at INTEGER NOT NULL,
            turns_json TEXT NOT NULL CHECK(length(turns_json) <= 1048576)
          )
        """)
        self.database.commit()
        path.chmod(0o600)

    def save_turns(self, turns: list[dict[str, str]]) -> str:
        cleaned = [
            {"role": item["role"], "content": item["content"][:32_768]}
            for item in turns[-40:]
            if item.get("role") in {"user", "assistant"} and isinstance(item.get("content"), str)
        ]
        identifier = str(uuid.uuid4())
        self.database.execute(
            "INSERT INTO conversations (id, created_at, turns_json) VALUES (?, ?, ?)",
            (identifier, int(time.time()), json.dumps(cleaned, ensure_ascii=False)),
        )
        self.database.commit()
        return identifier

    def export_json(self, destination: Path) -> int:
        rows = self.database.execute(
            "SELECT id, created_at, turns_json FROM conversations ORDER BY created_at"
        ).fetchall()
        document = [
            {"id": row[0], "created_at": row[1], "turns": json.loads(row[2])}
            for row in rows
        ]
        destination.write_text(json.dumps(document, indent=2, ensure_ascii=False), encoding="utf-8")
        destination.chmod(0o600)
        return len(document)

    def delete_all(self) -> int:
        count = self.database.execute("SELECT count(*) FROM conversations").fetchone()[0]
        self.database.execute("DELETE FROM conversations")
        self.database.commit()
        self.database.execute("VACUUM")
        return int(count)

from __future__ import annotations

import json
import queue
import threading
import tkinter as tk
from dataclasses import dataclass
from pathlib import Path
from tkinter import messagebox, ttk
from typing import Callable

from lm_api import LmStudioClient, LmStudioError
from store import ConversationStore


@dataclass(frozen=True)
class PrivacySettings:
    save_history: bool = False
    include_previous_turns: bool = True
    max_saved_conversations: int = 50


class LocalChatApp:
    def __init__(self, root: tk.Tk, data_dir: Path) -> None:
        self.root = root
        self.root.title("Local Desk Chat")
        self.root.geometry("940x700")
        self.client = LmStudioClient.from_environment()
        self.store = ConversationStore(data_dir / "conversations.sqlite3")
        self.events: queue.Queue[tuple[str, str]] = queue.Queue()
        self.cancel_event = threading.Event()
        self.turns: list[dict[str, str]] = []
        self.save_history = tk.BooleanVar(value=False)
        self.model = tk.StringVar()
        self.status = tk.StringVar(value="LM Studio has not been checked")
        self._build()
        self.root.after(40, self._drain_events)

    def _build(self) -> None:
        toolbar = ttk.Frame(self.root, padding=12)
        toolbar.pack(fill=tk.X)
        ttk.Label(toolbar, text="Model").pack(side=tk.LEFT)
        self.models = ttk.Combobox(toolbar, textvariable=self.model, state="readonly", width=45)
        self.models.pack(side=tk.LEFT, padx=(8, 16))
        ttk.Button(toolbar, text="Refresh", command=self.refresh_models).pack(side=tk.LEFT)
        ttk.Checkbutton(
            toolbar,
            text="Save conversations on this device",
            variable=self.save_history,
        ).pack(side=tk.RIGHT)

        self.transcript = tk.Text(self.root, wrap=tk.WORD, state=tk.DISABLED, padx=14, pady=14)
        self.transcript.pack(fill=tk.BOTH, expand=True, padx=12)
        self.transcript.tag_configure("user", foreground="#4FC1FF")
        self.transcript.tag_configure("assistant", foreground="#DCDCAA")
        self.transcript.tag_configure("notice", foreground="#CE9178")

        composer = ttk.Frame(self.root, padding=12)
        composer.pack(fill=tk.X)
        self.input = tk.Text(composer, height=5, wrap=tk.WORD)
        self.input.pack(side=tk.LEFT, fill=tk.X, expand=True)
        actions = ttk.Frame(composer)
        actions.pack(side=tk.RIGHT, padx=(12, 0))
        ttk.Button(actions, text="Send", command=self.send).pack(fill=tk.X)
        ttk.Button(actions, text="Stop", command=self.stop).pack(fill=tk.X, pady=6)
        ttk.Button(actions, text="Delete local history", command=self.delete_history).pack(fill=tk.X)
        ttk.Label(self.root, textvariable=self.status, padding=(12, 0, 12, 12)).pack(anchor=tk.W)

    def refresh_models(self) -> None:
        def task() -> None:
            try:
                models = self.client.list_models()
                identifiers = [
                    item["key"] for item in models
                    if item.get("type") == "llm"
                ]
                self.events.put(("models", json.dumps(identifiers)))
            except LmStudioError as error:
                self.events.put(("error", str(error)))
        threading.Thread(target=task, daemon=True).start()

    def send(self) -> None:
        prompt = self.input.get("1.0", tk.END).strip()
        model = self.model.get().strip()
        if not prompt or not model:
            messagebox.showwarning("Missing input", "Choose a loaded model and enter a message.")
            return
        if len(prompt.encode("utf-8")) > 32_768:
            messagebox.showwarning("Input too large", "The message is limited to 32 KiB.")
            return
        self.input.delete("1.0", tk.END)
        self.turns.append({"role": "user", "content": prompt})
        self._append("You", prompt, "user")
        self._append("Assistant", "", "assistant")
        self.cancel_event.clear()
        self.status.set("Generating locally…")

        def task() -> None:
            collected: list[str] = []
            try:
                for token in self.client.stream_chat(model, self.turns, self.cancel_event):
                    collected.append(token)
                    self.events.put(("token", token))
                answer = "".join(collected)
                if answer:
                    self.turns.append({"role": "assistant", "content": answer})
                    if self.save_history.get():
                        self.store.save_turns(self.turns)
                self.events.put(("done", "Stopped" if self.cancel_event.is_set() else "Ready"))
            except LmStudioError as error:
                self.events.put(("error", str(error)))
        threading.Thread(target=task, daemon=True).start()

    def stop(self) -> None:
        self.cancel_event.set()
        self.status.set("Stopping after the current network read…")

    def delete_history(self) -> None:
        if messagebox.askyesno("Delete history", "Delete every conversation saved by this app?"):
            deleted = self.store.delete_all()
            self.status.set(f"Deleted {deleted} saved conversations")

    def _append(self, label: str, text: str, tag: str) -> None:
        self.transcript.configure(state=tk.NORMAL)
        self.transcript.insert(tk.END, f"\n{label}\n", tag)
        self.transcript.insert(tk.END, text, tag)
        self.transcript.see(tk.END)
        self.transcript.configure(state=tk.DISABLED)

    def _drain_events(self) -> None:
        try:
            while True:
                kind, payload = self.events.get_nowait()
                if kind == "models":
                    values = json.loads(payload)
                    self.models.configure(values=values)
                    if values:
                        self.model.set(values[0])
                    self.status.set(f"{len(values)} local model(s) available")
                elif kind == "token":
                    self.transcript.configure(state=tk.NORMAL)
                    self.transcript.insert(tk.END, payload, "assistant")
                    self.transcript.see(tk.END)
                    self.transcript.configure(state=tk.DISABLED)
                elif kind == "done":
                    self.status.set(payload)
                elif kind == "error":
                    self.status.set("Request failed")
                    self._append("Local error", payload, "notice")
        except queue.Empty:
            pass
        self.root.after(40, self._drain_events)


if __name__ == "__main__":
    home = Path.home() / ".local" / "share" / "local-desk-chat"
    home.mkdir(parents=True, exist_ok=True, mode=0o700)
    root = tk.Tk()
    LocalChatApp(root, home).refresh_models()
    root.mainloop()

import json
import sqlite3
import tempfile
import unittest
from pathlib import Path
from threading import Event
from unittest.mock import patch

from lm_api import LmStudioClient
from store import ConversationStore


class FakeResponse:
    status = 200
    def __init__(self) -> None:
        self.lines = iter([
            b'data: {"choices":[{"delta":{"content":"local"}}]}\n',
            b'data: {"choices":[{"delta":{"content":" answer"}}]}\n',
            b'data: [DONE]\n',
        ])
    def readline(self, _limit: int) -> bytes:
        return next(self.lines, b"")
    def read(self, _limit: int = -1) -> bytes:
        return json.dumps({"models": [{"key": "test/model", "type": "llm"}]}).encode()


class FakeConnection:
    def __init__(self, *_args, **_kwargs) -> None:
        self.response = FakeResponse()
    def request(self, *_args, **_kwargs) -> None:
        return None
    def getresponse(self) -> FakeResponse:
        return self.response
    def close(self) -> None:
        return None


class LocalDeskChatTests(unittest.TestCase):
    @patch("lm_api.http.client.HTTPConnection", FakeConnection)
    def test_stream_reassembles_sse_tokens(self) -> None:
        client = LmStudioClient("http://127.0.0.1:1234", "test-token")
        result = "".join(client.stream_chat(
            "test/model",
            [{"role": "user", "content": "hello"}],
            Event(),
        ))
        self.assertEqual(result, "local answer")

    def test_store_is_opt_in_and_exportable(self) -> None:
        with tempfile.TemporaryDirectory() as directory:
            store = ConversationStore(Path(directory) / "chat.sqlite3")
            identifier = store.save_turns([
                {"role": "user", "content": "private test"},
                {"role": "assistant", "content": "local reply"},
            ])
            output = Path(directory) / "export.json"
            self.assertEqual(store.export_json(output), 1)
            self.assertIn(identifier, output.read_text(encoding="utf-8"))
            self.assertEqual(store.delete_all(), 1)


if __name__ == "__main__":
    unittest.main()