diff --git a/stt/server/pyproject.toml b/stt/server/pyproject.toml index e992fb3..86202f7 100644 --- a/stt/server/pyproject.toml +++ b/stt/server/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "stt-server" -version = "0.1.0" +version = "0.2.0" description = "STT-server — orchestrateur AI du homelab Funk (API pour les clients STT)" requires-python = ">=3.11" diff --git a/stt/server/stt_server/__init__.py b/stt/server/stt_server/__init__.py index f614a39..fe833f3 100644 --- a/stt/server/stt_server/__init__.py +++ b/stt/server/stt_server/__init__.py @@ -1,3 +1,3 @@ """STT-server — orchestrateur AI in-cluster pour les clients STT.""" -__version__ = "0.1.0" +__version__ = "0.2.0" diff --git a/stt/server/stt_server/app.py b/stt/server/stt_server/app.py index 9865e72..f3fa7e9 100644 --- a/stt/server/stt_server/app.py +++ b/stt/server/stt_server/app.py @@ -7,21 +7,39 @@ Endpoints : from __future__ import annotations +import logging +import time +from contextlib import asynccontextmanager + import httpx -from fastapi import FastAPI, HTTPException +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.longterm import LongTermMemory from stt_server.memory import SessionStore -app = FastAPI(title="STT-server", version=__version__) +log = logging.getLogger("stt_server") + sessions = SessionStore() longterm = LongTermMemory() if settings.memory_longterm else None +@asynccontextmanager +async def lifespan(app: FastAPI): + yield + # Fermeture propre des clients HTTP persistants (pooling). + await brain.aclose() + if longterm: + await longterm.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 @@ -59,7 +77,7 @@ async def v1_reset(req: AskRequest) -> dict: @app.post("/v1/ask", response_model=AskReply) -async def v1_ask(req: AskRequest) -> 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") @@ -69,17 +87,30 @@ async def v1_ask(req: AskRequest) -> AskReply: status_code=400, detail=f"modèle '{model}' non autorisé ; dispo : {settings.allowed_models}", ) + t0 = time.perf_counter() history = sessions.history(req.session_id) if req.session_id else None - memories = await longterm.recall(text) if longterm else None + # recall : timeout serré, dégrade vite ; renvoie aussi le vecteur (réutilisé par store) + memories, qvec = await longterm.recall(text) if longterm else ([], None) + t_recall = time.perf_counter() try: reply = await brain_ask(text, model, history, memories) 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: - await longterm.remember(req.session_id or "anon", text) + background.add_task(longterm.store, req.session_id or "anon", text, qvec) + log.info( + "ask model=%s recall=%.0fms gen=%.0fms total=%.0fms mem=%d", + model, + (t_recall - t0) * 1000, + (t_gen - t_recall) * 1000, + (t_gen - t0) * 1000, + len(memories), + ) return AskReply(reply=reply, model=model) diff --git a/stt/server/stt_server/brain.py b/stt/server/stt_server/brain.py index 098dbf4..8fdfe6f 100644 --- a/stt/server/stt_server/brain.py +++ b/stt/server/stt_server/brain.py @@ -11,6 +11,23 @@ import httpx from stt_server.config import settings +# Client persistant (pooling + keep-alive) : évite un handshake TCP vers LiteLLM à chaque tour. +_client: httpx.AsyncClient | None = None + + +def _get_client() -> httpx.AsyncClient: + global _client + if _client is None or _client.is_closed: + _client = httpx.AsyncClient(timeout=settings.request_timeout) + return _client + + +async def aclose() -> None: + global _client + if _client is not None and not _client.is_closed: + await _client.aclose() + _client = None + async def ask( text: str, @@ -42,13 +59,12 @@ async def ask( "temperature": settings.temperature, } headers = {"Authorization": f"Bearer {settings.litellm_key}"} - async with httpx.AsyncClient(timeout=settings.request_timeout) as client: - r = await client.post(settings.litellm_url, json=payload, headers=headers) - r.raise_for_status() - msg = r.json()["choices"][0]["message"] - # Filet de sécurité : si un modèle « thinking » renvoie un content vide (tout parti - # en reasoning_content), on récupère le raisonnement plutôt que de renvoyer "". - content = (msg.get("content") or "").strip() - if not content: - content = (msg.get("reasoning_content") or "").strip() - return content + r = await _get_client().post(settings.litellm_url, json=payload, headers=headers) + r.raise_for_status() + msg = r.json()["choices"][0]["message"] + # Filet de sécurité : si un modèle « thinking » renvoie un content vide (tout parti + # en reasoning_content), on récupère le raisonnement plutôt que de renvoyer "". + content = (msg.get("content") or "").strip() + if not content: + content = (msg.get("reasoning_content") or "").strip() + return content diff --git a/stt/server/stt_server/config.py b/stt/server/stt_server/config.py index c1c7778..e6ed3b6 100644 --- a/stt/server/stt_server/config.py +++ b/stt/server/stt_server/config.py @@ -43,6 +43,11 @@ class Settings: embed_url: str = os.getenv("STT_EMBED_URL", "http://192.168.10.20:1238/v1/embeddings") embed_model: str = os.getenv("STT_EMBED_MODEL", "nomic-embed-text") memory_top_k: int = int(os.getenv("STT_MEMORY_TOPK", "3")) + # Le recall (embed + recherche) est sur le chemin de réponse : timeout SERRÉ pour qu'un + # embed lent/mort dégrade vite (souvenirs vides) au lieu d'ajouter des secondes au client. + # Le store tourne en tâche de fond (après la réponse) → timeout plus large toléré. + memory_recall_timeout: float = float(os.getenv("STT_MEMORY_RECALL_TIMEOUT", "4")) + memory_store_timeout: float = float(os.getenv("STT_MEMORY_STORE_TIMEOUT", "20")) settings = Settings() diff --git a/stt/server/stt_server/longterm.py b/stt/server/stt_server/longterm.py index 93d65e1..84a38b1 100644 --- a/stt/server/stt_server/longterm.py +++ b/stt/server/stt_server/longterm.py @@ -27,21 +27,37 @@ class LongTermMemory: self.embed_url = settings.embed_url self.embed_model = settings.embed_model self.top_k = settings.memory_top_k + self.recall_timeout = settings.memory_recall_timeout + self.store_timeout = settings.memory_store_timeout self._ready = False + # Client persistant : pooling + keep-alive (évite un handshake TCP par appel). + self._http: httpx.AsyncClient | None = None - async def _embed(self, client: httpx.AsyncClient, text: str) -> list[float]: - r = await client.post( + def _client(self) -> httpx.AsyncClient: + if self._http is None or self._http.is_closed: + self._http = httpx.AsyncClient() + return self._http + + async def aclose(self) -> None: + if self._http is not None and not self._http.is_closed: + await self._http.aclose() + + async def _embed(self, text: str, timeout: float) -> list[float]: + r = await self._client().post( self.embed_url, json={"model": self.embed_model, "input": text}, - timeout=30, + timeout=timeout, ) r.raise_for_status() return r.json()["data"][0]["embedding"] - async def _ensure_collection(self, client: httpx.AsyncClient, dim: int) -> None: + async def _ensure_collection(self, dim: int) -> None: if self._ready: return - r = await client.get(f"{self.qdrant}/collections/{self.collection}") + client = self._client() + r = await client.get( + f"{self.qdrant}/collections/{self.collection}", timeout=self.store_timeout + ) if r.status_code == 200: # Collection existante : si la dimension a changé (modèle d'embedding # différent, ex. Qwen3 4096 → nomic 768), on recrée — les anciens vecteurs @@ -57,30 +73,38 @@ class LongTermMemory: await client.put( f"{self.qdrant}/collections/{self.collection}", json={"vectors": {"size": dim, "distance": "Cosine"}}, + timeout=self.store_timeout, ) self._ready = True - async def recall(self, text: str) -> list[str]: - """Souvenirs pertinents (texte) ou [] si indisponible.""" + async def recall(self, text: str) -> tuple[list[str], list[float] | None]: + """Souvenirs pertinents + le vecteur de la requête (réutilisable pour `store`). + + Sur le chemin de réponse → timeout serré (`recall_timeout`) : si l'embed ou Qdrant + traîne, on dégrade vite en `([], None)` plutôt que de faire patienter le client. + Le vecteur renvoyé évite de ré-embedder le même texte au moment du `store`. + """ try: - async with httpx.AsyncClient(timeout=20) as client: - vec = await self._embed(client, text) - r = await client.post( - f"{self.qdrant}/collections/{self.collection}/points/search", - json={"vector": vec, "limit": self.top_k, "with_payload": True}, - ) - if r.status_code == 404: # collection pas encore créée - return [] - r.raise_for_status() - pts = r.json().get("result", []) - return [p["payload"]["text"] for p in pts if p.get("payload", {}).get("text")] + client = self._client() + vec = await self._embed(text, timeout=self.recall_timeout) + r = await client.post( + f"{self.qdrant}/collections/{self.collection}/points/search", + json={"vector": vec, "limit": self.top_k, "with_payload": True}, + timeout=self.recall_timeout, + ) + if r.status_code == 404: # collection pas encore créée → rien à rappeler + return [], vec + r.raise_for_status() + pts = r.json().get("result", []) + texts = [p["payload"]["text"] for p in pts if p.get("payload", {}).get("text")] + return texts, vec except Exception: # noqa: BLE001 — dégrade silencieusement - return [] + return [], None async def health(self) -> dict: """Diagnostic actif : sonde embeddings + Qdrant + collection, sans rien avaler. - Contrairement à recall/remember (qui dégradent en silence), expose les erreurs + Contrairement à recall/store (qui dégradent en silence), expose les erreurs pour pouvoir déboguer la mémoire long-terme sans `kubectl exec`. """ out: dict = { @@ -91,49 +115,60 @@ class LongTermMemory: "embed": {"ok": False}, "qdrant": {"ok": False}, } - async with httpx.AsyncClient(timeout=20) as client: - # 1) embeddings - try: - vec = await self._embed(client, "ping mémoire") - out["embed"] = {"ok": True, "dim": len(vec)} - except Exception as e: # noqa: BLE001 — on veut l'erreur - out["embed"] = {"ok": False, "error": f"{type(e).__name__}: {e}"} - # 2) Qdrant + collection - try: - r = await client.get(f"{self.qdrant}/collections") - r.raise_for_status() - names = [c["name"] for c in r.json().get("result", {}).get("collections", [])] - exists = self.collection in names - qdrant: dict = {"ok": True, "collections": names, "collection_exists": exists} - if exists: - cr = await client.get(f"{self.qdrant}/collections/{self.collection}") - if cr.status_code == 200: - qdrant["points_count"] = cr.json().get("result", {}).get("points_count") - out["qdrant"] = qdrant - except Exception as e: # noqa: BLE001 - out["qdrant"] = {"ok": False, "error": f"{type(e).__name__}: {e}"} + client = self._client() + # 1) embeddings + try: + vec = await self._embed("ping mémoire", timeout=self.store_timeout) + out["embed"] = {"ok": True, "dim": len(vec)} + except Exception as e: # noqa: BLE001 — on veut l'erreur + out["embed"] = {"ok": False, "error": f"{type(e).__name__}: {e}"} + # 2) Qdrant + collection + try: + r = await client.get(f"{self.qdrant}/collections", timeout=self.store_timeout) + r.raise_for_status() + names = [c["name"] for c in r.json().get("result", {}).get("collections", [])] + exists = self.collection in names + qdrant: dict = {"ok": True, "collections": names, "collection_exists": exists} + if exists: + cr = await client.get( + f"{self.qdrant}/collections/{self.collection}", timeout=self.store_timeout + ) + if cr.status_code == 200: + qdrant["points_count"] = cr.json().get("result", {}).get("points_count") + out["qdrant"] = qdrant + except Exception as e: # noqa: BLE001 + out["qdrant"] = {"ok": False, "error": f"{type(e).__name__}: {e}"} return out - async def remember(self, session_id: str, text: str) -> None: + async def store( + self, session_id: str, text: str, vec: list[float] | None = None + ) -> None: + """Mémorise un tour. Pensé pour tourner **en tâche de fond** (hors chemin de réponse). + + `vec` : si fourni (réutilisé depuis `recall`), on évite un 2ᵉ embed du même texte. + Pas de `?wait=true` : on n'attend pas le flush disque de Qdrant — le client a déjà + sa réponse, ce point peut être indexé en arrière-plan. + """ try: - async with httpx.AsyncClient(timeout=20) as client: - vec = await self._embed(client, text) - await self._ensure_collection(client, len(vec)) - await client.put( - f"{self.qdrant}/collections/{self.collection}/points?wait=true", - json={ - "points": [ - { - "id": str(uuid.uuid4()), - "vector": vec, - "payload": { - "text": text, - "session_id": session_id, - "ts": time.time(), - }, - } - ] - }, - ) + if vec is None: + vec = await self._embed(text, timeout=self.store_timeout) + await self._ensure_collection(len(vec)) + await self._client().put( + f"{self.qdrant}/collections/{self.collection}/points", + json={ + "points": [ + { + "id": str(uuid.uuid4()), + "vector": vec, + "payload": { + "text": text, + "session_id": session_id, + "ts": time.time(), + }, + } + ] + }, + timeout=self.store_timeout, + ) except Exception: # noqa: BLE001 — dégrade silencieusement pass