159 lines
4.9 KiB
Python
159 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
import ollama
|
|
|
|
|
|
ROOT = Path(__file__).parent
|
|
DATA_DIR = ROOT / "data"
|
|
CHATS_FILE = DATA_DIR / "chats.json"
|
|
MODEL = "qwen2.5:7b"
|
|
HOST = "0.0.0.0"
|
|
PORT = 8787
|
|
|
|
|
|
def now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def read_store() -> dict:
|
|
if not CHATS_FILE.exists():
|
|
return {"chats": []}
|
|
|
|
try:
|
|
return json.loads(CHATS_FILE.read_text(encoding="utf-8"))
|
|
except json.JSONDecodeError:
|
|
return {"chats": []}
|
|
|
|
|
|
def write_store(store: dict) -> None:
|
|
DATA_DIR.mkdir(exist_ok=True)
|
|
CHATS_FILE.write_text(json.dumps(store, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
|
|
def chat_title(messages: list[dict]) -> str:
|
|
for message in messages:
|
|
if message.get("role") == "user" and message.get("content"):
|
|
title = " ".join(str(message["content"]).split())
|
|
return title[:40] or "새 대화"
|
|
return "새 대화"
|
|
|
|
|
|
class Handler(SimpleHTTPRequestHandler):
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, directory=str(ROOT), **kwargs)
|
|
|
|
def do_GET(self) -> None:
|
|
if self.path == "/api/chats":
|
|
store = read_store()
|
|
chats = [
|
|
{
|
|
"id": chat["id"],
|
|
"title": chat.get("title", "새 대화"),
|
|
"updatedAt": chat.get("updatedAt", ""),
|
|
}
|
|
for chat in store.get("chats", [])
|
|
]
|
|
self._send_json(200, {"chats": chats})
|
|
return
|
|
|
|
if self.path.startswith("/api/chats/"):
|
|
chat_id = self.path.rsplit("/", 1)[-1]
|
|
store = read_store()
|
|
chat = next((item for item in store.get("chats", []) if item["id"] == chat_id), None)
|
|
if not chat:
|
|
self._send_json(404, {"error": "Chat not found"})
|
|
return
|
|
|
|
self._send_json(200, {"chat": chat})
|
|
return
|
|
|
|
super().do_GET()
|
|
|
|
def do_POST(self) -> None:
|
|
if self.path != "/api/chat":
|
|
self.send_error(404)
|
|
return
|
|
|
|
length = int(self.headers.get("content-length", 0))
|
|
data = json.loads(self.rfile.read(length) or b"{}")
|
|
chat_id = data.get("chatId")
|
|
messages = data.get("messages", [])
|
|
|
|
try:
|
|
response = ollama.chat(
|
|
model=MODEL,
|
|
messages=[
|
|
{
|
|
"role": "system",
|
|
"content": "You are a concise local assistant. Reply in Korean unless the user asks otherwise.",
|
|
},
|
|
*messages,
|
|
],
|
|
)
|
|
reply = response["message"]["content"]
|
|
saved_messages = [*messages, {"role": "assistant", "content": reply}]
|
|
chat = self._save_chat(chat_id, saved_messages)
|
|
payload = {"chatId": chat["id"], "reply": reply, "chat": chat}
|
|
self._send_json(200, payload)
|
|
except Exception as exc:
|
|
self._send_json(500, {"error": str(exc)})
|
|
|
|
def do_DELETE(self) -> None:
|
|
if not self.path.startswith("/api/chats/"):
|
|
self.send_error(404)
|
|
return
|
|
|
|
chat_id = self.path.rsplit("/", 1)[-1]
|
|
store = read_store()
|
|
store["chats"] = [item for item in store.get("chats", []) if item["id"] != chat_id]
|
|
write_store(store)
|
|
self._send_json(200, {"ok": True})
|
|
|
|
def _save_chat(self, chat_id: str | None, messages: list[dict]) -> dict:
|
|
store = read_store()
|
|
chats = store.setdefault("chats", [])
|
|
chat = next((item for item in chats if item["id"] == chat_id), None)
|
|
|
|
if not chat:
|
|
chat = {
|
|
"id": str(uuid.uuid4()),
|
|
"createdAt": now_iso(),
|
|
"title": chat_title(messages),
|
|
"messages": [],
|
|
}
|
|
chats.insert(0, chat)
|
|
|
|
chat["title"] = chat_title(messages)
|
|
chat["messages"] = messages
|
|
chat["updatedAt"] = now_iso()
|
|
chats.sort(key=lambda item: item.get("updatedAt", ""), reverse=True)
|
|
write_store(store)
|
|
return chat
|
|
|
|
def _send_json(self, status: int, payload: dict[str, str]) -> None:
|
|
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("content-type", "application/json; charset=utf-8")
|
|
self.send_header("content-length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
|
|
def main() -> None:
|
|
server = ThreadingHTTPServer((HOST, PORT), Handler)
|
|
local_ip = socket.gethostbyname(socket.gethostname())
|
|
print(f"LLM web is running locally at http://127.0.0.1:{PORT}")
|
|
print(f"LAN access is available at http://{local_ip}:{PORT}")
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|