132 lines
4.6 KiB
Python
132 lines
4.6 KiB
Python
"""Portal-native chat front for the Intranet knowledge base.
|
|
|
|
Proxies streamed answers from AnythingLLM's /api/v1/workspace/{slug}/stream-chat
|
|
endpoint through to the browser as Server-Sent Events. The browser never talks
|
|
to AnythingLLM directly, so the API token stays server-side and we can swap the
|
|
backend later without touching the frontend.
|
|
"""
|
|
import json
|
|
import logging
|
|
import os
|
|
import uuid
|
|
from typing import AsyncIterator
|
|
|
|
import httpx
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import FileResponse, StreamingResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
ANYTHINGLLM_URL = os.getenv("ANYTHINGLLM_URL", "http://anythingllm:3001")
|
|
ANYTHINGLLM_API_KEY = os.environ["ANYTHINGLLM_API_KEY"]
|
|
WORKSPACE = os.getenv("ANYTHINGLLM_WORKSPACE", "intranet")
|
|
|
|
app = FastAPI(title="Firemní asistent")
|
|
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"],
|
|
allow_headers=["*"])
|
|
|
|
|
|
class ChatRequest(BaseModel):
|
|
message: str
|
|
session_id: str | None = None
|
|
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return FileResponse("static/index.html")
|
|
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.post("/api/session")
|
|
async def new_session():
|
|
"""Returns an ID the frontend can use to thread a conversation. AnythingLLM
|
|
keys chat history by sessionId; we leave history persistence to it."""
|
|
return {"session_id": str(uuid.uuid4())}
|
|
|
|
|
|
async def _stream_from_anythingllm(
|
|
message: str, session_id: str,
|
|
) -> AsyncIterator[bytes]:
|
|
"""Open an SSE stream against AnythingLLM and forward each event downstream.
|
|
|
|
AnythingLLM emits JSON lines of shape:
|
|
{type: 'textResponseChunk', textResponse: '...', sources: [], close: false}
|
|
{type: 'finalizeResponseStream', sources: [...], close: true}
|
|
We re-emit them verbatim wrapped in SSE 'data:' frames.
|
|
"""
|
|
payload = {
|
|
"message": message,
|
|
"sessionId": session_id,
|
|
"mode": "query", # workspace-tuned to RAG-only with citation
|
|
"reset": False,
|
|
}
|
|
headers = {
|
|
"Authorization": f"Bearer {ANYTHINGLLM_API_KEY}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "text/event-stream",
|
|
}
|
|
url = f"{ANYTHINGLLM_URL}/api/v1/workspace/{WORKSPACE}/stream-chat"
|
|
|
|
timeout = httpx.Timeout(connect=10.0, read=300.0, write=30.0, pool=10.0)
|
|
async with httpx.AsyncClient(timeout=timeout) as client:
|
|
try:
|
|
async with client.stream("POST", url, json=payload,
|
|
headers=headers) as r:
|
|
if r.status_code != 200:
|
|
err = (await r.aread()).decode("utf-8", errors="replace")
|
|
logger.error("AnythingLLM %d: %s", r.status_code, err[:300])
|
|
yield _sse({"type": "error",
|
|
"error": f"AnythingLLM HTTP {r.status_code}"})
|
|
return
|
|
async for line in r.aiter_lines():
|
|
if not line:
|
|
continue
|
|
# AnythingLLM uses SSE format already: "data: {...}".
|
|
# Strip the prefix, parse, re-emit as our own SSE.
|
|
if line.startswith("data:"):
|
|
body = line[5:].strip()
|
|
if not body:
|
|
continue
|
|
try:
|
|
evt = json.loads(body)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
yield _sse(evt)
|
|
if evt.get("close"):
|
|
return
|
|
except httpx.HTTPError as exc:
|
|
logger.exception("upstream stream error")
|
|
yield _sse({"type": "error", "error": f"upstream: {exc}"})
|
|
|
|
|
|
def _sse(payload: dict) -> bytes:
|
|
return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n".encode("utf-8")
|
|
|
|
|
|
@app.post("/api/chat")
|
|
async def chat(req: ChatRequest):
|
|
if not req.message.strip():
|
|
raise HTTPException(400, "Empty message")
|
|
session_id = req.session_id or str(uuid.uuid4())
|
|
return StreamingResponse(
|
|
_stream_from_anythingllm(req.message, session_id),
|
|
media_type="text/event-stream",
|
|
headers={
|
|
"Cache-Control": "no-cache",
|
|
"Connection": "keep-alive",
|
|
"X-Accel-Buffering": "no", # disable proxy buffering (Caddy)
|
|
"X-Session-Id": session_id,
|
|
},
|
|
)
|
|
|
|
|
|
app.mount("/static", StaticFiles(directory="static"), name="static")
|