Spaces:
Runtime error
Runtime error
| """Help bot (bottom right widget). Answers how to questions from the help corpus only, never from a writer's own | |
| notebooks or posts (see support/scope.py). Streams over SSE. | |
| POST /api/support/chat/stream events: token, answer_done, done, error | |
| GET /api/support/conversations/latest restore the chat after a refresh | |
| POST /api/support/escalate "talk to a human" / feature request form, emailed to the team | |
| """ | |
| import json | |
| import logging | |
| import re | |
| import time | |
| import uuid | |
| from collections import defaultdict, deque | |
| from collections.abc import AsyncIterator | |
| from fastapi import APIRouter, Depends, HTTPException, status | |
| from fastapi.responses import StreamingResponse | |
| from pydantic import BaseModel, Field | |
| from sqlalchemy import select | |
| from app.auth import Ctx, get_ctx | |
| from app.config import settings | |
| from app.llm.postprocess import strip_em_dashes | |
| from app.models import SupportConversation, SupportMessage | |
| from app.models.base import utcnow | |
| from app.services.email import email_service | |
| from app.services.plans import effective_plan | |
| from app.support import escalation as esc | |
| from app.support import llm, memory, prompts, scope | |
| from app.support.llm import LLMError | |
| from app.support.retriever import get_retriever | |
| log = logging.getLogger("notestack.support") | |
| router = APIRouter(prefix="/api/support", tags=["support"]) | |
| MAX_ESCALATIONS = 3 | |
| DOC_BUDGET_CHARS = 3500 | |
| _EMAIL = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") | |
| _hits: dict[uuid.UUID, deque[float]] = defaultdict(deque) # per process; enough to stop a runaway client | |
| class ChatIn(BaseModel): | |
| message: str = Field(min_length=1, max_length=4000) | |
| page_path: str | None = Field(default=None, max_length=300) | |
| conversation_id: uuid.UUID | None = None | |
| class EscalateIn(BaseModel): | |
| email: str = Field(min_length=3, max_length=320) | |
| concern: str = Field(min_length=1, max_length=4000) | |
| reason: str | None = Field(default=None, max_length=32) | |
| page_path: str | None = Field(default=None, max_length=300) | |
| conversation_id: uuid.UUID | None = None | |
| def _rate_limit(user_id: uuid.UUID) -> None: | |
| now, window = time.monotonic(), _hits[user_id] | |
| while window and now - window[0] > 60: | |
| window.popleft() | |
| if len(window) >= settings.support_messages_per_minute: | |
| raise HTTPException(status.HTTP_429_TOO_MANY_REQUESTS, | |
| "You are sending messages quickly. Try again in a minute.") | |
| window.append(now) | |
| def _sse(event: str, data) -> str: | |
| return f"event: {event}\ndata: {json.dumps(data)}\n\n" | |
| def _excerpt(body: str, query: str, budget: int = DOC_BUDGET_CHARS) -> str: | |
| """Whole doc when it fits, otherwise the sections that share the most words with the question.""" | |
| if len(body) <= budget: | |
| return body | |
| words = set(re.findall(r"[a-z0-9]+", query.lower())) | |
| sections = re.split(r"(?m)^(?=#{1,3}\s)", body) | |
| ranked = sorted(range(len(sections)), | |
| key=lambda i: -len(words & set(re.findall(r"[a-z0-9]+", sections[i].lower())))) | |
| keep, used = set(), 0 | |
| for i in ranked: | |
| if used + len(sections[i]) > budget and keep: | |
| continue | |
| keep.add(i) | |
| used += len(sections[i]) | |
| return "".join(sections[i] for i in sorted(keep))[:budget] | |
| def _docs_block(scored, query: str) -> str: | |
| if not scored: | |
| return "(no matching documents)" | |
| return "\n\n".join(f"--- id: {s.doc.id}\ntitle: {s.doc.title}\n{_excerpt(s.doc.body, query)}" for s in scored) | |
| def _owned(ctx: Ctx, conversation_id: uuid.UUID) -> SupportConversation: | |
| conv = ctx.db.get(SupportConversation, conversation_id) | |
| if not conv or conv.user_id != ctx.user.id: | |
| raise HTTPException(404, "Conversation not found") | |
| return conv | |
| def _message_out(m: SupportMessage) -> dict: | |
| return {"id": str(m.id), "role": m.role, "content": m.content, "page_path": m.page_path, | |
| "cited_docs": m.cited_docs or [], "created_at": m.created_at.isoformat()} | |
| async def chat_stream(body: ChatIn, ctx: Ctx = Depends(get_ctx)) -> StreamingResponse: | |
| if not settings.support_enabled: | |
| raise HTTPException(404, "Help chat is off") | |
| _rate_limit(ctx.user.id) | |
| db, message = ctx.db, body.message.strip() | |
| page = scope.scrub_path(body.page_path) # ids never reach the prompt, the database or the logs | |
| conv = _owned(ctx, body.conversation_id) if body.conversation_id else None | |
| if conv is None: | |
| conv = SupportConversation(user_id=ctx.user.id, title=message[:120], summary="", session_state={}) | |
| db.add(conv) | |
| db.flush() | |
| conv_id = conv.id | |
| plan = effective_plan(db, ctx.workspace).id | |
| recent = memory.recent_messages(db, conv_id) | |
| question_reason = esc.classify_question(message) | |
| out_of_scope = scope.about_user_data(message) | |
| short_circuit = esc.should_short_circuit(question_reason) and not out_of_scope | |
| scored, messages = [], [] | |
| if not out_of_scope and not short_circuit: | |
| scored = get_retriever().retrieve( | |
| message, history=memory.user_texts(recent), page_path=page, last_cited=memory.last_cited(recent) | |
| ) | |
| system = prompts.ANSWER_PROMPT.format( | |
| docs=_docs_block(scored, message), user_context=memory.state_block(conv.session_state or {}), | |
| summary=conv.summary or "(none)", | |
| ) | |
| messages = [{"role": "system", "content": system}] | |
| messages += [{"role": m.role, "content": m.content} for m in recent] | |
| messages.append({"role": "user", "content": f"[on {page}]\n{message}" if page else message}) | |
| async def generate() -> AsyncIterator[str]: | |
| answer, escalate_reason, meta = "", None, llm.Meta() | |
| if out_of_scope: | |
| answer, broken = scope.out_of_scope_reply(message) | |
| escalate_reason = esc.Reason.HUMAN if broken else None | |
| yield _sse("token", answer) | |
| elif short_circuit: | |
| drafted = "" | |
| try: # buffered, not streamed: the line is checked before the writer sees it | |
| async for tok in llm.stream_answer( | |
| [{"role": "system", "content": esc.handoff_prompt(question_reason)}, | |
| {"role": "user", "content": message}] | |
| ): | |
| drafted += tok | |
| except LLMError: | |
| drafted = "" | |
| drafted = strip_em_dashes(drafted.strip()) | |
| answer = (drafted if drafted and esc.handoff_line_is_safe(drafted) | |
| else esc.short_circuit_reply(question_reason, len(recent))) | |
| escalate_reason = question_reason | |
| yield _sse("token", answer) | |
| else: | |
| try: | |
| async for tok in llm.stream_answer(messages): | |
| answer += tok | |
| yield _sse("token", strip_em_dashes(tok)) | |
| except LLMError: | |
| yield _sse("error", "The help assistant is unavailable right now. Please try again.") | |
| return | |
| answer = strip_em_dashes(answer.strip()) or "Sorry, I could not form an answer." | |
| yield _sse("answer_done", {"conversation_id": str(conv_id)}) | |
| if not out_of_scope and not short_circuit: | |
| meta_system = prompts.META_PROMPT.format( | |
| doc_ids="\n".join(f"- {s.doc.id}: {s.doc.title}" for s in scored) or "(none)" | |
| ) | |
| try: | |
| meta = await llm.complete_meta( | |
| [{"role": "system", "content": meta_system}, {"role": "user", "content": message}, | |
| {"role": "assistant", "content": answer}] | |
| ) | |
| except LLMError: | |
| meta = llm.Meta() # the regex nets below still decide escalation | |
| escalate_reason = ( | |
| (question_reason if question_reason is esc.Reason.FEATURE else None) | |
| or esc.classify_answer(answer) | |
| or (esc.Reason.HUMAN if meta.escalate else None) | |
| ) | |
| valid = {s.doc.id for s in scored} | |
| citations = [c for c in meta.citations if c in valid] | |
| db.add(SupportMessage(conversation_id=conv_id, role="user", content=message, page_path=page)) | |
| db.add(SupportMessage(conversation_id=conv_id, role="assistant", content=answer, page_path=page, | |
| cited_docs=citations)) | |
| conv.session_state = memory.update_state(conv.session_state or {}, page_path=page, cited=citations, plan=plan) | |
| db.commit() | |
| yield _sse("done", {"conversation_id": str(conv_id), "citations": citations, | |
| "escalate": bool(escalate_reason), | |
| "escalate_reason": escalate_reason.value if escalate_reason else None}) | |
| # After `done`, so the extra call never delays the writer: fold older turns into the rolling summary. | |
| total = len(db.scalars(select(SupportMessage.id).where(SupportMessage.conversation_id == conv_id)).all()) | |
| if total >= memory.SUMMARIZE_AT: | |
| fold = memory.to_fold(db, conv_id, recent) | |
| if fold: | |
| transcript = "\n".join(f"{m.role.upper()}: {m.content}" for m in fold) | |
| try: | |
| conv.summary = strip_em_dashes(await llm.complete_text( | |
| [{"role": "system", "content": prompts.SUMMARY_PROMPT}, | |
| {"role": "user", "content": f"Previous summary: {conv.summary or '(none)'}\n\n" | |
| f"New turns:\n{transcript}"}], | |
| max_tokens=500))[:1200] | |
| db.commit() | |
| except LLMError: | |
| log.info("summary update skipped") | |
| return StreamingResponse(generate(), media_type="text/event-stream", | |
| headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) | |
| def latest(ctx: Ctx = Depends(get_ctx)): | |
| conv = ctx.db.scalar( | |
| select(SupportConversation).where(SupportConversation.user_id == ctx.user.id) | |
| .order_by(SupportConversation.updated_at.desc()).limit(1) | |
| ) | |
| if not conv: | |
| return {"conversation_id": None, "messages": []} | |
| msgs = ctx.db.scalars( | |
| select(SupportMessage).where(SupportMessage.conversation_id == conv.id) | |
| .order_by(SupportMessage.created_at).limit(60) | |
| ).all() | |
| return {"conversation_id": str(conv.id), "messages": [_message_out(m) for m in msgs]} | |
| def escalate(body: EscalateIn, ctx: Ctx = Depends(get_ctx)): | |
| from_email = body.email.strip() | |
| if not _EMAIL.match(from_email): | |
| raise HTTPException(422, "Please enter a valid email address.") | |
| concern = body.concern.strip() | |
| if not concern: | |
| raise HTTPException(422, "Please describe what you need.") | |
| transcript: list[tuple[str, str]] = [] | |
| if body.conversation_id: | |
| conv = _owned(ctx, body.conversation_id) | |
| if conv.escalation_count >= MAX_ESCALATIONS: | |
| raise HTTPException(429, "You have already sent this a few times. We will be in touch shortly.") | |
| rows = ctx.db.scalars( | |
| select(SupportMessage).where(SupportMessage.conversation_id == conv.id) | |
| .order_by(SupportMessage.created_at.desc()).limit(6) | |
| ).all() | |
| transcript = [(m.role, m.content[:500]) for m in reversed(rows)] | |
| conv.escalation_count += 1 | |
| conv.updated_at = utcnow() | |
| ctx.db.commit() | |
| sent = email_service.send_support_escalation( | |
| from_email=from_email, name=ctx.user.name, plan=effective_plan(ctx.db, ctx.workspace).id, | |
| reason=body.reason or "human", concern=concern, page=scope.scrub_path(body.page_path), transcript=transcript, | |
| ) | |
| if not sent: | |
| raise HTTPException(502, "We could not send that right now. Please try again.") | |
| return {"ok": True} | |