Working chat: auth, connections, streaming, folders
LLeMbas now runs end to end. Register, add an OpenAI-compatible connection, and hold a real streaming conversation organised into folders. Verified against the local llama-swap instance. Streaming is the one genuinely tricky part. Sending a message returns two HTML fragments -- the user bubble and an empty assistant bubble carrying an sse-connect -- and that attribute is the ONLY thing that starts a generation. Rendering an incomplete assistant message as a streaming shell falls out of the same template, which means loading a page whose last reply never finished simply picks it up again. Details worth knowing about, each commented where it matters: - SSE payloads are split across several data: lines. A raw newline in one data: line truncates the event, which shows up the first time a model emits a code block. - Markdown is rendered server-side by the same helper for both the page and the final streamed frame, so the two cannot disagree. The fence renderer is replaced outright rather than using markdown-it's highlight option, which re-wraps output in a second <pre>. - escape_text is html.escape, not nh3.clean_text: it escapes character by character, so escaping stream chunks separately equals escaping the whole string. - The stream opens its own session via session_scope(); it outlives the request handler and the dependency-scoped session may be closed. - Deleting a folder keeps the chats inside it (FK is SET NULL). Losing a conversation to a mis-clicked folder delete is unforgivable. - Login failures use one message for "no such account" and "wrong password" so the form cannot enumerate registered addresses. Also adds deploy/ for the gamebox install at https://chat.lan: system unit, nginx vhost with buffering off (buffering on turns streaming into one lump at the end), and install/update scripts following the same service-user and /srv bind-mount conventions as llama-swap and comfyui. 70 tests, ruff clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jaroslav Beneš
@@ -0,0 +1,290 @@
|
||||
"""Chat creation, messaging and the streaming reply endpoint."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from collections.abc import AsyncIterator
|
||||
|
||||
from fastapi import APIRouter, Form, HTTPException, Request, status
|
||||
from fastapi.responses import HTMLResponse, Response, StreamingResponse
|
||||
from sqlalchemy.orm import Session as DBSession
|
||||
|
||||
from lembas.api.deps import Db, RequiredUser
|
||||
from lembas.db.models import ROLE_ASSISTANT, ROLE_USER, Chat, Message, User
|
||||
from lembas.db.session import session_scope
|
||||
from lembas.services import chat as chat_service
|
||||
from lembas.services import sse
|
||||
from lembas.services.llm.openai_client import LLMError, delta_text, stream_chat
|
||||
from lembas.services.markdown import escape_text, render_markdown
|
||||
from lembas.web.templating import render, templates
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/api/chats", tags=["chats"])
|
||||
|
||||
|
||||
def _owned_chat(db: DBSession, chat_id: str, user_id: str) -> Chat:
|
||||
chat = db.get(Chat, chat_id)
|
||||
# 404 rather than 403 for someone else's chat: whether a given id exists is
|
||||
# not information this endpoint should hand out.
|
||||
if chat is None or chat.user_id != user_id:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "That chat no longer exists.")
|
||||
return chat
|
||||
|
||||
|
||||
@router.post("")
|
||||
async def create_chat(db: Db, user: RequiredUser, folder_id: str = Form("")) -> Response:
|
||||
chosen = chat_service.default_model(db)
|
||||
chat = Chat(
|
||||
user_id=user.id,
|
||||
folder_id=folder_id or None,
|
||||
model_id=chosen[0] if chosen else "",
|
||||
connection_id=chosen[1] if chosen else None,
|
||||
)
|
||||
db.add(chat)
|
||||
db.commit()
|
||||
|
||||
# HX-Redirect rather than a swap: a new chat is a new URL, and the address
|
||||
# bar has to follow so the chat can be reloaded or bookmarked.
|
||||
response = Response(status_code=status.HTTP_204_NO_CONTENT)
|
||||
response.headers["HX-Redirect"] = f"/chat/{chat.id}"
|
||||
return response
|
||||
|
||||
|
||||
@router.post("/{chat_id}/messages")
|
||||
async def post_message(
|
||||
request: Request,
|
||||
db: Db,
|
||||
user: RequiredUser,
|
||||
chat_id: str,
|
||||
content: str = Form(...),
|
||||
) -> Response:
|
||||
"""Persist the user's turn and hand back the pair of bubbles.
|
||||
|
||||
The assistant bubble comes back empty, carrying the sse-connect attribute
|
||||
that opens the stream below. Splitting it this way means the POST returns
|
||||
immediately and the slow part is a separate, resumable connection.
|
||||
"""
|
||||
chat = _owned_chat(db, chat_id, user.id)
|
||||
|
||||
content = content.strip()
|
||||
if not content:
|
||||
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
||||
|
||||
user_message = chat_service.create_message(db, chat, ROLE_USER, content)
|
||||
assistant_message = chat_service.create_message(
|
||||
db, chat, ROLE_ASSISTANT, "", complete_=False, model_id=chat.model_id
|
||||
)
|
||||
|
||||
# `user` is required by the shared message template, which renders both
|
||||
# roles; without it the user bubble's initial blows up.
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"chat/_turn.html",
|
||||
{
|
||||
"request": request,
|
||||
"user_message": user_message,
|
||||
"assistant_message": assistant_message,
|
||||
"chat": chat,
|
||||
"user": user,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.get("/{chat_id}/messages/{message_id}/stream")
|
||||
async def stream_message(
|
||||
db: Db,
|
||||
user: RequiredUser,
|
||||
chat_id: str,
|
||||
message_id: str,
|
||||
) -> Response:
|
||||
"""Stream the assistant's reply as server-sent events.
|
||||
|
||||
Emits `token` events carrying escaped text, then a single `done` event
|
||||
carrying the finished bubble rendered from Markdown, then `close`.
|
||||
"""
|
||||
chat = _owned_chat(db, chat_id, user.id)
|
||||
message = db.get(Message, message_id)
|
||||
if message is None or message.chat_id != chat.id:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "That message no longer exists.")
|
||||
|
||||
return StreamingResponse(
|
||||
_generate(chat.id, message.id),
|
||||
media_type="text/event-stream",
|
||||
headers={
|
||||
"Cache-Control": "no-cache, no-transform",
|
||||
"Connection": "keep-alive",
|
||||
# nginx buffers proxied responses by default, which turns a stream
|
||||
# into one delivery at the end. This is the documented opt-out.
|
||||
"X-Accel-Buffering": "no",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
async def _generate(chat_id: str, message_id: str) -> AsyncIterator[str]:
|
||||
"""Drive one completion and frame it as SSE.
|
||||
|
||||
Opens its own database session rather than using the request's: streaming
|
||||
outlives the request handler, and the dependency-scoped session may already
|
||||
be closed by the time the first token arrives.
|
||||
"""
|
||||
accumulated: list[str] = []
|
||||
error: str | None = None
|
||||
|
||||
with session_scope() as db:
|
||||
chat = db.get(Chat, chat_id)
|
||||
message = db.get(Message, message_id)
|
||||
if chat is None or message is None:
|
||||
yield sse.event("close", "")
|
||||
return
|
||||
|
||||
first_user_text = ""
|
||||
try:
|
||||
endpoint, model_id = chat_service.resolve_endpoint(db, chat)
|
||||
payload = chat_service.build_request(db, chat, upto=message)
|
||||
first_user_text = next(
|
||||
(m["content"] for m in reversed(payload["messages"]) if m["role"] == ROLE_USER),
|
||||
"",
|
||||
)
|
||||
|
||||
async for chunk in stream_chat(endpoint, payload):
|
||||
text = delta_text(chunk)
|
||||
if not text:
|
||||
continue
|
||||
accumulated.append(text)
|
||||
yield sse.event("token", escape_text(text))
|
||||
# Hand control back so the event is flushed rather than
|
||||
# batched behind a fast generator.
|
||||
await asyncio.sleep(0)
|
||||
|
||||
except LLMError as exc:
|
||||
error = exc.message
|
||||
log.info("generation failed for chat %s: %s", chat_id, exc.message)
|
||||
except asyncio.CancelledError:
|
||||
# The reader navigated away or closed the tab. Keep whatever was
|
||||
# produced so the partial reply is still there on reload.
|
||||
message.content = "".join(accumulated)
|
||||
message.complete = True
|
||||
db.commit()
|
||||
raise
|
||||
except Exception as exc: # noqa: BLE001 - must not kill the stream silently
|
||||
error = "Something went wrong while generating this reply."
|
||||
log.exception("unexpected generation failure for chat %s: %s", chat_id, exc)
|
||||
|
||||
message.content = "".join(accumulated)
|
||||
message.error = error or ""
|
||||
message.complete = True
|
||||
|
||||
if not chat.title_generated and (accumulated or error):
|
||||
chat.title = (
|
||||
await chat_service.generate_title(
|
||||
endpoint, model_id, first_user_text, message.content
|
||||
)
|
||||
if not error and first_user_text
|
||||
else chat_service.fallback_title(first_user_text)
|
||||
)
|
||||
chat.title_generated = True
|
||||
|
||||
db.commit()
|
||||
|
||||
final_html = templates.get_template("chat/_message.html").render(
|
||||
{
|
||||
"message": message,
|
||||
"body_html": render_markdown(message.content),
|
||||
"chat": chat,
|
||||
# Passed even though an assistant bubble never reads it: the
|
||||
# template shares both roles, and a missing `user` would only
|
||||
# blow up on whichever branch is not being exercised here.
|
||||
"user": db.get(User, chat.user_id),
|
||||
}
|
||||
)
|
||||
title_html = templates.get_template("chat/_title_oob.html").render(
|
||||
{"chat": chat}
|
||||
)
|
||||
|
||||
yield sse.event("done", final_html + title_html)
|
||||
yield sse.event("close", "")
|
||||
|
||||
|
||||
@router.patch("/{chat_id}")
|
||||
async def update_chat(
|
||||
db: Db,
|
||||
user: RequiredUser,
|
||||
chat_id: str,
|
||||
title: str | None = Form(None),
|
||||
folder_id: str | None = Form(None),
|
||||
model_id: str | None = Form(None),
|
||||
) -> Response:
|
||||
chat = _owned_chat(db, chat_id, user.id)
|
||||
|
||||
if title is not None:
|
||||
cleaned = title.strip()[:300]
|
||||
if cleaned:
|
||||
chat.title = cleaned
|
||||
# An explicit rename must not be overwritten by auto-titling later.
|
||||
chat.title_generated = True
|
||||
|
||||
if folder_id is not None:
|
||||
chat.folder_id = folder_id or None
|
||||
|
||||
if model_id is not None and model_id:
|
||||
chat.model_id = model_id
|
||||
match = next(
|
||||
(m for m in chat_service.available_models(db) if m.model_id == model_id), None
|
||||
)
|
||||
chat.connection_id = match.connection_id if match else None
|
||||
|
||||
db.commit()
|
||||
return Response(status_code=status.HTTP_204_NO_CONTENT)
|
||||
|
||||
|
||||
@router.delete("/{chat_id}")
|
||||
async def delete_chat(db: Db, user: RequiredUser, chat_id: str) -> Response:
|
||||
chat = _owned_chat(db, chat_id, user.id)
|
||||
db.delete(chat)
|
||||
db.commit()
|
||||
|
||||
response = Response(status_code=status.HTTP_204_NO_CONTENT)
|
||||
response.headers["HX-Redirect"] = "/chat"
|
||||
return response
|
||||
|
||||
|
||||
@router.get("/{chat_id}/messages/{message_id}/raw")
|
||||
async def raw_message(db: Db, user: RequiredUser, chat_id: str, message_id: str) -> HTMLResponse:
|
||||
"""The unrendered Markdown of a message, for the copy button."""
|
||||
_owned_chat(db, chat_id, user.id)
|
||||
message = db.get(Message, message_id)
|
||||
if message is None or message.chat_id != chat_id:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "That message no longer exists.")
|
||||
return HTMLResponse(escape_text(message.content))
|
||||
|
||||
|
||||
@router.post("/{chat_id}/messages/{message_id}/regenerate")
|
||||
async def regenerate(
|
||||
request: Request,
|
||||
db: Db,
|
||||
user: RequiredUser,
|
||||
chat_id: str,
|
||||
message_id: str,
|
||||
) -> Response:
|
||||
"""Discard an assistant reply and produce a fresh one in its place."""
|
||||
chat = _owned_chat(db, chat_id, user.id)
|
||||
message = db.get(Message, message_id)
|
||||
if message is None or message.chat_id != chat.id or message.role != ROLE_ASSISTANT:
|
||||
raise HTTPException(status.HTTP_404_NOT_FOUND, "That reply no longer exists.")
|
||||
|
||||
message.content = ""
|
||||
message.error = ""
|
||||
message.complete = False
|
||||
message.model_id = chat.model_id
|
||||
db.commit()
|
||||
|
||||
return templates.TemplateResponse(
|
||||
request,
|
||||
"chat/_message.html",
|
||||
{"request": request, "message": message, "chat": chat, "body_html": "", "user": user},
|
||||
)
|
||||
|
||||
|
||||
__all__ = ["render", "router"]
|
||||
Reference in New Issue
Block a user