"""STT-server — API FastAPI pour les clients STT. Endpoints : GET /healthz → état du service POST /v1/ask {text} → {reply} (requête AI, orchestrée vers LiteLLM) """ from __future__ import annotations import logging import time from contextlib import asynccontextmanager import httpx from fastapi import BackgroundTasks, FastAPI, HTTPException from pydantic import BaseModel from stt_server import __version__ from stt_server import brain from stt_server.brain import ask as brain_ask from stt_server.config import settings from stt_server.contexts import CONTEXTS, assemble, get_context from stt_server.knowledge import Knowledge from stt_server.longterm import LongTermMemory from stt_server.memory import SessionStore from stt_server.sources import fetch_blocks, ghostfolio_phrase log = logging.getLogger("stt_server") # uvicorn ne configure que ses propres loggers : on attache notre handler en INFO # pour que les lignes de timing (`ask … recall/gen/total`) sortent dans les logs du pod. if not log.handlers: log.setLevel(logging.INFO) _h = logging.StreamHandler() _h.setFormatter(logging.Formatter("%(levelname)s: %(name)s: %(message)s")) log.addHandler(_h) log.propagate = False sessions = SessionStore() longterm = LongTermMemory() if settings.memory_longterm else None knowledge = Knowledge() if settings.docs_rag else None @asynccontextmanager async def lifespan(app: FastAPI): yield # Fermeture propre des clients HTTP persistants (pooling). await brain.aclose() if longterm: await longterm.aclose() if knowledge: await knowledge.aclose() app = FastAPI(title="STT-server", version=__version__, lifespan=lifespan) class AskRequest(BaseModel): text: str model: str | None = None # alias LiteLLM ; défaut serveur si absent session_id: str | None = None # mémoire court-terme : fil de conversation context: str | None = None # contexte présélectionné (funk/ghostfolio/…) secrets: dict | None = None # jetons client (ex. {"ghostfolio_token": …}) class PortfolioRequest(BaseModel): token: str | None = None # jeton Ghostfolio fourni par le client (sinon env) class AskReply(BaseModel): reply: str model: str context_id: str # contexte effectivement utilisé context: dict | None = None # contexte assemblé (visualiseur HUD) @app.get("/healthz") async def healthz() -> dict: return {"status": "ok", "version": __version__} @app.get("/v1/models") async def v1_models() -> dict: return {"default": settings.model, "available": settings.allowed_models} @app.get("/v1/contexts") async def v1_contexts() -> dict: """Contextes présélectionnables (pour le sélecteur du HUD).""" return { "default": settings.default_context, "contexts": [ {"id": c.id, "label": c.label, "icon": c.icon, "description": c.description} for c in CONTEXTS.values() ], } @app.get("/v1/memory/health") async def v1_memory_health() -> dict: """État de la mémoire long-terme (embeddings + Qdrant + collection), erreurs exposées.""" if not longterm: return {"enabled": False} return await longterm.health() @app.post("/v1/portfolio") async def v1_portfolio(req: PortfolioRequest) -> dict: """Valeur du portefeuille Ghostfolio (intent vocal client → serveur). Le serveur fait l'appel Ghostfolio avec le jeton fourni par le client (sinon le jeton serveur `STT_GHOSTFOLIO_TOKEN`). Un seul code Ghostfolio, lecture seule. """ async with httpx.AsyncClient() as c: summary = await ghostfolio_phrase(c, req.token) return {"summary": summary} @app.post("/v1/reset") async def v1_reset(req: AskRequest) -> dict: if req.session_id: sessions.reset(req.session_id) return {"status": "reset"} @app.post("/v1/ask", response_model=AskReply) async def v1_ask(req: AskRequest, background: BackgroundTasks) -> AskReply: text = req.text.strip() if not text: raise HTTPException(status_code=400, detail="text vide") model = req.model or settings.model if model not in settings.allowed_models: raise HTTPException( status_code=400, detail=f"modèle '{model}' non autorisé ; dispo : {settings.allowed_models}", ) ctx = get_context(req.context) t0 = time.perf_counter() history = sessions.history(req.session_id) if req.session_id else None # recall : timeout serré, dégrade vite ; renvoie aussi le vecteur (réutilisé ci-dessous) memories, qvec = await longterm.recall(text) if longterm else ([], None) # RAG doc : seulement si le contexte le demande (réutilise qvec, même embedder nomic) docs = [] if "docs" in ctx.sources and knowledge: docs = await knowledge.search(text, qvec) # Sources live du contexte (Ghostfolio / Prometheus / Alertmanager) — best-effort blocks: list = [] live = tuple(s for s in ctx.sources if s != "docs") if live: async with httpx.AsyncClient() as c: blocks = await fetch_blocks(c, live, secrets=req.secrets) t_recall = time.perf_counter() system, ctx_debug = assemble(ctx, blocks=blocks, docs=docs, memories=memories) try: reply = await brain_ask(text, system, model, history) except httpx.HTTPError as e: raise HTTPException(status_code=502, detail=f"upstream LiteLLM : {e}") from e t_gen = time.perf_counter() if req.session_id: sessions.add(req.session_id, "user", text) sessions.add(req.session_id, "assistant", reply) # store : APRÈS la réponse (BackgroundTasks) → hors latence perçue, et on réutilise qvec if longterm: background.add_task(longterm.store, req.session_id or "anon", text, qvec) log.info( "ask ctx=%s model=%s recall+src=%.0fms gen=%.0fms total=%.0fms mem=%d docs=%d blocks=%d", ctx.id, model, (t_recall - t0) * 1000, (t_gen - t_recall) * 1000, (t_gen - t0) * 1000, len(memories), len(docs), len(blocks), ) return AskReply(reply=reply, model=model, context_id=ctx.id, context=ctx_debug) def run() -> None: """Entrypoint `stt-server` (dev local). En prod : uvicorn via le conteneur.""" import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000) # noqa: S104 — service interne au cluster