ProfessorNovaandClaude Opus 5
784b76baf7
Self-hosted knowledge management for SMEs: a split-screen Markdown editor whose sections an LLM refines while you write, and RAG question answering over the documents that result. FastAPI + Postgres/pgvector on the back, SvelteKit on the front, everything OpenAI-compatible and self-hostable. Squashed into a single commit; the development history stays local. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01CA43ZJda8Rbp2hKXNy8f6b
214 lines
9.2 KiB
Python
214 lines
9.2 KiB
Python
"""Query mode: permission-filtered retrieval → grounded streamed answer."""
|
|
|
|
import re
|
|
from collections.abc import AsyncIterator
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.authoring.context import summarize_transcript
|
|
from app.llm.client import chat_stream, role_config
|
|
from app.llm.errors import LLMError
|
|
from app.llm.gate import endpoint_busy
|
|
from app.models import Conversation, MessageRole, User
|
|
from app.modes.base import (
|
|
Degraded,
|
|
ModeEvent,
|
|
SourceChunk,
|
|
Sources,
|
|
StateChanged,
|
|
Token,
|
|
)
|
|
from app.modes.prompts import render_context_turn
|
|
from app.prompts.overrides import get_prompt
|
|
from app.rag.retrieval import (
|
|
SearchResult,
|
|
results_are_low_confidence,
|
|
search,
|
|
text_search,
|
|
)
|
|
|
|
HISTORY_TURNS = 8
|
|
EXCERPT_CHARS = 280
|
|
TOP_K = 5
|
|
|
|
|
|
def _topic_transcript(conversation: Conversation, user_message: str) -> str:
|
|
"""The recent turns plus the current message, as a transcript for the
|
|
topic summary. Returns '' when there is no earlier context — a first
|
|
message that misses is a genuine no-answer, not a lost topic."""
|
|
turns = [
|
|
(message.role, message.content)
|
|
for message in conversation.messages
|
|
if message.role in (MessageRole.user, MessageRole.assistant)
|
|
]
|
|
# The current message may or may not already be persisted into
|
|
# `conversation.messages`; append it only if it is not the last turn.
|
|
if not turns or turns[-1] != (MessageRole.user, user_message):
|
|
turns.append((MessageRole.user, user_message))
|
|
if len(turns) < 2:
|
|
return ""
|
|
return "\n".join(
|
|
f"{'User' if role is MessageRole.user else 'Assistant'}: {content}"
|
|
for role, content in turns[-HISTORY_TURNS:]
|
|
)
|
|
|
|
|
|
# Markdown reduced to prose, in the order the rules have to fire.
|
|
_TABLE_DIVIDER = re.compile(r"^\s*\|?[\s:|-]*-[\s:|-]*\|?\s*$", re.MULTILINE)
|
|
_BLOCK_MARKER = re.compile(r"^\s{0,3}(#{1,6}|[-*+]|\d+\.|>)\s+", re.MULTILINE)
|
|
_FENCE = re.compile(r"^\s*```.*$", re.MULTILINE)
|
|
_IMAGE = re.compile(r"!\[([^\]]*)\]\([^)]*\)")
|
|
_LINK = re.compile(r"\[([^\]]+)\]\([^)]*\)")
|
|
_EMPHASIS = re.compile(r"(\*\*\*|\*\*|\*|___|__|`)(?=\S)(.+?)(?<=\S)\1", re.DOTALL)
|
|
# Single underscores need word boundaries that the asterisk forms do not:
|
|
# `_kursiv_` is emphasis, but `result_document_id` is an identifier and the
|
|
# corpus is full of them.
|
|
_UNDERSCORE_EMPHASIS = re.compile(r"(?<!\w)_(?=\S)(.+?)(?<=\S)_(?!\w)", re.DOTALL)
|
|
|
|
|
|
def excerpt(content: str) -> str:
|
|
"""Short preview of a cited chunk for the citation popover.
|
|
|
|
The user has already passed the permission filter for this chunk, so
|
|
showing it back is safe, but keep it short: it is a hint, not the
|
|
document.
|
|
|
|
Markdown is reduced to prose rather than rendered. The popover is a
|
|
~320px hover surface showing a FRAGMENT, and rendering a fragment goes
|
|
wrong in exactly the cases that matter: a cited table becomes a real
|
|
table squeezed into the popover, a cited section heading renders at h2
|
|
size, and a list item arrives without its list. Clean prose answers the
|
|
only question the popover exists for, "is this the passage I want?".
|
|
Clicking the badge opens the document in the side panel, which renders
|
|
the Markdown properly through the sanitizing renderer.
|
|
|
|
Keeping this plain text is also what lets `Tooltip` promise that its
|
|
content is never markup: document text
|
|
reaches it through here and nowhere else.
|
|
"""
|
|
text = _FENCE.sub("", content)
|
|
# Divider rows first: they are pure punctuation and survive every other
|
|
# rule as a run of dashes and pipes.
|
|
text = _TABLE_DIVIDER.sub("", text)
|
|
text = _IMAGE.sub(r"\1", text)
|
|
text = _LINK.sub(r"\1", text)
|
|
text = _BLOCK_MARKER.sub("", text)
|
|
# Two passes: the outer run of `**bold _and_ italic**` has to go before
|
|
# the inner one is reachable.
|
|
for _ in range(2):
|
|
text = _EMPHASIS.sub(r"\2", text)
|
|
text = _UNDERSCORE_EMPHASIS.sub(r"\1", text)
|
|
# Remaining cell walls become sentence-ish separators, so a cited table
|
|
# reads as "Code · Meaning · Action" instead of a wall of pipes.
|
|
text = re.sub(r"\s*\|\s*", " · ", text)
|
|
text = re.sub(r"(?: · )+", " · ", text)
|
|
# A leading or trailing separator is what an empty first or last table
|
|
# cell leaves behind. Regex rather than str.strip: the latter treats the
|
|
# argument as a character set, which is not what this means.
|
|
text = re.sub(r"^(?:\s|·)+|(?:\s|·)+$", "", text)
|
|
|
|
flattened = " ".join(text.split())
|
|
if len(flattened) <= EXCERPT_CHARS:
|
|
return flattened
|
|
cut = flattened[:EXCERPT_CHARS]
|
|
head, separator, _ = cut.rpartition(" ")
|
|
return (head if separator else cut) + "…"
|
|
|
|
|
|
def _source(result: SearchResult, *, used: bool) -> SourceChunk:
|
|
return SourceChunk(
|
|
document_id=result.document_id,
|
|
title=result.title,
|
|
heading_path=result.heading_path,
|
|
excerpt=excerpt(result.content),
|
|
used=used,
|
|
review_pending=result.review_pending,
|
|
)
|
|
|
|
|
|
class QueryMode:
|
|
name = "query"
|
|
|
|
async def handle_turn(
|
|
self, conversation: Conversation, user_message: str, db: AsyncSession
|
|
) -> AsyncIterator[ModeEvent]:
|
|
user = await db.get(User, conversation.user_id)
|
|
assert user is not None
|
|
|
|
yield StateChanged(phase="searching")
|
|
try:
|
|
results = await search(db, user_message, user=user, top_k=TOP_K)
|
|
except LLMError:
|
|
# No embedding endpoint. The German full-text index finds documents
|
|
# on its own (keywords, not meaning), and the chat role is
|
|
# configured separately, so the turn can still end in a real answer.
|
|
results = await text_search(db, user_message, user=user, top_k=TOP_K)
|
|
else:
|
|
if results_are_low_confidence(results):
|
|
# A follow-up ("Hi", "and my earlier question?") loses the topic
|
|
# on its own, but the conversation's subject can recover it.
|
|
# Runs only on the low-confidence path, so a clear question pays
|
|
# no extra latency. (Eval: topic-summary 4/4 vs raw message 1/4.)
|
|
transcript = _topic_transcript(conversation, user_message)
|
|
if transcript:
|
|
topic = await summarize_transcript(transcript)
|
|
if topic and topic != user_message:
|
|
retry = await search(db, topic, user=user, top_k=TOP_K)
|
|
if not results_are_low_confidence(retry):
|
|
results = retry
|
|
|
|
grounded = not results_are_low_confidence(results)
|
|
if grounded:
|
|
yield StateChanged(phase="results", count=len(results))
|
|
else:
|
|
# Nothing solid to ground on: the answer gets no sources and the UI
|
|
# offers to capture the missing knowledge instead.
|
|
yield StateChanged(phase="no_answer", count=0)
|
|
# Every retrieved passage is reported for the "?" context inspector;
|
|
# `used` marks the ones that actually reached the prompt. On a no-answer
|
|
# they are all unused, which is exactly what explains "why no answer".
|
|
yield Sources(chunks=[_source(result, used=grounded) for result in results])
|
|
|
|
history = [
|
|
{"role": message.role.value, "content": message.content}
|
|
for message in conversation.messages[-HISTORY_TURNS:]
|
|
if message.role in (MessageRole.user, MessageRole.assistant)
|
|
]
|
|
# Only grounded passages ground the model; a no-answer sends none.
|
|
prompt_results = results if grounded else []
|
|
# Cache-friendly order: the static system prompt and the history stay
|
|
# byte-identical across a conversation's turns (so the endpoint's prompt
|
|
# cache reuses them); only this turn's excerpts + question are new.
|
|
messages = [
|
|
{"role": "system", "content": get_prompt("query_system")},
|
|
*history,
|
|
{
|
|
"role": "user",
|
|
"content": render_context_turn(prompt_results, user_message),
|
|
},
|
|
]
|
|
|
|
answered = False
|
|
try:
|
|
# Someone else may be holding every slot the endpoint has. Saying
|
|
# so beats a cursor that blinks for twenty seconds; the phase
|
|
# flips to "answering" the moment the first token arrives.
|
|
chat_base_url, _, _ = role_config("chat")
|
|
queued = endpoint_busy(chat_base_url)
|
|
yield StateChanged(phase="queued" if queued else "answering")
|
|
async for delta in chat_stream(messages, role="chat"):
|
|
if queued and not answered:
|
|
yield StateChanged(phase="answering")
|
|
answered = True
|
|
yield Token(text=delta)
|
|
except LLMError as exc:
|
|
if answered:
|
|
# Half an answer is already on screen: the router keeps it and
|
|
# reports the failure. There is nothing to fall back to.
|
|
raise
|
|
# No model at all. What retrieval found IS the reply now, as a plain
|
|
# list the user opens themselves. Nothing grounded anything, so the
|
|
# passages are re-sent unused (the second frame replaces the first).
|
|
yield Sources(chunks=[_source(result, used=False) for result in results])
|
|
yield Degraded(code=exc.code)
|